code: @Override
    public Tuple getNext() throws IOException {
      LzoW3CLogRecordReader w3CLogRecordReader = (LzoW3CLogRecordReader) reader;
      if (w3CLogRecordReader == null) {
        return null;
      }
      MapWritable value_;
      try {
        if (w3CLogRecordReader.nextKeyValue()
            && (value_ = w3CLogRecordReader.getCurrentValue()) != null) {
          Map<String, String> values = Maps.newHashMap();
  
          for (Writable key : value_.keySet()) {
            Writable value = value_.get(key);
            values.put(key.toString(), value != null ? value.toString() : null);
          }
          incrCounter(LzoW3CLogLoaderCounters.LinesW3CDecoded, 1L);
          incrCounter(LzoW3CLogLoaderCounters.UnparseableLines,
              w3CLogRecordReader.getBadRecordsSkipped());
          return tupleFactory_.newTuple(values);
        }
      } catch (InterruptedException e) {
        int errCode = 6018;
        String errMsg = "Error while reading input";
        throw new ExecException(errMsg, errCode,
            PigException.REMOTE_ENVIRONMENT, e);
      }
      return null;
    }
docs: Return every non-null line as a single-element tuple to Pig.