import org.apache.pig.newplan.logical.relational.LogicalSchema; //导入依赖的package包/类
/**
 * This function translates the new LogicalSchema into old Schema format required
 * by PhysicalOperators
 * @param schema LogicalSchema to be converted to Schema
 * @return Schema that is converted from LogicalSchema
 * @throws FrontendException 
 */
public static Schema translateSchema(LogicalSchema schema) {       
    if (schema == null) {
        return null;
    }
    
    Schema s2 = new Schema();
    List<LogicalSchema.LogicalFieldSchema> ll = schema.getFields();
    for (LogicalSchema.LogicalFieldSchema f: ll) {
        Schema.FieldSchema f2 = null;
        try {
            f2 = new Schema.FieldSchema(f.alias, translateSchema(f.schema), f.type);
            f2.canonicalName = ((Long)f.uid).toString();
            s2.add(f2);
        } catch (FrontendException e) {
        }
    }
    
    return s2;
}
 

import org.apache.pig.newplan.logical.relational.LogicalSchema; //导入依赖的package包/类
@Test
public void testLoadEqualityDifferentNumFuncSpecCstorArgs() throws FrontendException {
    LogicalPlan lp = new LogicalPlan();

    LogicalSchema aschema1 = new LogicalSchema();
    aschema1.addField(new LogicalSchema.LogicalFieldSchema(
            "x", null, DataType.INTEGER));
    LOLoad load1 = newLOLoad(new FileSpec("/abc",
            new FuncSpec(DummyLoad.class.getName(), new String[] { "x", "y" })), aschema1, lp,
            conf);
    lp.add(load1);

    LOLoad load3 = newLOLoad(new FileSpec("/abc",
            new FuncSpec(DummyLoad.class.getName(), "x")), aschema1, lp, conf);
    lp.add(load3);

    assertFalse(load1.isEqual(load3));
}
 

import org.apache.pig.newplan.logical.relational.LogicalSchema; //导入依赖的package包/类
/**
 * This function translates the new LogicalSchema into old Schema format required
 * by PhysicalOperators
 * @param schema LogicalSchema to be converted to Schema
 * @return Schema that is converted from LogicalSchema
 * @throws FrontendException
 */
public static Schema translateSchema(LogicalSchema schema) {
    if (schema == null) {
        return null;
    }

    Schema s2 = new Schema();
    List<LogicalSchema.LogicalFieldSchema> ll = schema.getFields();
    for (LogicalSchema.LogicalFieldSchema f: ll) {
        Schema.FieldSchema f2 = null;
        try {
            f2 = new Schema.FieldSchema(f.alias, translateSchema(f.schema), f.type);
            f2.canonicalName = ((Long)f.uid).toString();
            s2.add(f2);
        } catch (FrontendException e) {
        }
    }

    return s2;
}
 

import org.apache.pig.newplan.logical.relational.LogicalSchema; //导入依赖的package包/类
/**
 * @return a dummy logical relational operator
 */
private LogicalRelationalOperator createDummyRelOpWithAlias() {
    class DummyRelOp extends LogicalRelationalOperator{
        DummyRelOp(){
            super("dummy", new LogicalPlan());
            this.alias = "dummy";
        }

        @Override
        public LogicalSchema getSchema() throws FrontendException {
                      return null;
        }

        @Override
        public void accept(PlanVisitor v) throws FrontendException {

        }

        @Override
        public boolean isEqual(Operator operator) throws FrontendException {
            return false;
        }

    }
    return new DummyRelOp();
}
 

import org.apache.pig.newplan.logical.relational.LogicalSchema; //导入依赖的package包/类
@SuppressWarnings("unchecked")
private void addForEachIfNecessary(LogicalRelationalOperator op) throws FrontendException {
    Set<Long> outputUids = (Set<Long>)op.getAnnotation(ColumnPruneHelper.OUTPUTUIDS);
    if (outputUids!=null) {
        LogicalSchema schema = op.getSchema();
        Set<Integer> columnsToDrop = new HashSet<Integer>();

        for (int i=0;i<schema.size();i++) {
            if (!outputUids.contains(schema.getField(i).uid))
                columnsToDrop.add(i);
        }

        if (!columnsToDrop.isEmpty()) {
            LOForEach foreach = Util.addForEachAfter((LogicalPlan)op.getPlan(), op, 0, columnsToDrop);
            foreach.getSchema();
        }
    }
}
 

