hasMultipleKeys = true;
break;
}
}
LogicalFieldSchema groupKeySchema = null;
// Generate the groupField Schema
if( hasMultipleKeys ) {
LogicalSchema keySchema = new LogicalSchema();
// We sort here to maintain the correct order of inputs
for( Integer key : mExpressionPlans.keySet()) {
Collection<LogicalExpressionPlan> plans =
mExpressionPlans.get(key);
for( LogicalExpressionPlan plan : plans ) {
LogicalFieldSchema fieldSchema = getPlanSchema(plan);
// if any plan schema is null, that means we can't calculate
// further schemas so we bail out
if( fieldSchema == null ) {
schema = null;
return schema;
}
fieldSchema = new LogicalFieldSchema(fieldSchema);
keySchema.addField(fieldSchema);
}
// We only need fields from one input and not all
break;
}
groupKeySchema = new LogicalFieldSchema(GROUP_COL_NAME, keySchema, DataType.TUPLE);
} else {
// We sort here to maintain the correct order of inputs
for( Integer key : mExpressionPlans.keySet() ) {
Collection<LogicalExpressionPlan> plans = mExpressionPlans.get(key);
for( LogicalExpressionPlan plan : plans ) {
groupKeySchema = getPlanSchema(plan);
// if any plan schema is null, that means we cannot figure out
// the arity of keys, just give an empty tuple
if( groupKeySchema == null ) {
groupKeySchema = new LogicalSchema.LogicalFieldSchema("group", null, DataType.TUPLE);
break;
}
groupKeySchema = new LogicalSchema.LogicalFieldSchema(groupKeySchema);
// Change the uid of this field
groupKeySchema.alias = GROUP_COL_NAME;
break;
}
break;
}
}
if(mExpressionPlans.size() > 1){
//reset the uid, because the group column is associated with more
// than one input
groupKeySchema.resetUid();
}
if (groupKeySchema==null) {
throw new FrontendException(this, "Cannot get group key schema for " + this, 2234);
}
groupKeyUidOnlySchema = groupKeySchema.mergeUid(groupKeyUidOnlySchema);
fieldSchemaList.add( groupKeySchema );
// Generate the Bag Schema
int counter = 0;
for (Operator op : inputs) {
LogicalSchema inputSchema = ((LogicalRelationalOperator)op).getSchema();
// Check if we already have calculated Uid for this bag for given
// input operator
long bagUid;
if (generatedInputUids.get(counter)!=null)
bagUid = generatedInputUids.get(counter);
else {
bagUid = LogicalExpression.getNextUid();
generatedInputUids.put( counter, bagUid );
}
LogicalFieldSchema newTupleFieldSchema = new LogicalFieldSchema(
null, inputSchema, DataType.TUPLE, LogicalExpression.getNextUid());
LogicalSchema bagSchema = new LogicalSchema();
bagSchema.addField(newTupleFieldSchema);
LogicalFieldSchema newBagFieldSchema = new LogicalFieldSchema(
((LogicalRelationalOperator)op).getAlias(), bagSchema,
DataType.BAG, bagUid);
fieldSchemaList.add( newBagFieldSchema );
counter ++;