import org.apache.pig.impl.util.Pair; //导入依赖的package包/类
@Override
public void visitSort(POSort op) throws VisitorException {
    try{
        FileSpec fSpec = getTempFileSpec();
        MapReduceOper mro = endSingleInputPlanWithStr(fSpec);
        FileSpec quantFile = getTempFileSpec();
        int rp = op.getRequestedParallelism();
        Pair<POProject, Byte>[] fields = getSortCols(op.getSortPlans());
        Pair<MapReduceOper, Integer> quantJobParallelismPair =
            getQuantileJob(op, mro, fSpec, quantFile, rp);
        curMROp = getSortJob(op, quantJobParallelismPair.first, fSpec, quantFile,
                quantJobParallelismPair.second, fields);

        if(op.isUDFComparatorUsed){
            curMROp.UDFs.add(op.getMSortFunc().getFuncSpec().toString());
            curMROp.isUDFComparatorUsed = true;
        }
        phyToMROpMap.put(op, curMROp);
    }catch(Exception e){
        int errCode = 2034;
        String msg = "Error compiling operator " + op.getClass().getSimpleName();
        throw new MRCompilerException(msg, errCode, PigException.BUG, e);
    }
}
 

import org.apache.pig.impl.util.Pair; //导入依赖的package包/类
@SuppressWarnings("unchecked")
private Comparable<Object> getFirstKeyOfNextSplit(final int curSplitIdx, final List<Pair<Integer,Tuple>> index) throws IOException{

    // First find out the index entry corresponding to our current split.
    int i;
    for(i=0; i < index.size(); i++){
        if(index.get(i).first.equals(curSplitIdx))
            break;
    }

    // Now read key of the very next index entry.
    if(i < index.size()-1){
        Tuple keyTuple =  index.get(i+1).second;
        return keyTuple.size() == 1 ? (Comparable<Object>)keyTuple.get(0) : keyTuple;
    }

    // If we are here it implies, current split is the last split.
    return null;


}
 

import org.apache.pig.impl.util.Pair; //导入依赖的package包/类
/**
 * Disconnect two operators in the plan.
 * @param from Operator edge is coming from
 * @param to Operator edge is going to
 * @return pair of positions, indicating the position in the from and
 * to arrays.
 * @throws FrontendException if the two operators aren't connected.
 */
public Pair<Integer, Integer> disconnect(Operator from,
                                         Operator to) throws FrontendException {
    Pair<Operator, Integer> f = fromEdges.removeWithPosition(from, to);
    if (f == null) { 
        throw new FrontendException("Attempt to disconnect operators " + 
            from.getName() + " and " + to.getName() +
            " which are not connected.", 2219);
    }
    
    Pair<Operator, Integer> t = toEdges.removeWithPosition(to, from);
    if (t == null) { 
        throw new FrontendException("Plan in inconssistent state " + 
            from.getName() + " and " + to.getName() +
            " connected in fromEdges but not toEdges.", 2220);
    }
    
    markDirty();
    return new Pair<Integer, Integer>(f.second, t.second);
}
 

import org.apache.pig.impl.util.Pair; //导入依赖的package包/类
/**
 * Merge a MapKeysInfo structure to existing required map keys list
 * @param input
 *            The input of the field to merge
 * @param column
 *            The column of the field to merge
 * @param mapKeysInfo
 *            The MapKeysInfo structure to merge
 */
public void mergeMapKeysInfo(int input, int column, MapKeysInfo mapKeysInfo)
{
    if (mapKeysInfo == null)
        return;
    int index = mFields.indexOf(new Pair<Integer, Integer>(input, column));
    if (index==-1)
        return;
    if (mMapKeysInfoList.get(index)==null)
    {
        mMapKeysInfoList.set(index, mapKeysInfo);
        return;
    }
    if (mapKeysInfo.needAllKeys)
    {
        mMapKeysInfoList.get(index).needAllKeys = true;
        mMapKeysInfoList.get(index).keys = null;
    }
    if (mapKeysInfo.keys!=null)
    {
        for (String key : mapKeysInfo.keys)
            mergeMapKey(input, column, key);
    }
}
 