import org.apache.pig.newplan.logical.relational.LogicalSchema; //导入依赖的package包/类
protected Set<Integer> getColumns(LogicalSchema schema, Set<Long> uids) throws FrontendException {
    if (schema == null) {
        throw new SchemaNotDefinedException("Schema is not defined.");
    }

    Set<Integer> cols = new HashSet<Integer>();
    Iterator<Long> iter = uids.iterator();
    while(iter.hasNext()) {
        long uid = iter.next();
        int index = schema.findField(uid);
        if (index == -1) {
            throw new FrontendException("UID " + uid + " is not found in the schema " + schema, 2241);
        }

        cols.add(index);
    }

    return cols;
}
 

import org.apache.pig.newplan.logical.relational.LogicalSchema; //导入依赖的package包/类
@Override
public void visit(LODistinct distinct) throws FrontendException {
    setOutputUids(distinct);
    
    Set<Long> input = new HashSet<Long>();

    // Every field is required
    LogicalSchema s = distinct.getSchema();
    if (s == null) {
        throw new SchemaNotDefinedException("Schema for " + distinct.getName() + " is not defined.");
    }

    for(int i=0; i<s.size(); i++) {
        input.add(s.getField(i).uid);
    }
    distinct.annotate(INPUTUIDS, input);
}
 

import org.apache.pig.newplan.logical.relational.LogicalSchema; //导入依赖的package包/类
private void collectUids(LogicalRelationalOperator currentOp, LogicalExpressionPlan exp, Set<Long> uids) throws FrontendException {
    List<Operator> ll = exp.getSinks();
    for(Operator op: ll) {
        if (op instanceof ProjectExpression) {
            if (!((ProjectExpression)op).isRangeOrStarProject()) {
                long uid = ((ProjectExpression)op).getFieldSchema().uid;
                uids.add(uid);
            } else {
                LogicalRelationalOperator ref = ((ProjectExpression)op).findReferent();
                LogicalSchema s = ref.getSchema();
                if (s == null) {
                    throw new SchemaNotDefinedException("Schema not defined for " + ref.getAlias());
                }
                for(LogicalFieldSchema f: s.getFields()) {
                    uids.add(f.uid);
                }
            }
        }
    }
}
 

import org.apache.pig.newplan.logical.relational.LogicalSchema; //导入依赖的package包/类
@Test
public void testNewMergeNullSchemas() throws Throwable {
    LogicalSchema a = Utils.parseSchema( "a1:bytearray, b1:(b11:int, b12:float), c1:long" );
    LogicalSchema b = Utils.parseSchema( "a2:bytearray, b2:(), c2:int" );

    LogicalSchema mergedSchema = LogicalSchema.merge(a, b, LogicalSchema.MergeMode.Union);
    LogicalSchema expected = Utils.parseSchema( "a1:bytearray, b1:(), c1:long" );
    assertTrue(LogicalSchema.equals(mergedSchema, expected, false, false));

    mergedSchema = LogicalSchema.merge(a, b, LogicalSchema.MergeMode.LoadForEach);
    expected = Utils.parseSchema( "a1:bytearray, b1:(b11:int, b12:float), c1:long" );
    assertTrue(LogicalSchema.equals(mergedSchema, expected, false, false));

    mergedSchema = LogicalSchema.merge(b, a, LogicalSchema.MergeMode.LoadForEach);
    expected = Utils.parseSchema( "a2:bytearray, b2:(b11:int,b12:float), c2:int" );
    assertTrue(LogicalSchema.equals(mergedSchema, expected, false, false));
}
 

import org.apache.pig.newplan.logical.relational.LogicalSchema; //导入依赖的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.newplan.logical.relational.LogicalSchema; //导入依赖的package包/类
private List<Integer> translateAliasToPos(LogicalSchema schema, List<Object> rawColumns) throws FrontendException {
    List<Integer> columns = new ArrayList<Integer>();
    for( Object rawColumn : rawColumns ) {
        if( rawColumn instanceof Integer ) {
        	if (schema!=null && ((Integer)rawColumn>=schema.size() || (Integer)rawColumn<0)) {
        	    throw new FrontendException("Index "+rawColumn + " out of range in schema:" + schema.toString(false), 1127);
        	}
            columns.add( (Integer)rawColumn );
        } else {
            int pos = schema.getFieldPosition((String)rawColumn);
            if( pos != -1) {
                columns.add( pos );
                continue;
            } else {
                throw new FrontendException("Cannot find field " + rawColumn + " in " + schema.toString(false), 1128);
            }
        }
    }
    return columns;
}
 

