@Override
public void transform(OperatorPlan matched) throws FrontendException {
subPlan = new OperatorSubPlan(currentPlan);
LOFilter filter = (LOFilter)matched.getSources().get(0);
// This is the one that we will insert filter btwn it and it's input.
Operator predecessor = this.findNonFilterPredecessor( filter );
subPlan.add( predecessor) ;
// Disconnect the filter in the plan without removing it from the plan.
Operator predec = currentPlan.getPredecessors( filter ).get( 0 );
Operator succed;
if (currentPlan.getSuccessors(filter)!=null)
succed = currentPlan.getSuccessors(filter).get(0);
else
succed = null;
Pair<Integer, Integer> p1 = currentPlan.disconnect(predec, filter);
if (succed!=null) {
subPlan.add(succed);
Pair<Integer, Integer> p2 = currentPlan.disconnect(filter, succed);
currentPlan.connect(predec, p1.first, succed, p2.second);
}
if( predecessor instanceof LOSort || predecessor instanceof LODistinct ||
( predecessor instanceof LOCogroup && currentPlan.getPredecessors( predecessor ).size() == 1 ) ) {
// For sort, put the filter in front of it.
Operator prev = currentPlan.getPredecessors( predecessor ).get( 0 );
insertFilter( prev, predecessor, filter );
return;
}
// Find the predecessor of join that contains all required uids.
LogicalExpressionPlan filterPlan = filter.getFilterPlan();
List<Operator> preds = currentPlan.getPredecessors( predecessor );
Map<Integer, Operator> inputs = findInputsToAddFilter( filterPlan, predecessor, preds );
LOFilter newFilter = null;
for( Entry<Integer, Operator> entry : inputs.entrySet() ) {
int inputIndex = entry.getKey();
Operator pred = entry.getValue();
// Find projection field offset
int columnOffset = 0;
if( predecessor instanceof LOJoin || predecessor instanceof LOCross ) {
for( int i = 0; i < inputIndex; i++ ) {
columnOffset += ( (LogicalRelationalOperator)preds.get( i ) ).getSchema().size();
}
}
// Reuse the filter for the first match. For others, need to make a copy of the filter
// and add it between input and predecessor.
newFilter = newFilter == null ? filter : new LOFilter( (LogicalPlan)currentPlan );
currentPlan.add( newFilter );
subPlan.add( newFilter );
subPlan.add( pred );
LogicalExpressionPlan fPlan = filterPlan.deepCopy();
List<Operator> sinks = fPlan.getSinks();
List<ProjectExpression> projExprs = new ArrayList<ProjectExpression>();
for( Operator sink : sinks ) {
if( sink instanceof ProjectExpression )
projExprs.add( (ProjectExpression)sink );
}
if( predecessor instanceof LOCogroup ) {
for( ProjectExpression projExpr : projExprs ) {
// Need to merge filter condition and cogroup by expression;
LogicalExpressionPlan plan = ((LOCogroup) predecessor).getExpressionPlans().get( inputIndex ).iterator().next();
LogicalExpressionPlan copy = plan.deepCopy();
LogicalExpression root = (LogicalExpression)copy.getSinks().get( 0 );
List<Operator> predecessors = fPlan.getPredecessors( projExpr );
if( predecessors == null || predecessors.size() == 0 ) {
fPlan.remove( projExpr );
fPlan.add( root );
} else {
fPlan.add( root );
Operator pred1 = predecessors.get( 0 );
Pair<Integer, Integer> pair = fPlan.disconnect( pred1, projExpr );
fPlan.connect( pred1, pair.first, root, pair.second );
fPlan.remove( projExpr );
}
}
}
// Now, reset the projection expressions in the new filter plan.
sinks = fPlan.getSinks();
for( Operator sink : sinks ) {
if( sink instanceof ProjectExpression ) {
ProjectExpression projE = (ProjectExpression)sink;
projE.setAttachedRelationalOp( newFilter );
projE.setInputNum( 0 );
projE.setColNum( projE.getColNum() - columnOffset );
}
}
newFilter.setFilterPlan( fPlan );
insertFilter( pred, predecessor, newFilter );
}
}