public class JoinQuery implements Serializable {
    private static final long serialVersionUID = 1L;

    private Predicate[] predicates;
    private String table;
    private int joinAttribute;

    public JoinQuery(String queryString) {
        String[] parts = queryString.split("\\|");
        this.table = parts[0];
        this.joinAttribute = Integer.parseInt(parts[1]);
        if (parts.length > 2) {
            String predString = parts[2].trim();
            String[] predParts = predString.split(";");
            this.predicates = new Predicate[predParts.length];
            for (int i = 0; i < predParts.length; i++) {
                this.predicates[i] = new Predicate(predParts[i]);
            }
        } else {
            this.predicates = new Predicate[0];
        }
    }

    public JoinQuery(String table, int joinAttribute, Predicate[] predicates) {
        this.table = table;
        this.joinAttribute = joinAttribute;
        this.predicates = predicates;
    }

    public Predicate[] getPredicates() {
        return this.predicates;
    }

    public String getTable() {
        return this.table;
    }

    public int getJoinAttribute(){
        return this.joinAttribute;
    }

    public Query castToQuery(){
        return new Query(table, predicates);
    }

    public void write(DataOutput out) throws IOException {
        Text.writeString(out, toString());
    }



    @Override
    public String toString() {
        String stringPredicates = "";
        if (predicates.length != 0)
            stringPredicates = Joiner.on(";").join(predicates);
        return table + "|" + joinAttribute  + "|" +  stringPredicates;
    }
} public class JoinRepartitionIterator extends PartitionIterator {

    private JRNode newIndexTree;
    protected String zookeeperHosts;
    private int indexPartition;

    protected Map<Integer, Partition> newPartitions = new HashMap<Integer, Partition>();

    public JoinRepartitionIterator() {
    }

    public JoinRepartitionIterator(String iteratorString) {
        try {
            readFields(ByteStreams.newDataInput(iteratorString.getBytes()));
        } catch (IOException e) {
            e.printStackTrace();
            throw new RuntimeException("Failed to read the fields");
        }
    }

    public JoinRepartitionIterator(Query query, int partition) {
        super(query);
        this.indexPartition = partition;
    }

    public void setZookeeper(String zookeeperHosts) {
        this.zookeeperHosts = zookeeperHosts;
    }

    public Query getQuery() {
        return this.query;
    }

    public int getIndexPartition(){
        return indexPartition;
    }

    public JRNode getIndexTree() {
        return this.newIndexTree;
    }

    /**
     * Gets a HDFS Partition as input.
     * Loads the new index. PartitionWriter::setPartition does the rest.
     */
    @Override
    public void setPartition(Partition partition) {
        super.setPartition(partition);
        if (newIndexTree == null) {
            String path = FilenameUtils.getPathNoEndSeparator(partition
                    .getPath());

            if (FilenameUtils.getBaseName(path).contains("partitions")
                    || FilenameUtils.getBaseName(path).contains("repartition")) { // hack
                path = FilenameUtils.getPathNoEndSeparator(FilenameUtils.getPath(path));
            }

            if (FilenameUtils.getBaseName(path).contains("data")) { // hack
                path = FilenameUtils.getPathNoEndSeparator(FilenameUtils.getPath(path));
            }

            String pathToIndex = path + "/index";

            if (indexPartition != -1){
                pathToIndex = pathToIndex + "." + indexPartition;
            }

            // Initialize JoinRobustTree.
            byte[] indexBytes = HDFSUtils.readFile(
                    ((HDFSPartition) partition).getFS(), pathToIndex);
            JoinRobustTree tree = new JoinRobustTree(Globals.getTableInfo(query.getTable()));
            tree.unmarshall(indexBytes);
            newIndexTree = tree.getRoot();
        }
    }

    @Override
    protected boolean isRelevant(IteratorRecord record) {
        int id = newIndexTree.getBucketId(record);

        Partition p;
        if (newPartitions.containsKey(id)) {
            p = newPartitions.get(id);
        } else {
            p = partition.clone();
            p.setPartitionId(id);
            newPartitions.put(id, p);
        }

        p.write(record.getBytes(), 0, record.getBytes().length);

        if (p.size() > 2 * 1024 * 1024 ){ // 2mb
            p.store(true);
            newPartitions.remove(id);
        }

        return query.qualifies(record);
    }

    @Override
    public void finish() {
        if (zookeeperHosts != null) {

            for (Partition p : newPartitions.values()) {
                System.out.println("storing partition id " + p.getPartitionId());
                p.store(true);
            }

            newPartitions = Maps.newHashMap();
        } else {
            System.out.println("INFO: Zookeeper Hosts NULL");
        }
    }

    @Override
    public void write(DataOutput out) throws IOException {
        query.write(out);
        Text.writeString(out, zookeeperHosts);
        Text.writeString(out, Integer.toString(indexPartition));
    }

    @Override
    public void readFields(DataInput in) throws IOException {
        String predicateString = Text.readString(in);
        query = new Query(predicateString);
        zookeeperHosts = Text.readString(in);
        indexPartition = Integer.parseInt(Text.readString(in));
    }

    public static JoinRepartitionIterator read(DataInput in) throws IOException {
        JoinRepartitionIterator it = new JoinRepartitionIterator();
        it.readFields(in);
        return it;
    }
} public class JoinPlanner {

    private FileSystem fs;
    private SparkJoinQueryConf queryConf;
    private String dataset1, dataset2;
    private JoinQuery dataset1_query, dataset2_query;
    private List<JoinQuery> dataset1_queryWindow, dataset2_queryWindow;
    private TableInfo dataset1_tableInfo, dataset2_tableInfo;

    private Map<Integer, MDIndex.BucketInfo> dataset1_bucketInfo, dataset2_bucketInfo;

    private HPJoinInput dataset1_hpinput, dataset2_hpinput;
    private Map<Integer, JoinAccessMethod> dataset1_am, dataset2_am;
    private Map<Integer, ArrayList<Integer>> dataset1_scan_blocks, dataset2_scan_blocks;
    private Map<Integer, Integer> dataset1_iterator_type, dataset2_iterator_type;
    private Map<Integer, Integer> dataset1_belong, dataset2_belong;
    private Map<Integer, ArrayList<Integer>> overlap_chunks;

    private ArrayList<PartitionSplit> hyperJoinSplit, shuffleJoinSplit1, shuffleJoinSplit2;
    public String hyperjoin, shufflejoin1, shufflejoin2;

    public JoinPlanner(Configuration conf) {

        System.out.println("INFO: in JoinPlanner constructor");

        queryConf = new SparkJoinQueryConf(conf);
        fs = HDFSUtils.getFSByHadoopHome(queryConf.getHadoopHome());

        dataset1 = conf.get("DATASET1");
        dataset2 = conf.get("DATASET2");

        dataset1_query = new JoinQuery(conf.get("DATASET1_QUERY"));
        dataset2_query = new JoinQuery(conf.get("DATASET2_QUERY"));

        dataset1_queryWindow = loadQueries(dataset1, queryConf);
        dataset2_queryWindow = loadQueries(dataset2, queryConf);

        dataset1_queryWindow.add(dataset1_query);
        dataset2_queryWindow.add(dataset2_query);

        persistQueryToDisk(dataset1_query, queryConf);
        persistQueryToDisk(dataset2_query, queryConf);

        dataset1_hpinput = new HPJoinInput();
        dataset2_hpinput = new HPJoinInput();
        dataset1_am = new HashMap<Integer, JoinAccessMethod>();
        dataset2_am = new HashMap<Integer, JoinAccessMethod>();

        dataset1_iterator_type = new HashMap<Integer, Integer>();
        dataset2_iterator_type = new HashMap<Integer, Integer>();
        dataset1_belong = new HashMap<Integer, Integer>();
        dataset2_belong = new HashMap<Integer, Integer>();

        dataset1_bucketInfo = new HashMap<Integer, MDIndex.BucketInfo>();
        dataset2_bucketInfo = new HashMap<Integer, MDIndex.BucketInfo>();

        overlap_chunks = new HashMap<Integer, ArrayList<Integer>>();

        dataset1_scan_blocks = new HashMap<Integer, ArrayList<Integer>>();
        dataset2_scan_blocks = new HashMap<Integer, ArrayList<Integer>>();

        hyperJoinSplit = new ArrayList<PartitionSplit>();
        shuffleJoinSplit1 = new ArrayList<PartitionSplit>();
        shuffleJoinSplit2 = new ArrayList<PartitionSplit>();

        hyperjoin = "";
        shufflejoin1 = "";
        shufflejoin2 = "";

        // set up table info and get partitions for each table
        setupTableInfo();

        int[] dataset1_partitions = dataset1_tableInfo.partitions;
        int[] dataset2_partitions = dataset2_tableInfo.partitions;

        SmoothRepartition(dataset1, dataset1_query, dataset1_queryWindow, dataset1_tableInfo, dataset1_hpinput, dataset1_am, dataset1_scan_blocks, dataset1_iterator_type, dataset1_belong, dataset1_bucketInfo, queryConf, fs);
        SmoothRepartition(dataset2, dataset2_query, dataset2_queryWindow, dataset2_tableInfo, dataset2_hpinput, dataset2_am, dataset2_scan_blocks, dataset2_iterator_type, dataset2_belong, dataset2_bucketInfo, queryConf, fs);

        int dataset1_join_attr = dataset1_query.getJoinAttribute();
        int dataset2_join_attr = dataset2_query.getJoinAttribute();

        boolean multiple_iterator_type = false;

        // do we need to check dataset1_iterator_type?

        for (int block : dataset2_iterator_type.keySet()) {
            if (dataset2_iterator_type.get(block) != dataset2_join_attr) {
                multiple_iterator_type = true;
            }
        }

        boolean dataset1PartitionHasJoinAttr = ArrayUtils.contains(dataset1_partitions, dataset1_join_attr);
        boolean dataset2PartitionHasJoinAttr = ArrayUtils.contains(dataset2_partitions, dataset2_join_attr);

        System.out.println(dataset1 + "##" + dataset1_join_attr + "##" + dataset2 + "##" + +dataset2_join_attr + "##" + dataset1PartitionHasJoinAttr + "##" + dataset2PartitionHasJoinAttr + "##" + (dataset2_partitions.length == 1) + "##" + (multiple_iterator_type == false));

        // separate read-only and repartitioned. Read only go first.

        if (dataset1PartitionHasJoinAttr && dataset2PartitionHasJoinAttr && dataset2_partitions.length == 1 && multiple_iterator_type == false) {

            // read_index && init over_lap

            //read_index(dataset1_bucketInfo, dataset1_am.get(dataset1_join_attr).getIndex(), dataset1_join_attr);
            //read_index(dataset2_bucketInfo, dataset2_am.get(dataset2_join_attr).getIndex(), dataset2_join_attr);

            init_overlap();

            if (dataset1_partitions.length == 1) { // hyper join only

                // extract hyper join, update dataset2_iterator_type, shuffle scan on dataset2 in case some data blocks are left.

                extractHyperJoin(dataset1_query, hyperJoinSplit, dataset1_scan_blocks, dataset1_iterator_type, dataset1_belong, dataset1_hpinput.getPartitionIdSizeMap(), queryConf.getMaxSplitSize());

                hyperjoin = getHyperJoinInput();
                // clear dataset2_scan_blocks

                dataset2_scan_blocks.clear();

                // in case some blocks are left, probably it's empty
                extractShuffleJoin(dataset2_query, shuffleJoinSplit2, dataset2_scan_blocks, dataset2_iterator_type, dataset2_hpinput.getPartitionIdSizeMap(), queryConf.getMaxSplitSize(), queryConf.getWorkerNum());
            } else { // partial hyper join + partial shuffle join

                Map<Integer, Integer> dataset2_iterator_type_copy = new HashMap<Integer, Integer>();
                for (int key : dataset2_iterator_type.keySet()) {
                    dataset2_iterator_type_copy.put(key, dataset2_iterator_type.get(key));
                }

                // copy dataset2_iterator_type,  extract hyper join, update dataset2_iterator_type, any remaining block need repartitioned. all others need to be scanned (using the copy to update scan blocks).

                extractHyperJoin(dataset1_query, hyperJoinSplit, dataset1_scan_blocks, dataset1_iterator_type, dataset1_belong, dataset1_hpinput.getPartitionIdSizeMap(), queryConf.getMaxSplitSize());

                hyperjoin = getHyperJoinInput();

                // remaining blocks need to be repartitioned, otherwise scan

                for (int key : dataset2_iterator_type_copy.keySet()) {
                    if (dataset2_iterator_type.containsKey(key) == false) {
                        dataset2_scan_blocks.get(dataset2_join_attr).add(dataset2_iterator_type_copy.get(key));
                    }
                }

                // remove joinAttribute from dataset 1

                dataset1_scan_blocks.remove(dataset1_join_attr);

                extractShuffleJoin(dataset1_query, shuffleJoinSplit1, dataset1_scan_blocks, dataset1_iterator_type, dataset1_hpinput.getPartitionIdSizeMap(), queryConf.getMaxSplitSize(), queryConf.getWorkerNum());
                extractShuffleJoin(dataset2_query, shuffleJoinSplit2, dataset2_scan_blocks, dataset2_iterator_type, dataset2_hpinput.getPartitionIdSizeMap(), queryConf.getMaxSplitSize(), queryConf.getWorkerNum());
            }


        } else { // full shuffle join
            extractShuffleJoin(dataset1_query, shuffleJoinSplit1, dataset1_scan_blocks, dataset1_iterator_type, dataset1_hpinput.getPartitionIdSizeMap(), queryConf.getMaxSplitSize(), queryConf.getWorkerNum());
            extractShuffleJoin(dataset2_query, shuffleJoinSplit2, dataset2_scan_blocks, dataset2_iterator_type, dataset2_hpinput.getPartitionIdSizeMap(), queryConf.getMaxSplitSize(), queryConf.getWorkerNum());
        }

        shufflejoin1 = getShuffleJoinInput1();
        shufflejoin2 = getShuffleJoinInput2();
    }

    public static void persistQueryToDisk(JoinQuery q, SparkJoinQueryConf queryConf) {
        String pathToQueries = queryConf.getWorkingDir() + "/" + q.getTable() + "/queries";
        FileSystem fs = HDFSUtils.getFSByHadoopHome(queryConf.getHadoopHome());
        HDFSUtils.safeCreateFile(fs, pathToQueries,
                queryConf.getHDFSReplicationFactor());
        HDFSUtils.appendLine(fs, pathToQueries, q.toString());
    }


    private static void persistIndex(JoinRobustTree index, int partition, TableInfo tableInfo, SparkJoinQueryConf queryConf, FileSystem fs) {
        String pathToIndex = queryConf.getWorkingDir() + "/" + tableInfo.tableName + "/index";

        if (partition != -1) {
            pathToIndex = pathToIndex + "." + partition;
        }

        try {
            if (fs.exists(new Path(pathToIndex))) {
                // If index file exists, move it to a new filename
                long currentMillis = System.currentTimeMillis();
                String oldIndexPath = pathToIndex + "." + currentMillis;
                boolean successRename = fs.rename(new Path(pathToIndex),
                        new Path(oldIndexPath));
                if (!successRename) {
                    System.out.println("Index rename to " + oldIndexPath
                            + " failed");
                }
            }
            HDFSUtils.safeCreateFile(fs, pathToIndex, queryConf.getHDFSReplicationFactor());
        } catch (IOException e) {
            System.out.println("ERR: Writing Index failed: " + e.getMessage());
            e.printStackTrace();
        }

        byte[] indexBytes = index.marshall();
        HDFSUtils.writeFile(fs,
                pathToIndex, queryConf.getHDFSReplicationFactor(), indexBytes, 0,
                indexBytes.length, false);
    }

    private static int[] updatePartitions(HashSet<Integer> partititons, int latestPartition) {

        if (partititons.contains(latestPartition) == false) {
            throw new RuntimeException("latestPartition " + latestPartition + " is not in partititons.");
        }

        int[] ret = new int[partititons.size()];
        ret[0] = latestPartition;

        int index = 1;
        for (int parId : partititons) {
            if (parId != latestPartition) {
                ret[index++] = parId;
            }

        }
        return ret;
    }

    private static ArrayList<Integer> ArrayToArrayList(int[] arr) {
        ArrayList<Integer> result = new ArrayList<Integer>();
        for (int i = 0; i < arr.length; i++) {
            result.add(arr[i]);
        }
        return result;
    }

    private static int[] ArrayListToArray(ArrayList<Integer> arr_list) {
        int[] arr = new int[arr_list.size()];
        for (int i = 0; i < arr.length; i++) {
            arr[i] = arr_list.get(i);
        }
        return arr;
    }

    private static int[] GetNonEmptyDataBlocks(int[] blocks, HPJoinInput hpinput) {
        ArrayList<Integer> result = new ArrayList<Integer>();
        Map<Integer, Long> sizeMap = hpinput.getPartitionIdSizeMap();

        for (int i = 0; i < blocks.length; i++) {
            if (sizeMap.containsKey(blocks[i])) {
                result.add(blocks[i]);
            }
        }
        return ArrayListToArray(result);
    }

    private static void updateRepartitionBlocks(PartitionSplit[] splits, Map<Integer, Integer> iterator_type, Map<Integer, Integer> belong, HPJoinInput hpinput, int partition) {
        if (splits.length > 1) {
            int[] nonEmptyBidsFromRepartitionSplit = GetNonEmptyDataBlocks(splits[1].getPartitions(), hpinput);
            for (int j = 0; j < nonEmptyBidsFromRepartitionSplit.length; j++) {
                iterator_type.put(nonEmptyBidsFromRepartitionSplit[j], partition);
                belong.put(nonEmptyBidsFromRepartitionSplit[j], partition);
            }
        }
    }

    public static void SmoothRepartition(String tableName, JoinQuery query, List<JoinQuery> queryWindow, TableInfo tableInfo, HPJoinInput hpinput, Map<Integer, JoinAccessMethod> am_map, Map<Integer, ArrayList<Integer>> scanBlocks, Map<Integer, Integer> iterator_type, Map<Integer, Integer> belong, Map<Integer, MDIndex.BucketInfo> dataset1_bucketInfo, SparkJoinQueryConf queryConf, FileSystem fs) {

        int joinAttribute = query.getJoinAttribute();
        int[] partitions = tableInfo.partitions;

        int latestPartition = partitions[0];

        HashSet<Integer> newPartition = new HashSet<Integer>();
        for (int i = 0; i < partitions.length; i++) {
            newPartition.add(partitions[i]);
        }

        /*
            calculate expected weight for each partition
            expectedWeights contains information about how many times each join attribute appears in the query window.

            NOTE: the query window has the upcoming query which has not been executed.
        */

        Map<Integer, Integer> expectedWeights = new HashMap<Integer, Integer>();

        for (int i = 0; i < queryWindow.size(); i++) {
            int attr = queryWindow.get(i).getJoinAttribute();
            if (expectedWeights.containsKey(attr) == false) {
                expectedWeights.put(attr, 0);
            }
            expectedWeights.put(attr, expectedWeights.get(attr) + 1);
        }

        /*
            There is a limit on the size of query window, which means there may be partitions for a join attribute
            but the attribute does not appear in the query window.
            The expected size should be zero.

         */

        for (int i = 0; i < partitions.length; i++) {
            if (expectedWeights.containsKey(partitions[i]) == false) {
                expectedWeights.put(partitions[i], 0);
            }
        }


        /*
            is total_weight always equal to |query_window|?
         */

        int total_weight = 0;

        for (int attr : expectedWeights.keySet()) {
            total_weight += expectedWeights.get(attr);
        }

        /*
        *   this will only happen at the beginning
        *   If the total number of queries is less than Globals.QUERY_WINDOW_SIZE,
        *   the summation of expectedWeights is also less than Globals.QUERY_WINDOW_SIZE
        *   We should allocate the remaining quota to noJoinAttr(-1).
        * */

        int noJoinAttr = -1;

        if (queryWindow.size() < Globals.QUERY_WINDOW_SIZE) {
            if (expectedWeights.containsKey(noJoinAttr) == false) {
                expectedWeights.put(noJoinAttr, 0);
            }
            expectedWeights.put(noJoinAttr, Globals.QUERY_WINDOW_SIZE - total_weight);
        }

        // calculate partition size for each index

        Map<Integer, Long> partition_sizes = new HashMap<Integer, Long>();

        // init HPJoinInput && am_map

        queryConf.setJoinQuery(query);

        String input_dir = queryConf.getWorkingDir() + "/" + tableName + "/data";
        hpinput.initialize(listStatus(fs, input_dir));

        /*
        *   Compute the table size, which is a summation over all partitions.
        *   Load the partition corresponding the current join attr into dataset1_bucketInfo
        *   am_map is a mapping from partition id to JoinAccessMethod.
        *   JoinAccessMethod has PartitionIdSizeMap and partitionIdFileMap
        * */

        long table_size = 0;

        for (int i = 0; i < partitions.length; i++) {
            JoinAccessMethod am = new JoinAccessMethod();
            am.init(queryConf, partitions[i]);
            am_map.put(partitions[i], am);

            int[] bucket_ids = am.getIndex().getAllBuckets();
            Map<Integer, Long> bucket_size = hpinput.getPartitionIdSizeMap();
            long total_size = 0;
            for (int j = 0; j < bucket_ids.length; j++) {
                int bid = bucket_ids[j];
                if (bucket_size.containsKey(bid)) {
                    total_size += bucket_size.get(bid);
                }
            }
            partition_sizes.put(partitions[i], total_size);
            table_size += total_size;

            if (partitions[i] == joinAttribute) {
                read_index(dataset1_bucketInfo, am.getIndex(), joinAttribute);
            }
        }

        boolean partitionExists = ArrayUtils.contains(partitions, joinAttribute);

        if (partitionExists == false) {
            // create a new partitioning tree for the new joinAttribute.

            // step 1: read sample

            ParsedTupleList sample = new ParsedTupleList(tableInfo.getTypeArray());

            String pathToSample = queryConf.getWorkingDir() + "/" + tableName + "/sample";

            try {
                if (fs.exists(new Path(pathToSample))) {
                    byte[] sampleBytes = HDFSUtils.readFile(fs, pathToSample);
                    sample.unmarshall(sampleBytes, tableInfo.delimiter);
                } else {
                    throw new RuntimeException();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }

            // step 2: create a new partitioning tree

            // Construct the index from the sample.
            JoinRobustTree index = new JoinRobustTree(tableInfo);
            index.joinAttributeDepth = (tableInfo.depth + 1) / 2; // ceiling
            index.maxBuckets = 1 << tableInfo.depth;
            index.sample = sample;
            index.numAttributes = tableInfo.getTypeArray().length;
            index.dimensionTypes = tableInfo.getTypeArray();
            index.initProbe(query);


            // step 3: add joinAttribute into tableInfo's partitions, newPartition will persist at the end of smoothPartition

            newPartition.add(joinAttribute);
            latestPartition = joinAttribute;

            // step 4: persist index

            persistIndex(index, joinAttribute, tableInfo, queryConf, fs);
        }

        // adjust tree -> adapt or delete + scan

        double join_expected_weight = 1.0 * expectedWeights.get(joinAttribute) / Globals.QUERY_WINDOW_SIZE;

        /*
        * join_weight is calculated based on how many queries with this join attribute appear in the query window,
        * which is equal to |# of queries with join attr| / |query_window_size|
        * */

        double join_weight = 0;
        if (partition_sizes.containsKey(joinAttribute)) {
            join_weight = 1.0 * partition_sizes.get(joinAttribute) / table_size;
        }

        //only repartition data into new partition.

        if (join_weight < join_expected_weight && joinAttribute == latestPartition) { // repartition other indexes to join attribute
            // iterate through each partition and repartition some partitions into the join attr partition
            for (int i = 0; i < partitions.length; i++) {

                double actualWeight = 1.0 * partition_sizes.get(partitions[i]) / table_size;
                double expectedWeight = 1.0 * expectedWeights.get(partitions[i]) / Globals.QUERY_WINDOW_SIZE;

                JoinAccessMethod am = am_map.get(partitions[i]);
                JoinRobustTree rt = am.getIndex();

                if (actualWeight > expectedWeight) {  // delete

                    long deleteSize = (long) ((actualWeight - expectedWeight) * table_size);
                    boolean deleteAll = expectedWeights.get(partitions[i]) == 0;

                    // two partitionSplits, one for scan, one for delete

                    PartitionSplit scanSplit = am.getPartitionSplits(query, true, partitions[i])[0];
                    PartitionSplit deleteSplit = rt.delete(deleteSize, deleteAll, query, partitions[i], hpinput.getPartitionIdSizeMap());

                    /*
                    * Update or delete the index
                    * */

                    if (deleteAll || rt.getRoot() == null) {
                        String pathToIndex = queryConf.getWorkingDir() + "/" + tableInfo.tableName + "/index";

                        if (partitions[i] != -1) {
                            pathToIndex = pathToIndex + "." + partitions[i];
                        }

                        HDFSUtils.deleteFile(fs, pathToIndex, true);
                        newPartition.remove(partitions[i]);
                    } else {
                        persistIndex(rt, partitions[i], tableInfo, queryConf, fs);
                    }


                    /*
                    * some blocks may be empty, need a filter
                    * The index tree for each partition is created based on the sample file,
                    * If only partial dataset is repartitioned until deletion, some blocks may be empty.
                    * */


                    HashSet<Integer> deleteBids = new HashSet<Integer>();
                    int[] nonEmptyBidsFromDeleteSplit = GetNonEmptyDataBlocks(deleteSplit.getPartitions(), hpinput);


                    for (int j = 0; j < nonEmptyBidsFromDeleteSplit.length; j++) {
                        deleteBids.add(nonEmptyBidsFromDeleteSplit[j]);
                        iterator_type.put(nonEmptyBidsFromDeleteSplit[j], joinAttribute);
                        belong.put(nonEmptyBidsFromDeleteSplit[j], partitions[i]);
                    }

                    /*
                    *  filter scanSplit by deleteSplit
                    *  If a block appears in both scanSplit and deleteSplit, remove it from scanSplit
                    * */

                    ArrayList<Integer> scanBidsAfterfiltering = new ArrayList<Integer>();

                    int[] nonEmptyBidsFromScanSplit = GetNonEmptyDataBlocks(scanSplit.getPartitions(), hpinput);
                    for (int j = 0; j < nonEmptyBidsFromScanSplit.length; j++) {
                        if (deleteBids.contains(nonEmptyBidsFromScanSplit[j]) == false) {
                            scanBidsAfterfiltering.add(nonEmptyBidsFromScanSplit[j]);
                        }
                    }
                    scanBlocks.put(partitions[i], scanBidsAfterfiltering);
                } else { // scan
                    PartitionSplit[] splits = am.getPartitionSplits(query, false, partitions[i]);
                    scanBlocks.put(partitions[i], ArrayToArrayList(GetNonEmptyDataBlocks(splits[0].getPartitions(), hpinput)));
                    // There are blocks to be repartitioned
                    updateRepartitionBlocks(splits, iterator_type, belong, hpinput, partitions[i]);
                }
            }
        } else {
            for (int i = 0; i < partitions.length; i++) {
                JoinAccessMethod am = am_map.get(partitions[i]);
                PartitionSplit[] splits = am.getPartitionSplits(query, false, partitions[i]);
                scanBlocks.put(partitions[i], ArrayToArrayList(GetNonEmptyDataBlocks(splits[0].getPartitions(), hpinput)));
                updateRepartitionBlocks(splits, iterator_type, belong, hpinput, partitions[i]);
            }
        }

        // update partition
        tableInfo.partitions = updatePartitions(newPartition, joinAttribute);
        tableInfo.save(queryConf.getWorkingDir(), queryConf.getHDFSReplicationFactor(), fs);
    }

    public void setupTableInfo() {
        Globals.loadTableInfo(dataset1, queryConf.getWorkingDir(), fs);
        Globals.loadTableInfo(dataset2, queryConf.getWorkingDir(), fs);

        dataset1_tableInfo = Globals.getTableInfo(dataset1);
        dataset2_tableInfo = Globals.getTableInfo(dataset2);

    }

    public static List<JoinQuery> loadQueries(String tableName, SparkJoinQueryConf queryConf) {
        FileSystem fs = HDFSUtils.getFSByHadoopHome(queryConf.getHadoopHome());
        String pathToQueries = queryConf.getWorkingDir() + "/" + tableName + "/queries";

        List<JoinQuery> queryWindow = new ArrayList<JoinQuery>();

        try {
            if (fs.exists(new Path(pathToQueries))) {
                byte[] queryBytes = HDFSUtils.readFile(fs, pathToQueries);
                String queries = new String(queryBytes);
                Scanner sc = new Scanner(queries);
                while (sc.hasNextLine()) {
                    String query = sc.nextLine();
                    JoinQuery f = new JoinQuery(query);
                    queryWindow.add(f);
                }

                if (queryWindow.size() > Globals.QUERY_WINDOW_SIZE - 1) {
                    // set windows size
                    queryWindow = queryWindow.subList(queryWindow.size() - (Globals.QUERY_WINDOW_SIZE - 1), queryWindow.size());
                }
                sc.close();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }

        return queryWindow;
    }

    public void extractHyperJoin(JoinQuery dataset_query, ArrayList<PartitionSplit> hyperJoinSplit, Map<Integer, ArrayList<Integer>> scan_blocks, Map<Integer, Integer> iterator_type, Map<Integer, Integer> belong, Map<Integer, Long> partitionSizes, long maxSplitSize) {
        int join_attribute = dataset_query.getJoinAttribute();

        ArrayList<Integer> alist = scan_blocks.get(join_attribute);
        PartitionIterator pi = new PostFilterIterator(dataset_query.castToQuery());


        int total = 0;
        ArrayList<PartitionSplit> hyper_splits = groupSplits(pi, ArrayListToArray(alist), partitionSizes, maxSplitSize);
        for (PartitionSplit hs : hyper_splits) {
            hyperJoinSplit.add(hs);
            total += hs.getPartitions().length;
        }

        if (total != alist.size()) {
            throw new RuntimeException("some partition is lost");
        }


        // only handle join attribute


        ArrayList<Integer> blocks = new ArrayList<Integer>();

        for (int block : iterator_type.keySet()) {
            if (belong.get(block) == join_attribute) {
                //System.out.println("In blocks: " + block);
                blocks.add(block);
            }
        }
        // remove self repartitioned block
        for (int block : blocks) {
            iterator_type.remove(block);
            belong.remove(block);
        }


        pi = new JoinRepartitionIterator(dataset_query.castToQuery(), join_attribute);
        total = 0;

        hyper_splits = groupSplits(pi, ArrayListToArray(blocks), partitionSizes, maxSplitSize);
        for (PartitionSplit hs : hyper_splits) {
            hyperJoinSplit.add(hs);
            total += hs.getPartitions().length;
        }

        if (total != blocks.size()) {
            throw new RuntimeException("some partition is lost");
        }

    }

    public static void extractShuffleJoin(JoinQuery dataset_query, ArrayList<PartitionSplit> shuffleJoinSplit, Map<Integer, ArrayList<Integer>> scan_blocks, Map<Integer, Integer> iterator_type, Map<Integer, Long> partitionSizes, long maxSplitSize, int worker_num) {

        for (int index : scan_blocks.keySet()) {
            // construct a PostFilterIterator
            PartitionIterator pi = new PostFilterIterator(dataset_query.castToQuery());
            ArrayList<PartitionSplit> shuffle_splits = resizeSplits(pi, ArrayListToArray(scan_blocks.get(index)), partitionSizes, maxSplitSize, dataset_query, -1);
            int total = 0;
            for (PartitionSplit hs : shuffle_splits) {
                shuffleJoinSplit.add(hs);
                total += hs.getPartitions().length;
            }
            if (total != scan_blocks.get(index).size()) {
                throw new RuntimeException("some partition is lost");
            }

        }

        // for each index, construct a JoinRepartitionIterator

        Map<Integer, ArrayList<Integer>> repartition_blocks = new HashMap<Integer, ArrayList<Integer>>();

        for (int block : iterator_type.keySet()) {
            int index = iterator_type.get(block);
            if (repartition_blocks.containsKey(index) == false) {
                repartition_blocks.put(index, new ArrayList<Integer>());
            }
            repartition_blocks.get(index).add(block);
        }

        for (int index : repartition_blocks.keySet()) {
            ArrayList<Integer> alist = repartition_blocks.get(index);
            PartitionIterator pi = new JoinRepartitionIterator(dataset_query.castToQuery(), index);
            ArrayList<PartitionSplit> shuffle_splits = resizeSplits(pi, ArrayListToArray(alist), partitionSizes, maxSplitSize, dataset_query, worker_num);

            int total = 0;
            for (PartitionSplit hs : shuffle_splits) {
                shuffleJoinSplit.add(hs);
                total += hs.getPartitions().length;
            }
            if (total != repartition_blocks.get(index).size()) {
                throw new RuntimeException("some partition is lost");
            }
        }
    }

    private PartitionSplit[] getPartitionSplits(int[] bids, JoinQuery q) {
        PostFilterIterator pi = new PostFilterIterator(q.castToQuery());
        PartitionSplit psplit = new PartitionSplit(bids, pi);
        PartitionSplit[] ps = new PartitionSplit[1];
        ps[0] = psplit;
        return ps;
    }


    public String getHyperJoinInput() {
        // iter1_type, id1:len1:id2:len2:... , id1:len1:iter2_type:id2:len2:iter2_type... ;...
        StringBuilder sb = new StringBuilder();

        for (PartitionSplit split : hyperJoinSplit) {

            if (split.getPartitions().length == 0) continue;

            if (sb.length() > 0) {
                sb.append(";");
            }

            PartitionIterator iter = split.getIterator();
            if (iter instanceof PostFilterIterator) {
                sb.append(-2 + ",");
            } else {
                JoinRepartitionIterator pi = (JoinRepartitionIterator) iter;
                sb.append(pi.getIndexPartition() + ",");
            }

            int[] bucketIds = split.getPartitions();
            int[] dep_bucketIDs = getOverlappedSplits(bucketIds);

            long[] bucket_lens = dataset1_hpinput.getLengths(bucketIds);
            long[] dep_bucket_lens = dataset2_hpinput.getLengths(dep_bucketIDs);

            for (int i = 0; i < bucketIds.length; i++) {
                if (i > 0) {
                    sb.append(":");
                }
                sb.append(bucketIds[i] + ":" + bucket_lens[i]);
            }
            sb.append(",");

            for (int i = 0; i < dep_bucketIDs.length; i++) {
                if (i > 0) {
                    sb.append(":");
                }
                int type = -2;

                if (dataset2_iterator_type.containsKey(dep_bucketIDs[i])) {
                    type = dataset2_iterator_type.get(dep_bucketIDs[i]);
                    dataset2_iterator_type.remove(dep_bucketIDs[i]);// we can only use JoinRepartitionIterator once.
                }

                sb.append(dep_bucketIDs[i] + ":" + dep_bucket_lens[i] + ":" + type);
            }
        }

        return sb.toString();
    }

    public static String getShuffleJoinInputHelper(ArrayList<PartitionSplit> shuffleJoinSplit, HPJoinInput hpinput) {
        // iter,1:100,2:120; ...

        StringBuilder sb = new StringBuilder();

        for (PartitionSplit split : shuffleJoinSplit) {

            if (split.getPartitions().length == 0) continue;

            if (sb.length() > 0) {
                sb.append(";");
            }

            PartitionIterator iter = split.getIterator();
            if (iter instanceof PostFilterIterator) {
                sb.append(-2 + ",");
            } else {
                JoinRepartitionIterator pi = (JoinRepartitionIterator) iter;
                sb.append(pi.getIndexPartition() + ",");
            }

            int[] bucketIds = split.getPartitions();

            long[] bucket_lens = hpinput.getLengths(bucketIds);

            for (int i = 0; i < bucketIds.length; i++) {
                if (i > 0) {
                    sb.append(",");
                }
                sb.append(bucketIds[i] + ":" + bucket_lens[i]);
            }
        }

        return sb.toString();
    }

    public String getShuffleJoinInput1() {
        return getShuffleJoinInputHelper(shuffleJoinSplit1, dataset1_hpinput);
    }

    public String getShuffleJoinInput2() {
        //1:100:type;2:120:type
        return getShuffleJoinInputHelper(shuffleJoinSplit2, dataset2_hpinput);
    }

    public JoinQuery getDataset1_query() {
        return dataset1_query;
    }

    public JoinQuery getDataset2_query() {
        return dataset2_query;
    }

    private Path[] getPaths(int dataset_id, int[] split) {
        if (dataset_id == 1) {
            return dataset1_hpinput.getPaths(split);
        } else {
            return dataset2_hpinput.getPaths(split);
        }
    }

    private int[] getOverlappedSplits(int[] split) {

        HashSet<Integer> overlappedSplits = new HashSet<Integer>();
        for (int i = 0; i < split.length; i++) {
            int id = split[i];
            if (overlap_chunks.containsKey(id)) {
                overlappedSplits.addAll(overlap_chunks.get(id));
            }
        }

        int[] final_split = new int[overlappedSplits.size()];
        int it = 0;
        for (int i : overlappedSplits) {
            final_split[it++] = i;
        }

        return final_split;
    }


    public static List<FileStatus> listStatus(FileSystem fs, String path) {
        ArrayList<FileStatus> input = new ArrayList<FileStatus>();
        FileStatus[] input_array = null;
        try {
            input_array = fs.listStatus(new Path(path));
            for (int i = 0; i < input_array.length; i++) {
                input.add(input_array[i]);
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        return input;
    }

    public static int[] expandPartitionSplit(PartitionSplit[] splits) {
        int totalLength = 0;
        for (int i = 0; i < splits.length; i++) {
            totalLength += splits[i].getPartitions().length;
        }
        int[] partitions = new int[totalLength];
        for (int i = 0, k = 0; i < splits.length; i++) {
            int[] par = splits[i].getPartitions();
            for (int j = 0; j < par.length; j++) {
                partitions[k++] = par[j];
            }
        }
        return partitions;
    }

    private static void read_index(Map<Integer, MDIndex.BucketInfo> info, JoinRobustTree rt, int attr) {
        // If a particular bucket is not in the following map, then the range is (-oo,+oo).

        Map<Integer, MDIndex.BucketInfo> bucketRanges = rt.getBucketRanges(attr);

        int[] buckets = rt.getAllBuckets();

        for (int i = 0; i < buckets.length; i++) {
            int bucket_id = buckets[i];
            if (bucketRanges.containsKey(bucket_id) == false) {
                // hard code, the join key can only be int.

                info.put(bucket_id, new MDIndex.BucketInfo(TypeUtils.TYPE.LONG, null, null));
            } else {
                //System.out.println(">>> " + bucket_id + " " + bucketRanges.get(bucket_id));
                info.put(bucket_id, bucketRanges.get(bucket_id));
            }
        }
    }


    private void init_overlap() {
        // filtering

        HashSet<Integer> splits1 = new HashSet<Integer>();
        int join_attribute = dataset1_query.getJoinAttribute();

        for (int block : dataset1_scan_blocks.get(join_attribute)) {
            //System.out.println("Put " + block);
            splits1.add(block);
        }


        for (int block : dataset1_iterator_type.keySet()) {
            if (dataset1_belong.get(block) == join_attribute) {
                //System.out.println("Put " + block);
                splits1.add(block);
            }
        }

        HashSet<Integer> splits2 = new HashSet<Integer>();

        for (int block : dataset2_scan_blocks.get(dataset2_query.getJoinAttribute())) {
            splits2.add(block);
        }

        for (int block : dataset2_iterator_type.keySet()) {
            splits2.add(block);
        }


        for (Integer i : dataset1_bucketInfo.keySet()) {
            if (splits1.contains(i) == false) continue;

            MDIndex.BucketInfo info_i = dataset1_bucketInfo.get(i);

            for (Integer j : dataset2_bucketInfo.keySet()) {
                if (splits2.contains(j) == false) continue;

                MDIndex.BucketInfo info_j = dataset2_bucketInfo.get(j);

                //System.out.println(i + " from " + dataset1 + " intersects with " + j +  " from "+  dataset2 + " result: " + info_i.overlap(info_j));

                if (info_i.overlap(info_j)) {
                    if (overlap_chunks.containsKey(i) == false) {
                        overlap_chunks.put(i, new ArrayList<Integer>());
                    }
                    overlap_chunks.get(i).add(j);
                }
            }
        }

    }


    private void init_bucketInfo(JoinAccessMethod jam1, JoinAccessMethod jam2) {
        dataset1_bucketInfo = new HashMap<Integer, MDIndex.BucketInfo>();
        dataset2_bucketInfo = new HashMap<Integer, MDIndex.BucketInfo>();

        overlap_chunks = new HashMap<Integer, ArrayList<Integer>>();

        read_index(dataset1_bucketInfo, jam1.opt.getIndex(), dataset1_query.getJoinAttribute());
        read_index(dataset2_bucketInfo, jam2.opt.getIndex(), dataset2_query.getJoinAttribute());
    }

    private int[] getIntPartitions(PartitionSplit[] splits) {
        HashSet<Integer> allIds = new HashSet<Integer>();

        for (PartitionSplit split : splits) {
            for (int id : split.getPartitions()) {
                allIds.add(id);
            }
        }

        return Ints.toArray(allIds);
    }

    private int getIntersectionSize(HashSet<Integer> setValues, ArrayList<Integer> listValues) {
        int size = 0;
        for (int i = 0; i < listValues.size(); i++) {
            if (setValues.contains(listValues.get(i))) {
                size++;
            }
        }
        return size;
    }


    public static ArrayList<PartitionSplit> resizeSplits(PartitionIterator partitionIter, int[] bids, Map<Integer, Long> partitionSizes, long maxSplitSize, JoinQuery dataset_query, int worker_num) {

        // the size of a split could be larger than maxSplitSize, just in case a singel file is largen than this.

        ArrayList<PartitionSplit> resizedSplits = new ArrayList<PartitionSplit>();

        if (worker_num != -1) {
            // split input into WORKER_NUM splits

            ArrayList<ArrayList<Integer>> array_bids = new ArrayList<ArrayList<Integer>>();

            for (int i = 0; i < worker_num; i++) {
                array_bids.add(new ArrayList<Integer>());
            }

            for (int i = 0; i < bids.length; i++) {
                array_bids.get(i % worker_num).add(bids[i]);
            }

            for (int i = 0; i < worker_num; i++) {

                ArrayList<Integer> array_bid = array_bids.get(i);

                int[] split_bids = new int[array_bid.size()];
                for (int j = 0; j < array_bid.size(); j++) {
                    split_bids[j] = array_bid.get(j);
                }
                PartitionSplit split = new PartitionSplit(split_bids, partitionIter);
                resizedSplits.add(split);
            }
            return resizedSplits;
        } else {
            int it = 0;
            long totalsize = 0;
            ArrayList<Integer> cur_split = new ArrayList<Integer>();

            while (it < bids.length) {
                if (partitionSizes.get(bids[it]) == null) {
                    System.out.println(bids[it] + " is empty!");
                }
                long cur_size = partitionSizes.get(bids[it]);

                totalsize += cur_size;
                cur_split.add(bids[it]);

                if (totalsize >= maxSplitSize) {
                    int[] split_bids = new int[cur_split.size()];
                    for (int i = 0; i < split_bids.length; i++) {
                        split_bids[i] = cur_split.get(i);
                    }
                    PartitionSplit split = new PartitionSplit(split_bids, partitionIter);
                    resizedSplits.add(split);
                    cur_split = new ArrayList<Integer>();
                    totalsize = 0;
                }
                it++;
            }

            if (cur_split.size() > 0) {
                int[] split_bids = new int[cur_split.size()];
                for (int i = 0; i < split_bids.length; i++) {
                    split_bids[i] = cur_split.get(i);
                }
                PartitionSplit split = new PartitionSplit(split_bids, partitionIter);
                resizedSplits.add(split);
            }

            int sum = 0;

            for (PartitionSplit split : resizedSplits) {
                sum += split.getPartitions().length;
            }

            if (sum != bids.length) {
                throw new RuntimeException("miss some data blocks in resizeSplits");
            }

            return resizedSplits;
        }

    }


    /* the following code uses heuristic grouping
       TODO: add different grouping algos
    */

    private ArrayList<PartitionSplit> groupSplits(PartitionIterator partitionIter, int[] bids, Map<Integer, Long> partitionSizes, long maxSplitSize) {
        ArrayList<PartitionSplit> resizedSplits = new ArrayList<PartitionSplit>();

        Random rand = new Random();
        rand.setSeed(0); // Making things more deterministic.

        LinkedList<Integer> buckets = new LinkedList<Integer>();

        int size = bids.length;
        for (int i = 0; i < bids.length; i++) {
            buckets.add(bids[i]);
        }

        while (size > 0) {
            ArrayList<Integer> cur_split = new ArrayList<Integer>();
            HashSet<Integer> chunks = new HashSet<Integer>();
            long splitAvailableSize = maxSplitSize;
            while (size > 0 && splitAvailableSize > 0) {
                int maxIntersection = -1;
                int best_offset = -1;

                ListIterator<Integer> it = buckets.listIterator();
                int offset = 0;

                while (it.hasNext()) {
                    int value = it.next();
                    if (maxIntersection == -1) {
                        //System.out.println("#getting " + value);
                        maxIntersection = getIntersectionSize(chunks, overlap_chunks.get(value));
                        best_offset = offset;
                    } else {
                        //System.out.println("#getting " + value);
                        int curIntersection = getIntersectionSize(chunks, overlap_chunks.get(value));
                        if (curIntersection > maxIntersection) {
                            maxIntersection = curIntersection;
                            best_offset = offset;
                        }
                    }
                    offset++;
                }
                int bucket = buckets.get(best_offset);

                if (partitionSizes.get(bucket) == null) {
                    System.out.println(bucket + " is empty!");
                }

                splitAvailableSize -= partitionSizes.get(bucket);


                cur_split.add(bucket);
                for (int rhs : overlap_chunks.get(bucket)) {
                    chunks.add(rhs);
                }
                buckets.remove(best_offset);
                size--;

            }

            int[] split_bids = new int[cur_split.size()];
            for (int i = 0; i < split_bids.length; i++) {
                split_bids[i] = cur_split.get(i);
            }

            PartitionSplit split = new PartitionSplit(split_bids, partitionIter);
            resizedSplits.add(split);
            //System.out.println("split size: " + totalSize + " " + Arrays.toString(split_bids));
        }

        int sum = 0;

        for (PartitionSplit split : resizedSplits) {
            sum += split.getPartitions().length;
        }

        if (sum != bids.length) {
            throw new RuntimeException("miss some data blocks in groupSplits");
        }


        return resizedSplits;
    }


    private long getPartitionSplitSize(PartitionSplit split,
                                       Map<Integer, Long> partitionIdSizeMap) {

        long size = 0;
        for (int pid : split.getPartitions()) {
            if (partitionIdSizeMap.containsKey(pid))
                size += partitionIdSizeMap.get(pid);
            else
                System.err.println("partitoion size not found: " + pid);
        }
        return size;
    }
} public class SparkJoinQueryConf {

    public final static String WORKING_DIR = "WORKING_DIR";
    public final static String FULL_SCAN = "FULL_SCAN";
    public final static String REPARTITION_SCAN = "REPARTITION_SCAN";
    public final static String JUST_ACCESS = "JUST_ACCESS";
    public final static String QUERY = "QUERY";
    public final static String MAX_SPLIT_SIZE = "MAX_SPLIT_SIZE";
    public final static String MIN_SPLIT_SIZE = "MIN_SPLIT_SIZE";
    public final static String ZOOKEEPER_HOSTS = "ZOOKEEPER_HOSTS";
    public final static String HADOOP_HOME = "HADOOP_HOME";
    public final static String REPLICA_ID = "REPLICA_ID";
    public final static String HDFS_REPLICATION_FACTOR = "HDFS_REPLICATION_FACTOR";
    public final static String SCHEMA = "SCHEMA";
    public final static String CARTILAGE_PROPERTIES = "CARTILAGE_PROPERTIES";
    public final static String WORKER_NUM = "WORKER_NUM";

    private Configuration conf;

    public SparkJoinQueryConf(Configuration conf) {
        this.conf = conf;
    }

    public void setWorkingDir(String dataset) {
        conf.set(WORKING_DIR, dataset);
    }

    public String getWorkingDir() {
        return conf.get(WORKING_DIR);
    }

    public void setFullScan(boolean flag) {
        conf.setBoolean(FULL_SCAN, flag);
    }

    public boolean getFullScan() {
        return conf.getBoolean(FULL_SCAN, false); // don't full scan by default
    }

    public void setJustAccess(boolean flag) {
        conf.setBoolean(JUST_ACCESS, flag);
    }

    public boolean getJustAccess() {
        return conf.getBoolean(JUST_ACCESS, true); // don't adapt by default,
        // i.e. just access
    }

    public void setJoinQuery(JoinQuery query) {
        conf.set(QUERY, query.toString());
    }

    public JoinQuery getQuery() {
        if (conf.get(QUERY) == null || conf.get(QUERY).equals("")) {
            throw new RuntimeException("No query set in query conf.");
        }

        return new JoinQuery(conf.get(QUERY));
    }

    public void setMaxSplitSize(long maxSplitSize) {
        conf.set(MAX_SPLIT_SIZE, "" + maxSplitSize);
    }

    public long getMaxSplitSize() {
        return Long.parseLong(conf.get(MAX_SPLIT_SIZE));
    }



    public void setMinSplitSize(long minSplitSize) {
        conf.set(MIN_SPLIT_SIZE, "" + minSplitSize);
    }

    public void setWorkerNum(int num){
        conf.set(WORKER_NUM, "" + num);
    }

    public int getWorkerNum(){
        return Integer.parseInt(conf.get(WORKER_NUM));
    }

    public long getMinSplitSize() {
        return Long.parseLong(conf.get(MIN_SPLIT_SIZE));
    }

    public void setZookeeperHosts(String hosts) {
        conf.set(ZOOKEEPER_HOSTS, "" + hosts);
    }

    public String getZookeeperHosts() {
        return conf.get(ZOOKEEPER_HOSTS);
    }

    public void setHadoopHome(String home) {
        conf.set(HADOOP_HOME, home);
    }

    public String getHadoopHome() {
        return conf.get(HADOOP_HOME);
    }

    public void setHDFSReplicationFactor(short f) {
        conf.set(HDFS_REPLICATION_FACTOR, Short.toString(f));
    }

    public short getHDFSReplicationFactor() {
        return Short.parseShort(conf.get(HDFS_REPLICATION_FACTOR));
    }

    public void setReplicaId(int numReplicas) {
        conf.set(REPLICA_ID, String.valueOf(numReplicas));
    }

    public int getReplicaId() {
        return Integer.parseInt(conf.get(REPLICA_ID));
    }

    public Configuration getConf() {
        return conf;
    }
} public class Globals {
    // Re-partition cost multiplier.
    static public final int c = 4;

    // Query window size.
    static public final int window_size = 20;
    static public final int QUERY_WINDOW_SIZE = 10;

	static Map<String, TableInfo> tableInfos = new HashMap<String, TableInfo>();

    public static TableInfo getTableInfo(String tableName) {
        return tableInfos.get(tableName);
    }

    public static void addTableInfo(TableInfo tableInfo) {
        tableInfos.put(tableInfo.tableName, tableInfo);
    }

	public static void saveTableInfo(String tableName, String hdfsWorkingDir, short replication, FileSystem fs) {
        TableInfo tableInfo = tableInfos.get(tableName);
        tableInfo.save(hdfsWorkingDir, replication, fs);
	}

	public static void loadTableInfo(String tableName, String hdfsWorkingDir, FileSystem fs) {
        if (tableInfos.get(tableName) == null) {
            TableInfo tableInfo = new TableInfo(tableName);
            tableInfo.load(hdfsWorkingDir, fs);
            tableInfos.put(tableName, tableInfo);
        }
	}
} public class TableInfo {
    // Name of table.
    public String tableName;

    // Total number of tuples in dataset.
    public double numTuples;

    // TPC-H generated files use '|'. CSV uses ','.
    public char delimiter;

    // Schema of data set.
    public Schema schema;

    // -1 for non-join specific partition
    public int[] partitions;

    public int depth;

    public TableInfo(String tableName) {
        this(tableName, 0, '|', null);
    }

    public TableInfo(String tableName, double numTuples, char delimiter, Schema schema) {
        this.tableName = tableName;
        this.numTuples = numTuples;
        this.delimiter = delimiter;
        this.schema = schema;
        this.partitions = new int[]{-1};
        this.depth = 0;
    }

    private String partitionsString() {
        String res = "";
        for (int i = 0; i < partitions.length; i++) {
            if (i != 0) {
                res += ";";
            }
            res += Integer.toString(partitions[i]);
        }
        return res;
    }

    public TypeUtils.TYPE[] getTypeArray() {
        return schema.getTypeArray();
    }

    public void save(String hdfsWorkingDir, short replication, FileSystem fs) {
        String saveContent = "TOTAL_NUM_TUPLES: " + numTuples + "\n" +
                "DELIMITER: " + delimiter + "\n" +
                "SCHEMA: " + schema.toString() + "\n" +
                "PARTITION: " + partitionsString() + "\n" +
                "DEPTH: " + depth + "\n";
        byte[] saveContentBytes = saveContent.getBytes();
        String path = hdfsWorkingDir + "/" + tableName + "/info";
        HDFSUtils.writeFile(fs, path, replication,
                saveContentBytes, 0, saveContentBytes.length, false);
    }

    public void load(String hdfsWorkingDir, FileSystem fs) {
        String path = hdfsWorkingDir + "/" + tableName + "/info";
        byte[] fileContent = HDFSUtils.readFile(fs, path);
        String content = new String(fileContent);

        String[] settings = content.split("\n");

        if (settings.length != 5) {
            throw new RuntimeException();
        }

        for (int i = 0; i < settings.length; i++) {
            String setting = settings[i];
            String[] parts = setting.split(":");
            switch (parts[0].trim()) {
                case "TOTAL_NUM_TUPLES":
                    numTuples = Double.parseDouble(parts[1].trim());
                    break;
                case "DELIMITER":
                    delimiter = parts[1].trim().charAt(0);
                    break;
                case "SCHEMA":
                    schema = Schema.createSchema(parts[1].trim());
                    break;
                case "PARTITION":
                    String[] strPartitions = parts[1].trim().split(";");
                    partitions = new int[strPartitions.length];
                    for (int k = 0; k < strPartitions.length; k++) {
                        partitions[k] = Integer.parseInt(strPartitions[k]);
                    }
                    break;
                case "DEPTH":
                    depth = Integer.parseInt(parts[1].trim());
                    break;
                default:
                    System.out.println("Unknown setting found: " + parts[0].trim());
            }
        }
    }

    public void gc(String hdfsWorkingDir, FileSystem fs) {
        String path = hdfsWorkingDir + "/" + tableName;
        String pathToData = path + "/data";

        for (int i = 0; i < partitions.length; i++) {
            String pathToIndex;

            if (partitions[i] == -1) {
                pathToIndex = path + "/index";
            } else {
                pathToIndex = path + "/index." + partitions[i];
            }

            byte[] indexBytes = HDFSUtils.readFile(fs, pathToIndex);
            JoinRobustTree rt = new JoinRobustTree(this);
            rt.unmarshall(indexBytes);
            int[] bids = rt.getAllBuckets();
            HashSet<Integer> buckets = new HashSet<Integer>();
            for (int k = 0; k < bids.length; k++) {
                buckets.add(bids[k]);
            }
            try {
                FileStatus[] existingFiles = fs.listStatus(new Path(pathToData));
                for (int k = 0; k < existingFiles.length; k++) {
                    Path fp = existingFiles[k].getPath();
                    String fileName = FilenameUtils.getName(fp.toString());
                    int id = Integer.parseInt(fileName);
                    if (buckets.contains(id) == false) {
                        fs.delete(fp, false);
                    }
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
} public class JRNode {

    public int attribute;
    public TYPE type;
    public Object value;


    public JRNode parent;
    public JRNode leftChild;
    public JRNode rightChild;

    public boolean updated;
    public boolean fullAccessed;

    public Bucket bucket;

    public Map<Integer, BucketInfo> rangesByAttribute;

    public JRNode() {
        attribute = -1;
        type = null;
        value = null;

        parent = null;
        leftChild = null;
        rightChild = null;

        updated = false;
        fullAccessed = false;

        bucket = null;

        rangesByAttribute = new HashMap<Integer, BucketInfo>();
    }

    @Override
    public JRNode clone() {
        JRNode r = new JRNode();
        r.attribute = this.attribute;
        r.type = this.type;
        r.value = this.value;

        r.parent = this.parent;
        r.leftChild = this.leftChild;
        r.rightChild = this.rightChild;

        r.updated = this.updated;
        r.fullAccessed = this.fullAccessed;

        r.bucket = this.bucket;
        return r;
    }

    @Override
    public boolean equals(Object val) {
        if (val instanceof JRNode) {
            JRNode r = (JRNode) val;
            boolean allGood = true;
            allGood &= this.attribute == r.attribute;
            allGood &= this.type == r.type;
            allGood &= TypeUtils.compareTo(this.value, r.value, this.type) == 0;

            if (!allGood)
                return false;

            allGood = this.leftChild == r.leftChild;
            if (!allGood)
                return false;

            allGood = this.rightChild == r.rightChild;
            if (!allGood)
                return false;

            return true;
        }

        return false;
    }

    public void setValues(int dimension, TYPE type, RawIndexKey key) {
        this.attribute = dimension;
        this.type = type;
        this.value = getValue(dimension, type, key);
    }

    private Object getValue(int dimension, TYPE type, RawIndexKey key) {
        switch (type) {
            case INT:
                return key.getIntAttribute(dimension);
            case LONG:
                return key.getLongAttribute(dimension);
            case DOUBLE:
                return key.getDoubleAttribute(dimension);
            case DATE:
                return key.getDateAttribute(dimension);
            case STRING:
                return key.getStringAttribute(dimension);
            default:
                throw new RuntimeException("Unknown dimension type: " + type);
        }
    }

    private int compareKey(Object value, int dimension, TYPE type,
                           RawIndexKey key) {
        switch (type) {
            case INT:
                return ((Integer) value).compareTo(key.getIntAttribute(dimension));
            case LONG:
                return ((Long) value).compareTo(key.getLongAttribute(dimension));
            case DOUBLE:
                return ((Double) value).compareTo(key.getDoubleAttribute(dimension));
            case DATE:
                return ((SimpleDate) value).compareTo(key
                        .getDateAttribute(dimension));
            case STRING:
                return ((String) value).compareTo(key.getStringAttribute(dimension));
            default:
                throw new RuntimeException("Unknown dimension type: " + type);
        }
    }

    public int getBucketId(RawIndexKey key) {
        if (this.bucket != null) {
            return bucket.getBucketId();
        } else {
            if (compareKey(value, attribute, type, key) >= 0) {
                return leftChild.getBucketId(key);
            } else {
                return rightChild.getBucketId(key);
            }
        }
    }

    public List<JRNode> search(Predicate[] ps) {
        if (bucket == null) {
            boolean goLeft = true;
            boolean goRight = true;
            for (int i = 0; i < ps.length; i++) {
                Predicate p = ps[i];
                if (p.attribute == attribute) {
                    switch (p.predtype) {
                        case GEQ:
                            if (TypeUtils.compareTo(p.value, value, type) > 0)
                                goLeft = false;
                            break;
                        case LEQ:
                            if (TypeUtils.compareTo(p.value, value, type) <= 0)
                                goRight = false;
                            break;
                        case GT:
                            if (TypeUtils.compareTo(p.value, value, type) >= 0)
                                goLeft = false;
                            break;
                        case LT:
                            if (TypeUtils.compareTo(p.value, value, type) <= 0)
                                goRight = false;
                            break;
                        case EQ:
                            if (TypeUtils.compareTo(p.value, value, type) <= 0)
                                goRight = false;
                            else
                                goLeft = false;
                            break;
                    }
                }
            }


            List<JRNode> ret = null;
            if (goLeft) {
                ret = leftChild.search(ps);
            }

            if (goRight) {
                if (ret == null) {
                    ret = rightChild.search(ps);
                } else {
                    ret.addAll(rightChild.search(ps));
                }
            }

            if (ret == null) {
                String nStr = String.format("n %d %s %s\n", attribute,
                        type.toString(),
                        TypeUtils.serializeValue(value, type));
                System.out.println("ERR:" + goLeft + " " + goRight);
                System.out.println("ERR: " + nStr);
            }

            return ret;
        } else {
            List<JRNode> ret = new LinkedList<JRNode>();
            ret.add(this);
            return ret;
        }
    }

    public double numTuplesInSubtree() {
        LinkedList<JRNode> stack = new LinkedList<JRNode>();
        stack.add(this);
        double total = 0;
        while (stack.size() > 0) {
            JRNode t = stack.removeLast();
            if (t.bucket != null) {
                total += t.bucket.getEstimatedNumTuples();
            } else {
                stack.add(t.rightChild);
                stack.add(t.leftChild);
            }
        }

        return total;
    }

    public String marshall() {
        String ret = "";
        LinkedList<JRNode> stack = new LinkedList<JRNode>();
        stack.add(this);
        while (stack.size() != 0) {
            JRNode n = stack.removeLast();
            String nStr;
            if (n.bucket != null) {
                nStr = String.format("b %d\n", n.bucket.getBucketId());
            } else {
                nStr = String.format("n %d %s %s\n", n.attribute,
                        n.type.toString(),
                        TypeUtils.serializeValue(n.value, n.type));

                stack.add(n.rightChild);
                stack.add(n.leftChild);
            }
            ret += nStr;
        }
        return ret;
    }

    public void unmarshall(byte[] bytes) {
        String tree = new String(bytes);
        Scanner sc = new Scanner(tree);

        this.parseNode(sc);
    }

    public JRNode parseNode(Scanner sc) {
        String type = sc.next();
        if (type.equals("n")) {
            this.attribute = sc.nextInt();
            this.type = TYPE.valueOf(sc.next());
            // For string tokens; we may have to read more than one token, so
            // read till end of line
            this.value = TypeUtils.deserializeValue(this.type, sc.nextLine()
                    .trim());

            boolean exists = this.rangesByAttribute.containsKey(this.attribute);

            this.leftChild = new JRNode();
            this.leftChild.parent = this;

            this.rightChild = new JRNode();
            this.rightChild.parent = this;

            for (Map.Entry<Integer, BucketInfo> entry : this.rangesByAttribute
                    .entrySet()) {
                this.leftChild.rangesByAttribute.put(entry.getKey(), entry
                        .getValue().clone());
                this.rightChild.rangesByAttribute.put(entry.getKey(), entry
                        .getValue().clone());
            }
            this.rangesByAttribute = null;

            if (exists) {
                this.leftChild.rangesByAttribute.get(this.attribute).intersect(
                        new BucketInfo(this.type, null, this.value));
                this.rightChild.rangesByAttribute.get(this.attribute)
                        .intersect(new BucketInfo(this.type, this.value, null));
            } else {
                this.leftChild.rangesByAttribute.put(this.attribute,
                        new BucketInfo(this.type, null, this.value));
                this.rightChild.rangesByAttribute.put(this.attribute,
                        new BucketInfo(this.type, this.value, null));
            }

            this.leftChild.parseNode(sc);
            this.rightChild.parseNode(sc);

        } else if (type.equals("b")) {
            Bucket b = new Bucket(sc.nextInt());
            this.bucket = b;
            for (BucketInfo info : this.rangesByAttribute.values()) {
                info.setId(b.getBucketId());
            }
        } else {
            System.out.println("Bad things have happened in unmarshall");
            System.out.println(type);
        }

        return this;
    }
} public class JoinRobustTree implements MDIndex {
    public TableInfo tableInfo;

    public int maxBuckets;
    public int numAttributes;
    public int joinAttributeDepth;
    public ParsedTupleList sample;

    public TYPE[] dimensionTypes;
    JRNode root;

    public static Random randGenerator = new Random();

    public JoinRobustTree(TableInfo tableInfo) {
        this.root = new JRNode();
        this.tableInfo = tableInfo;
    }

    public JoinRobustTree() {
    }

    public void setMaxBuckets(int maxBuckets) {
        this.maxBuckets = maxBuckets;
    }

    public int getMaxBuckets() {
        return maxBuckets;
    }

    @Override
    public MDIndex clone() throws CloneNotSupportedException {
        throw new CloneNotSupportedException();
    }

    public JRNode getRoot() {
        return root;
    }

    // Only used for testing
    public void setRoot(JRNode root) {
        this.root = root;
    }

    @Override
    public boolean equals(Object obj) {
        if (obj instanceof JoinRobustTree) {
            JoinRobustTree rhs = (JoinRobustTree) obj;
            boolean allGood = true;
            allGood &= rhs.numAttributes == this.numAttributes;
            allGood &= rhs.maxBuckets == this.maxBuckets;
            allGood &= rhs.dimensionTypes.length == this.dimensionTypes.length;
            if (!allGood)
                return false;

            for (int i = 0; i < this.dimensionTypes.length; i++) {
                allGood &= this.dimensionTypes[i] == rhs.dimensionTypes[i];
            }
            if (!allGood)
                return false;

            allGood = this.root == rhs.root;
            return allGood;
        }
        return false;
    }

    public PartitionSplit delete(long delteSize, boolean deleteAll,JoinQuery query, int indexPartition, Map<Integer, Long> partitionIdSizeMap) {

        LinkedList<JRNode> nodeQueue = new LinkedList<JRNode>();
        LinkedList<JRNode> deleteNodes = new LinkedList<JRNode>();

        List<Integer> bucket_ids = new ArrayList<Integer>();


        PartitionIterator pi = new JoinRepartitionIterator(query.castToQuery(), indexPartition);

        // Initialize root with attribute 0
        nodeQueue.add(this.getRoot());

        while (nodeQueue.size() > 0) {
            JRNode t = nodeQueue.pollFirst();

            if (t.bucket != null){
                continue;
            }

            deleteNodes.add(t);

            nodeQueue.add(t.leftChild);
            nodeQueue.add(t.rightChild);
        }

        // reverse deleteNodes

        Collections.reverse(deleteNodes);


        if (deleteAll){

            int[] buckets = getAllBuckets();
            PartitionSplit psplit = new PartitionSplit(buckets, pi);
            return psplit;
        } else {
            for (JRNode node: deleteNodes){

                if (delteSize <= 0) break;

                if (node.leftChild.bucket == null || node.rightChild.bucket == null){
                    throw new RuntimeException();
                }

                if (node == this.getRoot()){

                    int lbid = node.leftChild.bucket.getBucketId();
                    int rbid = node.rightChild.bucket.getBucketId();
                    if (partitionIdSizeMap.containsKey(lbid))
                    {
                        delteSize -= partitionIdSizeMap.get(lbid);
                    }
                    if (partitionIdSizeMap.containsKey(rbid))
                    {
                        delteSize -= partitionIdSizeMap.get(rbid);
                    }

                    bucket_ids.add(lbid);
                    bucket_ids.add(rbid);

                    this.root = null;

                } else{
                    // delete left child
                    JRNode parent = node.parent;

                    if (parent.rightChild == node){
                        parent.rightChild = node.rightChild;
                    } else {
                        parent.leftChild = node.rightChild;
                    }

                    int bid = node.leftChild.bucket.getBucketId();
                    if (partitionIdSizeMap.containsKey(bid))
                    {
                        delteSize -= partitionIdSizeMap.get(bid);
                    }
                    bucket_ids.add(bid);

                }
            }

            int[] buckets = new int[bucket_ids.size()];

            for(int i =0 ;i < buckets.length; i ++){
                buckets[i] = bucket_ids.get(i);
            }

            PartitionSplit psplit = new PartitionSplit(buckets, pi);
            return psplit;

        }

    }

    public class Task {
        JRNode node;
        int depth;
        ParsedTupleList sample;
    }

    /**
     * Created the tree based on the histograms
     */
    @Override
    public void initProbe() {
        System.out.println("method not implemented!");
    }


    @Override
    public void initProbe(int joinAttribute) {
        System.out.println(this.sample.size() + " keys inserted");

        // Computes log(this.maxBuckets)
        int maxDepth = 31 - Integer.numberOfLeadingZeros(this.maxBuckets);
        double allocationPerAttribute = RobustTree.nthroot(
                this.numAttributes, this.maxBuckets);
        System.out.println("Max allocation: " + allocationPerAttribute);

        double[] allocations = new double[this.numAttributes];
        for (int i = 0; i < this.numAttributes; i++) {
            allocations[i] = allocationPerAttribute;
        }

        /**
         * Do a level-order traversal
         */
        LinkedList<Task> nodeQueue = new LinkedList<Task>();
        // Initialize root with attribute 0
        Task initialTask = new Task();
        initialTask.node = root;
        initialTask.sample = this.sample;
        initialTask.depth = 1;
        nodeQueue.add(initialTask);

        while (nodeQueue.size() > 0) {
            Task t = nodeQueue.pollFirst();


            if (t.depth > maxDepth) {
                Bucket b = new Bucket();
                b.setSample(t.sample);
                t.node.bucket = b;
                continue;
            }

            int dim = -1;
            Pair<ParsedTupleList, ParsedTupleList> halves = null;

            if (t.depth <= this.joinAttributeDepth) {
                dim = joinAttribute;
                allocations[dim] -= 2.0 / Math.pow(2, t.depth - 1);
                halves = t.sample.sortAndSplit(dim);
            } else if (t.depth <= maxDepth) {

                boolean[] validDims = new boolean[numAttributes];
                Arrays.fill(validDims, true);

                for (int i = 0; i < numAttributes; i++) {
                    int testDim = getLeastAllocated(allocations, validDims);
                    halves = t.sample.sortAndSplit(testDim);
                    if (halves.first.size() > 0 && halves.second.size() > 0) {
                        dim = testDim;
                        allocations[dim] -= 2.0 / Math.pow(2, t.depth - 1);
                        break;
                    } else {
                        validDims[testDim] = false;
                        //System.err.println("WARN: Skipping attribute " + testDim);
                    }
                }

            }

            if (dim == -1) {
                //System.err.println("ERR: No attribute to partition on");
                Bucket b = new Bucket();
                b.setSample(t.sample);
                t.node.bucket = b;
            } else {
                t.node.attribute = dim;
                t.node.type = this.dimensionTypes[dim];
                t.node.value = halves.first.getLast(dim); // Need to traverse up for range.

                t.node.leftChild = new JRNode();
                t.node.leftChild.parent = t.node;
                Task tl = new Task();
                tl.node = t.node.leftChild;
                tl.depth = t.depth + 1;
                tl.sample = halves.first;
                nodeQueue.add(tl);

                t.node.rightChild = new JRNode();
                t.node.rightChild.parent = t.node;
                Task tr = new Task();
                tr.node = t.node.rightChild;
                tr.depth = t.depth + 1;
                tr.sample = halves.second;
                nodeQueue.add(tr);
            }
        }

        System.out.println("Final Allocations: " + Arrays.toString(allocations));
    }

    public void initProbe(JoinQuery q) {
        System.out.println(this.sample.size() + " keys inserted");
        randGenerator.setSeed(0);

        int joinAttribute = q.getJoinAttribute();

        // Computes log(this.maxBuckets)
        int maxDepth = 31 - Integer.numberOfLeadingZeros(this.maxBuckets);
        double allocationPerAttribute = Math.pow(this.maxBuckets, 1.0 / this.numAttributes);

        double[] allocations = new double[this.numAttributes];
        for (int i = 0; i < this.numAttributes; i++) {
            allocations[i] = allocationPerAttribute;
        }

        /**
         * Do a level-order traversal
         */
        LinkedList<Task> nodeQueue = new LinkedList<Task>();
        // Initialize root with attribute 0
        Task initialTask = new Task();
        initialTask.node = root;
        initialTask.sample = this.sample;
        initialTask.depth = 1;
        nodeQueue.add(initialTask);

        while (nodeQueue.size() > 0) {
            Task t = nodeQueue.pollFirst();


            if (t.depth > maxDepth) {
                Bucket b = new Bucket();
                b.setSample(t.sample);
                t.node.bucket = b;
                continue;
            }

            int dim = -1;
            Pair<ParsedTupleList, ParsedTupleList> halves = null;

            if (t.depth <= this.joinAttributeDepth) {
                dim = joinAttribute;
                allocations[dim] -= 2.0 / Math.pow(2, t.depth - 1);
                halves = t.sample.sortAndSplit(dim);
            } else if (t.depth <= maxDepth) {

                // use attributes from the query

                int numAttributes = this.numAttributes;
                boolean[] validDims = new boolean[numAttributes];
                Arrays.fill(validDims, false);

                Predicate[] ps = q.getPredicates();

                int numPredicates = 0;

                for (int i = 0; i < ps.length; i++) {
                    if (validDims[ps[i].attribute] == false) {
                        validDims[ps[i].attribute] = true;
                        numPredicates++;
                    }

                }

                dim = -1;

                for (int i = 0; i < numPredicates; i++) {
                    int testDim = getLeastAllocated(allocations, validDims);
                    halves = t.sample.sortAndSplit(testDim);
                    if (halves.first.size() > 0 && halves.second.size() > 0) {
                        dim = testDim;
                        allocations[dim] -= 2.0 / Math.pow(2, t.depth - 1);
                        break;
                    } else {
                        validDims[testDim] = false;
                    }
                }

                numPredicates = this.numAttributes;

                if (dim == -1) {
                    Arrays.fill(validDims, true);
                    for (int i = 0; i < numAttributes; i++) {
                        int testDim = getLeastAllocated(allocations, validDims);
                        halves = t.sample.sortAndSplit(testDim);
                        if (halves.first.size() > 0 && halves.second.size() > 0) {
                            dim = testDim;
                            allocations[dim] -= 2.0 / Math.pow(2, t.depth - 1);
                            break;
                        } else {
                            validDims[testDim] = false;
                            //System.err.println("WARN: Skipping attribute " + testDim);
                        }
                    }
                }

            }

            if (dim == -1) {
                //System.err.println("ERR: No attribute to partition on");
                Bucket b = new Bucket();
                b.setSample(t.sample);
                t.node.bucket = b;
            } else {
                t.node.attribute = dim;
                t.node.type = this.dimensionTypes[dim];
                t.node.value = halves.first.getLast(dim); // Need to traverse up for range.

                t.node.leftChild = new JRNode();
                t.node.leftChild.parent = t.node;
                Task tl = new Task();
                tl.node = t.node.leftChild;
                tl.depth = t.depth + 1;
                tl.sample = halves.first;
                nodeQueue.add(tl);

                t.node.rightChild = new JRNode();
                t.node.rightChild.parent = t.node;
                Task tr = new Task();
                tr.node = t.node.rightChild;
                tr.depth = t.depth + 1;
                tr.sample = halves.second;
                nodeQueue.add(tr);
            }
        }

        System.out.println("Final Allocations: " + Arrays.toString(allocations));
    }

    /**
     * Return the dimension which has the maximum allocation unfulfilled
     */
    public static int getLeastAllocated(double[] allocations, boolean[] validDims) {

        boolean validFound = false;
        double minAlloc = 0;

        ArrayList<Integer> leastAllocated = new ArrayList<Integer>();

        for (int i = 0; i < allocations.length; i++) {
            if (validDims[i] == false) {
                continue;
            }
            if (validFound == false) {
                validFound = true;
                minAlloc = allocations[i];
                leastAllocated.add(i);
            } else {
                if (allocations[i] > minAlloc) {
                    minAlloc = allocations[i];
                    leastAllocated.clear();
                    leastAllocated.add(i);
                } else if (allocations[i] == minAlloc) {
                    leastAllocated.add(i);
                }
            }
        }

        if (leastAllocated.size() == 1) {
            return leastAllocated.get(0);
        } else {
            int r = randGenerator.nextInt(leastAllocated.size());
            return leastAllocated.get(r);
        }
    }

    /**
     * Used in the 2nd phase of upfront to assign each tuple to the right
     */
    @Override
    public Object getBucketId(RawIndexKey key) {
        return root.getBucketId(key);
    }


    private void getAllBucketsHelper(JRNode node, ArrayList<Integer> ids) {
        if (node.bucket != null) {
            ids.add(node.bucket.getBucketId());
            return;
        }
        getAllBucketsHelper(node.leftChild, ids);
        getAllBucketsHelper(node.rightChild, ids);
    }

    public int[] getAllBuckets() {
        ArrayList<Integer> ids = new ArrayList<Integer>();
        getAllBucketsHelper(root, ids);
        int[] buckets = new int[ids.size()];
        for (int i = 0; i < buckets.length; i++) {
            buckets[i] = ids.get(i);
        }
        return buckets;
    }

    /***************************************************
     * **************** RUNTIME METHODS *****************
     ***************************************************/

    public List<JRNode> getMatchingBuckets(Predicate[] predicates) {
        List<JRNode> results = root.search(predicates);
        return results;
    }

    /**
     * Serializes the index to string Very brittle - Consider rewriting
     */
    @Override
    public byte[] marshall() {
        // JVM optimizes shit so no need to use string builder / buffer
        // Format:
        // maxBuckets, numAttributes
        // types
        // nodes in pre-order

        String robustTree = "";
        robustTree += String.format("%d %d %d\n", this.maxBuckets,
                this.numAttributes, this.joinAttributeDepth);

        String types = "";
        for (int i = 0; i < this.numAttributes; i++) {
            types += this.dimensionTypes[i].toString() + " ";
        }
        types += "\n";
        robustTree += types;

        robustTree += this.root.marshall();

        return robustTree.getBytes();
    }

    @Override
    public void unmarshall(byte[] bytes) {
        String tree = new String(bytes);
        Scanner sc = new Scanner(tree);
        this.maxBuckets = sc.nextInt();
        this.numAttributes = sc.nextInt();
        this.joinAttributeDepth = sc.nextInt();

        this.dimensionTypes = new TYPE[this.numAttributes];
        for (int i = 0; i < this.numAttributes; i++) {
            this.dimensionTypes[i] = TYPE.valueOf(sc.next());
        }

        this.root = new JRNode();
        this.root.parseNode(sc);
    }

    public void loadSample(TableInfo tableInfo, byte[] bytes) {
        this.sample = new ParsedTupleList(tableInfo.getTypeArray());
        this.sample.unmarshall(bytes, tableInfo.delimiter);
        this.dimensionTypes = this.sample.getTypes();
        this.numAttributes = this.dimensionTypes.length;
        this.initializeBucketSamplesAndCounts(this.root, this.sample, this.sample.size(), tableInfo.numTuples);
    }

    public void loadSample(ParsedTupleList sample) {
        this.sample = sample;
        this.dimensionTypes = this.sample.getTypes();
        this.numAttributes = this.dimensionTypes.length;
    }

    public void initializeBucketSamplesAndCounts(JRNode n,
                                                 ParsedTupleList sample, final double totalSamples,
                                                 final double totalTuples) {

        if (n.bucket != null) {
            long sampleSize = sample.size();
            double numTuples = (sampleSize * totalTuples) / totalSamples;
            n.bucket.setSample(sample);
            n.bucket.setEstimatedNumTuples(numTuples);
        } else {
            // By sorting we avoid memory allocation
            // Will most probably be faster
            sample.sort(n.attribute);
            Pair<ParsedTupleList, ParsedTupleList> halves = sample
                    .splitAt(n.attribute, n.value);
            initializeBucketSamplesAndCounts(n.leftChild, halves.first,
                    totalSamples, totalTuples);
            initializeBucketSamplesAndCounts(n.rightChild, halves.second,
                    totalSamples, totalTuples);
        }
    }

    /**
     * Prints the tree created. Call only after initProbe is done.
     */
    public void printTree() {
        printNode(root);
    }

    public static void printNode(JRNode node) {
        if (node.bucket != null) {
            System.out.println("B " + node.bucket.getBucketId());
        } else {
            /*
            System.out.format("Node: %d %s { ", node.attribute,
                    node.value.toString());
            printNode(node.leftChild);
            System.out.print(" }{ ");
            printNode(node.rightChild);
            System.out.print(" }");
            */
            printNode(node.leftChild);
            printNode(node.rightChild);
        }
    }

    public Map<Integer, BucketInfo> getBucketRanges(int attribute) {
        return getBucketRangeSubtree(this.getRoot(), attribute);
    }

    private static Map<Integer, BucketInfo> getBucketRangeSubtree(JRNode root,
                                                                  int attribute) {
        if (root.bucket != null) {
            Map<Integer, BucketInfo> bucketRanges = new HashMap<Integer, BucketInfo>();
            if (root.rangesByAttribute.get(attribute) != null) {
                bucketRanges.put(root.bucket.getBucketId(),
                        root.rangesByAttribute.get(attribute));
            }
            return bucketRanges;
        } else {
            Map<Integer, BucketInfo> bucketRanges = getBucketRangeSubtree(
                    root.leftChild, attribute);
            bucketRanges.putAll(getBucketRangeSubtree(root.rightChild,
                    attribute));
            return bucketRanges;
        }
    }

    private void getAllocations_helper(JRNode node, int depth, double[] allocArray) {
        if (node.bucket != null) {
            return;
        }
        allocArray[node.attribute] += 2.0 / Math.pow(2, depth);
        getAllocations_helper(node.leftChild, depth + 1, allocArray);
        getAllocations_helper(node.rightChild, depth + 1, allocArray);
    }

    public double[] getAllocations() {
        double[] allocArray = new double[numAttributes];
        Arrays.fill(allocArray, 0);
        getAllocations_helper(root, 0, allocArray);
        return allocArray;
    }

    private void clearFlags_helper(JRNode node) {
        if (node.bucket != null) {
            return;
        }
        node.updated = false;
        node.fullAccessed = false;
        clearFlags_helper(node.leftChild);
        clearFlags_helper(node.rightChild);
    }

    public void clearFlags() {
        clearFlags_helper(root);
    }

    private boolean isUpdated_helper(JRNode node) {
        if (node.bucket != null) {
            return node.updated;
        }
        if (node.updated) {
            return true;
        }
        if (isUpdated_helper(node.leftChild)) {
            return true;
        } else {
            return isUpdated_helper(node.rightChild);
        }
    }

    public boolean isUpdated() {
        return isUpdated_helper(root);
    }
} public class Predicate {
	public enum PREDTYPE {
		LEQ, GEQ, GT, LT, EQ
	};

	public int attribute;
	public TYPE type;
	public Object value;
	public PREDTYPE predtype;

	public Predicate(int attr, TYPE t, Object val, PREDTYPE predtype) {
		this.attribute = attr;
		this.type = t;
		this.value = val;
		this.predtype = predtype;
	}

	public Predicate(TableInfo tableInfo, String attr, TYPE t, Object val, PREDTYPE predtype) {
		this.attribute = tableInfo.schema.getAttributeId(attr);
		this.type = t;
		this.value = val;
		this.predtype = predtype;
	}

	public Predicate(String predString) {
		String[] tokens = predString.split(":");
		this.attribute = Integer.parseInt(tokens[0]);
		this.type = TYPE.valueOf(tokens[1]);
		this.value = TypeUtils.deserializeValue(this.type, tokens[2]);
		this.predtype = PREDTYPE.valueOf(tokens[3]);
	}

	public void normalizePredicate() {
		if (this.predtype == PREDTYPE.GEQ) {
			this.predtype = PREDTYPE.GT;
			this.value = getHelpfulCutpoint();
		} else if (this.predtype == PREDTYPE.LT) {
			this.predtype = PREDTYPE.LEQ;
			this.value = getHelpfulCutpoint();
		}
	}

	public List<Predicate> getNormalizedPredicates() {
		List<Predicate> preds = new ArrayList<Predicate>();
		if (this.type == TYPE.INT || this.type == TYPE.DATE) {
            if (this.predtype == PREDTYPE.GEQ) {
                this.predtype = PREDTYPE.GT;
				this.value = TypeUtils.deltaLess(value, type);
				preds.add(this);
            } else if (this.predtype == PREDTYPE.LT) {
                this.predtype = PREDTYPE.LEQ;
				this.value = TypeUtils.deltaLess(value, type);
				preds.add(this);
            } else if (this.predtype == PREDTYPE.EQ) {
				Predicate pred1 = new Predicate(this.attribute, this.type, this.value, PREDTYPE.LEQ);
				Predicate pred2 = new Predicate(this.attribute, this.type, TypeUtils.deltaLess(value, type), PREDTYPE.GT);
				preds.add(pred1);
				preds.add(pred2);
			} else {
				preds.add(this);
			}
		} else {
			preds.add(this);
		}
		return preds;
	}

	/**
	 * Check if tuple with value for attribute is accepted (true) or rejected
	 * (false) by predicate
	 *
	 * @param value
	 * @return
	 */
	public boolean isRelevant(Object value) {
		switch (this.predtype) {
		case GEQ:
			if (TypeUtils.compareTo(this.value, value, this.type) <= 0)
				return true;
			break;
		case LEQ:
			if (TypeUtils.compareTo(this.value, value, this.type) >= 0)
				return true;
			break;
		case GT:
			if (TypeUtils.compareTo(this.value, value, this.type) < 0)
				return true;
			break;
		case LT:
			if (TypeUtils.compareTo(this.value, value, this.type) > 0)
				return true;
			break;
		case EQ:
			if (TypeUtils.compareTo(this.value, value, this.type) == 0)
				return true;
		}

		return false;
	}

	@Override
	public String toString() {
		return "" + attribute + ":" + type.toString() + ":"
				+ TypeUtils.serializeValue(value, type) + ":"
				+ predtype.toString();
	}

	public Object getHelpfulCutpoint() {
		switch (this.predtype) {
		case EQ:
		case GT:
		case LEQ:
			return value;
		case GEQ:
			return TypeUtils.deltaLess(value, type);

        case LT:
			if (type == TYPE.INT || type == TYPE.DATE)
				return TypeUtils.deltaLess(value, type);
			else
				return value;
		default:
			break;
		}
		return value;
	}
} public static class PartitionSplit {
	private int[] partitionIds;
	private PartitionIterator iterator;

	public PartitionSplit(int[] partitionIds, PartitionIterator iterator) {
		this.partitionIds = partitionIds;
		this.iterator = iterator;
	}

	public int[] getPartitions() {
		return this.partitionIds;
	}

	public PartitionIterator getIterator() {
		return this.iterator;
	}
} public class PartitionIterator implements Iterator<IteratorRecord> {

	public enum ITERATOR {
		SCAN, FILTER, REPART
	};

	protected IteratorRecord record;
	protected byte[] recordBytes;

	protected static char newLine = '\n';

	protected byte[] bytes;
	protected int bytesLength, offset, previous;

	protected Partition partition;

	protected Predicate[] predicates;

	protected Query query;

	public PartitionIterator() {

	}

	public PartitionIterator(Query q) {
		this.query = q;
	}

	public void setPartition(Partition partition) {
		this.partition = partition;
		if (Globals.getTableInfo(query.getTable()) == null) {
			String path = FilenameUtils.getPathNoEndSeparator(partition
					.getPath());

			if (FilenameUtils.getBaseName(path).contains("partitions")
					|| FilenameUtils.getBaseName(path).contains("repartition")) { // hack
				path = FilenameUtils.getPathNoEndSeparator(FilenameUtils.getPath(path));
			}

			if (FilenameUtils.getBaseName(path).contains("data")) { // hack
				path = FilenameUtils.getPathNoEndSeparator(FilenameUtils.getPath(path));
			}

			// To remove table name.
			path = FilenameUtils.getPathNoEndSeparator(FilenameUtils.getPath(path));

			// Initialize Globals.
			Globals.loadTableInfo(query.getTable(), path, ((HDFSPartition) partition).getFS());
		}

		TableInfo tableInfo = Globals.getTableInfo(query.getTable());
		record = new IteratorRecord(tableInfo.delimiter);
		bytes = partition.getBytes();
		// bytesLength = partition.getSize();
		bytesLength = bytes == null ? 0 : bytes.length;
		offset = 0;
		previous = 0;
	}

	@Override
	public boolean hasNext() {
		for (; offset < bytesLength; offset++) {
			if (bytes[offset] == newLine) {
				// record.setBytes(bytes, previous, offset-previous);
				recordBytes = ArrayUtils.subarray(bytes, previous, offset);

				try {
					record.setBytes(recordBytes);
				} catch (ArrayIndexOutOfBoundsException e) {
					System.out.println("Index out of bounds while setting bytes: "
									+ (new String(recordBytes)));
					throw e;
				}

				previous = ++offset;
				if (isRelevant(record)) {
					// System.out.println("relevant record found ..");
					return true;
				} else
					continue;
			}
		}
        return false;
	}

	protected boolean isRelevant(IteratorRecord record) {
		return true;
	}

	@Override
	public void remove() {
		next();
	}

	@Override
	public IteratorRecord next() {
		return record;
	}

	public void finish() {
	}

	public void write(DataOutput out) throws IOException {
	}

	public void readFields(DataInput in) throws IOException {
	}

	// public static PartitionIterator read(DataInput in) throws IOException {
	// PartitionIterator it = new PartitionIterator();
	// it.readFields(in);
	// return it;
	// }

	public static String iteratorToString(PartitionIterator itr) {
		ByteArrayDataOutput dat = ByteStreams.newDataOutput();
		try {
			itr.write(dat);
		} catch (IOException e) {
			e.printStackTrace();
			throw new RuntimeException("Failed to serialize the partitioner");
		}
		return itr.getClass().getName() + "@" + new String(dat.toByteArray());
	}

	public static PartitionIterator stringToIterator(String itrString) {
		String[] tokens = itrString.split("@", 2);
		return (PartitionIterator) ReflectionUtils.getInstance(tokens[0],
				new Class<?>[] { String.class }, new Object[] { tokens[1] });
	}
} public class PostFilterIterator extends PartitionIterator implements
		Serializable {
	private static final long serialVersionUID = 1L;

	public PostFilterIterator() {

	}

	public PostFilterIterator(Query q) {
		super(q);
	}

	public PostFilterIterator(String iteratorString) {
		try {
			readFields(ByteStreams.newDataInput(iteratorString.getBytes()));
		} catch (IOException e) {
			e.printStackTrace();
			throw new RuntimeException("Failed to read the fields");
		}
	}

	@Override
	protected boolean isRelevant(IteratorRecord record) {
		return query.qualifies(record);
	}

	@Override
	public void write(DataOutput out) throws IOException {
		query.write(out);
	}

	@Override
	public void readFields(DataInput in) throws IOException {
		String predicateString = Text.readString(in);
		query = new Query(predicateString);
	}

	public static PostFilterIterator read(DataInput in) throws IOException {
		PostFilterIterator it = new PostFilterIterator();
		it.readFields(in);
		return it;
	}
} public class RepartitionIterator extends PartitionIterator {
	private RNode newIndexTree;
	protected String zookeeperHosts;

	protected Map<Integer, Partition> newPartitions = new HashMap<Integer, Partition>();
	protected Map<Integer, Partition> oldPartitions = new HashMap<Integer, Partition>();

	public RepartitionIterator() {
	}

	public RepartitionIterator(String iteratorString) {
		try {
			readFields(ByteStreams.newDataInput(iteratorString.getBytes()));
		} catch (IOException e) {
			e.printStackTrace();
			throw new RuntimeException("Failed to read the fields");
		}
	}

	public RepartitionIterator(Query query) {
		super(query);
	}
	
	public RepartitionIterator(Query query, RNode tree) {
		this.query = query;
		this.newIndexTree = tree;
	}

	public void setZookeeper(String zookeeperHosts) {
		this.zookeeperHosts = zookeeperHosts;
	}

	public Query getQuery() {
		return this.query;
	}

	public RNode getIndexTree() {
		return this.newIndexTree;
	}

	/**
	 * Gets a HDFS Partition as input.
	 * Loads the new index. PartitionWriter::setPartition does the rest.
	 */
	@Override
	public void setPartition(Partition partition) {
		super.setPartition(partition);
		if (newIndexTree == null) {
			String path = FilenameUtils.getPathNoEndSeparator(partition
					.getPath());

			if (FilenameUtils.getBaseName(path).contains("partitions")
					|| FilenameUtils.getBaseName(path).contains("repartition")) { // hack
				path = FilenameUtils.getPathNoEndSeparator(FilenameUtils.getPath(path));
			}

			if (FilenameUtils.getBaseName(path).contains("data")) { // hack
				path = FilenameUtils.getPathNoEndSeparator(FilenameUtils.getPath(path));
			}

			// Initialize RobustTree.
			byte[] indexBytes = HDFSUtils.readFile(
					((HDFSPartition) partition).getFS(), path + "/index");
			RobustTree tree = new RobustTree(Globals.getTableInfo(query.getTable()));
			tree.unmarshall(indexBytes);
			newIndexTree = tree.getRoot();
		}
		oldPartitions.put(partition.getPartitionId(), partition);
	}

	@Override
	protected boolean isRelevant(IteratorRecord record) {
		int id = newIndexTree.getBucketId(record);
		Partition p;
		if (newPartitions.containsKey(id)) {
			p = newPartitions.get(id);
		} else {
			p = partition.clone();
			p.setPartitionId(id);
			newPartitions.put(id, p);
		}

		p.write(record.getBytes(), 0, record.getBytes().length);
		return query.qualifies(record);
	}

	@Override
	public void finish() {
		if (zookeeperHosts != null) {
			for (Partition p : newPartitions.values()) {
				System.out.println("storing partition id " + p.getPartitionId());
				p.store(true);
			}

			for (Partition p : oldPartitions.values()) {
				System.out.println("dropping old partition id "
						+ p.getPartitionId());
				p.drop();
			}

			oldPartitions = Maps.newHashMap();
			newPartitions = Maps.newHashMap();
		} else {
			System.out.println("INFO: Zookeeper Hosts NULL");
		}
	}

	@Override
	public void write(DataOutput out) throws IOException {
		query.write(out);
		Text.writeString(out, zookeeperHosts);
	}

	@Override
	public void readFields(DataInput in) throws IOException {
		String predicateString = Text.readString(in);
		query = new Query(predicateString);
		zookeeperHosts = Text.readString(in);
	}

	public static RepartitionIterator read(DataInput in) throws IOException {
		RepartitionIterator it = new RepartitionIterator();
		it.readFields(in);
		return it;
	}
} final class Bucket {
	/* Actual Values */
	int bucketId;
	ParsedTupleList sample;

	/* Estimates */
	private double estimatedTuples = 0;

	public static int maxBucketId = 0;

	private static HashMap<Integer, Double> counter = new HashMap<Integer, Double>();

	public Bucket() {
		bucketId = maxBucketId;
		maxBucketId += 1;
	}

	public Bucket(int id) {
		bucketId = id;
		maxBucketId = Math.max(bucketId + 1, maxBucketId);
	}

	public double getEstimatedNumTuples() {
		// TODO: This call used when restoring a replaced tree in Optimizer.
		// Can't use the assert below.
		// Assert.assertNotEquals(estimatedTuples, 0.0);
		return estimatedTuples;
	}

	public static double getEstimatedNumTuples(int bucketId) {
		Double val = counter.get(bucketId);
		return val;
	}

	public void setEstimatedNumTuples(double num) {
		estimatedTuples = num;
		counter.put(bucketId, num);
	}

	public int getBucketId() {
		return bucketId;
	}

	public void updateId() {
		this.bucketId = maxBucketId;
		maxBucketId += 1;
		estimatedTuples = 0;
	}

	public ParsedTupleList getSample() {
		return sample;
	}

	public void setSample(ParsedTupleList sample) {
		this.sample = sample;
	}
} public class ParsedTupleList {

	private List<Object[]> values;
	private TYPE[] types;
	private int sampleSize;

    public ParsedTupleList(TYPE[] types) {
        this.types = types;
        this.values = Lists.newArrayList();
        sampleSize = 0;
    }

	/**
	 * Instantiate a key set with a given set of keys.
	 *
	 * @param values
	 */
	public ParsedTupleList(List<Object[]> values, TYPE[] types) {
		this.values = values;
		this.types = types;
		this.sampleSize = 0;
	}

	/**
	 * Insert into the key set.
	 *
	 * @param key
	 */
	public void insert(RawIndexKey key) {
		Object[] keyValues = new Object[types.length];
		for (int i = 0; i < types.length; i++) {
			switch (types[i]) {
			case INT:
				keyValues[i] = key.getIntAttribute(i);
				break;
			case LONG:
				keyValues[i] = key.getLongAttribute(i);
				break;
			case DOUBLE:
				keyValues[i] = key.getDoubleAttribute(i);
				break;
			case DATE:
				keyValues[i] = key.getDateAttribute(i, new SimpleDate(0, 0, 0));
				break;
			case STRING:
				keyValues[i] = key.getStringAttribute(i);
				break;
			case VARCHAR:
				break; // skip partitioning on varchar attribute
			default:
				throw new RuntimeException("Unknown dimension type:");
			}
		}

		values.add(keyValues);
		sampleSize++;
	}

	/**
	 * Return the last entry in the keyset
	 */
	public Object getFirst(int dim) {
		assert values.size() > 0;
		assert dim < types.length;
		return values.get(0)[dim];
	}

	/**
	 * Return the last entry in the keyset
	 */
	public Object getLast(int dim) {
		assert values.size() > 0;
		assert dim < types.length;
		return values.get(values.size() - 1)[dim];
	}

	public Object[] getCutpoints(int dim, int numParts) {
		Object[] cutpoints = new Object[numParts + 1];
		this.sort(dim);
		cutpoints[0] = this.getFirst(dim);
		cutpoints[numParts] = this.getLast(dim);
		int partLength = this.values.size() / numParts;
		for (int i = 1; i <= numParts - 1; i++) {
			cutpoints[i] = this.values.get(i * partLength)[dim];
		}
		return cutpoints;
	}

	public Comparator<Object[]> getComparatorForType(TYPE type,
			final int attributeIdx) {
		switch (type) {
		case INT:
			return new Comparator<Object[]>() {
				@Override
				public int compare(Object[] o1, Object[] o2) {
					return ((Integer) o1[attributeIdx])
							.compareTo((Integer) o2[attributeIdx]);
				}
			};
		case LONG:
			return new Comparator<Object[]>() {
				@Override
				public int compare(Object[] o1, Object[] o2) {
					return ((Long) o1[attributeIdx])
							.compareTo((Long) o2[attributeIdx]);
				}
			};
		case DOUBLE:
			return new Comparator<Object[]>() {
				@Override
				public int compare(Object[] o1, Object[] o2) {
					return ((Double) o1[attributeIdx])
							.compareTo((Double) o2[attributeIdx]);
				}
			};
		case DATE:
			return new Comparator<Object[]>() {
				@Override
				public int compare(Object[] o1, Object[] o2) {
					return ((SimpleDate) o1[attributeIdx])
							.compareTo((SimpleDate) o2[attributeIdx]);
				}
			};
		case STRING:
			return new Comparator<Object[]>() {
				@Override
				public int compare(Object[] o1, Object[] o2) {
					return ((String) o1[attributeIdx])
							.compareTo((String) o2[attributeIdx]);
				}
			};
		case VARCHAR:
			throw new RuntimeException("sorting over varchar is not supported"); // skip
																					// partitioning
																					// on
																					// varchar
																					// attribute
		default:
			throw new RuntimeException("Unknown dimension type: " + type);
		}
	}

	/**
	 * Sort the key set on the given attribute.
	 *
	 * @param attributeIdx
	 *            -- the index of the sort attribute relative to the key
	 *            attributes.
	 *
	 *            Example -- if we have 5 attributes ids (0,1,2,3,4) and we
	 *            extract 3 of them in the CartilageIndexKey, e.g. (1,3,4) and
	 *            now if we want to sort on attribute 3 then the attributeIdx
	 *            parameter will pass 1 (sorting on the second key attribute).
	 *
	 */
	public void sort(final int attributeIdx) {
		final TYPE sortType = types[attributeIdx];
		try {
			Collections.sort(values,
					this.getComparatorForType(sortType, attributeIdx));
		} catch (ClassCastException e) {
			System.out.println("EXCEPTION: " + attributeIdx + " "
					+ sortType.toString());
			e.printStackTrace();
		}
	}

	/**
	 * Split this key set into two equal sized key sets. If the length is odd,
	 * the extra value goes in the second set. This function assumes that the
	 * key set has already been sorted.
	 *
	 * Note that we do not copy the keys into a new object (we simply create a
	 * view using the subList() method). Therefore, the original key set must
	 * not be destroyed after the split.
	 *
	 * @return
	 */
	public Pair<ParsedTupleList, ParsedTupleList> splitInTwo() {
		ParsedTupleList k1 = new ParsedTupleList(values.subList(0,
				values.size() / 2), types);
		ParsedTupleList k2 = new ParsedTupleList(values.subList(
				values.size() / 2, values.size()), types);
		return new Pair<ParsedTupleList, ParsedTupleList>(k1, k2);
	}


	/**
	 * Split this key set by the median on the specified attribute. All keys
	 * with the specified attribute larger than the median go in the second set.
	 *
	 * @return
	 */
	public Pair<ParsedTupleList, ParsedTupleList> splitByMedianLarger(
			int attributeIdx) {
		Object medianVal = values.get(values.size() / 2)[attributeIdx];
		Comparator<Object> comp = TypeUtils.getComparatorForType(types[attributeIdx]);

		int lo = 0;
		int hi = values.size() - 1;
		int res = -1;
		while (lo <= hi) {
			int mid = (lo + hi) / 2;
			Object midVal = this.values.get(mid)[attributeIdx];
			try {
				if (comp.compare(midVal, medianVal) > 0) {
					res = mid;
					hi = mid - 1;
				}
				else
				{
					lo = mid + 1;
				}

			} catch (Exception e) {
				System.out.println(midVal.toString() + " " + medianVal.toString() + " " + attributeIdx + " " + medianVal.getClass().getSimpleName() + " " + midVal.getClass().getSimpleName());
				e.printStackTrace();
			}
		}

		ParsedTupleList k1 = new ParsedTupleList(values.subList(0,
				res), types);
		ParsedTupleList k2 = new ParsedTupleList(values.subList(
				res, values.size()), types);
		return new Pair<ParsedTupleList, ParsedTupleList>(k1, k2);
	}

	/**
	 * Split this key set by the median on the specified attribute. All keys
	 * with the specified attribute equal to the median go in the second set.
	 *
	 * @return
	 */
	public Pair<ParsedTupleList, ParsedTupleList> splitByMedian(
			int attributeIdx) {
		Object medianVal = values.get(values.size() / 2)[attributeIdx];
		Comparator<Object> comp = TypeUtils.getComparatorForType(types[attributeIdx]);

		int lo = 0;
		int hi = values.size() - 1;
		int res = -1;
		while (lo <= hi) {
			int mid = (lo + hi) / 2;
			Object midVal = this.values.get(mid)[attributeIdx];
			try {
				if (comp.compare(midVal, medianVal) >= 0) {
					res = mid;
					hi = mid - 1;
				}
				else
				{
					lo = mid + 1;
				}

			} catch (Exception e) {
				System.out.println(midVal.toString() + " " + medianVal.toString() + " " + attributeIdx + " " + medianVal.getClass().getSimpleName() + " " + midVal.getClass().getSimpleName());
				e.printStackTrace();
			}
		}

		ParsedTupleList k1 = new ParsedTupleList(values.subList(0,
				res), types);
		ParsedTupleList k2 = new ParsedTupleList(values.subList(
				res, values.size()), types);
		return new Pair<ParsedTupleList, ParsedTupleList>(k1, k2);
	}

	/**
	 * Split this key set by the value on the specified attribute.
	 *
	 * @return
	 */
	public Pair<ParsedTupleList, ParsedTupleList> splitAt(
			int attributeIdx, Object value) {
		final TYPE sortType = types[attributeIdx];

		Comparator<Object> comp = TypeUtils.getComparatorForType(sortType);

		// Finds the least k such that k > value
		int lo = 0;
		int hi = values.size() - 1;
		int res = values.size(); // assume every value is smaller than or equal to the given value
		while (lo <= hi) {
			int mid = (lo + hi) / 2;
			Object midVal = this.values.get(mid)[attributeIdx];
			try {
				if (comp.compare(midVal, value) > 0) {
					res = mid;
					hi = mid - 1;
				}
				else
				{
					lo = mid + 1;
				}

			} catch (Exception e) {
				System.out.println(midVal.toString() + " " + value.toString() + " " + attributeIdx + " " + value.getClass().getSimpleName() + " " + midVal.getClass().getSimpleName());
				e.printStackTrace();
			}
		}


		ParsedTupleList	k1 = new ParsedTupleList(values.subList(0, res), types);
		ParsedTupleList	k2 = new ParsedTupleList(values.subList(res, values.size()), types);

		return new Pair<ParsedTupleList, ParsedTupleList>(k1, k2);
	}

	/**
	 * Sort and then split the key set.
	 *
	 * @param attributeIdx
	 * @return
	 */
	public Pair<ParsedTupleList, ParsedTupleList> sortAndSplit(
			final int attributeIdx) {
		sort(attributeIdx);

		Pair<ParsedTupleList, ParsedTupleList> halves = splitByMedian(attributeIdx);

		if (halves.first.size() == 0 && !halves.second.getFirst(attributeIdx).equals(halves.second.getLast(attributeIdx))){
			halves = splitByMedianLarger(attributeIdx);
		}
		return halves;
	}

	public List<Object[]> getValues() {
		return values;
	}

	public void addValues(List<Object[]> e) {
		this.values.addAll(e);
	}

	public void setTypes(TYPE[] types) {
		this.types = types;
	}

	public TYPE[] getTypes() {
		return this.types;
	}

	public int size() {
		return values.size();
	}

	public void reset() {
		values = Lists.newArrayList();
		types = null;
	}

	/**
	 * Iterate over the keys in the key set.
	 *
	 * @return
	 */
	public KeySetIterator iterator() {
		return new KeySetIterator(values, types);
	}

	/**
	 * An iterator to iterate over the key in a key set. The Iterator returns an
	 * instance of ParsedIndexKey which extends CartilageIndexKey.
	 *
	 * @author alekh
	 *
	 */
	public static class KeySetIterator implements Iterator<ParsedIndexKey> {
		private Iterator<Object[]> valueItr;
		private ParsedIndexKey key;

		public KeySetIterator(List<Object[]> values, TYPE[] types) {
			this.valueItr = values.iterator();
			key = new ParsedIndexKey(types);
		}

		@Override
		public boolean hasNext() {
			return valueItr.hasNext();
		}

		@Override
		public ParsedIndexKey next() {
			key.setValues(valueItr.next());
			return key;
		}

		@Override
		public void remove() {
			next();
		}
	}

	public byte[] marshall(char delimiter) {
		List<byte[]> byteArrays = Lists.newArrayList();

		int initialSize = sampleSize * 100; // assumption: each record could be
											// of max size 100 bytes
		byte[] recordBytes = new byte[initialSize];

		int offset = 0;
		for (Object[] v : values) {
			byte[] vBytes = Joiner.on(delimiter).join(v).getBytes();
			if (offset + vBytes.length + 1 > recordBytes.length) {
				byteArrays.add(BinaryUtils.resize(recordBytes, offset));
				recordBytes = new byte[initialSize];
				offset = 0;
			}
			BinaryUtils.append(recordBytes, offset, vBytes);
			offset += vBytes.length;
			recordBytes[offset++] = '\n';
		}

		byteArrays.add(BinaryUtils.resize(recordBytes, offset));

		byte[][] finalByteArrays = new byte[byteArrays.size()][];
		for (int i = 0; i < finalByteArrays.length; i++)
			finalByteArrays[i] = byteArrays.get(i);

		return Bytes.concat(finalByteArrays);
	}

	public void unmarshall(byte[] bytes, char delimiter) {
		RawIndexKey record = new RawIndexKey(delimiter);
		int offset = 0, previous = 0;
		for (; offset < bytes.length; offset++) {
			if (bytes[offset] == '\n') {
				byte[] lineBytes = ArrayUtils.subarray(bytes, previous, offset);
				record.setBytes(lineBytes);
				try {
					insert(record);
				} catch (ArrayIndexOutOfBoundsException e) {
					System.out.println("Exception in ParsedTupleList::unmarshall");
					e.printStackTrace();
				}
				previous = ++offset;
			}
		}
	}
} public class ConfUtils {
	private Properties p;

	public ConfUtils(String propertiesFile) {
		p = new Properties();
		try {
			p.load(new FileInputStream(propertiesFile));
		} catch (FileNotFoundException e) {
			throw new RuntimeException("could not read the properties file!");
		} catch (IOException e) {
			throw new RuntimeException("could not read the properties file!");
		}
	}

	/**
	 * Choose whether to enable optimizer If false, just runs the query If true,
	 * tries to optimize the storage layout
	 *
	 * @return
	 */
	public boolean getENABLE_OPTIMIZER() {
		return Boolean.parseBoolean(p.getProperty("ENABLE_OPTIMIZER").trim());
	}

	public String getMACHINE_ID() {
		String machineId = p.getProperty("MACHINE_ID").trim()
				.replaceAll("\\W+", "");
		assert !machineId.equals("");
		return machineId;
	}

	/********** SPARK CONFIG **************/
	public String getSPARK_MASTER() {
		return p.getProperty("SPARK_MASTER").trim();
	}

	public String getSPARK_HOME() {
		return p.getProperty("SPARK_HOME").trim();
	}

	/**
	 * @return spark application jar filepath
	 */
	public String getSPARK_APPLICATION_JAR() {
		return p.getProperty("SPARK_APPLICATION_JAR").trim();
	}

	public String getSPARK_EXECUTOR_MEMORY() {
		return p.getProperty("SPARK_EXECUTOR_MEMORY").trim();
	}

	public String getSPARK_DRIVER_MEMORY() {
		return p.getProperty("SPARK_DRIVER_MEMORY").trim();
	}

	public String getSPARK_TASK_CPUS() {
		return p.getProperty("SPARK_TASK_CPUS").trim();
	}

	/********** ZOOKEEPER CONFIG **************/
	public String getZOOKEEPER_HOSTS() {
		return p.getProperty("ZOOKEEPER_HOSTS").trim();
	}

	/********** HADOOP CONFIG **************/
	/**
	 * Path to hadoop installation
	 *
	 * @return
	 */
	public String getHADOOP_HOME() {
		return p.getProperty("HADOOP_HOME");
	}

	/**
	 * Path to hadoop namenode, eg: hdfs://localhost:9000
	 *
	 * @return
	 */
	public String getHADOOP_NAMENODE() {
		return p.getProperty("HADOOP_NAMENODE").trim();
	}

	/**
	 * Get working directory in HDFS.
	 *
	 * @return filepath to hdfs working dir
	 */
	public String getHDFS_WORKING_DIR() {
		return p.getProperty("HDFS_WORKING_DIR").trim();
	}
	public void setHDFS_WORKING_DIR(String path){
		p.setProperty("HDFS_WORKING_DIR", path);
	}

	/**
	 * Get HDFS Replication Factor
	 *
	 * @return
	 */
	public short getHDFS_REPLICATION_FACTOR() {
		return Short
				.parseShort(p.getProperty("HDFS_REPLICATION_FACTOR").trim());
	}
} public class HDFSUtils {

    public static class HDFSData {
        protected static FileSystem fs;
        protected String hadoopConfDir;

        public HDFSData(FileSystem fs) {
            if (HDFSData.fs == null)
                HDFSData.fs = fs;
        }

        public HDFSData(String hadoopConfDir) {
            if (fs == null) {
                Configuration conf = new Configuration();
                conf.addResource(new Path(hadoopConfDir + "/core-site.xml"));
                conf.addResource(new Path(hadoopConfDir + "/hdfs-site.xml"));
                try {
                    fs = FileSystem.get(conf);
                } catch (IOException e) {
                    throw new RuntimeException("could not get the filesystem: "
                            + e.getMessage());
                }
            }
            this.hadoopConfDir = hadoopConfDir;
        }

        protected void delete(Path dataPath) {
            try {
                fs.delete(dataPath, true);
            } catch (IOException e) {
                throw new RuntimeException("could not delete the data path: "
                        + dataPath + ", " + e.getMessage());
            }
        }
    }

    public static class HDFSDir extends HDFSData {
        private Path dataPath;

        public HDFSDir(String hadoopConfDir, String dataPath) {
            super(hadoopConfDir);
            this.dataPath = new Path(dataPath);
        }

        public List<HDFSFile> getFiles() {
            try {
                List<HDFSFile> files = Lists.newArrayList();
                for (FileStatus fileStatus : fs.listStatus(dataPath))
                    files.add(new HDFSFile(hadoopConfDir, fileStatus));
                return files;
            } catch (IOException e) {
                throw new RuntimeException(
                        "failed to list the files in directory " + dataPath
                                + " ! " + e.getMessage());
            }
        }

        public void delete() {
            super.delete(dataPath);
        }
    }

    public static class HDFSFile extends HDFSData {
        private FileStatus status;

        public HDFSFile(String hadoopConfDir, FileStatus status) {
            super(hadoopConfDir);
            this.status = status;
        }

        public HDFSFile() {
            super(HDFSData.fs);
        }

        public boolean isCorrupted() {
            try {
                for (BlockLocation blk : fs.getFileBlockLocations(status, 0,
                        status.getLen())) {
                    if (blk.isCorrupt())
                        return true;
                }
                return false;
            } catch (IOException e) {
                throw new RuntimeException(
                        "failed to get the blocks locations of file "
                                + status.getPath() + " ! " + e.getMessage());
            }
        }

        public void delete() {
            super.delete(status.getPath());
        }

        public List<HDFSFile> getSiblingFiles(String prefix) {
            final String siblingPrefix = prefix;
            String parentDir = FilenameUtils
                    .getPath(status.getPath().getName());
            try {
                FileStatus[] siblingStatuses = fs.listStatus(
                        new Path(parentDir), new PathFilter() {
                            @Override
                            public boolean accept(Path path) {
                                return path.getName().startsWith(siblingPrefix);
                            }
                        });
                List<HDFSFile> siblingFiles = Lists.newArrayList();
                for (FileStatus status : siblingStatuses)
                    siblingFiles.add(new HDFSFile(hadoopConfDir, status));
                return siblingFiles;
            } catch (IOException e) {
                throw new RuntimeException("Failed to get the siblings of "
                        + status.getPath() + ", " + e.getMessage());
            }
        }

        public String getPath() {
            return status.getPath().getName();
        }

        public void increaseReplication(int increment) {
            Path p = status.getPath();
            try {
                fs.setReplication(p,
                        (short) (status.getReplication() + increment));
            } catch (IOException e) {
                throw new RuntimeException(
                        "Failed to increase the replication of " + getPath()
                                + " by " + increment + ", " + e.getMessage());
            }
        }

        public byte[] getBytes() {
            return null; // TODO
        }

        public void putBytes(byte[] bytes) {
            // TODO
        }
    }

    private static FileSystem fs;

    private static FileSystem getFS(String coreSitePath) {
        if (fs == null) {
            try {
                Configuration conf = new Configuration();
                conf.addResource(new Path(coreSitePath));
                fs = FileSystem.get(conf);
            } catch (Exception e) {
                throw new RuntimeException(
                        "Failed to get the HDFS Filesystem! " + e.getMessage());
            }
        }
        return fs;
    }

    public static FileSystem getFSByHadoopHome(String hadoopHome) {
        return getFS(hadoopHome + "/etc/hadoop/core-site.xml");
    }

    public static OutputStream getHDFSOutputStream(FileSystem hdfs,
                                                   String filename, short replication, int bufferSize) {
        try {
            Path path = new Path(filename);
            if (hdfs.exists(path)) {
                return new BufferedOutputStream(hdfs.append(path, replication),
                        bufferSize);
            } else {
                return new BufferedOutputStream(hdfs.create(path, replication),
                        bufferSize);
            }
        } catch (FileNotFoundException e) {
            throw new RuntimeException("Could not open the file:" + filename);
        } catch (IOException e) {
            throw new RuntimeException("Could not open the file:" + filename
                    + ", " + e.getMessage());
        }
    }

    public static OutputStream getBufferedHDFSOutputStream(
            FileSystem fs, String filename, short replication, int bufferSize,
            CuratorFramework client) {
        return new HDFSBufferedOutputStream(fs, filename, replication,
                bufferSize, client);
    }

    public static InputStream getHDFSInputStream(FileSystem hdfs,
                                                 String filename) {
        try {
            return hdfs.open(new Path(filename));
        } catch (IOException e) {
            e.printStackTrace();
            throw new RuntimeException("Could not read from file:" + filename);
        }
    }

    public static void writeFile(FileSystem hdfs, String filename,
                                 short replication, byte[] bytes, int offset, int length,
                                 boolean append) {
        try {
            Path path = new Path(filename);
            FSDataOutputStream os;
            if (append && hdfs.exists(path))
                os = hdfs.append(path);
            else
                os = hdfs.create(new Path(filename), replication);
            os.write(bytes, offset, length);
            os.flush();
            os.close();
        } catch (IOException e) {
            e.printStackTrace();
            throw new RuntimeException("Could not write to file:" + filename);
        }
    }

    public static byte[] readFile(FileSystem hdfs, String filename) {
        try {
            FSDataInputStream in = hdfs.open(new Path(filename));
            byte[] bytes = ByteStreams.toByteArray(in);
            in.close();
            return bytes;
        } catch (IOException e) {
            e.printStackTrace();
            throw new RuntimeException("Could not read from file:" + filename);
        }
    }

    public static void closeHDFSInputStream(FSDataInputStream fsInputStream) {
        try {
            fsInputStream.close();
        } catch (IOException e) {
            throw new RuntimeException("Cannot close file. " + e.getMessage());
        }
    }

    public static void safeCreateFile(FileSystem fs, String path,
                                      short replication) {
        try {
            Path e = new Path(path);
            FSDataOutputStream os = fs.create(e, false, fs.getConf().getInt("io.file.buffer.size", 4096),
                    replication, fs.getDefaultBlockSize(e));
            os.close();

        } catch (IOException e) {
        }
    }


    public static void safeCreateDirectory(FileSystem fs, String path) {
        try {
            Path p = new Path(path);
            if (!fs.exists(p)) {
                fs.mkdirs(p);
            }
        } catch (IOException e) {
        }
    }

    public static boolean deleteFile(FileSystem fs, String filename,
                                     boolean recursive) {
        try {
            return fs.delete(new Path(filename), recursive);
        } catch (IOException e) {
            return false;
        }
    }

    public static void copyFile(FileSystem fs, String origin, String destination,
                                short replication) {
        byte[] contents = readFile(fs, origin);
        writeFile(fs, destination, replication, contents, 0, contents.length, false);
    }

    public static List<String> readHDFSLines(String hadoopHome, String filename) {
        try {
            Configuration conf = new Configuration();
            conf.addResource(new Path(hadoopHome + "/etc/hadoop/core-site.xml"));
            FileSystem fs = FileSystem.get(conf);
            Path path = new Path(filename);
            if (!fs.exists(path))
                return Lists.newArrayList();

            BufferedReader br = new BufferedReader(new InputStreamReader(
                    fs.open(path)));
            String line;
            List<String> lines = Lists.newArrayList();
            while ((line = br.readLine()) != null)
                lines.add(line);
            br.close();

            return lines;
        } catch (IOException e) {
            throw new RuntimeException("could not read the inputstream!");
        }
    }

    public static List<String> readHDFSLines(FileSystem fs, String filename) {
        try {
            Path path = new Path(filename);
            if (!fs.exists(path))
                return Lists.newArrayList();

            BufferedReader br = new BufferedReader(new InputStreamReader(
                    fs.open(path)));
            String line;
            List<String> lines = Lists.newArrayList();
            while ((line = br.readLine()) != null)
                lines.add(line);
            br.close();

            return lines;
        } catch (IOException e) {
            throw new RuntimeException("could not read the inputstream!");
        }
    }

    public static void appendLine(FileSystem fs, String filepath,
                                  String line) {
        try {
            FSDataOutputStream fout = fs.append(
                    new Path(filepath));
            fout.write(line.getBytes());
            fout.write('\n');
            fout.close();
        } catch (IOException e) {
            e.printStackTrace();
            throw new RuntimeException("could not append to file: " + filepath);
        }
    }

    public static void appendBytes(String hadoopHome, String filePath,
                                   byte[] bytes, int start, int length) {
        appendBytes(getFSByHadoopHome(hadoopHome), filePath, bytes, start,
                length);
    }

    public static void appendBytes(FileSystem fs, String filepath,
                                   byte[] bytes, int start, int length) {
        try {
            FSDataOutputStream fout = fs.append(new Path(filepath));
            fout.write(bytes, start, length);
            fout.close();
        } catch (IOException e) {
            //e.printStackTrace();
            System.out.println("Could not append to file: " + filepath);
            //throw new RuntimeException("Could not append to file: " + filepath);
        }
    }

    public static void writeHDFSLines(String hadoopHome, String filename,
                                      List<String> lines) {
        try {
            Configuration conf = new Configuration();
            conf.addResource(new Path(hadoopHome + "/etc/hadoop/core-site.xml"));
            FileSystem fs = FileSystem.get(conf);
            Path path = new Path(filename);
            if (fs.exists(path))
                fs.delete(path, true);

            FSDataOutputStream fout = fs.create(path);
            for (String line : lines) {
                fout.write(line.getBytes());
                fout.write('\n');
            }

            fout.close();
            fs.close();

        } catch (IOException e) {
            e.printStackTrace();
            throw new RuntimeException("failed to write the hdfs file: "
                    + filename + ", " + e.getMessage());
        }
    }

    public static OutputStream getOutputStreamWithRetry(FileSystem fs,
                                                        String filename, long retryIntervalMs, int maxRetryCount) {
        return getOutputStreamWithRetry(fs, filename, (short) 3,
                retryIntervalMs, maxRetryCount);
    }

    public static OutputStream getOutputStreamWithRetry(FileSystem fs,
                                                        String filename, short replication, long retryIntervalMs,
                                                        int maxRetryCount) {
        FSDataOutputStream fout = null;
        Path path = new Path(filename);

        int retryCount = 0;
        while (retryCount < maxRetryCount) {
            try {
                if (!fs.exists(path))
                    fout = fs.create(path, replication);
                else
                    fout = fs.append(path);
                break;
            } catch (IOException e) {
                try {
                    Thread.sleep(retryIntervalMs);
                } catch (InterruptedException i) {
                    throw new RuntimeException("Failed to sleep the thread: "
                            + i.getMessage());
                }
            }
        }
        if (fout == null)
            throw new RuntimeException("failed to write the hdfs file: "
                    + filename);
        else
            return fout;
    }


    public static List<String> getDataNodes(String hadoopHome) {
        List<String> lines = Lists.newArrayList();
        BufferedReader reader = null;
        try {
            reader = new BufferedReader(new FileReader(hadoopHome
                    + "/conf/slaves"));
        } catch (FileNotFoundException e) {
            try {
                reader = new BufferedReader(new FileReader(hadoopHome
                        + "/etc/hadoop/slaves"));
            } catch (FileNotFoundException e1) {
                throw new RuntimeException("Failed to read the slaves file: "
                        + e.getMessage());
            }
        } finally {
            try {
                String line;
                while ((line = reader.readLine()) != null)
                    lines.add(line);
                reader.close();
            } catch (IOException e) {
                throw new RuntimeException("Failed to read the slaves file: "
                        + e.getMessage());
            }
        }
        return lines;
    }
} public class TypeUtils {
	public enum TYPE {
		BOOLEAN, INT, LONG, DOUBLE, STRING, DATE, VARCHAR
	};

	public static class SimpleDate implements Comparable<SimpleDate> {
		private int year, month, day;
		public static int[] daysPerMonth = new int[] { 31, 28, 31, 30, 31, 30,
				31, 31, 30, 31, 30, 31 };

		public SimpleDate(int year, int month, int day) {
			this.year = year;
			this.month = month;
			this.day = day;
		}

		public int getYear() {
			return year;
		}

		public void setYear(int year) {
			this.year = year;
		}

		public int getMonth() {
			return month;
		}

		public void setMonth(int month) {
			this.month = month;
		}

		public int getDay() {
			return day;
		}

		public void setDay(int day) {
			this.day = day;
		}

		// Though expensive to create the calendar and find out, it works.
		// Infrequent operation, not worth optimizing.
		public SimpleDate oneDayLess() {
			// Month in the gregorian calendar class is 0 based.
			Calendar c = new GregorianCalendar(year, month - 1, day);
			c.add(Calendar.DAY_OF_MONTH, -1);
			return new SimpleDate(c.get(Calendar.YEAR), c.get(Calendar.MONTH) + 1,
					c.get(Calendar.DAY_OF_MONTH));
		}

		@Override
		public int compareTo(SimpleDate d) {
			if (d.getYear() < year
					|| (d.getYear() == year && d.getMonth() < month)
					|| (d.getYear() == year && d.getMonth() == month && d
							.getDay() < day))
				return 1;
			else if (d.getYear() == year && d.getMonth() == month
					&& d.getDay() == day)
				return 0;
			else
				return -1;
		}

		@Override
		public boolean equals(Object obj) {
			SimpleDate d = (SimpleDate) obj;
			return (d.getYear() == year && d.getMonth() == month && d.getDay() == day);
		}

		@Override
		public String toString() {
			String ret = "" + year + "-";
			ret += (month < 10 ? "0" + month : month) + "-";
			ret += (day < 10 ? "0" + day : day);
			return ret;
		}
	}

	public static int compareTo(Object x, Object y, TYPE t) {
		switch (t) {
		case INT:
			return ((Integer) x).compareTo((Integer) y);
		case DOUBLE:
			return ((Double) x).compareTo((Double) y);
		case LONG:
			return ((Long) x).compareTo((Long) y);
		case DATE:
			return ((SimpleDate) x).compareTo((SimpleDate) y);
		case STRING:
			return ((String) x).compareTo((String) y);
		default:
			System.err.println("Unknown TYPE in compareTo");
			return 0;
		}
	}

	public static Comparator<Object> getComparatorForType(TYPE type) {
		switch (type) {
		case INT:
			return new Comparator<Object>() {
				@Override
				public int compare(Object o1, Object o2) {
					return ((Integer) o1).compareTo((Integer) o2);
				}
			};
		case LONG:
			return new Comparator<Object>() {
				@Override
				public int compare(Object o1, Object o2) {
					return ((Long) o1).compareTo((Long) o2);
				}
			};
		case DOUBLE:
			return new Comparator<Object>() {
				@Override
				public int compare(Object o1, Object o2) {
					return ((Double) o1).compareTo((Double) o2);
				}
			};
		case DATE:
			return new Comparator<Object>() {
				@Override
				public int compare(Object o1, Object o2) {
					return ((SimpleDate) o1).compareTo((SimpleDate) o2);
				}
			};
		case STRING:
			return new Comparator<Object>() {
				@Override
				public int compare(Object o1, Object o2) {
					return ((String) o1).compareTo((String) o2);
				}
			};
		case VARCHAR:
			throw new RuntimeException("sorting over varchar is not supported"); // skip
																					// partitioning
																					// on
																					// varchar
																					// attribute
		default:
			throw new RuntimeException("Unknown dimension type: " + type);
		}
	}

	public static String serializeValue(Object value, TYPE t) {
		return value.toString();
	}

	public static Object deserializeValue(TYPE t, String token) {
		switch (t) {
		case INT:
			return Integer.parseInt(token);
		case LONG:
			return Long.parseLong(token);
		case DOUBLE:
			return Double.parseDouble(token);
		case DATE:
			byte[] bytes = token.getBytes();
			int off = 0;
			int year = 1000 * (bytes[off] - '0') + 100 * (bytes[off + 1] - '0')
					+ 10 * (bytes[off + 2] - '0') + (bytes[off + 3] - '0');
			int month = 10 * (bytes[off + 5] - '0') + (bytes[off + 6] - '0');
			int day = 10 * (bytes[off + 8] - '0') + (bytes[off + 9] - '0');

			return new SimpleDate(year, month, day);
		case BOOLEAN:
			return Boolean.parseBoolean(token);
		case STRING:
			return token;
		case VARCHAR:
			return token;
		default:
			return token;
		}
	}

	public static Object deltaLess(Object value, TYPE t) {
		switch (t) {
		case INT:
			return (Integer) value - 1;
		case LONG:
			return (Long) value - 1;
		case DOUBLE:
			return (Double) value - 0.001;
		case DATE:
			SimpleDate d = (SimpleDate) value;
			return d.oneDayLess();
		case BOOLEAN:
			return false;
		case STRING:
			System.err.println("ERR: Called deltaLess on string");
			return value;
		case VARCHAR:
			System.err.println("ERR: Called deltaLess on string");
			return value;
		default:
			System.err.println("ERR: Called deltaLess on unknown type");
			return value;
		}
	}
} public enum TYPE {
	BOOLEAN, INT, LONG, DOUBLE, STRING, DATE, VARCHAR
};
import java.io.IOException;
import java.util.*;
import core.adapt.JoinQuery;
import core.adapt.iterator.JoinRepartitionIterator;
import core.adapt.spark.join.JoinPlanner;
import core.adapt.spark.join.SparkJoinQueryConf;
import core.common.globals.Globals;
import core.common.globals.TableInfo;
import core.common.index.JRNode;
import core.common.index.JoinRobustTree;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import core.adapt.Predicate;
import core.adapt.AccessMethod.PartitionSplit;
import core.adapt.iterator.PartitionIterator;
import core.adapt.iterator.PostFilterIterator;
import core.adapt.iterator.RepartitionIterator;
import core.common.index.MDIndex.Bucket;
import core.common.key.ParsedTupleList;
import core.utils.ConfUtils;
import core.utils.HDFSUtils;
import core.utils.Pair;
import core.utils.TypeUtils;
import core.utils.TypeUtils.TYPE;
package core.adapt.opt;






/**
 * Created by ylu on 1/21/16.
 */


/**
 * Optimizer creates the execution plans for the queries.
 * It uses the incoming query predicates as hints for what should be added
 * into the partitioning tree. If we find a plan which has benefit >
 * cost, we do the repartitioning. Else we just do a scan.
 *
 * @author yilu
 */
public class JoinOptimizer {
    private static final double WRITE_MULTIPLIER = 1.5;

    private JoinRobustTree rt;

    // Properties extracted from ConfUtils
    private String workingDir;
    private String hadoopHome;
    private short fileReplicationFactor;

    private TableInfo tableInfo;

    private List<JoinQuery> queryWindow = new ArrayList<JoinQuery>();