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 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 RobustTree implements MDIndex {
	public TableInfo tableInfo;

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

	public TYPE[] dimensionTypes;
	RNode root;

	public static class Bound {
		public double upper;
		public double lower;

		public Bound(double lower, double upper) {
			this.upper = upper;
			this.lower = lower;
		}
	}

	public RobustTree(TableInfo tableInfo) {
		this.root = new RNode();
		this.tableInfo = tableInfo;
	}

	public RobustTree(){
	}

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

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

	public RNode getRoot() {
		return root;
	}

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

	@Override
	public boolean equals(Object obj) {
		if (obj instanceof RobustTree) {
			RobustTree rhs = (RobustTree) 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 class Task {
		RNode node;
		int depth;
		ParsedTupleList sample;
	}

	/**
	 * Created the tree based on the histograms
	 */
	@Override
	public void initProbe() {
		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 = 0;
		nodeQueue.add(initialTask);

		while (nodeQueue.size() > 0) {
			Task t = nodeQueue.pollFirst();
			if (t.depth < maxDepth) {
				int dim = -1;
				int round = 0;
				Pair<ParsedTupleList, ParsedTupleList> halves = null;

				while (dim == -1 && round < allocations.length) {
					int testDim = getLeastAllocated(allocations);
					allocations[testDim] -= 2.0 / Math.pow(2, t.depth);

					// TODO: For low cardinality values, it might be better to
					// choose some set of values on each side.
					// TPCH attribute 9 for example has only two distinct values
					// TODO: This might repeatedly use the same attribute
					halves = t.sample.sortAndSplit(testDim);
					if (halves.first.size() > 0 && halves.second.size() > 0) {
						dim = testDim;
					} else {
						System.err.println("WARN: Skipping attribute "
								+ testDim);
					}

					round++;
				}

				if (dim == -1) {
					System.err.println("ERR: No attribute to partition on");
					Bucket b = new Bucket();
					b.setSample(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 RNode();
					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 RNode();
					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);
				}
			} else {
				Bucket b = new Bucket();
				b.setSample(sample);
				t.node.bucket = b;
			}
		}

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


	@Override
	public void initProbe(int joinAttribute) {
		System.out.println("method not implemented!");
	}


	// TODO: Add capacity
	static List<Integer> leastAllocated = new ArrayList<>(20);
	public static Random randGenerator = new Random();

	/**
	 * Return the dimension which has the maximum allocation unfulfilled
	 */
	public static int getLeastAllocated(double[] allocations) {
		int numAttributes = allocations.length;

		leastAllocated.clear();
		leastAllocated.add(0);

		double alloc = allocations[0];
		for (int i = 1; i < numAttributes; i++) {
			if (allocations[i] > alloc) {
				alloc = allocations[i];
				leastAllocated.clear();
				leastAllocated.add(i);
			} else if (allocations[i] == alloc) {
				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(RNode 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<RNode> getMatchingBuckets(Predicate[] predicates) {
		List<RNode> 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\n", this.maxBuckets,
				this.numAttributes);

		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.dimensionTypes = new TYPE[this.numAttributes];
		for (int i = 0; i < this.numAttributes; i++) {
			this.dimensionTypes[i] = TYPE.valueOf(sc.next());
		}

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

	public void loadSample(TableInfo tableInfo, byte[] bytes) {
		this.sample = new ParsedTupleList(tableInfo.getTypeArray());
		this.sample.unmarshall(bytes, tableInfo.delimiter);
		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(RNode 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);
		}
	}

	public static double nthroot(int n, double A) {
		return nthroot(n, A, .001);
	}

	public static double nthroot(int n, double A, double p) {
		if (A < 0) {
			System.err.println("A < 0");// we handle only real positive numbers
			return -1;
		} else if (A == 0) {
			return 0;
		}

		double x_prev = A;
		double x = A / n; // starting "guessed" value...
		while (Math.abs(x - x_prev) > p) {
			x_prev = x;
			x = ((n - 1.0) * x + A / Math.pow(x, n - 1.0)) / n;
		}

		return x;
	}

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

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

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

	private static Map<Integer, BucketInfo> getBucketRangeSubtree(RNode 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;
		}
	}

	public double[] getAllocations() {
		List<RNode> queue = new ArrayList<RNode>();
		Map<Integer, Double> allocs = new HashMap<Integer, Double>();
		queue.add(this.getRoot());

		int nodeNum = 0;
		RNode filler = new RNode();
		filler.attribute = -1;
		int lastNode = 0;
		while (queue.size() > 0) {
			nodeNum++;
			RNode node = queue.remove(0);
			if (node.bucket != null || node.attribute == -1) {
				if (nodeNum > lastNode * 2) {
					break;
				}
				queue.add(filler);
				queue.add(filler);
				continue;
			}

			lastNode = nodeNum;
			if (!(allocs.containsKey(node.attribute))) {
				allocs.put(node.attribute, 0.0);
			}

			double addedAlloc = Math.pow(2, -1 * Math.floor(Math.log(nodeNum) / Math.log(2)) + 1);
			allocs.put(node.attribute, allocs.get(node.attribute) + addedAlloc);
			queue.add(node.leftChild);
			queue.add(node.rightChild);
		}

		double[] allocArray = new double[numAttributes];
		for (int i = 0; i < numAttributes; i++) {
			if (!(allocs.containsKey(i))) {
				allocArray[i] = 0;
			} else {
				allocArray[i] = allocs.get(i);
			}
		}
		System.out.println(nodeNum);
		return allocArray;
	}
} 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 RawIndexKey implements Cloneable {

    private SimpleDate dummyDate = new SimpleDate(0, 0, 0);

    protected byte[] bytes;
    protected int offset, length;

    protected int numAttrs;
    protected int[] attributeOffsets;

    protected char delimiter;

    public RawIndexKey(char delimiter) {
        this.delimiter = delimiter;
    }

    public RawIndexKey(String keyString) {
        String[] tokens = keyString.trim().split(",");
        this.delimiter = tokens[0].charAt(0);
    }

    @Override
    public RawIndexKey clone() throws CloneNotSupportedException {
        RawIndexKey k = (RawIndexKey) super.clone();
        k.dummyDate = new SimpleDate(0, 0, 0);
        return k;
    }

    private void setNumAttrs(byte[] bytes, int offset, int length) {
        numAttrs = 1;
        for (int i = offset; i < offset + length; i++) {
            if (bytes[i] == delimiter) {
                numAttrs += 1;
            }
        }
    }


    public void setBytes(byte[] bytes) {
        setBytes(bytes, 0, bytes.length);
    }

    public void setBytes(byte[] bytes, int offset, int length) {
        this.bytes = bytes;
        this.offset = offset;
        this.length = length;

        if (attributeOffsets == null) {
            setNumAttrs(bytes, offset, length);
            attributeOffsets = new int[numAttrs];
        }

        int previous = offset;
        int attrIdx = 0;
        for (int i = offset; i < offset + length; i++) {
            if (bytes[i] == delimiter) {
                try {
                    attributeOffsets[attrIdx++] = previous;
                    previous = i + 1;
                } catch (Exception e) {
                    System.out.println(delimiter + " " + new String(bytes));
                    e.printStackTrace();
                }
            }
        }
        if (attrIdx < attributeOffsets.length)
            attributeOffsets[attrIdx] = previous;
    }

    public String getKeyString() {
        return new String(bytes, offset, length);
    }

    public String getStringAttribute(int index) {
        int off = attributeOffsets[index];
        int strSize;
        if (index < attributeOffsets.length - 1)
            strSize = attributeOffsets[index + 1] - off - 1;
        else
            strSize = offset + length - off;

        return new String(bytes, off, strSize);
    }

    public int getIntAttribute(int index) {
        int off = attributeOffsets[index];
        int len;
        if (index < attributeOffsets.length - 1)
            len = attributeOffsets[index + 1] - off - 1;
        else
            len = offset + length - off;

        int ret = Integer.parseInt(new String(bytes, off, len));

        return ret;
    }

    public long getLongAttribute(int index) {
        int off = attributeOffsets[index];
        int len;
        if (index < attributeOffsets.length - 1)
            len = attributeOffsets[index + 1] - off - 1;
        else
            len = offset + length - off;

        long ret = Long.parseLong(new String(bytes, off, len));

        return ret;
    }

    public double getDoubleAttribute(int index) {
        int off = attributeOffsets[index];
        int len;
        if (index < attributeOffsets.length - 1)
            len = attributeOffsets[index + 1] - off - 1;
        else
            len = offset + length - off;

        double ret = Double.parseDouble(new String(bytes, off, len));
        return ret;
    }

    /*
	 * Parse date assuming the format: "yyyy-MM-dd".
	 * Skips anything after that.
	 */
    public SimpleDate getDateAttribute(int index) {
        int off = attributeOffsets[index];
        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');

        dummyDate.setYear(year);
        dummyDate.setMonth(month);
        dummyDate.setDay(day);

        return dummyDate;
    }

    public SimpleDate getDateAttribute(int index, SimpleDate date) {
        // parse date assuming the format: "yyyy-MM-dd"
        int off = attributeOffsets[index];
        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');

        date.setYear(year);
        date.setMonth(month);
        date.setDay(day);

        return date;
    }

    /**
     * Assumes that the boolean data is represented as a single character in the
     * ascii file.
     *
     * @param index
     * @return
     */
    public boolean getBooleanAttribute(int index) {
        int off = attributeOffsets[index];

        if (bytes[off] == '1' || bytes[off] == 't')
            return true;
        else if (bytes[off] == '0' || bytes[off] == 'f')
            return false;
        else
            throw new RuntimeException("Cannot parse the boolean attribute: "
                    + bytes[off]);
    }

    @Override
    public String toString() {
        String result = String.valueOf(delimiter);
        return result;
    }
} public class IndexBuilder {

	int bucketSize = 64 << 20; // 64 MB.

	/*
	public void sparkSampleInput(double sampleRate, String inputDirectory, String outputSamplePath,
								 FileSystem fs, ConfUtils cfg){

		SparkConf sconf = SparkUtils.getSparkConf(this.getClass().getName(), cfg);
		JavaSparkContext ctx = new JavaSparkContext(sconf);
		ctx.hadoopConfiguration().setBoolean(
				FileInputFormat.INPUT_DIR_RECURSIVE, true);
		ctx.hadoopConfiguration().set("fs.hdfs.impl",
				org.apache.hadoop.hdfs.DistributedFileSystem.class.getName());

		long startTime = System.nanoTime();

		JavaRDD<String> input = ctx.textFile(inputDirectory);
		JavaRDD<String> sample = input.sample(false, sampleRate);

		String sparkTempOutput = outputSamplePath + ".temp";
		sample.saveAsTextFile(sparkTempOutput);

		//Merge files from sparkTempOutput into a single sample file.
		try {
			FileUtil.copyMerge(fs, new Path(sparkTempOutput),
					fs, new Path(outputSamplePath), true,
					ctx.hadoopConfiguration(), null);
		} catch (IOException e) {
			e.printStackTrace();
		}
		double timeScanAndWrite = (System.nanoTime() - startTime) / 1e9;
		System.out.println("Scanning, sampling and write time: " + timeScanAndWrite + " sec");
	}
	*/

	// Reads the files from inputDirectory and samples them using blockSampling.
	// Writes out the samples to SAMPLES_DIR/sample.machineId.
	public void blockSampleInput(double samplingRate, RawIndexKey key, String inputDirectory,
			String outputSamplePath, FileSystem fs) {
		InputReader r = new InputReader(null, key);
		File[] files = new File(inputDirectory).listFiles();
		OutputStream out = HDFSUtils.getHDFSOutputStream(fs, outputSamplePath, (short) 1, 50 << 20);

		long startTime = System.nanoTime();
		try {
			for (File f : files) {
				System.out.println("Scanning " + f.getName());
				r.scanWithBlockSampling(f.getPath(), samplingRate, out);
			}
			out.close();
		} catch (IOException e) {
			e.printStackTrace();
		}
		double timeScanAndWrite = (System.nanoTime() - startTime) / 1e9;
		System.out.println("Scanning, sampling and write time: " + timeScanAndWrite + " sec");
	}

	// Build index from the samples collected.
	public void buildIndexFromSample(ParsedTupleList sample, int numBuckets, MDIndex index, PartitionWriter writer) {
		index.setMaxBuckets(numBuckets);
		((RobustTree) index).loadSample(sample);

		long startTime = System.nanoTime();
		index.initProbe();
		System.out.println("BUILD: index building time = " + ((System.nanoTime() - startTime) / 1E9));

		startTime = System.nanoTime();
		byte[] indexBytes = index.marshall();
		System.out.println("Index Size: " + indexBytes.length);
		writer.writeToPartition("index", indexBytes, 0, indexBytes.length);
		writer.flush();
		System.out.println("BUILD: index writing time = " + ((System.nanoTime() - startTime) / 1E9));
	}

	// Build join index from the samples collected.
	public void buildJoinIndexFromSample(ParsedTupleList sample, int numBuckets, int joinAttribute, MDIndex index, PartitionWriter writer) {
		index.setMaxBuckets(numBuckets);
		((JoinRobustTree) index).loadSample(sample);

		long startTime = System.nanoTime();
		index.initProbe(joinAttribute);
		System.out.println("BUILD: index building time = " + ((System.nanoTime() - startTime) / 1E9));

		startTime = System.nanoTime();
		byte[] indexBytes = index.marshall();
		System.out.println("Index Size: " + indexBytes.length);
		writer.writeToPartition("index", indexBytes, 0, indexBytes.length);
		writer.flush();
		System.out.println("BUILD: index writing time = " + ((System.nanoTime() - startTime) / 1E9));
	}

	// Given an index that has already been constructed, and a directory of
	// files to partition, writes out the appropriate partition files.
	public void buildDistributedFromIndex(MDIndex index, RawIndexKey key, String inputDirectory,
			PartitionWriter writer) {
		InputReader r = new InputReader(index, key);
		r.firstPass = false;
		File[] files = new File(inputDirectory).listFiles();

		long startTime = System.nanoTime();
		for (File f : files) {
			r.scan(f.getPath(), writer);
		}
		writer.flush();
		double time3 = (System.nanoTime() - startTime) / 1E9;
		System.out.println("Index Probe Time = " + time3 + " sec");
	}
} public abstract class PartitionWriter implements Cloneable {
	protected int bufferPartitionSize = 2 * 1024 * 1024;  // 2 MB
	protected String partitionDir;

	protected Map<String, OutputStream> buffer;
	protected Map<String, MutableInt> partitionRecordCount;
	private long writingTime = 0;

	public PartitionWriter(String partitionDir, int bufferPartitionSize) {
		this(partitionDir);
		this.bufferPartitionSize = bufferPartitionSize;
	}

	public PartitionWriter(String partitionDir) {
		this.partitionDir = partitionDir;
		this.buffer = Maps.newHashMap();
		this.partitionRecordCount = Maps.newHashMap();
	}

	@Override
	public PartitionWriter clone() throws CloneNotSupportedException {
		PartitionWriter w = (PartitionWriter) super.clone();
		w.buffer = Maps.newHashMap();
		w.partitionRecordCount = Maps.newHashMap();
		return w;
	}

	public void setPartitionDir(String partitionDir) {
		this.partitionDir = partitionDir;
	}

	public String getPartitionDir() {
		return this.partitionDir;
	}

	public void writeToPartition(String partitionId, byte[] bytes,
			int b_offset, int b_length) {
		long start = System.nanoTime();
		OutputStream b = buffer.get(partitionId);
		if (b == null) {
			b = getOutputStream(partitionDir + "/" + partitionId);
			buffer.put(partitionId, b);
		}

		try {
			b.write(bytes, b_offset, b_length);
			writingTime += System.nanoTime() - start;
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	protected abstract OutputStream getOutputStream(String path);

	public abstract void createPartitionDir();

	public void flush() {
		flush(buffer.size());
	}

	protected void flush(int numPartitions) {
		System.out.println("SCAN: partition writing time = " + writingTime
				/ 1E9);
		long start = System.nanoTime();
		int flushCount = 0;
		Set<String> keys = new HashSet<String>(buffer.keySet());
		for (String k : keys) {
			try {
				buffer.get(k).close();
				buffer.remove(k);
			} catch (IOException e) {
				e.printStackTrace();
			}
			flushCount++;
			if (flushCount > numPartitions)
				break;
		}
		System.out.println("SCAN: flushing time = "
				+ (System.nanoTime() - start) / 1E9);
	}
} 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 CuratorUtils {

	/**
	 * Settings
	 */
	private static int baseSleepTimeMills = 10000;
	private static int maxRetries = 4;
	private static int waitTimeSeconds = 10000;

	private static int sessionTimeout = 200000;
	private static int connectionTimeout = 20000;

	/*
	 * Curator client utils
	 */

	// public static class MyStateListener implements ConnectionStateListener{
	// public void stateChanged(CuratorFramework arg0, ConnectionState arg1) {
	// }
	// }

	public static CuratorFramework createClient(String zkHosts) {
		RetryPolicy retryPolicy = new ExponentialBackoffRetry(
				baseSleepTimeMills, maxRetries);
		CuratorFramework client = CuratorFrameworkFactory.newClient(zkHosts,
				sessionTimeout, connectionTimeout, retryPolicy);
		// try {
		// client.getZookeeperClient().blockUntilConnectedOrTimedOut();
		// } catch (InterruptedException e) {
		// e.printStackTrace();
		// }
		return client;
	}

	public static CuratorFramework createAndStartClient(String zkHosts) {
		CuratorFramework client = createClient(zkHosts);
		client.start();
		return client;
	}

	public static void stopClient(CuratorFramework client) {
		client.close();
	}

	/*
	 * Lock utils
	 */

	public static InterProcessSemaphoreMutex acquireLock(
			CuratorFramework client, String lockPath) {

		// make sure that the zookeeper client is connected
		int connTries = 0, maxConnTries = 100;
		while (!(client.getZookeeperClient().isConnected())
				&& connTries < maxConnTries) {
			ThreadUtils.sleep(500);
			connTries++;
		}
		if (!(client.getZookeeperClient().isConnected()))
			throw new RuntimeException("The zookeeper cient is not connected!");

		InterProcessSemaphoreMutex lock = new InterProcessSemaphoreMutex(
				client, lockPath);

		try {
			if (lock.acquire(waitTimeSeconds, TimeUnit.SECONDS))
				return lock;
			else {
				System.out.println("Time out: Failed to obtain lock: "
						+ lockPath);
				// return null;
				throw new RuntimeException("Failed to obtain lock: " + lockPath);
			}
		} catch (Exception e) {
			e.printStackTrace();
			// return null;
			throw new RuntimeException("Failed to obtain lock: " + lockPath
					+ "\n " + e.getMessage());
		}
	}

	public static void releaseLock(InterProcessSemaphoreMutex lock) {
		try {
			lock.release();
		} catch (Exception e) {
			e.printStackTrace();
			throw new RuntimeException("Failed to unlock " + lock + "\n "
					+ e.getMessage());
		}
	}

	/*
	 * Counter utils
	 */
	public static SharedCount createCounter(String hosts, String counterPath) {
		CuratorFramework client = createClient(hosts);
		return new SharedCount(client, counterPath, 0);
	}

	public static SharedCount createAndStartCounter(String zkHosts,
			String counterPath) {
		SharedCount c = createCounter(zkHosts, counterPath);
		try {
			c.start();
			return c;
		} catch (Exception e) {
			e.printStackTrace();
			throw new RuntimeException("Failed to start the counter: "
					+ counterPath + "\n" + e.getMessage());
		}
	}

	public static SharedCount createAndStartCounter(CuratorFramework client,
			String counterPath) {
		SharedCount c = new SharedCount(client, counterPath, 0);
		try {
			c.start();
			return c;
		} catch (Exception e) {
			e.printStackTrace();
			throw new RuntimeException("Failed to start the counter: "
					+ counterPath + "\n" + e.getMessage());
		}
	}

	public static int getCounter(SharedCount c) {
		return c.getCount();
	}

	public static int getCounter(CuratorFramework client, String counterPath) {
		SharedCount c = new SharedCount(client, counterPath, 0);
		try {
			c.start();
			int val = c.getCount();
			c.close();
			return val;

		} catch (Exception e) {
			e.printStackTrace();
			throw new RuntimeException("Failed to add the counter: "
					+ counterPath + "\n" + e.getMessage());
		}
	}

	public static void addCounter(SharedCount c, int increment) {
		int oldVal = c.getCount();
		try {
			c.setCount(oldVal + increment);
		} catch (Exception e) {
			e.printStackTrace();
			throw new RuntimeException("Failed to increment the counter: " + c);
		}
	}

	public static void addCounter(CuratorFramework client, String counterPath,
			int increment) {
		SharedCount c = new SharedCount(client, counterPath, 0);
		try {
			c.start();
			int oldVal = c.getCount();
			c.setCount(oldVal + increment);
			c.close();
		} catch (Exception e) {
			e.printStackTrace();
			throw new RuntimeException("Failed to add the counter: "
					+ counterPath + "\n" + e.getMessage());
		}
	}

	public static void setCounter(CuratorFramework client, String counterPath,
			int value) {
		SharedCount c = new SharedCount(client, counterPath, 0);
		try {
			c.start();
			c.setCount(value);
			c.close();
		} catch (Exception e) {
			e.printStackTrace();
			throw new RuntimeException("Failed to add the counter: "
					+ counterPath + "\n" + e.getMessage());
		}
	}

	public static void deleteAll(CuratorFramework client, String path,
			String prefix) {
		try {
			// client.start();
			List<String> children = client.getChildren().forPath(path);
			// System.out.println(children);
			for (String child : children) {
				if (child.startsWith(prefix)) {
					// System.out.println("deleting: "+path + child);
					client.delete().deletingChildrenIfNeeded()
							.forPath(path + child);
				}
			}
			// client.close();

		} catch (Exception e) {
			e.printStackTrace();
			throw new RuntimeException("Failed to delete path: " + path);
		}
	}

	public static void printAll(CuratorFramework client, String path,
			String prefix) {
		client.start();
		List<String> children;
		try {
			children = client.getChildren().forPath(path);
			for (String child : children) {
				if (child.startsWith(prefix)) {
					SharedCount c = new SharedCount(client, path + child, 0);
					c.start();
					System.out.println(path + child + "\t" + c.getCount());
					c.close();
				}
			}
			client.close();
		} catch (Exception e) {
			e.printStackTrace();

		}
	}

	public static void main(String[] args) {
		String zookeeperHost = "localhost";
		String deletePath = "/";
		String prefix = "partition-count";

		CuratorUtils.printAll(CuratorUtils.createClient(zookeeperHost),
				deletePath, prefix);
	}
} 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;
    }
}
import core.common.globals.Globals;
import core.common.globals.TableInfo;
import core.common.index.JoinRobustTree;
import core.common.index.RobustTree;
import core.common.key.ParsedTupleList;
import core.common.key.RawIndexKey;
import core.upfront.build.HDFSPartitionWriter;
import core.upfront.build.IndexBuilder;
import core.upfront.build.PartitionWriter;
import core.utils.ConfUtils;
import core.utils.CuratorUtils;
import core.utils.HDFSUtils;
import org.apache.curator.framework.CuratorFramework;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.LocatedFileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.RemoteIterator;
import java.io.File;
import java.io.IOException;
import java.io.OutputStream;
package perf.benchmark;



/**
 * Builds the index. Captures time taken by the different steps in index
 * building.
 * Make sure that the table info has been created.
 * TODO: Compute sampling fraction instead of taking as an option.
 *
 * @author anil
 *
 */
public class RunIndexBuilder {
	RawIndexKey key;
	IndexBuilder builder;

	int partitionBufferSize;

	ConfUtils cfg;

	// Table name.
	String tableName;
	
	// Directory on local file system containing the inputs.
	String inputsDir;

	// Specifies which method should be run.
	// See 'main' for method numbers.
	int method = -1;

	// Sampling probability.
	double samplingRate = 0.0;

	// Number of buckets in the index.
	int numBuckets = -1;

	// depth of join attribute
	int joinAttributeDepth = 0;

	// join attribute
	int joinAttribute = 0;

	// Directory corresponding to table on HDFS.
	String tableHDFSDir;

	// HDFS Filesystem.
	FileSystem fs;

    // Table Info.
    TableInfo tableInfo;

    // Zookeeper client.
    CuratorFramework client;

	public void setUp() {

		partitionBufferSize = 2 * 1024 * 1024;

		cfg = new ConfUtils(BenchmarkSettings.conf);
		fs = HDFSUtils.getFSByHadoopHome(cfg.getHADOOP_HOME());

		// Load table info.
		Globals.loadTableInfo(tableName, cfg.getHDFS_WORKING_DIR(), fs);
		tableInfo = Globals.getTableInfo(tableName);
		assert tableInfo != null;

		builder = new IndexBuilder();
		key = new RawIndexKey(tableInfo.delimiter);
		
		tableHDFSDir = cfg.getHDFS_WORKING_DIR() + "/" + tableName;

        client = CuratorUtils.createAndStartClient(