You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by ta...@apache.org on 2019/10/04 21:57:55 UTC
[impala] branch master updated: IMPALA-2138: part 3: refactor
analytic planning
This is an automated email from the ASF dual-hosted git repository.
tarmstrong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
The following commit(s) were added to refs/heads/master by this push:
new 7b3da1d IMPALA-2138: part 3: refactor analytic planning
7b3da1d is described below
commit 7b3da1d6d27320be92839d4594896b7515f2f91f
Author: Tim Armstrong <ta...@cloudera.com>
AuthorDate: Mon Sep 30 16:10:48 2019 -0700
IMPALA-2138: part 3: refactor analytic planning
This is a minor refactor of the analytic planner that moves
the logic to generate a "buffered tuple descriptor" and
equality expressions to the AnalyticEvalNode.
This is required for projection, because the buffered tuple
layout is required to match the input tuple layout, so needs
to be updated if the input tuple is projected.
After the refactoring, this can be correctly updated simply
by regenerating the tuple descriptor and expressions,
instead of attempting to "fix up" the existing expressions
with smaps, etc.
I think the plan node is actually a more approprate place for
this logic regardless, since it's tightly entwined with the
backend's assumption that the two tuple layouts are identical.
Change-Id: I691f1509a85cad51d2654278c0a4ba70a10b2706
Reviewed-on: http://gerrit.cloudera.org:8080/14325
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
.../main/java/org/apache/impala/analysis/Expr.java | 10 ++
.../apache/impala/planner/AnalyticEvalNode.java | 117 ++++++++++++++++++--
.../org/apache/impala/planner/AnalyticPlanner.java | 123 +--------------------
3 files changed, 123 insertions(+), 127 deletions(-)
diff --git a/fe/src/main/java/org/apache/impala/analysis/Expr.java b/fe/src/main/java/org/apache/impala/analysis/Expr.java
index e4dfeaa..84ee9b9 100644
--- a/fe/src/main/java/org/apache/impala/analysis/Expr.java
+++ b/fe/src/main/java/org/apache/impala/analysis/Expr.java
@@ -1319,6 +1319,16 @@ abstract public class Expr extends TreeNode<Expr> implements ParseNode, Cloneabl
}
/**
+ * Return true if and only if all exprs in 'exprs' are constant.
+ */
+ public static boolean allConstant(List<Expr> exprs) {
+ for (Expr p: exprs) {
+ if (!p.isConstant()) return false;
+ }
+ return true;
+ }
+
+ /**
* Return true if this expr is a scalar subquery.
*/
public boolean isScalarSubquery() {
diff --git a/fe/src/main/java/org/apache/impala/planner/AnalyticEvalNode.java b/fe/src/main/java/org/apache/impala/planner/AnalyticEvalNode.java
index d019619..82e0f87 100644
--- a/fe/src/main/java/org/apache/impala/planner/AnalyticEvalNode.java
+++ b/fe/src/main/java/org/apache/impala/planner/AnalyticEvalNode.java
@@ -25,10 +25,18 @@ import org.slf4j.LoggerFactory;
import org.apache.impala.analysis.AnalyticWindow;
import org.apache.impala.analysis.Analyzer;
+import org.apache.impala.analysis.BinaryPredicate;
+import org.apache.impala.analysis.BoolLiteral;
+import org.apache.impala.analysis.CompoundPredicate;
import org.apache.impala.analysis.Expr;
import org.apache.impala.analysis.ExprSubstitutionMap;
+import org.apache.impala.analysis.IsNullPredicate;
import org.apache.impala.analysis.OrderByElement;
+import org.apache.impala.analysis.SlotDescriptor;
+import org.apache.impala.analysis.SlotRef;
import org.apache.impala.analysis.TupleDescriptor;
+import org.apache.impala.analysis.TupleId;
+import org.apache.impala.analysis.CompoundPredicate.Operator;
import org.apache.impala.thrift.TAnalyticNode;
import org.apache.impala.thrift.TExplainLevel;
import org.apache.impala.thrift.TPlanNode;
@@ -62,18 +70,18 @@ public class AnalyticEvalNode extends PlanNode {
// physical output slots in outputTupleDesc_
private final ExprSubstitutionMap logicalToPhysicalSmap_;
- // predicates constructed from partitionExprs_/orderingExprs_ to
- // compare input to buffered tuples
- private final Expr partitionByEq_;
- private final Expr orderByEq_;
- private final TupleDescriptor bufferedTupleDesc_;
+ // Predicates constructed from partitionExprs_/orderingExprs_ to
+ // compare input to buffered tuples. Initialized by constructEqExprs().
+ // Must be recomputed if the sort tuple is changed, e.g. by projection.
+ private Expr partitionByEq_;
+ private Expr orderByEq_;
+ private TupleDescriptor bufferedTupleDesc_;
public AnalyticEvalNode(
PlanNodeId id, PlanNode input, List<Expr> analyticFnCalls,
List<Expr> partitionExprs, List<OrderByElement> orderByElements,
AnalyticWindow analyticWindow, TupleDescriptor intermediateTupleDesc,
- TupleDescriptor outputTupleDesc, ExprSubstitutionMap logicalToPhysicalSmap,
- Expr partitionByEq, Expr orderByEq, TupleDescriptor bufferedTupleDesc) {
+ TupleDescriptor outputTupleDesc, ExprSubstitutionMap logicalToPhysicalSmap) {
super(id, "ANALYTIC");
Preconditions.checkState(!tupleIds_.contains(outputTupleDesc.getId()));
analyticFnCalls_ = analyticFnCalls;
@@ -83,9 +91,6 @@ public class AnalyticEvalNode extends PlanNode {
intermediateTupleDesc_ = intermediateTupleDesc;
outputTupleDesc_ = outputTupleDesc;
logicalToPhysicalSmap_ = logicalToPhysicalSmap;
- partitionByEq_ = partitionByEq;
- orderByEq_ = orderByEq;
- bufferedTupleDesc_ = bufferedTupleDesc;
children_.add(input);
computeTupleIds();
}
@@ -131,9 +136,101 @@ public class AnalyticEvalNode extends PlanNode {
substitutedPartitionExprs_ = Expr.substituteList(partitionExprs_, childSmap,
analyzer, false);
orderByElements_ = OrderByElement.substitute(orderByElements_, childSmap, analyzer);
+ constructEqExprs(analyzer);
if (LOG.isTraceEnabled()) LOG.trace("evalnode: " + debugString());
}
+ /**
+ * Create partition-by (pb) and order-by (ob) less-than predicates between the input
+ * tuple (the output of the preceding sort, which is always the first tuple in the
+ * input row) and a buffered tuple that is identical to the input tuple. We need a
+ * different tuple descriptor for the buffered tuple because the generated predicates
+ * should compare two different tuple instances from the same input stream (i.e., the
+ * predicates should be evaluated over a row that is composed of the input and the
+ * buffered tuple).
+ *
+ * Requires 'substitutedPartitionExprs_' and 'orderByElements_' to be initialized.
+ * Sets 'partitionByEq_', 'orderByEq_' and 'bufferedTupleDesc_'.
+ */
+ private void constructEqExprs(Analyzer analyzer) {
+ // we need to remap the pb/ob exprs to a) the sort output, b) our buffer of the
+ // sort input
+ ExprSubstitutionMap bufferedSmap = new ExprSubstitutionMap();
+ TupleId sortTupleId = getChild(0).getTupleIds().get(0);
+ boolean hasActivePartition = !Expr.allConstant(substitutedPartitionExprs_);
+ boolean hasOrderBy = !orderByElements_.isEmpty();
+ if (hasActivePartition || hasOrderBy) {
+ // create bufferedTupleDesc and bufferedSmap
+ bufferedTupleDesc_ =
+ analyzer.getDescTbl().copyTupleDescriptor(sortTupleId, "buffered-tuple");
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("desctbl: " + analyzer.getDescTbl().debugString());
+ }
+
+ List<SlotDescriptor> inputSlots = analyzer.getTupleDesc(sortTupleId).getSlots();
+ List<SlotDescriptor> bufferedSlots = bufferedTupleDesc_.getSlots();
+ for (int i = 0; i < inputSlots.size(); ++i) {
+ bufferedSmap.put(
+ new SlotRef(inputSlots.get(i)), new SlotRef(bufferedSlots.get(i)));
+ }
+ }
+ if (hasActivePartition) {
+ partitionByEq_ =
+ createNullMatchingEquals(analyzer, substitutedPartitionExprs_, sortTupleId, bufferedSmap);
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("partitionByEq: " + partitionByEq_.debugString());
+ }
+ }
+ if (hasOrderBy) {
+ orderByEq_ = createNullMatchingEquals(analyzer,
+ OrderByElement.getOrderByExprs(orderByElements_), sortTupleId, bufferedSmap);
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("orderByEq: " + orderByEq_.debugString());
+ }
+ }
+ }
+
+ /**
+ * Create a predicate that checks if all exprs are equal or both sides are null.
+ */
+ private Expr createNullMatchingEquals(Analyzer analyzer, List<Expr> exprs,
+ TupleId inputTid, ExprSubstitutionMap bufferedSmap) {
+ Preconditions.checkState(!exprs.isEmpty());
+ Expr result = createNullMatchingEqualsAux(analyzer, exprs, 0, inputTid, bufferedSmap);
+ result.analyzeNoThrow(analyzer);
+ return result;
+ }
+
+ /**
+ * Create an unanalyzed predicate that checks if elements >= i are equal or
+ * both sides are null.
+ *
+ * The predicate has the form
+ * ((lhs[i] is null && rhs[i] is null) || (
+ * lhs[i] is not null && rhs[i] is not null && lhs[i] = rhs[i]))
+ * && <createEqualsAux(i + 1)>
+ */
+ private Expr createNullMatchingEqualsAux(Analyzer analyzer, List<Expr> elements, int i,
+ TupleId inputTid, ExprSubstitutionMap bufferedSmap) {
+ if (i > elements.size() - 1) return new BoolLiteral(true);
+
+ // compare elements[i]
+ Expr lhs = elements.get(i);
+ Preconditions.checkState(lhs.isBound(inputTid));
+ Expr rhs = lhs.substitute(bufferedSmap, analyzer, false);
+
+ Expr bothNull = new CompoundPredicate(
+ Operator.AND, new IsNullPredicate(lhs, false), new IsNullPredicate(rhs, false));
+ Expr lhsEqRhsNotNull = new CompoundPredicate(Operator.AND,
+ new CompoundPredicate(
+ Operator.AND, new IsNullPredicate(lhs, true), new IsNullPredicate(rhs, true)),
+ new BinaryPredicate(BinaryPredicate.Operator.EQ, lhs, rhs));
+ Expr remainder =
+ createNullMatchingEqualsAux(analyzer, elements, i + 1, inputTid, bufferedSmap);
+ return new CompoundPredicate(CompoundPredicate.Operator.AND,
+ new CompoundPredicate(Operator.OR, bothNull, lhsEqRhsNotNull), remainder);
+ }
+
@Override
protected void computeStats(Analyzer analyzer) {
super.computeStats(analyzer);
diff --git a/fe/src/main/java/org/apache/impala/planner/AnalyticPlanner.java b/fe/src/main/java/org/apache/impala/planner/AnalyticPlanner.java
index f45d8fb..57baa58 100644
--- a/fe/src/main/java/org/apache/impala/planner/AnalyticPlanner.java
+++ b/fe/src/main/java/org/apache/impala/planner/AnalyticPlanner.java
@@ -27,13 +27,8 @@ import org.apache.impala.analysis.AnalyticExpr;
import org.apache.impala.analysis.AnalyticInfo;
import org.apache.impala.analysis.AnalyticWindow;
import org.apache.impala.analysis.Analyzer;
-import org.apache.impala.analysis.BinaryPredicate;
-import org.apache.impala.analysis.BoolLiteral;
-import org.apache.impala.analysis.CompoundPredicate;
-import org.apache.impala.analysis.CompoundPredicate.Operator;
import org.apache.impala.analysis.Expr;
import org.apache.impala.analysis.ExprSubstitutionMap;
-import org.apache.impala.analysis.IsNullPredicate;
import org.apache.impala.analysis.OrderByElement;
import org.apache.impala.analysis.SlotDescriptor;
import org.apache.impala.analysis.SlotRef;
@@ -84,18 +79,6 @@ public class AnalyticPlanner {
}
/**
- * Return true if and only if exprs is non-empty and contains non-constant
- * expressions.
- */
- private boolean activeExprs(List<Expr> exprs) {
- if (exprs.isEmpty()) return false;
- for (Expr p: exprs) {
- if (!p.isConstant()) { return true; }
- }
- return false;
- }
-
- /**
* Return plan tree that augments 'root' with plan nodes that implement single-node
* evaluation of the AnalyticExprs in analyticInfo.
* This plan takes into account a possible hash partition of its input on
@@ -248,7 +231,7 @@ public class AnalyticPlanner {
// remove the non-partitioning group from partitionGroups
PartitionGroup nonPartitioning = null;
for (PartitionGroup pg: partitionGroups) {
- if (!activeExprs(pg.partitionByExprs)) {
+ if (Expr.allConstant(pg.partitionByExprs)) {
nonPartitioning = pg;
break;
}
@@ -349,12 +332,7 @@ public class AnalyticPlanner {
List<Expr> partitionExprs) throws ImpalaException {
List<Expr> partitionByExprs = sortGroup.partitionByExprs;
List<OrderByElement> orderByElements = sortGroup.orderByElements;
- ExprSubstitutionMap sortSmap = null;
- TupleId sortTupleId = null;
- TupleDescriptor bufferedTupleDesc = null;
- // map from input to buffered tuple
- ExprSubstitutionMap bufferedSmap = new ExprSubstitutionMap();
- boolean activePartition = activeExprs(partitionByExprs);
+ boolean hasActivePartition = !Expr.allConstant(partitionByExprs);
// IMPALA-8069: Ignore something like ORDER BY 0
boolean isConstSort = true;
@@ -362,7 +340,7 @@ public class AnalyticPlanner {
isConstSort = isConstSort && elmt.getExpr().isConstant();
}
// sort on partition by (pb) + order by (ob) exprs and create pb/ob predicates
- if (activePartition || !isConstSort) {
+ if (hasActivePartition || !isConstSort) {
// first sort on partitionExprs (direction doesn't matter)
List<Expr> sortExprs = Lists.newArrayList(partitionByExprs);
List<Boolean> isAsc =
@@ -389,12 +367,12 @@ public class AnalyticPlanner {
// if this sort group does not have partitioning exprs, we want the sort
// to be executed like a regular distributed sort
- if (activePartition) sortNode.setIsAnalyticSort(true);
+ if (hasActivePartition) sortNode.setIsAnalyticSort(true);
if (partitionExprs != null) {
// create required input partition
DataPartition inputPartition = DataPartition.UNPARTITIONED;
- if (activePartition) {
+ if (hasActivePartition) {
inputPartition = DataPartition.hashPartitioned(partitionExprs);
}
sortNode.setInputPartition(inputPartition);
@@ -403,109 +381,20 @@ public class AnalyticPlanner {
root = sortNode;
root.init(analyzer_);
}
- if (activePartition || !orderByElements.isEmpty()) {
- sortSmap = root.getOutputSmap();
-
- // create bufferedTupleDesc and bufferedSmap
- sortTupleId = root.tupleIds_.get(0);
- bufferedTupleDesc =
- analyzer_.getDescTbl().copyTupleDescriptor(sortTupleId, "buffered-tuple");
- if (LOG.isTraceEnabled()) {
- LOG.trace("desctbl: " + analyzer_.getDescTbl().debugString());
- }
-
- List<SlotDescriptor> inputSlots = analyzer_.getTupleDesc(sortTupleId).getSlots();
- List<SlotDescriptor> bufferedSlots = bufferedTupleDesc.getSlots();
- for (int i = 0; i < inputSlots.size(); ++i) {
- bufferedSmap.put(
- new SlotRef(inputSlots.get(i)), new SlotRef(bufferedSlots.get(i)));
- }
- }
// create one AnalyticEvalNode per window group
for (WindowGroup windowGroup: sortGroup.windowGroups) {
- // Create partition-by (pb) and order-by (ob) less-than predicates between the
- // input tuple (the output of the preceding sort) and a buffered tuple that is
- // identical to the input tuple. We need a different tuple descriptor for the
- // buffered tuple because the generated predicates should compare two different
- // tuple instances from the same input stream (i.e., the predicates should be
- // evaluated over a row that is composed of the input and the buffered tuple).
-
- // we need to remap the pb/ob exprs to a) the sort output, b) our buffer of the
- // sort input
- Expr partitionByEq = null;
- if (activeExprs(windowGroup.partitionByExprs)) {
- partitionByEq = createNullMatchingEquals(
- Expr.substituteList(windowGroup.partitionByExprs, sortSmap, analyzer_, false),
- sortTupleId, bufferedSmap);
- if (LOG.isTraceEnabled()) {
- LOG.trace("partitionByEq: " + partitionByEq.debugString());
- }
- }
- Expr orderByEq = null;
- if (!windowGroup.orderByElements.isEmpty()) {
- orderByEq = createNullMatchingEquals(
- OrderByElement.getOrderByExprs(OrderByElement.substitute(
- windowGroup.orderByElements, sortSmap, analyzer_)),
- sortTupleId, bufferedSmap);
- if (LOG.isTraceEnabled()) {
- LOG.trace("orderByEq: " + orderByEq.debugString());
- }
- }
-
root = new AnalyticEvalNode(ctx_.getNextNodeId(), root,
windowGroup.analyticFnCalls, windowGroup.partitionByExprs,
windowGroup.orderByElements, windowGroup.window,
windowGroup.physicalIntermediateTuple, windowGroup.physicalOutputTuple,
- windowGroup.logicalToPhysicalSmap,
- partitionByEq, orderByEq, bufferedTupleDesc);
+ windowGroup.logicalToPhysicalSmap);
root.init(analyzer_);
}
return root;
}
/**
- * Create a predicate that checks if all exprs are equal or both sides are null.
- */
- private Expr createNullMatchingEquals(List<Expr> exprs, TupleId inputTid,
- ExprSubstitutionMap bufferedSmap) {
- Preconditions.checkState(!exprs.isEmpty());
- Expr result = createNullMatchingEqualsAux(exprs, 0, inputTid, bufferedSmap);
- result.analyzeNoThrow(analyzer_);
- return result;
- }
-
- /**
- * Create an unanalyzed predicate that checks if elements >= i are equal or
- * both sides are null.
- *
- * The predicate has the form
- * ((lhs[i] is null && rhs[i] is null) || (
- * lhs[i] is not null && rhs[i] is not null && lhs[i] = rhs[i]))
- * && <createEqualsAux(i + 1)>
- */
- private Expr createNullMatchingEqualsAux(List<Expr> elements, int i,
- TupleId inputTid, ExprSubstitutionMap bufferedSmap) {
- if (i > elements.size() - 1) return new BoolLiteral(true);
-
- // compare elements[i]
- Expr lhs = elements.get(i);
- Preconditions.checkState(lhs.isBound(inputTid));
- Expr rhs = lhs.substitute(bufferedSmap, analyzer_, false);
-
- Expr bothNull = new CompoundPredicate(Operator.AND,
- new IsNullPredicate(lhs, false), new IsNullPredicate(rhs, false));
- Expr lhsEqRhsNotNull = new CompoundPredicate(Operator.AND,
- new CompoundPredicate(Operator.AND,
- new IsNullPredicate(lhs, true), new IsNullPredicate(rhs, true)),
- new BinaryPredicate(BinaryPredicate.Operator.EQ, lhs, rhs));
- Expr remainder = createNullMatchingEqualsAux(elements, i + 1, inputTid, bufferedSmap);
- return new CompoundPredicate(CompoundPredicate.Operator.AND,
- new CompoundPredicate(Operator.OR, bothNull, lhsEqRhsNotNull),
- remainder);
- }
-
- /**
* Collection of AnalyticExprs that share the same partition-by/order-by/window
* specification. The AnalyticExprs are stored broken up into their constituent parts.
*/