// Path: src/main/java/org/apache/jena/grande/Utils.java
// public class Utils {
// 
//     public static ParserProfile createParserProfile(JobContext context, Path path) {
//         Prologue prologue = new Prologue(null, IRIResolver.createNoResolve()); 
//         LabelToNode labelMapping = new MapReduceLabelToNode(context, path);
//         return new ParserProfileBase(prologue, ErrorHandlerFactory.errorHandlerStd, labelMapping);
//     }
// 
//     public static String toString(String[] args) {
// 	    StringBuffer sb = new StringBuffer();
// 	    sb.append("{");
// 	    for ( String arg : args ) sb.append(arg).append(", ");
// 	    if ( sb.length() > 2 ) sb.delete(sb.length()-2, sb.length());
// 	    sb.append("}");
// 	    return sb.toString();
// 	}
//     
// 	public static void log(Job job, Logger log) throws ClassNotFoundException {
// 		log.debug ("{} -> {} ({}, {}) -> {}#{} ({}, {}) -> {}", 
// 			new Object[]{
// 				job.getInputFormatClass().getSimpleName(), job.getMapperClass().getSimpleName(), job.getMapOutputKeyClass().getSimpleName(), job.getMapOutputValueClass().getSimpleName(), 
// 				job.getReducerClass().getSimpleName(), job.getNumReduceTasks(), job.getOutputKeyClass().getSimpleName(), job.getOutputValueClass().getSimpleName(), job.getOutputFormatClass().getSimpleName()
// 			}
// 		);
// 		Path[] inputs = FileInputFormat.getInputPaths(job);
// 		Path output = FileOutputFormat.getOutputPath(job);
// 		log.debug("input: {}", inputs[0]);
// 		log.debug("output: {}", output);
// 	}
// 	
// 	public static void setReducers(Job job, Configuration configuration, Logger log) {
// 	    boolean runLocal = configuration.getBoolean(Constants.OPTION_RUN_LOCAL, Constants.OPTION_RUN_LOCAL_DEFAULT);
// 	    int num_reducers = configuration.getInt(Constants.OPTION_NUM_REDUCERS, Constants.OPTION_NUM_REDUCERS_DEFAULT);
// 	
// 	    // TODO: should we comment this out and let Hadoop decide the number of reducers?
// 	    if ( runLocal ) {
// 	    	if (log != null) log.debug("Setting number of reducers to {}", 1);
// 	        job.setNumReduceTasks(1);
// 	    } else {
//     		job.setNumReduceTasks(num_reducers);
// 	    	if (log != null) log.debug("Setting number of reducers to {}", num_reducers);
// 	    }
// 	}
//     
// }


import java.io.IOException;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.Seekable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.compress.CodecPool;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.CompressionCodecFactory;
import org.apache.hadoop.io.compress.Decompressor;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.util.LineReader;
import org.apache.jena.grande.Utils;
import org.openjena.riot.lang.LangNTriples;
import org.openjena.riot.system.ParserProfile;
import org.openjena.riot.tokens.Tokenizer;
import org.openjena.riot.tokens.TokenizerFactory;
/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.jena.grande.mapreduce.io;



public class TripleRecordReader extends RecordReader<LongWritable, TripleWritable> {

    private static final Log LOG = LogFactory.getLog(TripleRecordReader.class);
    public static final String MAX_LINE_LENGTH = "mapreduce.input.linerecordreader.line.maxlength";

    private LongWritable key = null;
    private Text value = null;
    private TripleWritable triple = null;
    
    private long start;
    private long pos;
    private long end;
    
    private LineReader in;
    private FSDataInputStream fileIn;
    private Seekable filePosition;

    private ParserProfile profile;
    
    private int maxLineLength;
//    private Counter inputByteCounter;

