// Path: src/main/java/org/apache/jena/grande/NodeEncoder.java
// public class NodeEncoder {
// 
// 	private static final PrefixMap emptyPrefixMap = new PrefixMap();
// 	
// 	public static String asString(Node node) {
//         StringWriter sw = new StringWriter() ;
//         NodeFmtLib.serialize(sw, node, null, emptyPrefixMap);
//         return sw.toString() ; 
// 	}
// 	
// 	public static String asString(Node node, PrefixMap prefixMap) {
//         StringWriter sw = new StringWriter() ;
//         NodeFmtLib.serialize(sw, node, null, prefixMap);
//         return sw.toString() ; 
// 	}
// 
// 	public static Node asNode(String str) {
// 		return NodeFactory.parseNode(str);
// 	}
// 
// }
// 
// Path: src/main/java/org/apache/jena/grande/mapreduce/io/QuadWritable.java
// public class QuadWritable extends BinaryComparable implements WritableComparable<BinaryComparable> {
// 
//     private Quad quad;
//     private byte[] bytes;
//     private int length;
//     
//     public QuadWritable(){
//     	this( new Quad(Node.ANY, Node.ANY, Node.ANY, Node.ANY) );
//     }
//     
//     public QuadWritable(Quad quad) {
//         this.quad = quad;
//         this.bytes = toBytes(quad);
//         this.length = bytes.length;
//     }
//     
//     @Override
//     public void readFields(DataInput in) throws IOException {
//         length = WritableUtils.readVInt(in);
//         bytes = new byte[length];
//         in.readFully(bytes, 0, length);
//         Tokenizer tokenizer = TokenizerFactory.makeTokenizerASCII(new String(bytes)) ;
//         LangNQuads parser = new LangNQuads(tokenizer, RiotLib.profile(Lang.NQUADS, null), null) ;
//         quad = parser.next();
//     }
// 
//     @Override
//     public void write(DataOutput out) throws IOException {
//         WritableUtils.writeVInt(out, length);
//         out.write(bytes, 0, length);
//     }
// 
//     @Override
//     public byte[] getBytes() {
//         return bytes;
//     }
// 
//     @Override
//     public int getLength() {
//         return length;
//     }
//     
//     public Quad getQuad() {
//         return quad;
//     }
//     
//     private byte[] toBytes(Quad quad) {
//         StringWriter out = new StringWriter();
//         out.getBuffer().length();
//         OutputLangUtils.output(out, quad, null, null);
//         try {
//             return out.toString().getBytes("UTF-8");
//         } catch (UnsupportedEncodingException e) {
//             throw new JenaGrandeException(e);
//         }
//     }
// 
//     @Override
//     public String toString() {
//         return quad.toString();
//     }
// }