import org.apache.pig.impl.util.Pair; //导入依赖的package包/类
private void reset(Operator newOp, Operator oldOp)
                throws FrontendException {
    List<Operator> p = plan.getPredecessors(oldOp);
    if (p != null) {
        Operator[] preds = p.toArray(new Operator[0]);
        for (Operator pred : preds) {
            Pair<Integer, Integer> pos = plan.disconnect(pred, oldOp);
            plan.connect(pred, pos.first, newOp, pos.second);
        }
    }
    List<Operator> s = plan.getSuccessors(oldOp);
    if (s != null) {
        Operator[] sucs = s.toArray(new Operator[0]);
        for (Operator suc : sucs) {
            plan.disconnect(oldOp, suc);
        }
    }
    plan.remove(oldOp);
}
 

import org.apache.pig.impl.util.Pair; //导入依赖的package包/类
/**
 * checks if a relational operator contains all of the specified uids
 * @param op LogicalRelational operator that should contain the uid
 * @param uids Uids to check for
 * @return true if given LogicalRelationalOperator has all the given uids
 */
private boolean hasAll(LogicalRelationalOperator op, Pair<List<Long>,
        List<Byte>> uidWithTypes) throws FrontendException {
    LogicalSchema schema = op.getSchema();

    if (schema==null)
        return false;

    List<Long> uids = uidWithTypes.first;
    List<Byte> types = uidWithTypes.second;

    for (int i=0;i<uids.size();i++) {
        boolean found = false;
        for (LogicalSchema.LogicalFieldSchema fs : schema.getFields()) {
            if (fs.uid==uids.get(i) && fs.type==types.get(i))
                found = true;
        }
        if (!found)
            return false;
    }
    return true;
}
 

import org.apache.pig.impl.util.Pair; //导入依赖的package包/类
/**
 * Moves all the results of a collection of MR jobs to the final
 * output directory. Some of the results may have been put into a
 * temp location to work around restrictions with multiple output
 * from a single map reduce job.
 *
 * This method should always be called after the job execution
 * completes.
 */
public void moveResults(List<Job> completedJobs) throws IOException {
    for (Job job: completedJobs) {
        Pair<List<POStore>, Path> pair = jobStoreMap.get(job);
        if (pair != null && pair.second != null) {
            Path tmp = pair.second;
            Path abs = new Path(tmp, "abs");
            Path rel = new Path(tmp, "rel");
            FileSystem fs = tmp.getFileSystem(conf);

            if (fs.exists(abs)) {
                moveResults(abs, abs.toUri().getPath(), fs);
            }

            if (fs.exists(rel)) {
                moveResults(rel, rel.toUri().getPath()+"/", fs);
            }
        }
    }
}
 

import org.apache.pig.impl.util.Pair; //导入依赖的package包/类
@Override
public void visitFilter(POFilter fl) throws VisitorException {
    PhysicalPlan filterPlan = fl.getPlan();
    if (filterPlan.size() == 1) {
        PhysicalOperator op = filterPlan.getRoots().get(0);
        if (op instanceof ConstantExpression) {
            ConstantExpression exp = (ConstantExpression)op;
            Object value = exp.getValue();
            if (value instanceof Boolean) {
                Boolean filterValue = (Boolean)value;
                if (filterValue) {
                    removalQ.add(new Pair<POFilter, PhysicalPlan>(fl, mCurrentWalker.getPlan()));
                }
            }
        }
    }
}
 

import org.apache.pig.impl.util.Pair; //导入依赖的package包/类
/**
 * 
 * @param fields
 *            the list of input columns that are required
 * @param needAllFields
 *            to indicate if this required fields needs all the fields from
 *            its input; cannot be true if needNoFields is true
 * @param needNoFields
 *            to indicate if this required fields needs no fields from
 *            its input; cannot be true if needAllFields is true
 */
