/*
 * 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.pig.piggybank.storage;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Properties;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

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.RecordWriter;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

import org.apache.log4j.Logger;

import org.apache.pig.LoadPushDown;
import org.apache.pig.PigException;
import org.apache.pig.ResourceSchema;
import org.apache.pig.ResourceSchema.ResourceFieldSchema;
import org.apache.pig.StoreFuncInterface;

import org.apache.pig.backend.executionengine.ExecException;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigSplit;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigTextInputFormat;

import org.apache.pig.builtin.PigStorage;
import org.apache.pig.bzip2r.Bzip2TextInputFormat;

import org.apache.pig.data.DataByteArray;
import org.apache.pig.data.Tuple;
import org.apache.pig.data.TupleFactory;

import org.apache.pig.impl.logicalLayer.FrontendException;
import org.apache.pig.impl.util.ObjectSerializer;
import org.apache.pig.impl.util.StorageUtil;
import org.apache.pig.impl.util.UDFContext;
import org.apache.pig.impl.util.Utils;

import org.apache.pig.parser.ParserException;

/**
 * CSV loading and storing with support for multi-line fields, 
 * and escaping of delimiters and double quotes within fields; 
 * uses CSV conventions of Excel 2007.
 * 
 * Arguments allow for control over:
 *
 * Which field delimiter to use (default = ',')
 * Whether line breaks are allowed inside of fields (YES_MULTILINE = yes, NO_MULTILINE = no, default = no)
 * How line breaks are to be written when storing (UNIX = LF, WINDOWS = CRLF, NOCHANGE = system default, default = system default)
 * What to do with header rows (first line of each file):
 *     On load: READ_INPUT_HEADER = read header rows, SKIP_INPUT_HEADER = do not read header rows, default = read header rows
 *     On store: WRITE_OUTPUT_HEADER = write a header row, SKIP_OUTPUT_HEADER = do not write a header row, default = do not write a header row
 *
 * Usage:
 *
 * STORE x INTO '<destFileName>'
 *         USING org.apache.pig.piggybank.storage.CSVExcelStorage(
 *              [DELIMITER[, 
 *                  {YES_MULTILINE | NO_MULTILINE}[, 
 *                      {UNIX | WINDOWS | NOCHANGE}[, 
 *                          {READ_INPUT_HEADER, SKIP_INPUT_HEADER, WRITE_OUTPUT_HEADER, SKIP_OUTPUT_HEADER}]]]]
 *         );
 * 
 * Linebreak settings are only used during store; during load, no conversion is performed.
 *
 * WARNING: A danger with enabling multiline fields during load is that unbalanced
 *          double quotes will cause slurping up of input until a balancing double
 *          quote is found, or until something breaks. If you are not expecting
 *          newlines within fields it is therefore more robust to use NO_MULTILINE,
 *          which is the default for that reason.
 * 
 * This is Adreas Paepcke's <paepcke@cs.stanford.edu> CSVExcelStorage with a few modifications.
 */

public class CSVExcelStorage extends PigStorage implements StoreFuncInterface, LoadPushDown {

    public static enum Linebreaks { UNIX, WINDOWS, NOCHANGE };
    public static enum Multiline { YES, NO };
    public static enum Headers { DEFAULT, READ_INPUT_HEADER, SKIP_INPUT_HEADER, WRITE_OUTPUT_HEADER, SKIP_OUTPUT_HEADER }

    protected final static byte LINEFEED = '\n';
    protected final static byte CARRIAGE_RETURN = '\r';
    protected final static byte DOUBLE_QUOTE = '"';
    protected final static byte RECORD_DEL = LINEFEED;

    private static final String FIELD_DELIMITER_DEFAULT_STR = ",";
    private static final String MULTILINE_DEFAULT_STR = "NO_MULTILINE";
    private static final String EOL_DEFAULT_STR = "NOCHANGE";
    private static final String HEADER_DEFAULT_STR = "DEFAULT";
    
    long end = Long.MAX_VALUE;

    private byte fieldDelimiter = ',';
    private Multiline multilineTreatment = Multiline.NO;
    private Linebreaks eolTreatment = Linebreaks.NOCHANGE;
    private Headers headerTreatment = Headers.DEFAULT;

    private ArrayList<Object> mProtoTuple = null;
    private TupleFactory mTupleFactory = TupleFactory.getInstance();
    private String udfContextSignature;
    private String loadLocation;
    private boolean[] mRequiredColumns = null;
    private boolean mRequiredColumnsInitialized = false;
    