    private CompressionCodecFactory compressionCodecs = null;
    private CompressionCodec codec;
    private Decompressor decompressor;
    
//    @SuppressWarnings({ "unchecked", "rawtypes" })
	@Override
    public void initialize(InputSplit genericSplit, TaskAttemptContext context) throws IOException, InterruptedException {
        FileSplit split = (FileSplit) genericSplit;
        
        // RIOT configuration 
// Path: src/main/java/org/apache/jena/grande/Utils.java
// public class Utils {
// 
//     public static ParserProfile createParserProfile(JobContext context, Path path) {
//         Prologue prologue = new Prologue(null, IRIResolver.createNoResolve()); 
//         LabelToNode labelMapping = new MapReduceLabelToNode(context, path);
//         return new ParserProfileBase(prologue, ErrorHandlerFactory.errorHandlerStd, labelMapping);
//     }
// 
//     public static String toString(String[] args) {
// 	    StringBuffer sb = new StringBuffer();
// 	    sb.append("{");
// 	    for ( String arg : args ) sb.append(arg).append(", ");
// 	    if ( sb.length() > 2 ) sb.delete(sb.length()-2, sb.length());
// 	    sb.append("}");
// 	    return sb.toString();
// 	}
//     
// 	public static void log(Job job, Logger log) throws ClassNotFoundException {
// 		log.debug ("{} -> {} ({}, {}) -> {}#{} ({}, {}) -> {}", 
// 			new Object[]{
// 				job.getInputFormatClass().getSimpleName(), job.getMapperClass().getSimpleName(), job.getMapOutputKeyClass().getSimpleName(), job.getMapOutputValueClass().getSimpleName(), 
// 				job.getReducerClass().getSimpleName(), job.getNumReduceTasks(), job.getOutputKeyClass().getSimpleName(), job.getOutputValueClass().getSimpleName(), job.getOutputFormatClass().getSimpleName()
// 			}
// 		);
// 		Path[] inputs = FileInputFormat.getInputPaths(job);
// 		Path output = FileOutputFormat.getOutputPath(job);
// 		log.debug("input: {}", inputs[0]);
// 		log.debug("output: {}", output);
// 	}
// 	
// 	public static void setReducers(Job job, Configuration configuration, Logger log) {
// 	    boolean runLocal = configuration.getBoolean(Constants.OPTION_RUN_LOCAL, Constants.OPTION_RUN_LOCAL_DEFAULT);
// 	    int num_reducers = configuration.getInt(Constants.OPTION_NUM_REDUCERS, Constants.OPTION_NUM_REDUCERS_DEFAULT);
// 	
// 	    // TODO: should we comment this out and let Hadoop decide the number of reducers?
// 	    if ( runLocal ) {
// 	    	if (log != null) log.debug("Setting number of reducers to {}", 1);
// 	        job.setNumReduceTasks(1);
// 	    } else {
//     		job.setNumReduceTasks(num_reducers);
// 	    	if (log != null) log.debug("Setting number of reducers to {}", num_reducers);
// 	    }
// 	}
//     
// }


// Path: src/main/java/org/apache/jena/grande/mapreduce/io/TripleRecordReader.java
import java.io.IOException;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.Seekable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.compress.CodecPool;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.CompressionCodecFactory;
import org.apache.hadoop.io.compress.Decompressor;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.util.LineReader;
import org.apache.jena.grande.Utils;
import org.openjena.riot.lang.LangNTriples;
import org.openjena.riot.system.ParserProfile;
import org.openjena.riot.tokens.Tokenizer;
import org.openjena.riot.tokens.TokenizerFactory;

/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.jena.grande.mapreduce.io;



public class TripleRecordReader extends RecordReader<LongWritable, TripleWritable> {

    private static final Log LOG = LogFactory.getLog(TripleRecordReader.class);
    public static final String MAX_LINE_LENGTH = "mapreduce.input.linerecordreader.line.maxlength";

    private LongWritable key = null;
    private Text value = null;
    private TripleWritable triple = null;
    
    private long start;
    private long pos;
    private long end;
    
    private LineReader in;
    private FSDataInputStream fileIn;
    private Seekable filePosition;

    private ParserProfile profile;
    
    private int maxLineLength;
//    private Counter inputByteCounter;

    private CompressionCodecFactory compressionCodecs = null;
    private CompressionCodec codec;
    private Decompressor decompressor;
    
//    @SuppressWarnings({ "unchecked", "rawtypes" })
	@Override
    public void initialize(InputSplit genericSplit, TaskAttemptContext context) throws IOException, InterruptedException {
        FileSplit split = (FileSplit) genericSplit;
        
        // RIOT configuration 
        profile = Utils.createParserProfile(context, split.getPath());