package com.github.dkanellis.skyspark.api.algorithms.bitmap;

import com.github.dkanellis.skyspark.api.algorithms.SkylineAlgorithm;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import scala.Tuple2;

import java.awt.geom.Point2D;
import java.io.Serializable;
import java.util.BitSet;
import java.util.List;

public class Bitmap implements SkylineAlgorithm, Serializable {

    @Override
    public List<Point2D> getSkylinePoints(JavaRDD<Point2D> points) {
        JavaRDD<Double> distinctPointsOfXDimension = getDistinctSorted(points);

        JavaPairRDD<Double, Long> distinctPointsOfYDimension = points
                .map(Point2D::getY)
                .distinct()
                .sortBy(Double::doubleValue, false, 1)
                .zipWithIndex();


        JavaRDD<BitSet> bitsetX = points.map(p -> getBitVectorRepresentation(p.getX(), distinctPointsOfYDimension));

        List<Double> x = distinctPointsOfXDimension.collect();
        List<Tuple2<Double, Long>> y = distinctPointsOfYDimension.collect();

        throw new UnsupportedOperationException("Bitmap is not supported yet.");
    }

    /**
     * We use ascending order because our points dominate each other when they are less in every dimension.
     *
     * @param points
     * @return
     */
    JavaRDD<Double> getDistinctSorted(JavaRDD<Point2D> points) {
        return points
                .map(Point2D::getX)
                .distinct()
                .sortBy(Double::doubleValue, true, 1);
    }

    private BitSet getBitVectorRepresentation(final double p, JavaPairRDD<Double, Long> distinctPoints) {
        BitSet bitSet = new BitSet();
        bitSet.set(0, (int) getPosition(p, distinctPoints));
        return bitSet;
    }

    private long getPosition(final double pY, JavaPairRDD<Double, Long> distinctPointsOfXDimension) {
        return distinctPointsOfXDimension.count() - distinctPointsOfXDimension.filter(p -> p._1() >= pY).count();
    }

    @Override
    public String toString() {
        return "Bitmap";
    }
}

@@ -347,71 +347,430 @@
 t;%0A%0A
-public class Bitmap implements SkylineAlgorithm, Serializable %7B
+import static com.google.common.base.Preconditions.checkArgument;%0A%0Apublic class Bitmap implements SkylineAlgorithm, Serializable %7B%0A%0A    private final int numberOfPartitions;%0A%0A    public Bitmap() %7B%0A        this(4);%0A    %7D%0A%0A    public Bitmap(final int numberOfPartitions) %7B%0A        checkArgument(numberOfPartitions %3E 0, %22Partitions can't be less than 1.%22);%0A        %0A        this.numberOfPartitions = numberOfPartitions;%0A    %7D
 %0A%0A  
@@ -1117,17 +1117,34 @@
  false, 
-1
+numberOfPartitions
 )%0A      
@@ -1523,22 +1523,15 @@
  %7D%0A%0A
+%0A
     /
-**%0A     *
+/
  We 
@@ -1632,59 +1632,8 @@
 on.%0A
-     *%0A     * @param points%0A     * @return%0A     */%0A
     

Add constructor to provide partitions