    final Logger logger = Logger.getLogger(getClass().getName());

    @SuppressWarnings("rawtypes")
    protected RecordReader in = null;    

    // For replacing LF with CRLF (Unix --> Windows end-of-line convention):
    Pattern loneLFDetectorPattern = Pattern.compile("([^\r])\n", Pattern.DOTALL | Pattern.MULTILINE);
    Matcher loneLFDetector = loneLFDetectorPattern.matcher("");

    // For removing CR (Windows --> Unix):
    Pattern CRLFDetectorPattern = Pattern.compile("\r\n", Pattern.DOTALL | Pattern.MULTILINE);
    Matcher CRLFDetector = CRLFDetectorPattern.matcher("");


    // Pig Storage with COMMA as delimiter:
    TupleFactory tupleMaker = TupleFactory.getInstance();
    private boolean getNextInQuotedField;
    private int getNextFieldID;
    private boolean nextTupleSkipChar;

    // For handling headers
    private boolean loadingFirstRecord = true;
    private boolean storingFirstRecord = true;
    private String header = null;
    private int splitIndex;

    private static final String SCHEMA_SIGNATURE = "pig.csvexcelstorage.schema";
    protected ResourceSchema schema = null;

    /*-----------------------------------------------------
    | Constructors 
    ------------------------*/

    public CSVExcelStorage() {
        super(FIELD_DELIMITER_DEFAULT_STR);
        initializeInstance(FIELD_DELIMITER_DEFAULT_STR, MULTILINE_DEFAULT_STR, EOL_DEFAULT_STR, HEADER_DEFAULT_STR);
    }
    
    public CSVExcelStorage(String delimiter) {
        super(delimiter);
        initializeInstance(delimiter, MULTILINE_DEFAULT_STR, EOL_DEFAULT_STR, HEADER_DEFAULT_STR);
    }

    public CSVExcelStorage(String delimiter, String multilineTreatmentStr) {
        super(delimiter);
        initializeInstance(delimiter, multilineTreatmentStr, EOL_DEFAULT_STR, HEADER_DEFAULT_STR);
    }

    public CSVExcelStorage(String delimiter, String multilineTreatmentStr, String eolTreatmentStr) {
        super(delimiter);
        initializeInstance(delimiter, multilineTreatmentStr, eolTreatmentStr, HEADER_DEFAULT_STR);
    }

    public CSVExcelStorage(String delimiter, String multilineTreatmentStr, String eolTreatmentStr, String headerTreatmentStr) {
        super(delimiter);
        initializeInstance(delimiter, multilineTreatmentStr, eolTreatmentStr, headerTreatmentStr);
    }
    
    private void initializeInstance(String delimiter, String multilineTreatmentStr, String eolTreatmentStr, String headerTreatmentStr) {
        fieldDelimiter = StorageUtil.parseFieldDel(delimiter);

        multilineTreatment = canonicalizeMultilineTreatmentRequest(multilineTreatmentStr);
        eolTreatment = canonicalizeEOLTreatmentRequest(eolTreatmentStr);
        headerTreatment = canonicalizeHeaderTreatmentRequest(headerTreatmentStr);
    }
    
    private Multiline canonicalizeMultilineTreatmentRequest(String multilineTreatmentStr) {
        if (multilineTreatmentStr.equalsIgnoreCase("YES_MULTILINE"))
            return Multiline.YES;
        else if (multilineTreatmentStr.equalsIgnoreCase("NO_MULTILINE"))
            return Multiline.NO;

        throw new IllegalArgumentException(
                "Unrecognized multiline treatment argument " + multilineTreatmentStr + ". " +
                "Should be either 'YES_MULTILINE' or 'NO_MULTILINE'");
    }
    
    private Linebreaks canonicalizeEOLTreatmentRequest(String eolTreatmentStr) {
        if (eolTreatmentStr.equalsIgnoreCase("UNIX"))
            return Linebreaks.UNIX;
        else if (eolTreatmentStr.equalsIgnoreCase("WINDOWS"))
            return Linebreaks.WINDOWS;
        else if (eolTreatmentStr.equalsIgnoreCase("NOCHANGE"))
            return Linebreaks.NOCHANGE;

        throw new IllegalArgumentException(
                "Unrecognized end-of-line treatment argument " + eolTreatmentStr + ". " +
                "Should be one of 'UNIX', 'WINDOWS', or 'NOCHANGE'");
    }