import org.apache.pig.newplan.logical.relational.LogicalSchema; //导入依赖的package包/类
@Test
public void testRelationalEqualityOneNullOneNotNullSchema() throws FrontendException {
    LogicalPlan lp = new LogicalPlan();
    LogicalSchema aschema1 = new LogicalSchema();
    aschema1.addField(new LogicalSchema.LogicalFieldSchema(
            "x", null, DataType.INTEGER));
    LOLoad load1 = newLOLoad(new FileSpec("/abc",
            new FuncSpec(DummyLoad.class.getName(), new String[] { "x", "y" })), aschema1, lp,
            conf);
    lp.add(load1);

    // Test that one with schema and one without breaks equality
    LOLoad load9 = newLOLoad(new FileSpec("/abc",
            new FuncSpec(DummyLoad.class.getName(), new String[] { "x", "z" })), null, lp, conf);
    lp.add(load9);

    assertFalse(load1.isEqual(load9));
}
 

import org.apache.pig.newplan.logical.relational.LogicalSchema; //导入依赖的package包/类
private void readBaseData(List<Operator> loads) throws IOException, InterruptedException, FrontendException, ExecException {
    PhysicalPlan thisPhyPlan = new PhysicalPlan();
    for (Operator op : loads) {
        LogicalSchema schema = ((LOLoad) op).getSchema();
        if(schema == null) {
            throw new ExecException("Example Generator requires a schema. Please provide a schema while loading data.");
        }
        poLoadToSchemaMap.put((POLoad)logToPhyMap.get(op), schema);
        thisPhyPlan.add(logToPhyMap.get(op));
    }
    baseData = null;
    Map<Operator, DataBag> result = getData(thisPhyPlan);
    baseData = new HashMap<LOLoad, DataBag>();
    for (Operator lo : result.keySet()) {
        if (lo instanceof LOLoad) {
            baseData.put((LOLoad) lo, result.get(lo));
        }
    }
}
 

import org.apache.pig.newplan.logical.relational.LogicalSchema; //导入依赖的package包/类
public LogicalFieldSchema getFieldSchema() throws FrontendException {
    if (fieldSchema!=null)
        return fieldSchema;
    LogicalExpression successor = (LogicalExpression)plan.getSuccessors(this).get(0);
    LogicalFieldSchema predFS = successor.getFieldSchema();
    if (predFS!=null) {
        if (predFS.type==DataType.MAP && predFS.schema!=null) {
            return (predFS.schema.getField(0));
        }
        else {
            fieldSchema = new LogicalSchema.LogicalFieldSchema(null, null, DataType.BYTEARRAY);
            uidOnlyFieldSchema = fieldSchema.mergeUid(uidOnlyFieldSchema);
            return fieldSchema;
        }
    }
    return null;
}
 

import org.apache.pig.newplan.logical.relational.LogicalSchema; //导入依赖的package包/类
/**
 * Only for use by Pig internal code.
 * Construct a ResourceSchema from a {@link LogicalSchema}
 * @param pigSchema LogicalSchema to use
 * @param sortInfo information on how data is sorted
 */
@InterfaceAudience.Private
public ResourceSchema(LogicalSchema pigSchema, SortInfo sortInfo) {
    this(pigSchema);
    if (sortInfo!=null && sortInfo.getSortColInfoList().size()!=0) {
        sortKeys = new int[sortInfo.getSortColInfoList().size()];
        sortKeyOrders = new Order[sortInfo.getSortColInfoList().size()];
        for (int i=0;i<sortInfo.getSortColInfoList().size();i++) {
            SortColInfo colInfo = sortInfo.getSortColInfoList().get(i); 
            int index = colInfo.getColIndex();
            Order order;
            org.apache.pig.SortColInfo.Order origOrder = colInfo.getSortOrder();
            if (origOrder==org.apache.pig.SortColInfo.Order.ASCENDING) {
                order = Order.ASCENDING;
            } else {
                order = Order.DESCENDING;
            }
            sortKeys[i] = index;
            sortKeyOrders[i] = order;
        }
    }
}
 

import org.apache.pig.newplan.logical.relational.LogicalSchema; //导入依赖的package包/类
@Test
public void testEmptyBagInnerPlan() throws Exception {
    String query = "A = load 'x' as ( u:bag{} );" +
            "B = foreach A { B1 = filter u by $1==0; generate B1;};" +
            "store B into '111';";

    LogicalPlan lp = Util.parseAndPreprocess(query, pc);
    Util.optimizeNewLP(lp);
    LOStore loStore = (LOStore)lp.getSinks().get(0);
    LOForEach loForEach = (LOForEach)lp.getPredecessors(loStore).get(0);
    LogicalSchema schema = loForEach.getSchema();
    assertEquals(1, schema.size());
    LogicalFieldSchema bagFieldSchema = schema.getField(0);
    assertEquals(DataType.BAG, bagFieldSchema.type);
    LogicalFieldSchema tupleFieldSchema = bagFieldSchema.schema.getField(0);
    assertNull(tupleFieldSchema.schema);
}
 