import java.io.IOException;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.compress.BZip2Codec;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.GzipCodec;
import org.apache.hadoop.mapreduce.InputFormat;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.OutputFormat;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.jena.grande.NodeEncoder;
import org.apache.jena.grande.mapreduce.io.QuadWritable;
import org.apache.pig.FileInputLoadFunc;
import org.apache.pig.LoadFunc;
import org.apache.pig.ResourceSchema;
import org.apache.pig.StoreFunc;
import org.apache.pig.StoreFuncInterface;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigSplit;
import org.apache.pig.data.Tuple;
import org.apache.pig.data.TupleFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.hp.hpl.jena.graph.Node;
import com.hp.hpl.jena.sparql.core.Quad;
    @Override
    public void setLocation(String location, Job job) throws IOException {
    	log.debug("setLocation({}, {})", location, job);
        this.location = location;
        FileInputFormat.setInputPaths(job, location);
    }
    
	@Override
	public InputFormat<LongWritable, QuadWritable> getInputFormat() throws IOException {
		InputFormat<LongWritable, QuadWritable> inputFormat = new NQuadsPigInputFormat(); 
        log.debug("getInputFormat() --> {}", inputFormat);
        return inputFormat;
	}
	
	@SuppressWarnings("unchecked")
	@Override
	public void prepareToRead(@SuppressWarnings("rawtypes") RecordReader reader, PigSplit split) throws IOException {
		log.debug("prepareToRead({}, {})", reader, split);
		this.reader = reader;
	}

	private final TupleFactory tupleFactory = TupleFactory.getInstance();
	
	@Override
	public Tuple getNext() throws IOException {
		Tuple tuple = null;
        try {
    		if ( reader.nextKeyValue() ) {
                QuadWritable quad = reader.getCurrentValue();
                tuple = tupleFactory.newTuple(4);
// Path: src/main/java/org/apache/jena/grande/NodeEncoder.java
// public class NodeEncoder {
// 
// 	private static final PrefixMap emptyPrefixMap = new PrefixMap();
// 	
// 	public static String asString(Node node) {
//         StringWriter sw = new StringWriter() ;
//         NodeFmtLib.serialize(sw, node, null, emptyPrefixMap);
//         return sw.toString() ; 
// 	}
// 	
// 	public static String asString(Node node, PrefixMap prefixMap) {
//         StringWriter sw = new StringWriter() ;
//         NodeFmtLib.serialize(sw, node, null, prefixMap);
//         return sw.toString() ; 
// 	}
// 
// 	public static Node asNode(String str) {
// 		return NodeFactory.parseNode(str);
// 	}
// 
// }
// 
// Path: src/main/java/org/apache/jena/grande/mapreduce/io/QuadWritable.java
// public class QuadWritable extends BinaryComparable implements WritableComparable<BinaryComparable> {
// 
//     private Quad quad;
//     private byte[] bytes;
//     private int length;
//     
//     public QuadWritable(){
//     	this( new Quad(Node.ANY, Node.ANY, Node.ANY, Node.ANY) );
//     }
//     
//     public QuadWritable(Quad quad) {
//         this.quad = quad;
//         this.bytes = toBytes(quad);
//         this.length = bytes.length;
//     }
//     
//     @Override
//     public void readFields(DataInput in) throws IOException {
//         length = WritableUtils.readVInt(in);
//         bytes = new byte[length];
//         in.readFully(bytes, 0, length);
//         Tokenizer tokenizer = TokenizerFactory.makeTokenizerASCII(new String(bytes)) ;
//         LangNQuads parser = new LangNQuads(tokenizer, RiotLib.profile(Lang.NQUADS, null), null) ;
//         quad = parser.next();
//     }
// 
//     @Override
//     public void write(DataOutput out) throws IOException {
//         WritableUtils.writeVInt(out, length);
//         out.write(bytes, 0, length);
//     }
// 
//     @Override
//     public byte[] getBytes() {
//         return bytes;
//     }
// 
//     @Override
//     public int getLength() {
//         return length;
//     }
//     
//     public Quad getQuad() {
//         return quad;
//     }
//     
//     private byte[] toBytes(Quad quad) {
//         StringWriter out = new StringWriter();
//         out.getBuffer().length();
//         OutputLangUtils.output(out, quad, null, null);
//         try {
//             return out.toString().getBytes("UTF-8");
//         } catch (UnsupportedEncodingException e) {
//             throw new JenaGrandeException(e);
//         }
//     }
// 
//     @Override
//     public String toString() {
//         return quad.toString();
//     }
// }


// Path: src/main/java/org/apache/jena/grande/pig/RdfStorage.java
import java.io.IOException;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.compress.BZip2Codec;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.GzipCodec;
import org.apache.hadoop.mapreduce.InputFormat;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.OutputFormat;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.jena.grande.NodeEncoder;
import org.apache.jena.grande.mapreduce.io.QuadWritable;
import org.apache.pig.FileInputLoadFunc;
import org.apache.pig.LoadFunc;
import org.apache.pig.ResourceSchema;
import org.apache.pig.StoreFunc;
import org.apache.pig.StoreFuncInterface;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigSplit;
import org.apache.pig.data.Tuple;
import org.apache.pig.data.TupleFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.hp.hpl.jena.graph.Node;
import com.hp.hpl.jena.sparql.core.Quad;

    @Override
    public void setLocation(String location, Job job) throws IOException {
    	log.debug("setLocation({}, {})", location, job);
        this.location = location;
        FileInputFormat.setInputPaths(job, location);
    }
    
	@Override
	public InputFormat<LongWritable, QuadWritable> getInputFormat() throws IOException {
		InputFormat<LongWritable, QuadWritable> inputFormat = new NQuadsPigInputFormat(); 
        log.debug("getInputFormat() --> {}", inputFormat);
        return inputFormat;
	}
	
	@SuppressWarnings("unchecked")
	@Override
	public void prepareToRead(@SuppressWarnings("rawtypes") RecordReader reader, PigSplit split) throws IOException {
		log.debug("prepareToRead({}, {})", reader, split);
		this.reader = reader;
	}

	private final TupleFactory tupleFactory = TupleFactory.getInstance();
	
	@Override
	public Tuple getNext() throws IOException {
		Tuple tuple = null;
        try {
    		if ( reader.nextKeyValue() ) {
                QuadWritable quad = reader.getCurrentValue();
                tuple = tupleFactory.newTuple(4);
                tuple.set(0, NodeEncoder.asString(quad.getQuad().getGraph()));