package com.twitter.elephantbird.pig.load;

import java.io.IOException;

import org.apache.hadoop.mapreduce.InputFormat;
import org.apache.pig.PigException;
import org.apache.pig.backend.executionengine.ExecException;
import org.apache.pig.data.DataByteArray;
import org.apache.pig.data.Tuple;
import org.apache.pig.data.TupleFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.twitter.elephantbird.mapreduce.input.LzoTextInputFormat;

/**
 * Load the LZO file line by line, passing each line as a single-field Tuple to Pig.
 */
public class LzoTextLoader extends LzoBaseLoadFunc {
  private static final Logger LOG = LoggerFactory.getLogger(LzoTextLoader.class);

  private static final TupleFactory tupleFactory_ = TupleFactory.getInstance();
  protected enum LzoTextLoaderCounters { LinesRead }

  public LzoTextLoader() {
  }

  /**
   * Return every non-null line as a single-element tuple to Pig.
   */
  @Override
  public Tuple getNext() throws IOException {
	  if (reader_ == null) {
		  return null;
	  }
	  Tuple t = null;
	  try {
		  Object line = reader_.getCurrentValue();
		  if (line != null) {
			  incrCounter(LzoTextLoaderCounters.LinesRead, 1L);
			  t = tupleFactory_.newTuple(new DataByteArray(line.toString().getBytes()));
		  }
	  } catch (InterruptedException e) {
		  int errCode = 6018;
		  String errMsg = "Error while reading input";
		  throw new ExecException(errMsg, errCode,
				  PigException.REMOTE_ENVIRONMENT, e);
	  }
	  return t;
  }

  @Override
  public InputFormat getInputFormat() {
	  return new LzoTextInputFormat();
  }

}

@@ -1063,17 +1063,56 @@
  try %7B%0A%09
-%09
+    if ( reader_.nextKeyValue()) %7B%0A%09    
   Object
@@ -1147,17 +1147,20 @@
 lue();%0A%09
-%09
+    
   if (li
@@ -1174,18 +1174,22 @@
 ull) %7B%0A%09
-%09%09
+      
   incrCo
@@ -1233,18 +1233,22 @@
 , 1L);%0A%09
-%09%09
+      
   t = tu
@@ -1317,17 +1317,27 @@
 s()));%0A%09
-%09
+      %7D%0A%09  
   %7D%0A%09  %7D

Fix infinite loop in LzoTextLoader