// Path: binarypig/src/main/java/com/endgame/binarypig/util/StreamUtils.java
// public class StreamUtils {
// 
// 	public static String drainInputStream(InputStream stream)
// 	{	
// 		StringBuilder buf = new StringBuilder();
// 		try {
// 			while(stream.available() > 0)
// 			{
// 				buf.append((char)stream.read());
// 			}
// 		} catch (IOException e) {}
// 		return buf.toString();
// 	}
// 
// 	public static void writeToFile(BytesWritable value, File binaryFile) throws IOException
// 	{
// 		FileOutputStream fileOut = new FileOutputStream(binaryFile);
// 		fileOut.write(value.getBytes(), 0, value.getLength());
// 		fileOut.close();
// 	}
// 
// }


import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import org.apache.commons.io.FileUtils;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputFormat;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileRecordReader;
import org.apache.pig.FileInputLoadFunc;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigSplit;
import org.apache.pig.data.Tuple;
import org.apache.pig.data.TupleFactory;
import com.endgame.binarypig.util.StreamUtils;
import com.google.common.base.Stopwatch;
		System.err.println("dataDir     = "+dataDir);
		System.err.println("useDevShm     = "+useDevShm);
		System.err.println("numRecords     = "+numRecords);
		
		long usec = totalOverhead.elapsedTime(TimeUnit.MICROSECONDS);
		dumpStat("copyOverhead", copyOverhead, usec, numRecords);
		dumpStat("deleteOverhead", deleteOverhead, usec, numRecords);
		dumpStat("execOverhead", execOverhead, usec, numRecords);
		dumpStat("tupleCreationOverhead", tupleCreationOverhead, usec, numRecords);
		dumpStat("totalOverhead", totalOverhead, usec, numRecords);
	}

	@Override
	public Tuple getNext() throws IOException {
		if(!shouldContinue()){
			dumpStats();
			cleanUp();
			return null;
		}
		totalOverhead.start();
		
		key = reader.getCurrentKey();
		value = reader.getCurrentValue();
		
		++numRecords;
		
		copyOverhead.start();
		File binaryFile = new File(dataDir, key.toString());
		binaryFile.deleteOnExit();
		try {
// Path: binarypig/src/main/java/com/endgame/binarypig/util/StreamUtils.java
// public class StreamUtils {
// 
// 	public static String drainInputStream(InputStream stream)
// 	{	
// 		StringBuilder buf = new StringBuilder();
// 		try {
// 			while(stream.available() > 0)
// 			{
// 				buf.append((char)stream.read());
// 			}
// 		} catch (IOException e) {}
// 		return buf.toString();
// 	}
// 
// 	public static void writeToFile(BytesWritable value, File binaryFile) throws IOException
// 	{
// 		FileOutputStream fileOut = new FileOutputStream(binaryFile);
// 		fileOut.write(value.getBytes(), 0, value.getLength());
// 		fileOut.close();
// 	}
// 
// }


// Path: binarypig/src/main/java/com/endgame/binarypig/loaders/AbstractFileDroppingLoader.java
import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import org.apache.commons.io.FileUtils;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputFormat;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileRecordReader;
import org.apache.pig.FileInputLoadFunc;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigSplit;
import org.apache.pig.data.Tuple;
import org.apache.pig.data.TupleFactory;
import com.endgame.binarypig.util.StreamUtils;
import com.google.common.base.Stopwatch;

		System.err.println("dataDir     = "+dataDir);
		System.err.println("useDevShm     = "+useDevShm);
		System.err.println("numRecords     = "+numRecords);
		
		long usec = totalOverhead.elapsedTime(TimeUnit.MICROSECONDS);
		dumpStat("copyOverhead", copyOverhead, usec, numRecords);
		dumpStat("deleteOverhead", deleteOverhead, usec, numRecords);
		dumpStat("execOverhead", execOverhead, usec, numRecords);
		dumpStat("tupleCreationOverhead", tupleCreationOverhead, usec, numRecords);
		dumpStat("totalOverhead", totalOverhead, usec, numRecords);
	}

	@Override
	public Tuple getNext() throws IOException {
		if(!shouldContinue()){
			dumpStats();
			cleanUp();
			return null;
		}
		totalOverhead.start();
		
		key = reader.getCurrentKey();
		value = reader.getCurrentValue();
		
		++numRecords;
		
		copyOverhead.start();
		File binaryFile = new File(dataDir, key.toString());
		binaryFile.deleteOnExit();
		try {
			StreamUtils.writeToFile((BytesWritable) value, binaryFile);