import org.apache.pig.newplan.logical.relational.LogicalSchema; //导入依赖的package包/类
@Test
public void testNewNormalNestedMerge1() throws Exception {
    LogicalSchema a = org.apache.pig.newplan.logical.Util.translateSchema(Utils.getSchemaFromString(
        "a1:bytearray, b1:(b11:int, b12:float), c1:long"));
    LogicalSchema b = org.apache.pig.newplan.logical.Util.translateSchema(Utils.getSchemaFromString(
        "a2:bytearray, b2:(b21:double, b22:long), c2:int"));

    LogicalSchema mergedSchema = LogicalSchema.merge(a, b, LogicalSchema.MergeMode.Union);
    LogicalSchema expected = org.apache.pig.newplan.logical.Util.translateSchema(Utils.getSchemaFromString(
        "a1:bytearray, b1:(), c1:long"));
    expected.getField(1).schema = new LogicalSchema();
    assertTrue(LogicalSchema.equals(mergedSchema, expected, false, false));

    mergedSchema = LogicalSchema.merge(a, b, LogicalSchema.MergeMode.LoadForEach);
    expected = org.apache.pig.newplan.logical.Util.translateSchema(Utils.getSchemaFromString(
        "a1:bytearray, b1:(b11:int, b12:float), c1:long"));
    assertTrue(LogicalSchema.equals(mergedSchema, expected, false, false));

    mergedSchema = LogicalSchema.merge(b, a, LogicalSchema.MergeMode.LoadForEach);
    expected = org.apache.pig.newplan.logical.Util.translateSchema(Utils.getSchemaFromString(
        "a2:bytearray, b2:(b21:double, b22:long), c2:int"));
    assertTrue(LogicalSchema.equals(mergedSchema, expected, false, false));
}
 

import org.apache.pig.newplan.logical.relational.LogicalSchema; //导入依赖的package包/类
public ExampleGenerator(LogicalPlan plan, PigContext hadoopPigContext) {
        this.plan = plan;
//        pigContext = new PigContext(ExecType.LOCAL, hadoopPigContext
//                .getProperties());
        pigContext = hadoopPigContext;
        // pigContext.setExecType(ExecType.LOCAL);
        FileLocalizer.setInitialized(false);
        try {
            pigContext.connect();
        } catch (ExecException e) {
            log.error("Error connecting to the cluster "
                    + e.getLocalizedMessage());

        }
        execEngine = new HExecutionEngine(pigContext);
        localMRRunner = new LocalMapReduceSimulator();
        poLoadToSchemaMap = new HashMap<POLoad, LogicalSchema>();
    }
 

import org.apache.pig.newplan.logical.relational.LogicalSchema; //导入依赖的package包/类
private boolean inInput(Tuple newTuple, DataBag input, LogicalSchema schema) throws ExecException {
    boolean result;
    for (Iterator<Tuple> iter = input.iterator(); iter.hasNext();) {
        result = true;
        Tuple tmp = iter.next();
        for (int i = 0; i < schema.size(); ++i)
            if (!newTuple.get(i).equals(tmp.get(i)))
            {
                result = false;
                break;
            }
        if (result)
            return true;
    }
    return false;
}
 

import org.apache.pig.newplan.logical.relational.LogicalSchema; //导入依赖的package包/类
@Test
public void testLoadEqualityDifferentFunctionNames() throws FrontendException {
    LogicalPlan lp = new LogicalPlan();

    LogicalSchema aschema1 = new LogicalSchema();
    aschema1.addField(new LogicalSchema.LogicalFieldSchema(
            "x", null, DataType.INTEGER));
    LOLoad load1 = newLOLoad(new FileSpec("/abc",
            new FuncSpec(DummyLoad.class.getName(), new String[] { "x", "y" })), aschema1, lp,
            conf);
    lp.add(load1);

    // Different function names in FuncSpec
    LOLoad load4 = newLOLoad(new FileSpec("/abc",
            new FuncSpec(DummyLoad.class.getName(), new String[] { "x", "z" })), aschema1, lp,
            conf);
    lp.add(load4);

    assertFalse(load1.isEqual(load4));
}