public class RangePartitioner extends Partitioner {

    long[] ranges;

    public RangePartitioner(String ranges){
        this.ranges = RangePartitionerUtils.getLongCutPoints(ranges);
    }

    @Override
    public int numPartitions() {
        return ranges.length + 1;
    }

    @Override
    public int getPartition(Object key) {
        //hard code, key can only be long

        long longKey = ((LongWritable) key).get();
        for(int i = 0 ;i < ranges.length; i ++){
            if(longKey <= ranges[i]){
                return i;
            }
        }

        return ranges.length;
    }
} public class SparkJoinCopartitionedInputFormat extends
        FileInputFormat<LongWritable, Text> implements Serializable {


    public static class SparkJoinCopartitionedFileSplit extends CombineFileSplit implements
            Serializable {
        private static final long serialVersionUID = 1L;

        private PartitionIterator iter1, iter2;
        public SparkJoinCopartitionedFileSplit() {

        }

        public SparkJoinCopartitionedFileSplit(Path[] files, long[] lengths, PartitionIterator iter1, PartitionIterator iter2) {
            super(files, lengths);
            this.iter1 = iter1;
            this.iter2 = iter2;
        }

        public PartitionIterator getIterator1() {
            return iter1;
        }
        public PartitionIterator getIterator2() {
            return iter2;
        }

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

            Text.writeString(out, iter1.getClass().getName());
            iter1.write(out);
            Text.writeString(out, iter2.getClass().getName());
            iter2.write(out);
        }

        @Override
        public void readFields(DataInput in) throws IOException {
            super.readFields(in);

            String type = Text.readString(in);
            iter1 = (PartitionIterator) ReflectionUtils.getInstance(type);
            iter1.readFields(in);
            type = Text.readString(in);
            iter2 = (PartitionIterator) ReflectionUtils.getInstance(type);
            iter2.readFields(in);

        }

        public static SparkJoinCopartitionedFileSplit read(DataInput in) throws IOException {
            SparkJoinCopartitionedFileSplit s = new SparkJoinCopartitionedFileSplit();
            s.readFields(in);
            return s;
        }
    }


    @Override
    public List<InputSplit> getSplits(JobContext job) throws IOException {

        System.out.println("In SparkJoinCopartitionedInputFormat getSplits");

        List<InputSplit> finalSplits = new ArrayList<InputSplit>();

        Configuration conf = job.getConfiguration();

        SparkQueryConf queryConf = new SparkQueryConf(conf);


        JoinQuery jq1 =  new JoinQuery(conf.get("DATASET1_QUERY"));
        JoinQuery jq2 =  new JoinQuery(conf.get("DATASET2_QUERY"));;

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

        int partition_num = Integer.parseInt(conf.get("PARTITION_NUM"));

        String hdfsDefaultName = conf.get("fs.default.name");
        String workingDir = queryConf.getWorkingDir();

        PartitionIterator iter1 = new PostFilterIterator(jq1.castToQuery());
        PartitionIterator iter2 = new PostFilterIterator(jq2.castToQuery());

        FileSystem fs = HDFSUtils.getFSByHadoopHome(queryConf.getHadoopHome());

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

            String path1 = hdfsDefaultName + workingDir +  "/" + dataset1 + "/" + i;
            String path2 = hdfsDefaultName + workingDir +  "/" + dataset2 + "/" + i;

            Path[] paths = new Path[2];
            paths[0] = new Path(path1);
            paths[1] = new Path(path2);

            long[] lengths = new long[2];

            lengths[0] = fs.getFileStatus(paths[0]).getLen();
            lengths[1] = fs.getFileStatus(paths[1]).getLen();

            SparkJoinCopartitionedFileSplit split = new SparkJoinCopartitionedFileSplit(paths, lengths, iter1, iter2);
            finalSplits.add(split);
        }

        job.getConfiguration().setLong(NUM_INPUT_FILES, finalSplits.size());
        System.out.println("done with getting splits");
        return finalSplits;
    }

    @Override
    public RecordReader<LongWritable, Text> createRecordReader(InputSplit split, TaskAttemptContext context) throws IOException, InterruptedException {
        return new SparkJoinCopartitionedReader();
    }
} public class SparkJoinQuery {

    public static class Mapper implements PairFunction<Tuple2<LongWritable, Tuple2<Text, Text>>, LongWritable, Text> {
        String Delimiter;
        String Splitter;
        int partitionKey;
        public Mapper(String Delimiter, int partitionKey) {
            this.Delimiter = Delimiter;
            this.partitionKey = partitionKey;
            if (Delimiter.equals("|"))
                Splitter = "\\|";
            else
                Splitter = Delimiter;
        }

        @Override
        public Tuple2<LongWritable, Text> call(Tuple2<LongWritable, Tuple2<Text, Text>> x) throws Exception {
            String s1 = x._2()._1().toString();
            String s2 = x._2()._2().toString();
            String value = s1 + Delimiter + s2;
            long key = Long.parseLong(value.split(Splitter)[partitionKey]);
            return new Tuple2<LongWritable, Text>(new LongWritable(key), new Text(value));
        }
    }

    private SparkJoinQueryConf queryConf;
    private JavaSparkContext ctx;
    private ConfUtils cfg;

    private String Delimiter = "|";
    private String joinStrategy = "Heuristic";

    public SparkJoinQuery(ConfUtils config) {
        this.cfg = config;
        SparkConf sconf = SparkUtils.getSparkConf(this.getClass().getName(), config);

        try {
            sconf.registerKryoClasses(new Class<?>[]{
                    Class.forName("org.apache.hadoop.io.LongWritable"),
                    Class.forName("org.apache.hadoop.io.Text")
            });
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        }

        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());
        queryConf = new SparkJoinQueryConf(ctx.hadoopConfiguration());

        queryConf.setHDFSReplicationFactor(cfg.getHDFS_REPLICATION_FACTOR());
        queryConf.setHadoopHome(cfg.getHADOOP_HOME());
        queryConf.setZookeeperHosts(cfg.getZOOKEEPER_HOSTS());
        queryConf.setMaxSplitSize(4L * 1024 * 1024 * 1024); // 4 GB
        //queryConf.setMaxSplitSize(400L * 1024 * 1024); // 400 MB
        queryConf.setWorkerNum(8);
    }

    public JavaSparkContext getSparkContext(){
        return ctx;
    }

    public void setBufferSize(long size){
        queryConf.setMaxSplitSize(size);
    }

    public void setDelimiter(String delimiter) {
        Delimiter = delimiter;
    }

    public String getDelimiter() {
        return Delimiter;
    }

	/* SparkJoinQuery */

    public JavaPairRDD<LongWritable, Text> createJoinRDD(String hdfsPath) {
        queryConf.setReplicaId(0);

        return ctx.newAPIHadoopFile(cfg.getHADOOP_NAMENODE() + hdfsPath,
                SparkJoinInputFormat.class, LongWritable.class,
                Text.class, ctx.hadoopConfiguration());
    }


    public JavaPairRDD<LongWritable, Text> createJoinRDD(
            String dataset1, JoinQuery dataset1_query,
            String dataset2, JoinQuery dataset2_query,
            int partitionKey) {

        // set SparkQuery conf

        String hdfsPath = cfg.getHDFS_WORKING_DIR();

        queryConf.setWorkingDir(hdfsPath);
        queryConf.setJustAccess(false);

        Configuration conf = ctx.hadoopConfiguration();

        conf.set("DATASET1", dataset1);
        conf.set("DATASET2", dataset2);

        conf.set("DATASET1_QUERY", dataset1_query.toString());
        conf.set("DATASET2_QUERY", dataset2_query.toString());

        conf.set("PARTITION_KEY", Integer.toString(partitionKey)) ;

        conf.set("JOINALGO", joinStrategy);

        conf.set("DELIMITER", Delimiter);

        JoinPlanner planner = new JoinPlanner(conf);

        String hyperJoinInput = planner.hyperjoin;
        System.out.println("hyperJoinInput: " + hyperJoinInput);
        conf.set("DATASETINFO",hyperJoinInput);
        JavaPairRDD<LongWritable, Text> hyperjoin_rdd = createJoinRDD(hdfsPath);

        String input1 = planner.shufflejoin1;
        String input2 = planner.shufflejoin2;

        System.out.println("shuffleInput1: " + input1);
        System.out.println("shuffleInput2: " + input2);

        // set conf input;
        conf.set("DATASET_QUERY", dataset1_query.toString());
        conf.set("DATASETINFO", input1);
        conf.set("DATASET", dataset1);

        JavaPairRDD<LongWritable, Text> dataset1RDD = createSingleTableRDD(hdfsPath, dataset1_query);
        
        // set conf input;
        conf.set("DATASET_QUERY", dataset2_query.toString());
        conf.set("DATASETINFO", input2);
        conf.set("DATASET", dataset2);

        JavaPairRDD<LongWritable, Text> dataset2RDD = createSingleTableRDD(hdfsPath, dataset2_query);

        int numPartitions = 2000;

        JavaPairRDD<LongWritable, Tuple2<Text,Text> > join_rdd = dataset1RDD.join(dataset2RDD, numPartitions);

        JavaPairRDD<LongWritable, Text> shufflejoin_rdd = join_rdd.mapToPair(new Mapper(Delimiter, partitionKey));

        return hyperjoin_rdd.union(shufflejoin_rdd);
    }

    /* SingleTableScan  */

    public JavaPairRDD<LongWritable, Text> createSingleTableRDD(String hdfsPath,
                                                                JoinQuery q) {
        queryConf.setWorkingDir(hdfsPath);
        queryConf.setJoinQuery(q);

        return ctx.newAPIHadoopFile(cfg.getHADOOP_NAMENODE() + hdfsPath + "/" + q.getTable() + "/data",
                SparkScanInputFormat.class, LongWritable.class,
                Text.class, ctx.hadoopConfiguration());
    }

    /* for tpch 6 */

    public JavaPairRDD<LongWritable, Text> createScanRDD(String dataset, JoinQuery q) {

        // set SparkQuery conf

        String hdfsPath = cfg.getHDFS_WORKING_DIR();
        queryConf.setJustAccess(false);
        queryConf.setWorkingDir(hdfsPath);

        Configuration conf = ctx.hadoopConfiguration();


        conf.set("DATASET", dataset);
        conf.set("DATASET_QUERY", q.toString());
        conf.set("PARTITION_KEY", "0") ;
        conf.set("JOINALGO", joinStrategy);
        conf.set("DELIMITER", Delimiter);


        FileSystem fs = HDFSUtils.getFSByHadoopHome(queryConf.getHadoopHome());

        String workingDir = queryConf.getWorkingDir();
        //System.out.println("INFO working dir: " + workingDir);

        List<JoinQuery> dataset_queryWindow = JoinPlanner.loadQueries(dataset, queryConf);
        dataset_queryWindow.add(q);

        JoinPlanner.persistQueryToDisk(q, queryConf);

        Globals.loadTableInfo(dataset, queryConf.getWorkingDir(), fs);
        TableInfo dataset_tableInfo = Globals.getTableInfo(dataset);
        HPJoinInput dataset_hpinput = new HPJoinInput();
        Map<Integer, JoinAccessMethod> dataset_am = new HashMap<Integer, JoinAccessMethod>();
        Map<Integer, Integer> dataset_iterator_type = new HashMap<Integer, Integer>();
        Map<Integer, Integer> dataset_belong = new HashMap<Integer, Integer>();
        Map<Integer, ArrayList<Integer>> dataset_scan_blocks = new HashMap<Integer, ArrayList<Integer>>();
        Map<Integer, MDIndex.BucketInfo> bucketInfo = new HashMap<Integer, MDIndex.BucketInfo>();

        JoinPlanner.SmoothRepartition(dataset, q, dataset_queryWindow, dataset_tableInfo, dataset_hpinput, dataset_am, dataset_scan_blocks, dataset_iterator_type,dataset_belong, bucketInfo, queryConf, fs);
        ArrayList<PartitionSplit> shuffleJoinSplit = new  ArrayList<PartitionSplit>();
        JoinPlanner.extractShuffleJoin(q, shuffleJoinSplit, dataset_scan_blocks, dataset_iterator_type, dataset_hpinput.getPartitionIdSizeMap(), queryConf.getMaxSplitSize(), queryConf.getWorkerNum());

        String input = JoinPlanner.getShuffleJoinInputHelper(shuffleJoinSplit, dataset_hpinput);

        System.out.println("shuffleInput: " + input);

        conf.set("DATASETINFO", input);

        JavaPairRDD<LongWritable, Text> rdd = createSingleTableRDD(hdfsPath, q);
        return rdd;
    }

    public JavaPairRDD<LongWritable, Text> createCopartitionedRDD(String dataset1, JoinQuery dataset1_query,
                                                                  String dataset2, JoinQuery dataset2_query,
                                                                  int partitionKey){
        Configuration conf = ctx.hadoopConfiguration();

        conf.set("DATASET1", dataset1);
        conf.set("DATASET2", dataset2);

        conf.set("DATASET1_QUERY", dataset1_query.toString());
        conf.set("DATASET2_QUERY", dataset2_query.toString());

        conf.set("PARTITION_NUM", "200");

        conf.setInt("PARTITION_KEY",  partitionKey);

        conf.set("WORKING_DIR", cfg.getHDFS_WORKING_DIR());
        conf.set("HADOOP_HOME",cfg.getHADOOP_HOME());
        conf.set("ZOOKEEPER_HOSTS",cfg.getZOOKEEPER_HOSTS());
        conf.setInt("HDFS_REPLICATION_FACTOR",cfg.getHDFS_REPLICATION_FACTOR());
        conf.set("DELIMITER", "|");

        return ctx.newAPIHadoopFile(cfg.getHADOOP_NAMENODE() + cfg.getHDFS_WORKING_DIR(),
                SparkJoinCopartitionedInputFormat.class, LongWritable.class,
                Text.class, ctx.hadoopConfiguration());
    }
} 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 Schema {
	public Field[] fields;

	public static class Field {
		public String name;
		public TYPE type;
		int id;

		public Field(String name, TYPE type, int id) {
			this.name = name;
			this.type = type;
			this.id = id;

		}

		@Override
		public String toString() {
			return this.name + " " + this.type.toString();
		}
	}

	public Schema(Field[] fields) {
		this.fields = fields;
	}

	@Override
	public String toString() {
		String ret = "";
		for (int i = 0; i < this.fields.length; i++) {
			ret += this.fields[i].toString();
			if (i != this.fields.length - 1) {
				ret += ",";
			}
		}
		return ret;
	}

	public TYPE[] getTypeArray() {
		TYPE[] typeArray = new TYPE[fields.length];
		for (int i=0; i<fields.length; i++) {
			typeArray[i] = fields[i].type;
		}
		return typeArray;
	}

	public static Schema createSchema(String schemaString) {
		System.out.println(schemaString);
		String[] columns = schemaString.split(",");
		Field[] fieldList = new Field[columns.length];

		for (int i = 0; i < columns.length; i++) {
			String[] columnInfo = columns[i].trim().split(" ");
			String fieldName = columnInfo[0].trim();
			// enum valueOf is case sensitive, make everything uppercase.
			TYPE fieldType = TYPE.valueOf(columnInfo[1].trim().toUpperCase());
			fieldList[i] = new Field(fieldName, fieldType, i);
		}

		System.out.println("INFO: Created schema with " + fieldList.length + " fields");
		return new Schema(fieldList);
	}

	public int getAttributeId(String attrName) {
		for (Field f: fields) {
			if (f.name.equals(attrName)) {
				return f.id;
			}
		}

		return -1;
	}

	public TYPE getType(int attrId) {
		return fields[attrId].type;
	}
} 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 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 RangePartitionerUtils {

    public static ArrayList<Long> getKeys(ConfUtils cfg,  TableInfo tableInfo, String path, int dimension){
        FileSystem fs = HDFSUtils.getFSByHadoopHome(cfg.getHADOOP_HOME());

        // read all the sample files and put them into the sample key set
        ParsedTupleList sample = new ParsedTupleList(tableInfo.getTypeArray());
        byte[] bytes = HDFSUtils.readFile(fs, path);
        sample.unmarshall(bytes, tableInfo.delimiter);

        ArrayList<Long> keys = new ArrayList<Long>();

        for(Object[] o : sample.getValues()){
            if (o[dimension] instanceof Long) {
                keys.add((Long)o[dimension]);
            } else  if(o[dimension] instanceof Integer){
                int key = (Integer) o[dimension];
                keys.add((long)key);
            }
        }

        return keys;

    }

    public static String getCutPoints(ArrayList<Long> sampleKeys, int recordSize, long count){

        long chunkSize = 32 * 1024 * 1024; // 32 MB
        long totalSize = count * recordSize;
        int n = (int) ( (totalSize + chunkSize - 1)  / chunkSize); // ceiling

        long[] keys = new long[sampleKeys.size()];
        int i = 0;
        for(long lw : sampleKeys){
            keys[i++] = lw;
        }
        Arrays.sort(keys);

        int m = keys.length;
        long[] cutpoints = new long[n - 1];

        int small = n - m % n;
        int large = m % n;

        int pi = 0, pj = 0;
        for(int j = 0; j < small; j ++){
            pj += m / n;
            if(pi < cutpoints.length){
                cutpoints[pi ++ ] = keys[pj];
            }

        }
        for(int j = 0; j < large; j ++){
            pj += m / n + 1;
            if(pi < cutpoints.length){
                cutpoints[pi ++ ] = keys[pj];
            }
        }

        String strCutPoints = getStringCutPoints(cutpoints);
        return strCutPoints;
    }

    public static long[] getLongCutPoints(String cutpoints){
        if(cutpoints.equals("NULL")){
            return null;
        }
        if(cutpoints.length() == 0){
            return new long[0];
        }
        String[] values = cutpoints.split(",");
        long[] ranges = new long[values.length];
        for(int i = 0 ; i < ranges.length; i ++){
            ranges[i] = Long.parseLong(values[i]);
        }
        return ranges;
    }
    public static String getStringCutPoints(long[] cutpoints){
        StringBuilder sb  = new StringBuilder();
        for(int i = 0 ;i < cutpoints.length; i ++){
            if(sb.length() > 0){
                sb.append(",");
            }
            sb.append(cutpoints[i]);
        }
        return sb.toString();
    }
    public static int[] getSplits(long[] cutpoints){
        int[] splits = new int[cutpoints.length + 1];
        for(int i = 0 ;i < splits.length; i ++){
            splits[i] = i;
        }
        return splits;
    }
} 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 enum TYPE {
	BOOLEAN, INT, LONG, DOUBLE, STRING, DATE, VARCHAR
}; 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 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 enum PREDTYPE {
	LEQ, GEQ, GT, LT, EQ
}; 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 enum TYPE {
	BOOLEAN, INT, LONG, DOUBLE, STRING, DATE, VARCHAR
};
import core.adapt.spark.RangePartitioner;
import core.adapt.spark.join.SparkJoinCopartitionedInputFormat;
import core.adapt.spark.join.SparkJoinQuery;
import core.adapt.spark.join.SparkJoinQueryConf;
import core.common.globals.Schema;
import core.common.globals.TableInfo;
import core.utils.ConfUtils;
import core.utils.HDFSUtils;
import core.utils.RangePartitionerUtils;
import core.utils.SparkUtils;
import core.utils.TypeUtils.SimpleDate;
import core.utils.TypeUtils.TYPE;
import org.apache.commons.io.FilenameUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.spark.HashPartitioner;
import org.apache.spark.Partitioner;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.util.*;
import core.adapt.JoinQuery;
import core.adapt.Predicate;
import core.adapt.Predicate.PREDTYPE;
import core.utils.TypeUtils.SimpleDate;
import core.utils.TypeUtils.TYPE;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.PairFunction;
import scala.Tuple2;
package perf.benchmark;

/**
 * Created by ylu on 4/25/16.
 */





public class TPCHCopartitionedWorkload {


    public static class Mapper implements PairFunction<String, Long, String> {
        String Delimiter;
        String Splitter;
        int partitionKey;
        public Mapper(String Delimiter, int partitionKey) {
            this.Delimiter = Delimiter;
            this.partitionKey = partitionKey;
            if (Delimiter.equals("|"))
                Splitter = "\\|";
            else
                Splitter = Delimiter;
        }

        @Override
        public Tuple2<Long, String> call(String v1) throws Exception {
            long key = Long.parseLong(v1.split(Splitter)[partitionKey]);
            return new Tuple2<Long, String>(key, v1);
        }
    }

    private ConfUtils cfg;

    private SparkJoinQuery sq;