DerivedDataVisitor visitor = null;
LineageTracer lineage = null;
// create affinity groups
if (cg.getInputs().size() == 1) {
affinityGroups = new HashMap<IdentityHashSet<Tuple>, Integer>();
LogicalOperator childOp = cg.getInputs().get(0);
visitor = new DerivedDataVisitor(childOp, null, baseData,
LogToPhyMap, physPlan);
try {
visitor.visit();
} catch (VisitorException e) {
log.error(e.getMessage());
}
lineage = visitor.lineage;
DataBag bag = visitor.evaluateIsolatedOperator(cg);
for (Iterator<Tuple> it = bag.iterator(); it.hasNext();) {
DataBag field;
try {
field = (DataBag) it.next().get(1);
} catch (ExecException e) {
// TODO Auto-generated catch block
e.printStackTrace();
log.error(e.getMessage());
throw new VisitorException(
"Error trimming operator COGROUP operator "
+ cg.getAlias()
+ "in example generator");
}
IdentityHashSet<Tuple> set = new IdentityHashSet<Tuple>();
affinityGroups.put(set, 2);
for (Iterator<Tuple> it1 = field.iterator(); it1.hasNext();) {
set.add(it1.next());
}
}
// add the equivalence classes obtained from derived data
// creation
for (IdentityHashSet<Tuple> set : visitor.EqClasses) {
affinityGroups.put(set, 1);
}
AffinityGroups.put(cg.getInputs().get(0), affinityGroups);
Lineage.put(cg.getInputs().get(0), lineage);
} else {
List<DataBag> inputs = new LinkedList<DataBag>();
visitor = new DerivedDataVisitor(cg, null, baseData,
LogToPhyMap, physPlan);
affinityGroups = new HashMap<IdentityHashSet<Tuple>, Integer>();
for (int i = 0; i < cg.getInputs().size(); i++) {
// affinityGroups = new HashMap<IdentityHashSet<Tuple>,
// Integer>();
LogicalOperator childOp = cg.getInputs().get(i);
// visitor = new DerivedDataVisitor(cg.getInputs().get(i),
// null, baseData, LogToPhyMap, physPlan);
visitor.setOperatorToEvaluate(childOp);
try {
visitor.visit();