in hyracks-fullstack/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/PushSelectIntoJoinRule.java [59:213]
public boolean rewritePost(Mutable<ILogicalOperator> opRef, IOptimizationContext context)
throws AlgebricksException {
Collection<LogicalVariable> joinLiveVarsLeft = new HashSet<LogicalVariable>();
Collection<LogicalVariable> joinLiveVarsRight = new HashSet<LogicalVariable>();
Collection<LogicalVariable> liveInOpsToPushLeft = new HashSet<LogicalVariable>();
Collection<LogicalVariable> liveInOpsToPushRight = new HashSet<LogicalVariable>();
List<ILogicalOperator> pushedOnLeft = new ArrayList<ILogicalOperator>();
List<ILogicalOperator> pushedOnRight = new ArrayList<ILogicalOperator>();
List<ILogicalOperator> pushedOnEither = new ArrayList<ILogicalOperator>();
LinkedList<ILogicalOperator> notPushedStack = new LinkedList<ILogicalOperator>();
Collection<LogicalVariable> usedVars = new HashSet<LogicalVariable>();
Collection<LogicalVariable> producedVars = new HashSet<LogicalVariable>();
AbstractLogicalOperator op = (AbstractLogicalOperator) opRef.getValue();
if (op.getOperatorTag() != LogicalOperatorTag.SELECT) {
return false;
}
SelectOperator select = (SelectOperator) op;
Mutable<ILogicalOperator> opRef2 = op.getInputs().get(0);
AbstractLogicalOperator son = (AbstractLogicalOperator) opRef2.getValue();
AbstractLogicalOperator op2 = son;
boolean needToPushOps = false;
while (son.isMap()) {
needToPushOps = true;
Mutable<ILogicalOperator> opRefLink = son.getInputs().get(0);
son = (AbstractLogicalOperator) opRefLink.getValue();
}
if (son.getOperatorTag() != LogicalOperatorTag.INNERJOIN
&& son.getOperatorTag() != LogicalOperatorTag.LEFTOUTERJOIN) {
return false;
}
boolean isLoj = son.getOperatorTag() == LogicalOperatorTag.LEFTOUTERJOIN;
AbstractBinaryJoinOperator join = (AbstractBinaryJoinOperator) son;
Mutable<ILogicalOperator> joinBranchLeftRef = join.getInputs().get(0);
Mutable<ILogicalOperator> joinBranchRightRef = join.getInputs().get(1);
if (needToPushOps) {
ILogicalOperator joinBranchLeft = joinBranchLeftRef.getValue();
ILogicalOperator joinBranchRight = joinBranchRightRef.getValue();
VariableUtilities.getLiveVariables(joinBranchLeft, joinLiveVarsLeft);
VariableUtilities.getLiveVariables(joinBranchRight, joinLiveVarsRight);
Mutable<ILogicalOperator> opIterRef;
ILogicalOperator opIter = op2;
while (opIter != join) {
LogicalOperatorTag tag = opIter.getOperatorTag();
if (tag == LogicalOperatorTag.PROJECT) {
notPushedStack.addFirst(opIter);
} else {
VariableUtilities.getUsedVariables(opIter, usedVars);
VariableUtilities.getProducedVariables(opIter, producedVars);
if (usedVars.size() == 0) {
pushedOnEither.add(opIter);
} else if (joinLiveVarsLeft.containsAll(usedVars)) {
pushedOnLeft.add(opIter);
liveInOpsToPushLeft.addAll(producedVars);
} else if (joinLiveVarsRight.containsAll(usedVars)) {
pushedOnRight.add(opIter);
liveInOpsToPushRight.addAll(producedVars);
} else {
return false;
}
}
opIterRef = opIter.getInputs().get(0);
opIter = opIterRef.getValue();
}
if (isLoj && pushedOnLeft.isEmpty()) {
return false;
}
}
boolean intersectsAllBranches = true;
boolean[] intersectsBranch = new boolean[join.getInputs().size()];
LinkedList<LogicalVariable> selectVars = new LinkedList<>();
select.getCondition().getValue().getUsedVariables(selectVars);
int i = 0;
for (Mutable<ILogicalOperator> branch : join.getInputs()) {
LinkedList<LogicalVariable> branchVars = new LinkedList<>();
VariableUtilities.getLiveVariables(branch.getValue(), branchVars);
if (i == 0) {
branchVars.addAll(liveInOpsToPushLeft);
} else {
branchVars.addAll(liveInOpsToPushRight);
}
if (OperatorPropertiesUtil.disjoint(selectVars, branchVars)) {
intersectsAllBranches = false;
} else {
intersectsBranch[i] = true;
}
i++;
}
if (!intersectsBranch[0] && !intersectsBranch[1]) {
return false;
}
boolean planChanged;
if (needToPushOps) {
//We should push independent ops into the first branch that the selection depends on
planChanged =
pushOps(pushedOnEither, intersectsBranch[0] ? joinBranchLeftRef : joinBranchRightRef, context);
planChanged |= pushOps(pushedOnLeft, joinBranchLeftRef, context);
planChanged |= pushOps(pushedOnRight, joinBranchRightRef, context);
} else {
planChanged = false;
}
if (intersectsAllBranches) {
// add condition to the join condition only if we have IJ
if (isLoj) {
notPushedStack.addFirst(select);
} else {
addCondToJoin(select, join, context);
planChanged = true;
}
} else { // push down
Iterator<Mutable<ILogicalOperator>> branchIter = join.getInputs().iterator();
ILogicalExpression selectCondition = select.getCondition().getValue();
boolean lojToInner = false;
for (int j = 0; j < intersectsBranch.length; j++) {
Mutable<ILogicalOperator> branch = branchIter.next();
boolean inter = intersectsBranch[j];
if (!inter) {
continue;
}
if (j > 0 && isLoj) {
// if a LOJ and the select condition is not-missing filtering,
// we rewrite LOJ to IJ for this case.
FunctionIdentifier isMissingNullFunction = OperatorPropertiesUtil
.getIsMissingNullFunction(((LeftOuterJoinOperator) join).getMissingValue());
if (containsNotMissingFiltering(selectCondition, isMissingNullFunction)) {
lojToInner = true;
}
// Do not push conditions into the right branch of a LOJ;
notPushedStack.addFirst(select);
} else {
// Conditions for the left branch for IJ/LOJ or
// for the right branch of IJ can always be pushed into that branch.
// We don't push conditions into the right branch of LOJ at this point.
copySelectToBranch(select, branch, context);
planChanged = true;
}
}
if (lojToInner) {
// Rewrites left outer join to inner join.
InnerJoinOperator innerJoin = new InnerJoinOperator(join.getCondition());
innerJoin.getInputs().addAll(join.getInputs());
join = innerJoin;
context.computeAndSetTypeEnvironmentForOperator(join);
planChanged = true;
}
}
planChanged |= applyNonPushed(opRef, notPushedStack, join, context);
return planChanged;
}