private RequiredFields(List<Pair<Integer, Integer>> fields,
        boolean needAllFields,
        boolean needNoFields) {
    mFields = fields;
    if(needAllFields && needNoFields) {
        //both cannot be true
        //set both of them to false
        mNeedAllFields = false;
        mNeedNoFields = false;
    } else {
        mNeedAllFields = needAllFields;
        mNeedNoFields = needNoFields;
    }
    if (mFields!=null)
    {
        mMapKeysInfoList = new ArrayList<MapKeysInfo>();
        for (int i=0;i<mFields.size();i++)
            mMapKeysInfoList.add(null);
    }
}
 

import org.apache.pig.impl.util.Pair; //导入依赖的package包/类
/**
 * @param conf
 * @param mapStores
 * @return
 * @throws IOException 
 */
@SuppressWarnings("unchecked")
private List<Pair<OutputCommitter, POStore>> getCommitters(
        TaskAttemptContext context,
        List<POStore> stores) throws IOException {
    List<Pair<OutputCommitter, POStore>> committers = 
        new ArrayList<Pair<OutputCommitter,POStore>>();
    for (POStore store : stores) {
        StoreFuncInterface sFunc = store.getStoreFunc();
        
        TaskAttemptContext updatedContext = setUpContext(context, store);
        try {
            committers.add(new Pair<OutputCommitter, POStore>(
                    sFunc.getOutputFormat().getOutputCommitter(
                            updatedContext), store));
        } catch (InterruptedException e) {
            throw new IOException(e);
        }
    }
    return committers;
    
}
 

import org.apache.pig.impl.util.Pair; //导入依赖的package包/类
@Override
public void visitSort(POSort op) throws VisitorException {
    try{
        FileSpec fSpec = getTempFileSpec();
        MapReduceOper mro = endSingleInputPlanWithStr(fSpec);
        FileSpec quantFile = getTempFileSpec();
        int rp = op.getRequestedParallelism();
        Pair<POProject, Byte>[] fields = getSortCols(op.getSortPlans());
        Pair<MapReduceOper, Integer> quantJobParallelismPair = 
            getQuantileJob(op, mro, fSpec, quantFile, rp);
        curMROp = getSortJob(op, quantJobParallelismPair.first, fSpec, quantFile, 
                quantJobParallelismPair.second, fields);
        
        if(op.isUDFComparatorUsed){
            curMROp.UDFs.add(op.getMSortFunc().getFuncSpec().toString());
            curMROp.isUDFComparatorUsed = true;
        }
        phyToMROpMap.put(op, curMROp);
    }catch(Exception e){
        int errCode = 2034;
        String msg = "Error compiling operator " + op.getClass().getSimpleName();
        throw new MRCompilerException(msg, errCode, PigException.BUG, e);
    }
}
 

import org.apache.pig.impl.util.Pair; //导入依赖的package包/类
@Override
public Pair<Writable, List<Writable>> separate(List<Integer[]> bins) {
	MapIdxWritable def = null;
	List<MapIdxWritable> ret = new ArrayList<MapIdxWritable>(bins.size());
	
	for (Integer[] bin : bins) {
		ret.add(null);
	}
	
	for (Entry<Writable, Writable> ent : entrySet()) {
		IPigIdxState wrapped = (IPigIdxState) ent.getValue();
		
		Pair<Writable, List<Writable>> res = wrapped.separate(bins);
		// Set the appropriate bins in this return.
		if (res.first != null) {
			if (def == null) {
				def = new CombineWrapperState();
			}
			def.put(ent.getKey(), res.first);					
		}

		for(int i = 0; i < res.second.size(); i++) {
			if (res.second.get(i) != null) {
				if (ret.get(i) == null) {
					ret.set(i, new CombineWrapperState());
				}
				ret.get(i).put(ent.getKey(), res.second.get(i));
			}
		}
	}
	
	return new Pair(def, ret);
}
 