    private Headers canonicalizeHeaderTreatmentRequest(String headerTreatmentStr) {
        if (headerTreatmentStr.equalsIgnoreCase("DEFAULT"))
            return Headers.DEFAULT;
        else if (headerTreatmentStr.equalsIgnoreCase("READ_INPUT_HEADER"))
            return Headers.READ_INPUT_HEADER;
        else if (headerTreatmentStr.equalsIgnoreCase("SKIP_INPUT_HEADER"))
            return Headers.SKIP_INPUT_HEADER;
        else if (headerTreatmentStr.equalsIgnoreCase("WRITE_OUTPUT_HEADER"))
            return Headers.WRITE_OUTPUT_HEADER;
        else if (headerTreatmentStr.equalsIgnoreCase("SKIP_OUTPUT_HEADER"))
            return Headers.SKIP_OUTPUT_HEADER;

        throw new IllegalArgumentException(
            "Unrecognized header treatment argument " + headerTreatmentStr + ". " +
            "Should be one of 'READ_INPUT_HEADER', 'SKIP_INPUT_HEADER', 'WRITE_OUTPUT_HEADER', 'SKIP_OUTPUT_HEADER'");
    }
    
    // ---------------------------------------- STORAGE -----------------------------
    
    public void checkSchema(ResourceSchema s) throws IOException {
        // Not actually checking schema
        // Actually, just storing it to use in the backend
        
        UDFContext udfc = UDFContext.getUDFContext();
        Properties p =
            udfc.getUDFProperties(this.getClass(), new String[]{ udfContextSignature });
        p.setProperty(SCHEMA_SIGNATURE, s.toString());
    }

    public void prepareToWrite(RecordWriter writer) {
        // Get the schema string from the UDFContext object.
        UDFContext udfc = UDFContext.getUDFContext();
        Properties p =
            udfc.getUDFProperties(this.getClass(), new String[]{ udfContextSignature });

        String strSchema = p.getProperty(SCHEMA_SIGNATURE);
        if (strSchema != null) {
            // Parse the schema from the string stored in the properties object.
            try {
                schema = new ResourceSchema(Utils.getSchemaFromString(strSchema));
            } catch (ParserException pex) {
                logger.warn("Could not parse schema for storing.");
            }
        }

        if (headerTreatment == Headers.DEFAULT) {
            headerTreatment = Headers.SKIP_OUTPUT_HEADER;
        }

        // PigStorage's prepareToWrite()
        super.prepareToWrite(writer);
    }

    /* (non-Javadoc)
     * @see org.apache.pig.builtin.PigStorage#putNext(org.apache.pig.data.Tuple)
     * 
     * Given a tuple that corresponds to one record, write
     * it out as CSV, converting among Unix/Windows line
     * breaks as requested in the instantiation. Also take
     * care of escaping field delimiters, double quotes,
     * and linebreaks embedded within fields,
     * 
     */
    @Override
    public void putNext(Tuple tupleToWrite) throws IOException {
        // If WRITE_OUTPUT_HEADER, store a header record with the names of each field
        if (storingFirstRecord && headerTreatment == Headers.WRITE_OUTPUT_HEADER && schema != null) {
            ArrayList<Object> headerProtoTuple = new ArrayList<Object>();
            ResourceFieldSchema[] fields = schema.getFields();
            for (ResourceFieldSchema field : fields) {
                headerProtoTuple.add(field.getName());
            }
            super.putNext(tupleMaker.newTuple(headerProtoTuple));
        }
        storingFirstRecord = false;

        ArrayList<Object> mProtoTuple = new ArrayList<Object>();
        int embeddedNewlineIndex = -1;
        int embeddedCarriageReturn = -1;
        String fieldStr = null;
        // For good debug messages:
        int fieldCounter = -1;
        
        // Do the escaping:
        for (Object field : tupleToWrite.getAll()) {
            fieldCounter++;

            // Substitute a null value with an empty string. See PIG-2470.
            if (field == null) {
                fieldStr = null;
                mProtoTuple.add("");
                continue;
            }
            
            fieldStr = field.toString();
            
            // Embedded double quotes are replaced by two double quotes:
            fieldStr = fieldStr.replaceAll("[\"]", "\"\"");
            
            // If any field delimiters are in the field, or if we did replace
            // any double quotes with a pair of double quotes above,
            // or if the string includes a newline character (LF:\n:0x0A)
            //               or includes a carriage return (CR:\r:0x0D)
            // and we are to allow newlines in fields,
            // then the entire field must be enclosed in double quotes:
            embeddedNewlineIndex =  fieldStr.indexOf(LINEFEED);
            embeddedCarriageReturn = fieldStr.indexOf(CARRIAGE_RETURN);
            
            if ((fieldStr.indexOf(fieldDelimiter) != -1) || 
                (fieldStr.indexOf(DOUBLE_QUOTE) != -1) ||
                (multilineTreatment == Multiline.YES) && (embeddedNewlineIndex != -1 || embeddedCarriageReturn != -1))  {
                fieldStr = "\"" + fieldStr + "\"";
            }
            
            // If requested: replace any Linefeed-only (^J), with LF-Newline (^M^J),
            // This is needed for Excel to recognize a field-internal 
            // new line:

            if ((eolTreatment != Linebreaks.NOCHANGE) && (embeddedNewlineIndex != -1)) {
                if (eolTreatment == Linebreaks.WINDOWS) {
                    loneLFDetector.reset(fieldStr);
                    loneLFDetector.matches();
                    fieldStr = loneLFDetector.replaceAll("$1\r\n");
                } else if (eolTreatment == Linebreaks.UNIX) {
                    CRLFDetector.reset(fieldStr);
                    fieldStr = CRLFDetector.replaceAll("\n");
                }
            }

            mProtoTuple.add(fieldStr);          
        }
        // If Windows line breaks are requested, append 
        // a newline (0x0D a.k.a. ^M) to the last field
        // so that the row termination will end up being
        // \r\n, once the superclass' putNext() method
        // is done below:

        if ((eolTreatment == Linebreaks.WINDOWS) && (fieldStr != null))
            mProtoTuple.set(mProtoTuple.size() - 1, fieldStr + "\r"); 

        Tuple resTuple = tupleMaker.newTuple(mProtoTuple);
        super.putNext(resTuple);
    }

