Yingyi Bu has submitted this change and it was merged. Change subject: ASTERIXDB-1127: fix ExtractCommonOperatorsRule. ......................................................................
ASTERIXDB-1127: fix ExtractCommonOperatorsRule. Change-Id: I16933a4b72432b5fbd523ca80ce6426f6b6743a9 Reviewed-on: https://asterix-gerrit.ics.uci.edu/691 Tested-by: Jenkins <[email protected]> Reviewed-by: Pouria Pirzadeh <[email protected]> Reviewed-by: Till Westmann <[email protected]> --- M algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/jobgen/impl/JobBuilder.java M algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/rewriter/base/HeuristicOptimizer.java M algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/ExtractCommonOperatorsRule.java 3 files changed, 42 insertions(+), 31 deletions(-) Approvals: Pouria Pirzadeh: Looks good to me, but someone else must approve Till Westmann: Looks good to me, approved Jenkins: Verified diff --git a/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/jobgen/impl/JobBuilder.java b/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/jobgen/impl/JobBuilder.java index d130d4c..ac2ae5c 100644 --- a/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/jobgen/impl/JobBuilder.java +++ b/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/jobgen/impl/JobBuilder.java @@ -233,7 +233,7 @@ AlgebricksPartitionConstraint pc = partitionConstraintMap.get(parentOp); if (pc != null) { opConstraint = pc; - } else if (opInputs == null || opInputs.size() == 0) { + } else if ((opInputs == null || opInputs.size() == 0) && finalPass) { opConstraint = new AlgebricksCountPartitionConstraint(1); } } diff --git a/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/rewriter/base/HeuristicOptimizer.java b/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/rewriter/base/HeuristicOptimizer.java index 7b04bd5..2a28d2e 100644 --- a/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/rewriter/base/HeuristicOptimizer.java +++ b/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/rewriter/base/HeuristicOptimizer.java @@ -35,7 +35,7 @@ public class HeuristicOptimizer { - public static PhysicalOperatorTag[] hyraxOperators = new PhysicalOperatorTag[] { + public static PhysicalOperatorTag[] hyracksOperators = new PhysicalOperatorTag[] { PhysicalOperatorTag.DATASOURCE_SCAN, PhysicalOperatorTag.BTREE_SEARCH, PhysicalOperatorTag.EXTERNAL_GROUP_BY, PhysicalOperatorTag.HASH_GROUP_BY, PhysicalOperatorTag.HDFS_READER, PhysicalOperatorTag.HYBRID_HASH_JOIN, PhysicalOperatorTag.IN_MEMORY_HASH_JOIN, @@ -44,8 +44,8 @@ PhysicalOperatorTag.UNION_ALL }; public static PhysicalOperatorTag[] hyraxOperatorsBelowWhichJobGenIsDisabled = new PhysicalOperatorTag[] {}; - public static boolean isHyraxOp(PhysicalOperatorTag opTag) { - for (PhysicalOperatorTag t : hyraxOperators) { + public static boolean isHyracksOp(PhysicalOperatorTag opTag) { + for (PhysicalOperatorTag t : hyracksOperators) { if (t == opTag) { return true; } diff --git a/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/ExtractCommonOperatorsRule.java b/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/ExtractCommonOperatorsRule.java index f13187f..3b31f6d 100644 --- a/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/ExtractCommonOperatorsRule.java +++ b/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/ExtractCommonOperatorsRule.java @@ -47,6 +47,7 @@ import org.apache.hyracks.algebricks.core.algebra.operators.physical.OneToOneExchangePOperator; import org.apache.hyracks.algebricks.core.algebra.operators.physical.ReplicatePOperator; import org.apache.hyracks.algebricks.core.algebra.operators.physical.StreamProjectPOperator; +import org.apache.hyracks.algebricks.core.rewriter.base.HeuristicOptimizer; import org.apache.hyracks.algebricks.core.rewriter.base.IAlgebraicRewriteRule; public class ExtractCommonOperatorsRule implements IAlgebraicRewriteRule { @@ -67,8 +68,9 @@ && op.getOperatorTag() != LogicalOperatorTag.DISTRIBUTE_RESULT) { return false; } - if (!roots.contains(op)) + if (!roots.contains(op)) { roots.add(new MutableObject<ILogicalOperator>(op)); + } return false; } @@ -89,10 +91,12 @@ topDownMaterialization(roots); genCandidates(context); removeTrivialShare(); - if (equivalenceClasses.size() > 0) + if (equivalenceClasses.size() > 0) { changed = rewrite(context); - if (!rewritten) + } + if (!rewritten) { rewritten = changed; + } equivalenceClasses.clear(); childrenToParents.clear(); opToCandidateInputs.clear(); @@ -110,22 +114,27 @@ for (int i = candidates.size() - 1; i >= 0; i--) { Mutable<ILogicalOperator> opRef = candidates.get(i); AbstractLogicalOperator aop = (AbstractLogicalOperator) opRef.getValue(); - if (aop.getOperatorTag() == LogicalOperatorTag.EXCHANGE) + if (aop.getOperatorTag() == LogicalOperatorTag.EXCHANGE) { aop = (AbstractLogicalOperator) aop.getInputs().get(0).getValue(); - if (aop.getOperatorTag() == LogicalOperatorTag.EMPTYTUPLESOURCE) + } + if (aop.getOperatorTag() == LogicalOperatorTag.EMPTYTUPLESOURCE) { candidates.remove(i); + } } } - for (int i = equivalenceClasses.size() - 1; i >= 0; i--) - if (equivalenceClasses.get(i).size() < 2) + for (int i = equivalenceClasses.size() - 1; i >= 0; i--) { + if (equivalenceClasses.get(i).size() < 2) { equivalenceClasses.remove(i); + } + } } private boolean rewrite(IOptimizationContext context) throws AlgebricksException { boolean changed = false; for (List<Mutable<ILogicalOperator>> members : equivalenceClasses) { - if (rewriteForOneEquivalentClass(members, context)) + if (rewriteForOneEquivalentClass(members, context)) { changed = true; + } } return changed; } @@ -191,11 +200,13 @@ List<LogicalVariable> liveVarsNew = new ArrayList<LogicalVariable>(); VariableUtilities.getLiveVariables(candidate.getValue(), liveVarsNew); ArrayList<Mutable<ILogicalExpression>> assignExprs = new ArrayList<Mutable<ILogicalExpression>>(); - for (LogicalVariable liveVar : liveVarsNew) + for (LogicalVariable liveVar : liveVarsNew) { assignExprs.add(new MutableObject<ILogicalExpression>(new VariableReferenceExpression(liveVar))); + } for (Mutable<ILogicalOperator> ref : group) { - if (ref.equals(candidate)) + if (ref.equals(candidate)) { continue; + } ArrayList<LogicalVariable> liveVars = new ArrayList<LogicalVariable>(); Map<LogicalVariable, LogicalVariable> variableMappingBack = new HashMap<LogicalVariable, LogicalVariable>(); IsomorphismUtilities.mapVariablesTopDown(ref.getValue(), candidate.getValue(), variableMappingBack); @@ -227,20 +238,13 @@ for (Mutable<ILogicalOperator> parentOpRef : parentOpList) { AbstractLogicalOperator parentOp = (AbstractLogicalOperator) parentOpRef.getValue(); int index = parentOp.getInputs().indexOf(ref); - if (parentOp.getOperatorTag() == LogicalOperatorTag.EXCHANGE) { - AbstractLogicalOperator parentOpNext = (AbstractLogicalOperator) childrenToParents - .get(parentOpRef).get(0).getValue(); - if (parentOpNext.isMap()) { - index = parentOpNext.getInputs().indexOf(parentOpRef); - parentOp = parentOpNext; - } - } - ILogicalOperator childOp = parentOp.getOperatorTag() == LogicalOperatorTag.PROJECT ? assignOperator : projectOperator; - if (parentOp.isMap()) { + if (!HeuristicOptimizer.isHyracksOp(parentOp.getPhysicalOperator().getOperatorTag())) { parentOp.getInputs().set(index, new MutableObject<ILogicalOperator>(childOp)); } else { + // If the parent operator is a hyracks operator, + // an extra one-to-one exchange is needed. AbstractLogicalOperator exchg = new ExchangeOperator(); exchg.setPhysicalOperator(new OneToOneExchangePOperator()); exchg.setExecutionMode(childOp.getExecutionMode()); @@ -270,16 +274,19 @@ if (candidates.size() > 0) { for (Mutable<ILogicalOperator> opRef : candidates) { List<Mutable<ILogicalOperator>> refs = childrenToParents.get(opRef); - if (refs != null) + if (refs != null) { currentLevelOpRefs.addAll(refs); + } } } - if (currentLevelOpRefs.size() == 0) + if (currentLevelOpRefs.size() == 0) { continue; + } candidatesGrow(currentLevelOpRefs, candidates); } - if (currentLevelOpRefs.size() == 0) + if (currentLevelOpRefs.size() == 0) { break; + } prune(context); } if (equivalenceClasses.size() < 1 && previousEquivalenceClasses.size() > 0) { @@ -301,8 +308,9 @@ } opRefList.add(op); } - if (op.getValue().getInputs().size() == 0) + if (op.getValue().getInputs().size() == 0) { candidates.add(op); + } } if (equivalenceClasses.size() > 0) { equivalenceClasses.get(0).addAll(candidates); @@ -344,10 +352,12 @@ } } } - if (!validCandidate) + if (!validCandidate) { continue; - if (!candidates.contains(op)) + } + if (!candidates.contains(op)) { candidates.add(op); + } } } @@ -361,8 +371,9 @@ equivalenceClasses.clear(); for (List<Mutable<ILogicalOperator>> candidates : previousEquivalenceClasses) { boolean[] reserved = new boolean[candidates.size()]; - for (int i = 0; i < reserved.length; i++) + for (int i = 0; i < reserved.length; i++) { reserved[i] = false; + } for (int i = candidates.size() - 1; i >= 0; i--) { if (reserved[i] == false) { List<Mutable<ILogicalOperator>> equivalentClass = new ArrayList<Mutable<ILogicalOperator>>(); -- To view, visit https://asterix-gerrit.ics.uci.edu/691 To unsubscribe, visit https://asterix-gerrit.ics.uci.edu/settings Gerrit-MessageType: merged Gerrit-Change-Id: I16933a4b72432b5fbd523ca80ce6426f6b6743a9 Gerrit-PatchSet: 4 Gerrit-Project: hyracks Gerrit-Branch: master Gerrit-Owner: Yingyi Bu <[email protected]> Gerrit-Reviewer: Jenkins <[email protected]> Gerrit-Reviewer: Pouria Pirzadeh <[email protected]> Gerrit-Reviewer: Till Westmann <[email protected]> Gerrit-Reviewer: Yingyi Bu <[email protected]>
