<commit_before>package core.access.spark;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;

import org.apache.commons.io.FilenameUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.fs.BlockLocation;
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.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.CombineFileSplit;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

import com.google.common.collect.Lists;
import com.google.common.collect.Maps;

import core.access.AccessMethod;
import core.access.AccessMethod.PartitionSplit;
import core.access.Query.FilterQuery;
import core.access.iterator.DistributedRepartitionIterator;
import core.access.iterator.PartitionIterator;
import core.access.iterator.PartitionIterator.IteratorRecord;
import core.access.iterator.RepartitionIterator;
import core.utils.ReflectionUtils;

public class SparkInputFormat extends FileInputFormat<LongWritable, IteratorRecord>{

	private static final Log LOG = LogFactory.getLog(FileInputFormat.class);

	private SparkQueryConf queryConf;


	public static class SparkFileSplit extends CombineFileSplit{
		private PartitionIterator iterator;
		public SparkFileSplit(){
		}
		public SparkFileSplit(Path[] files, long[] start, long[] lengths, String[] locations, PartitionIterator iterator) {
			super(files, start, lengths, locations);
			this.iterator = iterator;
		}
		public PartitionIterator getIterator(){
			return this.iterator;
		}
		@Override
		public void write(DataOutput out) throws IOException{
			super.write(out);
			out.writeBytes(iterator.getClass().getName()+"\n");
			iterator.write(out);
		}
		@Override
		public void readFields(DataInput in) throws IOException{
			super.readFields(in);
			iterator = (PartitionIterator)ReflectionUtils.getInstance(in.readLine());
			iterator.readFields(in);
		}
	}


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

		List<InputSplit> finalSplits = new ArrayList<InputSplit>();
		List<FileStatus> files = listStatus(job);

		queryConf = new SparkQueryConf(job.getConfiguration());
		AccessMethod am = new AccessMethod();
		am.init(queryConf);
		// am.init(queryConf.getDataset(), queryConf.getHadoopHome());

		Map<Integer,FileStatus> partitionIdFileMap = Maps.newHashMap();
		for(FileStatus file: files)
			try{
				partitionIdFileMap.put(Integer.parseInt(FilenameUtils.getName(file.getPath().toString())), file);
			} catch (NumberFormatException e){
			}

		PartitionSplit[] splits = am.getPartitionSplits(new FilterQuery(queryConf.getPredicates()), queryConf.getWorkers(), false);<commit_msg>cyst for checking the number of partition splits from the optimizer<commit_after>package core.access.spark;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;

import org.apache.commons.io.FilenameUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.fs.BlockLocation;
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.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.CombineFileSplit;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

import com.google.common.collect.Lists;
import com.google.common.collect.Maps;

import core.access.AccessMethod;
import core.access.AccessMethod.PartitionSplit;
import core.access.Query.FilterQuery;
import core.access.iterator.DistributedRepartitionIterator;
import core.access.iterator.PartitionIterator;
import core.access.iterator.PartitionIterator.IteratorRecord;
import core.access.iterator.RepartitionIterator;
import core.utils.ReflectionUtils;

public class SparkInputFormat extends FileInputFormat<LongWritable, IteratorRecord>{

	private static final Log LOG = LogFactory.getLog(FileInputFormat.class);

	private SparkQueryConf queryConf;


	public static class SparkFileSplit extends CombineFileSplit{
		private PartitionIterator iterator;
		public SparkFileSplit(){
		}
		public SparkFileSplit(Path[] files, long[] start, long[] lengths, String[] locations, PartitionIterator iterator) {
			super(files, start, lengths, locations);
			this.iterator = iterator;
		}
		public PartitionIterator getIterator(){
			return this.iterator;
		}
		@Override
		public void write(DataOutput out) throws IOException{
			super.write(out);
			out.writeBytes(iterator.getClass().getName()+"\n");
			iterator.write(out);
		}
		@Override
		public void readFields(DataInput in) throws IOException{
			super.readFields(in);
			iterator = (PartitionIterator)ReflectionUtils.getInstance(in.readLine());
			iterator.readFields(in);
		}
	}


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

		List<InputSplit> finalSplits = new ArrayList<InputSplit>();
		List<FileStatus> files = listStatus(job);

		queryConf = new SparkQueryConf(job.getConfiguration());
		AccessMethod am = new AccessMethod();
		am.init(queryConf);
		// am.init(queryConf.getDataset(), queryConf.getHadoopHome());

		Map<Integer,FileStatus> partitionIdFileMap = Maps.newHashMap();
		for(FileStatus file: files)
			try{
				partitionIdFileMap.put(Integer.parseInt(FilenameUtils.getName(file.getPath().toString())), file);
			} catch (NumberFormatException e){
			}

		PartitionSplit[] splits = am.getPartitionSplits(new FilterQuery(queryConf.getPredicates()), queryConf.getWorkers(), false);
		System.out.println("Number of partition splits = "+splits.length);
{'fraction_non_alphanumeric': 0.11020603737422137, 'fraction_numerical': 0.0, 'mean_word_length': 6.004474272930649, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 15, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6709470', 'n_tokens_mistral': 1937, 'n_tokens_neox': 1792, 'n_words': 364}