    // ---------------------------------------- LOADING  -----------------------------  

    /* (non-Javadoc)
     * @see org.apache.pig.builtin.PigStorage#getNext()
     */
    @Override
    public Tuple getNext() throws IOException {
        // If SKIP_INPUT_HEADER and this is the first input split, skip header record
        // We store its value as a string though, so we can compare
        // further records to it. If they are the same (this would 
        // happen if multiple small files each with a header were combined
        // into one split), we know to skip the duplicate header record as well.
        if (loadingFirstRecord && headerTreatment == Headers.SKIP_INPUT_HEADER &&
                (splitIndex == 0 || splitIndex == -1)) {
            try {
                if (!in.nextKeyValue())
                    return null;
                header = ((Text) in.getCurrentValue()).toString();
            } catch (InterruptedException e) {
                int errCode = 6018;
                String errMsg = "Error while reading input";
                throw new ExecException(errMsg, errCode, 
                        PigException.REMOTE_ENVIRONMENT, e);
            }
        }
        loadingFirstRecord = false;

        mProtoTuple = new ArrayList<Object>();

        getNextInQuotedField = false;
        boolean evenQuotesSeen = true;
        boolean sawEmbeddedRecordDelimiter = false;
        byte[] buf = null;
        
        if (!mRequiredColumnsInitialized) {
            if (udfContextSignature != null) {
                Properties p = UDFContext.getUDFContext().getUDFProperties(this.getClass());
                mRequiredColumns = (boolean[]) ObjectSerializer.deserialize(p.getProperty(udfContextSignature));
            }
            mRequiredColumnsInitialized = true;
        }
        // Note: we cannot factor out the check for nextKeyValue() being null,
        // because that call overwrites buf with the new line, which is
        // bad if we have a field with a newline.

        try {
            int recordLen = 0;
            getNextFieldID = 0;
            
            while (sawEmbeddedRecordDelimiter || getNextFieldID == 0) {
                Text value = null;
                if (sawEmbeddedRecordDelimiter) {
                    
                    // Deal with pulling more records from the input, because
                    // a double quoted embedded newline was encountered in a field.
                    // Save the length of the record so far, plus one byte for the 
                    // record delimiter (usually newline) that's embedded in the field 
                    // we were working on before falling into this branch:
                    int prevLineLen = recordLen + 1;
                    
                    // Save previous line (the one with the field that has the newline) in a new array.
                    // The last byte will be random; we'll fill in the embedded
                    // record delimiter (usually newline) below:
                    byte[] prevLineSaved = Arrays.copyOf(buf, prevLineLen);
                    prevLineSaved[prevLineLen - 1] = RECORD_DEL;
                    
                    // Read the continuation of the record, unless EOF:
                    if (!in.nextKeyValue()) {
                        return null;
                    }                                                                                           
                    value = (Text) in.getCurrentValue();
                    recordLen = value.getLength();
                    // Grab the continuation's bytes:
                    buf = value.getBytes();
                    
                    // Combine the previous line and the continuation into a new array.
                    // The following copyOf() does half the job: it allocates all the
                    // space, and also copies the previous line into that space:
                    byte[] prevLineAndContinuation = Arrays.copyOf(prevLineSaved, prevLineLen + recordLen);
                    
                    // Now append the continuation. Parms: fromBuf, fromStartPos, toBuf, toStartPos, lengthToCopy:
                    System.arraycopy(buf, 0, prevLineAndContinuation, prevLineLen, recordLen);
                    
                    // We'll work with the combination now:
                    buf = prevLineAndContinuation;
                    
                    // Do the whole record over from the start:
                    mProtoTuple.clear();
                    getNextInQuotedField = false;
                    evenQuotesSeen = true;
                    getNextFieldID = 0;
                    recordLen = prevLineAndContinuation.length;
                    
                } else {
                    // Previous record finished cleanly: start with the next record,
                    // unless EOF:
                    if (!in.nextKeyValue()) {
                        return null;
                    }                                                                                           
                    value = (Text) in.getCurrentValue();

                    // if the line is a duplicate header and 'SKIP_INPUT_HEADER' is set, ignore it
                    // (this might happen if multiple files each with a header are combined into a single split)
                    if (headerTreatment == Headers.SKIP_INPUT_HEADER && value.toString().equals(header)) {
                        if (!in.nextKeyValue())
                            return null;
                        value = (Text) in.getCurrentValue();
                    }

                    buf = value.getBytes();
                    getNextFieldID = 0;
                    recordLen = value.getLength();
                }
                
                nextTupleSkipChar = false;

                ByteBuffer fieldBuffer = ByteBuffer.allocate(recordLen);

                sawEmbeddedRecordDelimiter = processOneInRecord(evenQuotesSeen,
                        buf, recordLen, fieldBuffer);

                // The last field is never delimited by a FIELD_DEL, but by
                // the end of the record. So we need to add that last field.
                // The '!sawEmbeddedRecordDelimiter' handles the case of
                // embedded newlines; we are amidst a field, not at
                // the final record:
                if (!sawEmbeddedRecordDelimiter) 
                    readField(fieldBuffer, getNextFieldID++);
            } // end while

        } catch (InterruptedException e) {
            int errCode = 6018;
            String errMsg = "Error while reading input";
            throw new ExecException(errMsg, errCode, 
                    PigException.REMOTE_ENVIRONMENT, e);
        }

        Tuple t =  mTupleFactory.newTupleNoCopy(mProtoTuple);
        return t;
    }