import org.apache.pig.impl.util.Pair; //导入依赖的package包/类
public static List<Pair<List<NullableTuple>, List<NullableTuple>>> getTupleBatches(
		CombineWrapperState cws) {
	
	IPigIdxState state = (IPigIdxState) cws.get(CUR);
	if (state == null) {
		throw new RuntimeException("No current state in CombineWrapperState: " + cws);
	}

	return state.getTupleBatches((IPigIdxState) cws.get(LAST));
}
 

import org.apache.pig.impl.util.Pair; //导入依赖的package包/类
static Pair<List<NullableTuple>, WindowBundle<NullableTuple>> extractStates(WindowCombineState state) {
	if (state == null) {
		return new Pair<List<NullableTuple>, WindowBundle<NullableTuple>>(null, null);
	}
	
	List<NullableTuple> fixed = new ArrayList<NullableTuple>();
	WindowBundle<NullableTuple> win = null;
	
	for (Entry<Writable, Writable> ent : state.entrySet()) {
		if (ent.getKey().getClass().isAssignableFrom(IntWritable.class)) {
			// This is a windowed element.
			if (win != null) {
				// If we have multiple windows on a particular node, I'm not sure how we would
				// sequence things together given the current implementation.
				throw new RuntimeException("ERROR: Only one windowed element per group is currently supported.");
			}
			
			win = (WindowBundle<NullableTuple>) ent.getValue();
		} else {
			// This is a counted element.
			int c = ((IntWritable) ent.getValue()).get();
			NullableTuple v = (NullableTuple) ent.getKey();
			for (int i = 0; i < c; i++) {
				fixed.add(v);
			}
		}
	}
	
	return new Pair<List<NullableTuple>, WindowBundle<NullableTuple>>(fixed, win);
}
 

import org.apache.pig.impl.util.Pair; //导入依赖的package包/类
@Override
		public Pair<Writable, List<Writable>> separate(List<Integer[]> bins) {
			MapIdxWritable def = null; // = new TriBasicPersistState();
			List<MapIdxWritable> ret = new ArrayList<MapIdxWritable>(bins.size());
			HashMap<Integer, Integer> idxMap = new HashMap<Integer, Integer>();
			
			for (Integer[] bin : bins) {
//				MapIdxWritable st = new TriBasicPersistState();
				
				for (Integer i : bin) {
					idxMap.put(i, ret.size());
				}
				ret.add(null);
			}
			
			for (Entry<Writable, Writable> ent : entrySet()) {
				NullableTuple v = (NullableTuple) ent.getKey();
				int idx = v.getIndex();

				// Look up the appropriate state.
				Integer mappedIdx = idxMap.get(idx);
				MapIdxWritable mapped;
				if (mappedIdx == null) {
					if (def == null) {
						def = new TriBasicPersistState();
					}
					mapped = def;
				} else {
					mapped = ret.get(mappedIdx);
					if (mapped == null) {
						mapped = new TriBasicPersistState();
						ret.set(mappedIdx, mapped);
					}
				}
				mapped.put(ent.getKey(), ent.getValue());
			}
			
			return new Pair(def, ret);
		}
 

import org.apache.pig.impl.util.Pair; //导入依赖的package包/类
/**
 * Get all uids from Projections of this FilterOperator
 * @param filter
 * @return Set of uid
 */
private Pair<List<Long>, List<Byte>> getFilterProjectionUids(LOFilter filter) throws FrontendException {
    List<Long> uids = new ArrayList<Long>();
    List<Byte> types = new ArrayList<Byte>();
    if( filter != null ) {
        LogicalExpressionPlan filterPlan = filter.getFilterPlan();
        Iterator<Operator> iter = filterPlan.getOperators();
        Operator op = null;
        while( iter.hasNext() ) {
            op = iter.next();
            if( op instanceof ProjectExpression ) {
                ProjectExpression proj = (ProjectExpression)op;
                if( proj.isProjectStar() ) {
                    //project-range is always expanded when schema is
                    //available, so nothing to do here for it
                    LogicalRelationalOperator pred = (LogicalRelationalOperator)filter.getPlan().getPredecessors(filter).get(0);
                    LogicalSchema predSchema = pred.getSchema();
                    if (predSchema!=null) {
                        for (int i=0;i<predSchema.size();i++) {
                            uids.add(predSchema.getField(i).uid);
                            types.add(predSchema.getField(i).type);
                        }
                    }
                } else {
                    uids.add(proj.getFieldSchema().uid);
                    types.add(proj.getFieldSchema().type);
                }
            }
        }

    }

    Pair<List<Long>, List<Byte>> result = new Pair<List<Long>, List<Byte>>(uids, types);
    return result;
}
 

