src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/POProject.java - pig - Git at Google

 /*
  * 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.backend.hadoop.executionengine.physicalLayer.expressionOperators;

 import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.List;

 import org.apache.pig.PigException;
 import org.apache.pig.PigWarning;
 import org.apache.pig.backend.executionengine.ExecException;
 import org.apache.pig.backend.hadoop.executionengine.physicalLayer.POStatus;
 import org.apache.pig.backend.hadoop.executionengine.physicalLayer.Result;
 import org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhyPlanVisitor;
 import org.apache.pig.data.DataBag;
 import org.apache.pig.data.DataType;
 import org.apache.pig.data.SingleTupleBag;
 import org.apache.pig.data.Tuple;
 import org.apache.pig.impl.plan.NodeIdGenerator;
 import org.apache.pig.impl.plan.OperatorKey;
 import org.apache.pig.impl.plan.VisitorException;

 /**
  * Implements the overloaded form of the project operator.
  * Projects the specified column from the input tuple.
  * However, if asked for tuples when the input is a bag,
  * the overloaded form is invoked and the project streams
  * the tuples through instead of the bag.
  */
 public class POProject extends ExpressionOperator {

     /**
      *
      */
     private static final long serialVersionUID = 1L;

     private boolean resultSingleTupleBag = false;

     //The column to project
     protected ArrayList<Integer> columns;

     //True if we are in the middle of streaming tuples
     //in a bag
     private boolean processingBagOfTuples = false;

     //The bag iterator used while straeming tuple
     private transient Iterator<Tuple> bagIterator = null;

     //Represents the fact that this instance of POProject
     //is overloaded to stream tuples in the bag rather
     //than passing the entire bag. It is the responsibility
     //of the translator to set this.
     protected boolean overloaded = false;


     protected boolean isProjectToEnd = false;
     protected int startCol;

     public POProject(OperatorKey k) {
         this(k,-1,0);
     }

     public POProject(OperatorKey k, int rp) {
         this(k, rp, 0);
     }

     public POProject(OperatorKey k, int rp, int col) {
         super(k, rp);
         columns = new ArrayList<Integer>();
         columns.add(col);
     }

     public POProject(OperatorKey k, int rp, ArrayList<Integer> cols) {
         super(k, rp);
         columns = cols;
     }

     public void setProjectToEnd(int startCol){
         this.isProjectToEnd = true;
         this.startCol = startCol;
         columns = new ArrayList<Integer>();
     }

     @Override
     public String name() {

         String str = "Project" + "[" + DataType.findTypeName(resultType) + "]";

         if(isStar()){
             str += "[*]";
         }else if(isProjectToEnd){
             str += "[" + startCol + " .. " + "]";
         }else{
             str += columns;
         }

         str += " - " + mKey.toString();
         return str;

     }

     @Override
     public boolean supportsMultipleInputs() {
         return false;
     }

     @Override
     public boolean supportsMultipleOutputs() {
         return false;
     }

     @Override
     public void visit(PhyPlanVisitor v) throws VisitorException {
         v.visitProject(this);
     }


     /**
      * Overridden since the attachment of the new input
      * should cause the old processing to end.
      */
     @Override
     public void attachInput(Tuple t) {
         super.attachInput(t);
         processingBagOfTuples = false;
     }

     /**
      * Fetches the input tuple and returns the requested
      * column
      * @return next value.
      * @throws ExecException
      */
     public Result getNext() throws ExecException{
         Result res = processInput();
         Tuple inpValue = (Tuple)res.result;

         Object ret;
         if(res.returnStatus != POStatus.STATUS_OK){
             return res;
         }
         if (isStar()) {
             illustratorMarkup(inpValue, res.result, -1);
             return res;
         } else if(columns.size() == 1) {
             if ( inpValue == null ) {
                 // the tuple is null, so a dereference should also produce a null
                 res.returnStatus = POStatus.STATUS_OK;
                 ret = null;
             } else if( inpValue.size() > columns.get(0) ) {
                 ret = inpValue.get(columns.get(0));
             } else {
                 if(pigLogger != null) {
                     pigLogger.warn(this,"Attempt to access field " +
                             "which was not found in the input", PigWarning.ACCESSING_NON_EXISTENT_FIELD);
                 }
                 res.returnStatus = POStatus.STATUS_OK;
                 ret = null;
             }
         } else if(isProjectToEnd){
             ret = getRangeTuple(inpValue);
         }
         else {
             ArrayList<Object> objList = new ArrayList<Object>(columns.size());

             for(int col : columns) {
                 addColumn(objList, inpValue, col);
             }
             ret = mTupleFactory.newTupleNoCopy(objList);
         }
         res.result = ret;
         illustratorMarkup(inpValue, res.result, -1);
         return res;
     }

     private boolean isRangeInvalid(int lastColIdx) {
         if(startCol > lastColIdx){
             // this must be happening because tuple is smaller than startCol
             if(pigLogger != null) {
                 pigLogger.warn(this, "Invalid range being projected," +
                         " startCol postition" + startCol + " greater than tuple size",
                         PigWarning.PROJECTION_INVALID_RANGE
                 );
             }
             return true;
         }
         return false;
     }

     /**
      * Add i'th column from inpValue to objList
      * @param objList
      * @param inpValue
      * @param i
      * @throws ExecException
      */
     private void addColumn(ArrayList<Object> objList, Tuple inpValue, int i)
     throws ExecException {
         if( inpValue == null ) {
             // the tuple is null, so a dereference should also produce a null
             objList.add(null);
         } else if( inpValue.size() > i ) {
             objList.add(inpValue.get(i));
         } else {
             if(pigLogger != null) {
                 pigLogger.warn(this,"Attempt to access field " + i +
                         " which was not found in the input", PigWarning.ACCESSING_NON_EXISTENT_FIELD);
             }
             objList.add(null);
         }
     }

     @Override
     public Result getNextDataBag() throws ExecException {

         Result res = processInputBag();
         if(res.returnStatus!=POStatus.STATUS_OK)
             return res;
         return(consumeInputBag(res));
     }

     /**
      * @param input
      * @throws ExecException
      */
     protected Result consumeInputBag(Result input) throws ExecException {
         if(isInputAttached() || isStar()){
             Result retVal = new Result();
             retVal.result = input.result;
             retVal.returnStatus = POStatus.STATUS_OK;
             detachInput();
             return retVal;
         }

         if (input.result instanceof DataBag) {
             DataBag inpBag = (DataBag) input.result;
             Result retVal = new Result();

             DataBag outBag;
             if(resultSingleTupleBag) {
                 // we have only one tuple in a bag - so create
                 // A SingleTupleBag for the result and fill it
                 // appropriately from the input bag
                 Tuple tuple = inpBag.iterator().next();
                 if(!isProjectToEnd){
                     ArrayList<Object> objList = new ArrayList<Object>(columns.size());
                     for (int col : columns) {
                         addColumn(objList, tuple, col);
                     }
                     outBag = new SingleTupleBag( mTupleFactory.newTupleNoCopy(objList) );
                 }else {
                     Tuple tmpTuple = getRangeTuple(tuple);
                     outBag = new SingleTupleBag(tmpTuple);
                 }
             } else {
                 outBag = mBagFactory.newDefaultBag();
                 for (Tuple tuple : inpBag) {
                     if(!isProjectToEnd){
                         ArrayList<Object> objList = new ArrayList<Object>(columns.size());
                         for (int col : columns) {
                             addColumn(objList, tuple, col);
                         }
                         outBag.add( mTupleFactory.newTupleNoCopy(objList) );
                     }else{
                         Tuple outTuple = getRangeTuple(tuple);
                         outBag.add(outTuple);
                     }
                 }
             }
             retVal.result = outBag;
             retVal.returnStatus = POStatus.STATUS_OK;
             return retVal;
         } else if (input.result instanceof Tuple) {
             // if input is tuple, columns should only have one item
             Result retVal = new Result();
             retVal.result = ((Tuple)input.result).get(columns.get(0));
             retVal.returnStatus = POStatus.STATUS_OK;
             return retVal;
         } else if (input.result==null) {
             Result retVal = new Result();
             retVal.result = null;
             retVal.returnStatus = POStatus.STATUS_OK;
             return retVal;
         } else {
             throw new ExecException("Cannot dereference a bag from " + input.result.getClass().getName(), 1129);
         }
     }

     private Tuple getRangeTuple(Tuple tuple) throws ExecException {
         int lastColIdx = tuple.size() - 1;
         Tuple outTuple;
         if(isRangeInvalid(lastColIdx)){
             //invalid range - return empty tuple
             outTuple = mTupleFactory.newTuple();
         }
         else {
             ArrayList<Object> objList = new ArrayList<Object>(lastColIdx - startCol + 1);
             for(int i = startCol; i <= lastColIdx ; i++){
                 addColumn(objList, tuple, i);
             }
             outTuple = mTupleFactory.newTupleNoCopy(objList);
         }
         return outTuple;
     }

     @Override
     public Result getNextDataByteArray() throws ExecException {
         return getNext();
     }

     @Override
     public Result getNextDouble() throws ExecException {
         return getNext();
     }

     @Override
     public Result getNextFloat() throws ExecException {
         return getNext();
     }

     @Override
     public Result getNextInteger() throws ExecException {
         return getNext();
     }

     @Override
     public Result getNextLong() throws ExecException {
         return getNext();
     }


     @Override
     public Result getNextBoolean() throws ExecException {
         return getNext();
     }

     @Override
     public Result getNextDateTime() throws ExecException {
         return getNext();
     }

     @Override
     public Result getNextMap() throws ExecException {
         return getNext();
     }

     @Override
     public Result getNextString() throws ExecException {
         return getNext();
     }

     @Override
     public Result getNextBigInteger() throws ExecException {
         return getNext();
     }

     @Override
     public Result getNextBigDecimal() throws ExecException {
         return getNext();
     }

     /**
      * Asked for Tuples. Check if the input is a bag.
      * If so, stream the tuples in the bag instead of
      * the entire bag.
      */
     @Override
     public Result getNextTuple() throws ExecException {
         Result res = new Result();
         if(!processingBagOfTuples){
             Tuple inpValue = null;
             res = processInput();
             if(res.returnStatus!=POStatus.STATUS_OK)
                 return res;
             if(isStar())
                 return res;

             inpValue = (Tuple)res.result;
             res.result = null;

             Object ret;

             if(columns.size() == 1) {
                 if( inpValue == null ) {
                     // the tuple is null, so a dereference should also produce a null
                     ret = null;
                 } else if( inpValue.size() > columns.get(0) ) {
                     ret = inpValue.get(columns.get(0));
                 } else {
                     if(pigLogger != null) {
                         pigLogger.warn(this,"Attempt to access field " +
                                 "which was not found in the input", PigWarning.ACCESSING_NON_EXISTENT_FIELD);
                     }
                     ret = null;
                 }
             } else if(isProjectToEnd) {
                 ret = getRangeTuple(inpValue);
             } else {
                 ArrayList<Object> objList = new ArrayList<Object>(columns.size());

                 for(int col: columns) {
                     if( inpValue == null ) {
                         // the tuple is null, so a dereference should also produce a null
                         objList.add(null);
                     } else if( inpValue.size() > col ) {
                         objList.add(inpValue.get(col));
                     } else {
                         if(pigLogger != null) {
                             pigLogger.warn(this,"Attempt to access field " +
                                     "which was not found in the input", PigWarning.ACCESSING_NON_EXISTENT_FIELD);
                         }
                         objList.add(null);
                     }
                 }
                 ret = mTupleFactory.newTuple(objList);
                 res.result = (Tuple)ret;
                 return res;
             }

             if(overloaded){
                 if (ret!=null) {
                     DataBag retBag = (DataBag)ret;
                     bagIterator = retBag.iterator();
                     if(bagIterator.hasNext()){
                         processingBagOfTuples = true;
                         res.result = bagIterator.next();
                     }
                     // If the bag contains no tuple, set the returnStatus to STATUS_EOP
                     if (!processingBagOfTuples)
                         res.returnStatus = POStatus.STATUS_EOP;
                 } else {
                     res.returnStatus = POStatus.STATUS_EOP;
                 }
             }
             else {
                 res.result = (Tuple)ret;
             }
             return res;
         }
         if(bagIterator.hasNext()){
             res.result = bagIterator.next();
             res.returnStatus = POStatus.STATUS_OK;
             return res;
         }
         else{
             //done processing the bag of tuples
             processingBagOfTuples = false;
             return getNextTuple();
         }
     }

     public ArrayList<Integer> getColumns(){
         if(isProjectToEnd) {
             throw new AssertionError("Internal error. Improper use of method getColumns() in "
                 + POProject.class.getSimpleName());
         }
         return columns;
     }

     public int getColumn() throws ExecException {
         if(columns.size() != 1 || isProjectToEnd) {
             int errCode = 2068;
             String msg = "Internal error. Improper use of method getColumn() in "
                 + POProject.class.getSimpleName();
             throw new ExecException(msg, errCode, PigException.BUG);
         }
         return columns.get(0);
     }

     public int getStartCol(){
         return startCol;
     }

     public void setColumns(ArrayList<Integer> cols) {
         if(isProjectToEnd){
             throw new AssertionError("Columns should not be set for range projection");
         }
         this.columns = cols;
     }

     public void setColumn(int col) {
         isProjectToEnd = false;
         if(null == columns) {
             columns = new ArrayList<Integer>();
         } else {
             columns.clear();
         }
         columns.add(col);
     }

     public boolean isOverloaded() {
         return overloaded;
     }

     public void setOverloaded(boolean overloaded) {
         this.overloaded = overloaded;
     }

     public boolean isStar() {
         return isProjectToEnd && startCol == 0;
     }

     public boolean isProjectToEnd(){
         return isProjectToEnd;
     }

     public void setStar(boolean star) {
         if(star){
             isProjectToEnd = true;
             startCol = 0;
         }else{
             isProjectToEnd = false;
         }
     }

     @Override
     public POProject clone() throws CloneNotSupportedException {
         ArrayList<Integer> cols = new ArrayList<Integer>(columns.size());
         // Can resuse the same Integer objects, as they are immutable
         for (Integer i : columns) {
             cols.add(i);
         }
         POProject clone = new POProject(new OperatorKey(mKey.scope,
             NodeIdGenerator.getGenerator().getNextNodeId(mKey.scope)),
             requestedParallelism, cols);
         clone.cloneHelper(this);
         clone.overloaded = overloaded;
         clone.startCol = startCol;
         clone.isProjectToEnd = isProjectToEnd;
         clone.resultType = resultType;
         return clone;
     }

     protected Result processInputBag() throws ExecException {

         Result res = new Result();
         if (input==null && (inputs == null || inputs.size()==0)) {
 //            log.warn("No inputs found. Signaling End of Processing.");
             res.returnStatus = POStatus.STATUS_EOP;
             return res;
         }

         //Should be removed once the model is clear
         if(getReporter()!=null) {
             getReporter().progress();
         }

         if(!isInputAttached()) {
             if (inputs.get(0).getResultType()==DataType.BAG)
                 return inputs.get(0).getNextDataBag();
             else
                 return inputs.get(0).getNextTuple();
         }
         else{
             res.result = (DataBag)input.get(columns.get(0));
             res.returnStatus = POStatus.STATUS_OK;
             return res;
         }
     }

     public void setResultSingleTupleBag(boolean resultSingleTupleBag) {
         this.resultSingleTupleBag = resultSingleTupleBag;
     }

     @Override
     public List<ExpressionOperator> getChildExpressions() {
         return null;
     }

     @Override
     public Tuple illustratorMarkup(Object in, Object out, int eqClassIndex) {
         if(illustrator != null) {

         }
         return null;
     }
 }
	/*
	* 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.backend.hadoop.executionengine.physicalLayer.expressionOperators;

	import java.util.ArrayList;
	import java.util.Iterator;
	import java.util.List;

	import org.apache.pig.PigException;
	import org.apache.pig.PigWarning;
	import org.apache.pig.backend.executionengine.ExecException;
	import org.apache.pig.backend.hadoop.executionengine.physicalLayer.POStatus;
	import org.apache.pig.backend.hadoop.executionengine.physicalLayer.Result;
	import org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhyPlanVisitor;
	import org.apache.pig.data.DataBag;
	import org.apache.pig.data.DataType;
	import org.apache.pig.data.SingleTupleBag;
	import org.apache.pig.data.Tuple;
	import org.apache.pig.impl.plan.NodeIdGenerator;
	import org.apache.pig.impl.plan.OperatorKey;
	import org.apache.pig.impl.plan.VisitorException;

	/**
	* Implements the overloaded form of the project operator.
	* Projects the specified column from the input tuple.
	* However, if asked for tuples when the input is a bag,
	* the overloaded form is invoked and the project streams
	* the tuples through instead of the bag.
	*/
	public class POProject extends ExpressionOperator {

	/**
	*
	*/
	private static final long serialVersionUID = 1L;

	private boolean resultSingleTupleBag = false;

	//The column to project
	protected ArrayList<Integer> columns;

	//True if we are in the middle of streaming tuples
	//in a bag
	private boolean processingBagOfTuples = false;

	//The bag iterator used while straeming tuple
	private transient Iterator<Tuple> bagIterator = null;

	//Represents the fact that this instance of POProject
	//is overloaded to stream tuples in the bag rather
	//than passing the entire bag. It is the responsibility
	//of the translator to set this.
	protected boolean overloaded = false;


	protected boolean isProjectToEnd = false;
	protected int startCol;

	public POProject(OperatorKey k) {
	this(k,-1,0);
	}

	public POProject(OperatorKey k, int rp) {
	this(k, rp, 0);
	}

	public POProject(OperatorKey k, int rp, int col) {
	super(k, rp);
	columns = new ArrayList<Integer>();
	columns.add(col);
	}

	public POProject(OperatorKey k, int rp, ArrayList<Integer> cols) {
	super(k, rp);
	columns = cols;
	}

	public void setProjectToEnd(int startCol){
	this.isProjectToEnd = true;
	this.startCol = startCol;
	columns = new ArrayList<Integer>();
	}

	@Override
	public String name() {

	String str = "Project" + "[" + DataType.findTypeName(resultType) + "]";

	if(isStar()){
	str += "[*]";
	}else if(isProjectToEnd){
	str += "[" + startCol + " .. " + "]";
	}else{
	str += columns;
	}

	str += " - " + mKey.toString();
	return str;

	}

	@Override
	public boolean supportsMultipleInputs() {
	return false;
	}

	@Override
	public boolean supportsMultipleOutputs() {
	return false;
	}

	@Override
	public void visit(PhyPlanVisitor v) throws VisitorException {
	v.visitProject(this);
	}


	/**
	* Overridden since the attachment of the new input
	* should cause the old processing to end.
	*/
	@Override
	public void attachInput(Tuple t) {
	super.attachInput(t);
	processingBagOfTuples = false;
	}

	/**
	* Fetches the input tuple and returns the requested
	* column
	* @return next value.
	* @throws ExecException
	*/
	public Result getNext() throws ExecException{
	Result res = processInput();
	Tuple inpValue = (Tuple)res.result;

	Object ret;
	if(res.returnStatus != POStatus.STATUS_OK){
	return res;
	}
	if (isStar()) {
	illustratorMarkup(inpValue, res.result, -1);
	return res;
	} else if(columns.size() == 1) {
	if ( inpValue == null ) {
	// the tuple is null, so a dereference should also produce a null
	res.returnStatus = POStatus.STATUS_OK;
	ret = null;
	} else if( inpValue.size() > columns.get(0) ) {
	ret = inpValue.get(columns.get(0));
	} else {
	if(pigLogger != null) {
	pigLogger.warn(this,"Attempt to access field " +
	"which was not found in the input", PigWarning.ACCESSING_NON_EXISTENT_FIELD);
	}
	res.returnStatus = POStatus.STATUS_OK;
	ret = null;
	}
	} else if(isProjectToEnd){
	ret = getRangeTuple(inpValue);
	}
	else {
	ArrayList<Object> objList = new ArrayList<Object>(columns.size());

	for(int col : columns) {
	addColumn(objList, inpValue, col);
	}
	ret = mTupleFactory.newTupleNoCopy(objList);
	}
	res.result = ret;
	illustratorMarkup(inpValue, res.result, -1);
	return res;
	}

	private boolean isRangeInvalid(int lastColIdx) {
	if(startCol > lastColIdx){
	// this must be happening because tuple is smaller than startCol
	if(pigLogger != null) {
	pigLogger.warn(this, "Invalid range being projected," +
	" startCol postition" + startCol + " greater than tuple size",
	PigWarning.PROJECTION_INVALID_RANGE
	);
	}
	return true;
	}
	return false;
	}

	/**
	* Add i'th column from inpValue to objList
	* @param objList
	* @param inpValue
	* @param i
	* @throws ExecException
	*/
	private void addColumn(ArrayList<Object> objList, Tuple inpValue, int i)
	throws ExecException {
	if( inpValue == null ) {
	// the tuple is null, so a dereference should also produce a null
	objList.add(null);
	} else if( inpValue.size() > i ) {
	objList.add(inpValue.get(i));
	} else {
	if(pigLogger != null) {
	pigLogger.warn(this,"Attempt to access field " + i +
	" which was not found in the input", PigWarning.ACCESSING_NON_EXISTENT_FIELD);
	}
	objList.add(null);
	}
	}

	@Override
	public Result getNextDataBag() throws ExecException {

	Result res = processInputBag();
	if(res.returnStatus!=POStatus.STATUS_OK)
	return res;
	return(consumeInputBag(res));
	}

	/**
	* @param input
	* @throws ExecException
	*/
	protected Result consumeInputBag(Result input) throws ExecException {
	if(isInputAttached() \|\| isStar()){
	Result retVal = new Result();
	retVal.result = input.result;
	retVal.returnStatus = POStatus.STATUS_OK;
	detachInput();
	return retVal;
	}

	if (input.result instanceof DataBag) {
	DataBag inpBag = (DataBag) input.result;
	Result retVal = new Result();

	DataBag outBag;
	if(resultSingleTupleBag) {
	// we have only one tuple in a bag - so create
	// A SingleTupleBag for the result and fill it
	// appropriately from the input bag
	Tuple tuple = inpBag.iterator().next();
	if(!isProjectToEnd){
	ArrayList<Object> objList = new ArrayList<Object>(columns.size());
	for (int col : columns) {
	addColumn(objList, tuple, col);
	}
	outBag = new SingleTupleBag( mTupleFactory.newTupleNoCopy(objList) );
	}else {
	Tuple tmpTuple = getRangeTuple(tuple);
	outBag = new SingleTupleBag(tmpTuple);
	}
	} else {
	outBag = mBagFactory.newDefaultBag();
	for (Tuple tuple : inpBag) {
	if(!isProjectToEnd){
	ArrayList<Object> objList = new ArrayList<Object>(columns.size());
	for (int col : columns) {
	addColumn(objList, tuple, col);
	}
	outBag.add( mTupleFactory.newTupleNoCopy(objList) );
	}else{
	Tuple outTuple = getRangeTuple(tuple);
	outBag.add(outTuple);
	}
	}
	}
	retVal.result = outBag;
	retVal.returnStatus = POStatus.STATUS_OK;
	return retVal;
	} else if (input.result instanceof Tuple) {
	// if input is tuple, columns should only have one item
	Result retVal = new Result();
	retVal.result = ((Tuple)input.result).get(columns.get(0));
	retVal.returnStatus = POStatus.STATUS_OK;
	return retVal;
	} else if (input.result==null) {
	Result retVal = new Result();
	retVal.result = null;
	retVal.returnStatus = POStatus.STATUS_OK;
	return retVal;
	} else {
	throw new ExecException("Cannot dereference a bag from " + input.result.getClass().getName(), 1129);
	}
	}

	private Tuple getRangeTuple(Tuple tuple) throws ExecException {
	int lastColIdx = tuple.size() - 1;
	Tuple outTuple;
	if(isRangeInvalid(lastColIdx)){
	//invalid range - return empty tuple
	outTuple = mTupleFactory.newTuple();
	}
	else {
	ArrayList<Object> objList = new ArrayList<Object>(lastColIdx - startCol + 1);
	for(int i = startCol; i <= lastColIdx ; i++){
	addColumn(objList, tuple, i);
	}
	outTuple = mTupleFactory.newTupleNoCopy(objList);
	}
	return outTuple;
	}

	@Override
	public Result getNextDataByteArray() throws ExecException {
	return getNext();
	}

	@Override
	public Result getNextDouble() throws ExecException {
	return getNext();
	}

	@Override
	public Result getNextFloat() throws ExecException {
	return getNext();
	}

	@Override
	public Result getNextInteger() throws ExecException {
	return getNext();
	}

	@Override
	public Result getNextLong() throws ExecException {
	return getNext();
	}



	@Override
	public Result getNextBoolean() throws ExecException {
	return getNext();
	}

	@Override
	public Result getNextDateTime() throws ExecException {
	return getNext();
	}

	@Override
	public Result getNextMap() throws ExecException {
	return getNext();
	}

	@Override
	public Result getNextString() throws ExecException {
	return getNext();
	}

	@Override
	public Result getNextBigInteger() throws ExecException {
	return getNext();
	}

	@Override
	public Result getNextBigDecimal() throws ExecException {
	return getNext();
	}

	/**
	* Asked for Tuples. Check if the input is a bag.
	* If so, stream the tuples in the bag instead of
	* the entire bag.
	*/
	@Override
	public Result getNextTuple() throws ExecException {
	Result res = new Result();
	if(!processingBagOfTuples){
	Tuple inpValue = null;
	res = processInput();
	if(res.returnStatus!=POStatus.STATUS_OK)
	return res;
	if(isStar())
	return res;

	inpValue = (Tuple)res.result;
	res.result = null;

	Object ret;

	if(columns.size() == 1) {
	if( inpValue == null ) {
	// the tuple is null, so a dereference should also produce a null
	ret = null;
	} else if( inpValue.size() > columns.get(0) ) {
	ret = inpValue.get(columns.get(0));
	} else {
	if(pigLogger != null) {
	pigLogger.warn(this,"Attempt to access field " +
	"which was not found in the input", PigWarning.ACCESSING_NON_EXISTENT_FIELD);
	}
	ret = null;
	}
	} else if(isProjectToEnd) {
	ret = getRangeTuple(inpValue);
	} else {
	ArrayList<Object> objList = new ArrayList<Object>(columns.size());

	for(int col: columns) {
	if( inpValue == null ) {
	// the tuple is null, so a dereference should also produce a null
	objList.add(null);
	} else if( inpValue.size() > col ) {
	objList.add(inpValue.get(col));
	} else {
	if(pigLogger != null) {
	pigLogger.warn(this,"Attempt to access field " +
	"which was not found in the input", PigWarning.ACCESSING_NON_EXISTENT_FIELD);
	}
	objList.add(null);
	}
	}
	ret = mTupleFactory.newTuple(objList);
	res.result = (Tuple)ret;
	return res;
	}

	if(overloaded){
	if (ret!=null) {
	DataBag retBag = (DataBag)ret;
	bagIterator = retBag.iterator();
	if(bagIterator.hasNext()){
	processingBagOfTuples = true;
	res.result = bagIterator.next();
	}
	// If the bag contains no tuple, set the returnStatus to STATUS_EOP
	if (!processingBagOfTuples)
	res.returnStatus = POStatus.STATUS_EOP;
	} else {
	res.returnStatus = POStatus.STATUS_EOP;
	}
	}
	else {
	res.result = (Tuple)ret;
	}
	return res;
	}
	if(bagIterator.hasNext()){
	res.result = bagIterator.next();
	res.returnStatus = POStatus.STATUS_OK;
	return res;
	}
	else{
	//done processing the bag of tuples
	processingBagOfTuples = false;
	return getNextTuple();
	}
	}

	public ArrayList<Integer> getColumns(){
	if(isProjectToEnd) {
	throw new AssertionError("Internal error. Improper use of method getColumns() in "
	+ POProject.class.getSimpleName());
	}
	return columns;
	}

	public int getColumn() throws ExecException {
	if(columns.size() != 1 \|\| isProjectToEnd) {
	int errCode = 2068;
	String msg = "Internal error. Improper use of method getColumn() in "
	+ POProject.class.getSimpleName();
	throw new ExecException(msg, errCode, PigException.BUG);
	}
	return columns.get(0);
	}

	public int getStartCol(){
	return startCol;
	}

	public void setColumns(ArrayList<Integer> cols) {
	if(isProjectToEnd){
	throw new AssertionError("Columns should not be set for range projection");
	}
	this.columns = cols;
	}

	public void setColumn(int col) {
	isProjectToEnd = false;
	if(null == columns) {
	columns = new ArrayList<Integer>();
	} else {
	columns.clear();
	}
	columns.add(col);
	}

	public boolean isOverloaded() {
	return overloaded;
	}

	public void setOverloaded(boolean overloaded) {
	this.overloaded = overloaded;
	}

	public boolean isStar() {
	return isProjectToEnd && startCol == 0;
	}

	public boolean isProjectToEnd(){
	return isProjectToEnd;
	}

	public void setStar(boolean star) {
	if(star){
	isProjectToEnd = true;
	startCol = 0;
	}else{
	isProjectToEnd = false;
	}
	}

	@Override
	public POProject clone() throws CloneNotSupportedException {
	ArrayList<Integer> cols = new ArrayList<Integer>(columns.size());
	// Can resuse the same Integer objects, as they are immutable
	for (Integer i : columns) {
	cols.add(i);
	}
	POProject clone = new POProject(new OperatorKey(mKey.scope,
	NodeIdGenerator.getGenerator().getNextNodeId(mKey.scope)),
	requestedParallelism, cols);
	clone.cloneHelper(this);
	clone.overloaded = overloaded;
	clone.startCol = startCol;
	clone.isProjectToEnd = isProjectToEnd;
	clone.resultType = resultType;
	return clone;
	}

	protected Result processInputBag() throws ExecException {

	Result res = new Result();
	if (input==null && (inputs == null \|\| inputs.size()==0)) {
	// log.warn("No inputs found. Signaling End of Processing.");
	res.returnStatus = POStatus.STATUS_EOP;
	return res;
	}

	//Should be removed once the model is clear
	if(getReporter()!=null) {
	getReporter().progress();
	}

	if(!isInputAttached()) {
	if (inputs.get(0).getResultType()==DataType.BAG)
	return inputs.get(0).getNextDataBag();
	else
	return inputs.get(0).getNextTuple();
	}
	else{
	res.result = (DataBag)input.get(columns.get(0));
	res.returnStatus = POStatus.STATUS_OK;
	return res;
	}
	}

	public void setResultSingleTupleBag(boolean resultSingleTupleBag) {
	this.resultSingleTupleBag = resultSingleTupleBag;
	}

	@Override
	public List<ExpressionOperator> getChildExpressions() {
	return null;
	}

	@Override
	public Tuple illustratorMarkup(Object in, Object out, int eqClassIndex) {
	if(illustrator != null) {

	}
	return null;
	}
	}