    /*
     * Service method for getNext().
     * Looks at char after char in the input record,
     * that was previously pulled in by getNext(),
     * and fills the fieldBuffer with those chars.
     * <p> 
     * If multilineTreatment is Multiline.YES, then
     * the return value indicates whether an embedded
     * newline was found in a field, and that newline
     * was in a field that opened with a double quote
     * that was not closed before the end of the 
     * record was reached. If multilineTreatment
     * is Multine.NO, then the return value is always false.
     * <p> 
     * A return value of true will cause the calling method
     * to continue pulling records from the input stream,
     * until a closing quote is found.
     * <p> 
     * Note that the recordReader that delivers records
     * to out getNext() method above considers record 
     * boundaries to be newlines. We therefore never see an actual
     * newline character embedded in a field. We just
     * run out of record. For Multiline.NO we just take
     * such an end of record at face value; the final 
     * resulting tuple will contain information only up
     * to the first newline that was found. 
     * <p> 
     * For Multiline.YES, when we run out of record 
     * in an open double quote, our return of true from
     * this method will cause the caller getNext() to
     * do its additional readings of records from the
     * stream, until the closing double quote is found.
     *  <p> 
     *  
     * @param evenQuotesSeen
     * @param sawEmbeddedRecordDelimiter
     * @param buf
     * @param recordLen
     * @param fieldBuffer
     * @return
     */
    private boolean processOneInRecord(boolean evenQuotesSeen,
                                       byte[] buf, int recordLen,
                                       ByteBuffer fieldBuffer) {
        for (int i = 0; i < recordLen; i++) {
            if (nextTupleSkipChar) {
                nextTupleSkipChar = false;
                continue;
            }
            byte b = buf[i];
            if (getNextInQuotedField) {
                if (b == DOUBLE_QUOTE) {
                    // Does a double quote immediately follow?
                    if ((i < recordLen-1) && (buf[i+1] == DOUBLE_QUOTE)) {
                        fieldBuffer.put(b);
                        nextTupleSkipChar = true;
                        continue;
                    }
                    evenQuotesSeen = !evenQuotesSeen;

                    // If the quote is ending the last field in a record,
                    // set the genNextInQuotedField flag to false,
                    // so the return statement conditional (see below)
                    // is false, indicating that we're ready for the next record
                    if (!evenQuotesSeen && i == recordLen - 1) {
                        getNextInQuotedField = false;
                    }

                    if (evenQuotesSeen) {
                        fieldBuffer.put(DOUBLE_QUOTE);
                    }
                } else if (!evenQuotesSeen && (b == fieldDelimiter || b == RECORD_DEL)) {
                    getNextInQuotedField = false;
                    readField(fieldBuffer, getNextFieldID++);
                } else {
                    fieldBuffer.put(b);
                }
            } else if (b == DOUBLE_QUOTE) {
                // Does a double quote immediately follow?                  
                if ((i < recordLen-1) && (buf[i+1] == DOUBLE_QUOTE) && (fieldBuffer.position() != 0)) {
                    fieldBuffer.put(b);
                    nextTupleSkipChar = true;
                    continue;
                }
                // If we are at the start of a field,
                // that entire field is quoted:
                getNextInQuotedField = true;
                evenQuotesSeen = true;
            } else if (b == fieldDelimiter) {
                readField(fieldBuffer, getNextFieldID++); // end of the field
            } else {
                evenQuotesSeen = true;
                fieldBuffer.put(b);
            }
        } // end for
        return getNextInQuotedField && (multilineTreatment == Multiline.YES);
    }

