// All customized packagers are introduced during MRCompilaition.
// Illustrate happens before that, so we only have to focus on the basic
// POPackage
if (illustrator != null) {
ExampleTuple tOut = new ExampleTuple((Tuple) out);
LineageTracer lineageTracer = illustrator.getLineage();
lineageTracer.insert(tOut);
boolean synthetic = false;
if (illustrator.getEquivalenceClasses() == null) {
LinkedList<IdentityHashSet<Tuple>> equivalenceClasses = new LinkedList<IdentityHashSet<Tuple>>();
for (int i = 0; i < numInputs; ++i) {
IdentityHashSet<Tuple> equivalenceClass = new IdentityHashSet<Tuple>();
equivalenceClasses.add(equivalenceClass);
}
illustrator.setEquivalenceClasses(equivalenceClasses, parent);
}
if (isDistinct()) {
int count = 0;
for (Tuple tmp : bags[0]){
count++;
if (!tmp.equals(tOut))
lineageTracer.union(tOut, tmp);
}
if (count > 1) // only non-distinct tuples are inserted into the
// equivalence class
illustrator.getEquivalenceClasses().get(eqClassIndex)
.add(tOut);
illustrator.addData((Tuple) tOut);
return (Tuple) tOut;
}
boolean outInEqClass = true;
try {
for (int i = 1; i < numInputs + 1; i++) {
DataBag dbs = (DataBag) ((Tuple) out).get(i);
Iterator<Tuple> iter = dbs.iterator();
if (dbs.size() <= 1 && outInEqClass) // all inputs have >= 2
// records
outInEqClass = false;
while (iter.hasNext()) {
Tuple tmp = iter.next();
// any of synthetic data in bags causes the output tuple
// to be synthetic
if (!synthetic && ((ExampleTuple) tmp).synthetic)
synthetic = true;
lineageTracer.union(tOut, tmp);
}
}
} catch (ExecException e) {
// TODO better exception handling
throw new RuntimeException("Illustrator exception :"