import org.apache.pig.impl.util.Pair; //导入依赖的package包/类
/**
 * Remove one value from an existing key and return which position in
 * the arraylist the value was at..  If that is the last value
 * for the key, then remove the key too.
 * @param key Key to remove the value from.
 * @param value Value to remove.
 * @return A pair containing the value being removed and an integer
 * indicating the position, or null if the key or value does
 * not exist.  Positions are zero based.
 */
public Pair<Operator, Integer> removeWithPosition(Operator key,
                                                  Operator value) {
    ArrayList<Operator> list = mMap.get(key);
    if (list == null) return null;

    int index = -1;
    Iterator<Operator> i = list.iterator();
    Operator keeper = null;
    for (int j = 0; i.hasNext(); j++) {
        keeper = i.next();
        //if (keeper.equals(value)) {
        if (keeper == value) {
            i.remove();
            index = j;
            break;
        }
    }
    
    if (index == -1) return null;

    if (list.size() == 0) {
        mMap.remove(key);
    }

    return new Pair<Operator, Integer>(keeper, index);
}
 

import org.apache.pig.impl.util.Pair; //导入依赖的package包/类
@Override
public void insertBetween(Operator pred, Operator operatorToInsert, Operator succ) throws FrontendException {
    add(operatorToInsert);
    Pair<Integer, Integer> pos = disconnect(pred, succ);
    connect(pred, pos.first, operatorToInsert, 0);
    connect(operatorToInsert, 0, succ, pos.second);
}
 

import org.apache.pig.impl.util.Pair; //导入依赖的package包/类
private void replace(Operator oldOp, Operator newOp) throws FrontendException {
	List<Operator> grandParents = plan.getPredecessors( oldOp );
	if( grandParents == null || grandParents.size() == 0 ) {
		plan.remove( oldOp );
		return;
	}
	Operator grandParent = plan.getPredecessors( oldOp ).get( 0 );
	Pair<Integer, Integer> pair = plan.disconnect( grandParent, oldOp );
	plan.add( newOp );
	plan.connect( grandParent, pair.first, newOp, pair.second );
	plan.remove( oldOp );
}
 

import org.apache.pig.impl.util.Pair; //导入依赖的package包/类
protected DataBag constructPROutput(List<Result> resLst, Tuple value) throws ExecException{
    Tuple t = super.constructLROutput(resLst, null, value);

    //Construct key
    Object key = t.get(1);

    // Construct an output bag and feed in the tuples
    DataBag opBag = mBagFactory.newDefaultBag();

    //Put the index, key, and value
    //in a tuple and return
    Pair <Integer, Integer> indexes = reducerMap.get(key);    // first -> min, second ->max

    // For non skewed keys, we set the partition index to be -1
    if (indexes == null) {
        indexes = new Pair <Integer, Integer>(-1,0);
    }

    for (Integer reducerIdx=indexes.first, cnt=0; cnt <= indexes.second; reducerIdx++, cnt++) {
        if (reducerIdx >= totalReducers) {
            reducerIdx = 0;
        }
        Tuple opTuple = mTupleFactory.newTuple(4);
        opTuple.set(0, t.get(0));
        // set the partition index
        opTuple.set(1, reducerIdx.intValue());
        opTuple.set(2, key);
        opTuple.set(3, t.get(2));

        opBag.add(opTuple);
    }

    return opBag;
}