    private void readField(ByteBuffer buf, int fieldID) {
        if (mRequiredColumns==null || (mRequiredColumns.length>fieldID && mRequiredColumns[fieldID])) {
            byte[] bytes = new byte[buf.position()];
            buf.rewind();
            buf.get(bytes, 0, bytes.length); 
            mProtoTuple.add(new DataByteArray(bytes));
        }
        buf.clear();
    }

    @Override
    public void setLocation(String location, Job job) throws IOException {
        loadLocation = location;
        FileInputFormat.setInputPaths(job, location);        
    }

    @SuppressWarnings("rawtypes")
    @Override
    public InputFormat getInputFormat() {
        if(loadLocation.endsWith(".bz2") || loadLocation.endsWith(".bz")) {
            return new Bzip2TextInputFormat();
        } else {
            return new PigTextInputFormat();
        }
    }

    @Override
    public void prepareToRead(@SuppressWarnings("rawtypes") RecordReader reader, PigSplit split) {
        in = reader;
        splitIndex = split.getSplitIndex();
        
        if (headerTreatment == Headers.DEFAULT) {
            headerTreatment = Headers.READ_INPUT_HEADER;
        }
    }

    @Override
    public RequiredFieldResponse pushProjection(RequiredFieldList requiredFieldList) throws FrontendException {
        if (requiredFieldList == null)
            return null;
        if (requiredFieldList.getFields() != null)
        {
            int lastColumn = -1;
            for (RequiredField rf: requiredFieldList.getFields())
            {
                if (rf.getIndex()>lastColumn)
                {
                    lastColumn = rf.getIndex();
                }
            }
            mRequiredColumns = new boolean[lastColumn+1];
            for (RequiredField rf: requiredFieldList.getFields())
            {
                if (rf.getIndex()!=-1)
                    mRequiredColumns[rf.getIndex()] = true;
            }
            Properties p = UDFContext.getUDFContext().getUDFProperties(this.getClass());
            try {
                p.setProperty(udfContextSignature, ObjectSerializer.serialize(mRequiredColumns));
            } catch (Exception e) {
                throw new RuntimeException("Cannot serialize mRequiredColumns");
            }
        }
        return new RequiredFieldResponse(true);
    }

    @Override
    public void setUDFContextSignature(String signature) {
        this.udfContextSignature = signature; 
    }

    @Override
    public void setStoreFuncUDFContextSignature(String signature) {
        this.udfContextSignature = signature; 
    }

    @Override
    public List<OperatorSet> getFeatures() {
        return Arrays.asList(LoadPushDown.OperatorSet.PROJECTION);
    }
}
