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 2018/08/10 23:35:03 UTC
[4/5] impala git commit: IMPALA-7231: group plan nodes into pipelines
IMPALA-7231: group plan nodes into pipelines
This adds some informational output to explain plans and
sends the information to the backend.
The idea is that this will make it easier to explain how Impala's
pipelined execution works and also enable future work on profile
analysis that can more intelligently group plan nodes.
Tests:
* Updated planner tests to include new output.
Change-Id: I1d10eb14d997242f445e5c5fc5362d5410370721
Reviewed-on: http://gerrit.cloudera.org:8080/10848
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
Project: http://git-wip-us.apache.org/repos/asf/impala/repo
Commit: http://git-wip-us.apache.org/repos/asf/impala/commit/b7d509d7
Tree: http://git-wip-us.apache.org/repos/asf/impala/tree/b7d509d7
Diff: http://git-wip-us.apache.org/repos/asf/impala/diff/b7d509d7
Branch: refs/heads/master
Commit: b7d509d76117940a8ee57eb6012063c1a93be59a
Parents: 4af3a78
Author: Tim Armstrong <ta...@cloudera.com>
Authored: Thu Jun 28 09:19:05 2018 -0700
Committer: Impala Public Jenkins <im...@cloudera.com>
Committed: Fri Aug 10 19:31:30 2018 +0000
----------------------------------------------------------------------
common/thrift/PlanNodes.thrift | 9 +
.../org/apache/impala/planner/JoinNode.java | 21 +-
.../impala/planner/PipelineMembership.java | 75 +++
.../org/apache/impala/planner/PlanFragment.java | 4 +
.../org/apache/impala/planner/PlanNode.java | 73 +++
.../java/org/apache/impala/planner/Planner.java | 1 +
.../org/apache/impala/planner/SubplanNode.java | 15 +
.../org/apache/impala/planner/UnionNode.java | 17 +
.../queries/PlannerTest/constant-folding.test | 34 ++
.../PlannerTest/fk-pk-join-detection.test | 40 ++
.../queries/PlannerTest/kudu-selectivity.test | 17 +
.../queries/PlannerTest/max-row-size.test | 39 ++
.../PlannerTest/min-max-runtime-filters.test | 20 +-
.../queries/PlannerTest/mt-dop-validation.test | 59 ++-
.../PlannerTest/parquet-filtering-disabled.test | 35 ++
.../queries/PlannerTest/parquet-filtering.test | 71 ++-
.../queries/PlannerTest/partition-pruning.test | 1 +
.../PlannerTest/resource-requirements.test | 500 ++++++++++++++++++-
.../PlannerTest/sort-expr-materialization.test | 21 +
.../PlannerTest/spillable-buffer-sizing.test | 104 +++-
.../queries/PlannerTest/tablesample.test | 15 +-
.../queries/PlannerTest/union.test | 8 +
.../queries/QueryTest/explain-level2.test | 5 +
.../queries/QueryTest/explain-level3.test | 5 +
.../queries/QueryTest/stats-extrapolation.test | 6 +
25 files changed, 1174 insertions(+), 21 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/impala/blob/b7d509d7/common/thrift/PlanNodes.thrift
----------------------------------------------------------------------
diff --git a/common/thrift/PlanNodes.thrift b/common/thrift/PlanNodes.thrift
index b45625f..e116c3b 100644
--- a/common/thrift/PlanNodes.thrift
+++ b/common/thrift/PlanNodes.thrift
@@ -546,6 +546,13 @@ struct TCardinalityCheckNode {
1: required string display_statement
}
+// See PipelineMembership in the frontend for details.
+struct TPipelineMembership {
+ 1: required Types.TPlanNodeId pipe_id
+ 2: required i32 height
+ 3: required TExecNodePhase phase
+}
+
// This is essentially a union of all messages corresponding to subclasses
// of PlanNode.
struct TPlanNode {
@@ -565,6 +572,8 @@ struct TPlanNode {
// node is codegen'd if the backend supports it.
8: required bool disable_codegen
+ 27: required list<TPipelineMembership> pipelines
+
// one field per PlanNode subclass
9: optional THdfsScanNode hdfs_scan_node
10: optional THBaseScanNode hbase_scan_node
http://git-wip-us.apache.org/repos/asf/impala/blob/b7d509d7/fe/src/main/java/org/apache/impala/planner/JoinNode.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/planner/JoinNode.java b/fe/src/main/java/org/apache/impala/planner/JoinNode.java
index 815b31f..e8adcfa 100644
--- a/fe/src/main/java/org/apache/impala/planner/JoinNode.java
+++ b/fe/src/main/java/org/apache/impala/planner/JoinNode.java
@@ -32,8 +32,8 @@ import org.apache.impala.catalog.ColumnStats;
import org.apache.impala.catalog.FeTable;
import org.apache.impala.common.ImpalaException;
import org.apache.impala.common.Pair;
+import org.apache.impala.thrift.TExecNodePhase;
import org.apache.impala.thrift.TJoinDistributionMode;
-import org.apache.impala.service.BackendConfig;
import org.apache.impala.thrift.TQueryOptions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -690,4 +690,23 @@ public abstract class JoinNode extends PlanNode {
finishedBuildProfile.sum(probeSideProfile.postOpenProfile);
return new ExecPhaseResourceProfiles(duringOpenProfile, probePhaseProfile);
}
+
+ @Override
+ public void computePipelineMembership() {
+ children_.get(0).computePipelineMembership();
+ children_.get(1).computePipelineMembership();
+ pipelines_ = Lists.newArrayList();
+ for (PipelineMembership probePipeline : children_.get(0).getPipelines()) {
+ if (probePipeline.getPhase() == TExecNodePhase.GETNEXT) {
+ pipelines_.add(new PipelineMembership(
+ probePipeline.getId(), probePipeline.getHeight() + 1, TExecNodePhase.GETNEXT));
+ }
+ }
+ for (PipelineMembership buildPipeline : children_.get(1).getPipelines()) {
+ if (buildPipeline.getPhase() == TExecNodePhase.GETNEXT) {
+ pipelines_.add(new PipelineMembership(
+ buildPipeline.getId(), buildPipeline.getHeight() + 1, TExecNodePhase.OPEN));
+ }
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/impala/blob/b7d509d7/fe/src/main/java/org/apache/impala/planner/PipelineMembership.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/planner/PipelineMembership.java b/fe/src/main/java/org/apache/impala/planner/PipelineMembership.java
new file mode 100644
index 0000000..35af71f
--- /dev/null
+++ b/fe/src/main/java/org/apache/impala/planner/PipelineMembership.java
@@ -0,0 +1,75 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.impala.planner;
+
+import org.apache.impala.thrift.TExecNodePhase;
+import org.apache.impala.thrift.TPipelineMembership;
+
+/**
+ * Represents that a plan node executes as part of a query pipeline in one of its
+ * execution phases (GetNext() or Open()). A pipeline is a set of plan nodes and
+ * data sinks that execute concurrently, streaming rows through from the bottom
+ * to the top of the pipeline. There will be some amount of buffering at different
+ * parts of the pipeline, but the general idea is that the top of the pipeline may
+ * start processing rows while all other nodes in the pipeline are producing rows.
+ * A pipeline can span fragments. Often a pipeline is a linear sequence of
+ * nodes, but subplans are an exception because the right tree of the subplan is
+ * repeatedly executes for each row streamed from the left tree of the subplan.
+ *
+ * If a plan node's Open() phase is part of a pipeline, that means that it is a blocking
+ * node at the top of the pipeline, accumulating rows from the pipeline. If a plan node's
+ * GetNext() phase is part of a pipeline, that means that it is below the top of the
+ * pipeline.
+ */
+public class PipelineMembership {
+ // The id of the bottom-most node in the pipeline. Used to identify the pipeline.
+ private final PlanNodeId id;
+
+ // The height of this node in the pipeline. Starts at 0.
+ private final int height;
+
+ // The phase of the PlanNode that executes as part of the pipeline.
+ private final TExecNodePhase phase;
+
+
+ public PipelineMembership(PlanNodeId id, int height, TExecNodePhase phase) {
+ this.id = id;
+ this.height = height;
+ this.phase = phase;
+ }
+
+ public PlanNodeId getId() {
+ return id;
+ }
+
+ public int getHeight() {
+ return height;
+ }
+
+ public TExecNodePhase getPhase() {
+ return phase;
+ }
+
+ public TPipelineMembership toThrift() {
+ return new TPipelineMembership(getId().asInt(), height, getPhase());
+ }
+
+ public String getExplainString() {
+ return getId().toString() + "(" + getPhase().toString() + ")";
+ }
+}
http://git-wip-us.apache.org/repos/asf/impala/blob/b7d509d7/fe/src/main/java/org/apache/impala/planner/PlanFragment.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/planner/PlanFragment.java b/fe/src/main/java/org/apache/impala/planner/PlanFragment.java
index 7a90c23..bd9b415 100644
--- a/fe/src/main/java/org/apache/impala/planner/PlanFragment.java
+++ b/fe/src/main/java/org/apache/impala/planner/PlanFragment.java
@@ -219,6 +219,10 @@ public class PlanFragment extends TreeNode<PlanFragment> {
}
}
+ public void computePipelineMembership() {
+ planRoot_.computePipelineMembership();
+ }
+
/**
* Compute the peak resource profile for an instance of this fragment. Must
* be called after all the plan nodes and sinks are added to the fragment and resource
http://git-wip-us.apache.org/repos/asf/impala/blob/b7d509d7/fe/src/main/java/org/apache/impala/planner/PlanNode.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/planner/PlanNode.java b/fe/src/main/java/org/apache/impala/planner/PlanNode.java
index a14c89a..927cc18 100644
--- a/fe/src/main/java/org/apache/impala/planner/PlanNode.java
+++ b/fe/src/main/java/org/apache/impala/planner/PlanNode.java
@@ -18,6 +18,7 @@
package org.apache.impala.planner;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
@@ -33,6 +34,7 @@ import org.apache.impala.common.ImpalaException;
import org.apache.impala.common.PrintUtils;
import org.apache.impala.common.TreeNode;
import org.apache.impala.planner.RuntimeFilterGenerator.RuntimeFilter;
+import org.apache.impala.thrift.TExecNodePhase;
import org.apache.impala.thrift.TExecStats;
import org.apache.impala.thrift.TExplainLevel;
import org.apache.impala.thrift.TPlan;
@@ -117,6 +119,9 @@ abstract public class PlanNode extends TreeNode<PlanNode> {
// computeResourceProfile().
protected ResourceProfile nodeResourceProfile_ = ResourceProfile.invalid();
+ // List of query execution pipelines that this node executes as a part of.
+ protected List<PipelineMembership> pipelines_;
+
// sum of tupleIds_' avgSerializedSizes; set in computeStats()
protected float avgRowSize_;
@@ -184,6 +189,7 @@ abstract public class PlanNode extends TreeNode<PlanNode> {
}
public PlanNodeId getId() { return id_; }
+ public List<PipelineMembership> getPipelines() { return pipelines_; }
public void setId(PlanNodeId id) {
Preconditions.checkState(id_ == null);
id_ = id;
@@ -322,6 +328,17 @@ abstract public class PlanNode extends TreeNode<PlanNode> {
expBuilder.append("\n");
}
+ if (detailLevel.ordinal() >= TExplainLevel.EXTENDED.ordinal()) {
+ expBuilder.append(detailPrefix);
+ expBuilder.append("in pipelines: ");
+ List<String> pipelines = Lists.newArrayList();
+ for (PipelineMembership pipe: pipelines_) {
+ pipelines.add(pipe.getExplainString());
+ }
+ expBuilder.append(Joiner.on(", ").join(pipelines) + "\n");
+ }
+
+
// Print the children. Do not traverse into the children of an Exchange node to
// avoid crossing fragment boundaries.
if (traverseChildren) {
@@ -410,6 +427,10 @@ abstract public class PlanNode extends TreeNode<PlanNode> {
msg.setDisable_codegen(disableCodegen_);
Preconditions.checkState(nodeResourceProfile_.isValid());
msg.resource_profile = nodeResourceProfile_.toThrift();
+ msg.pipelines = Lists.newArrayList();
+ for (PipelineMembership pipe : pipelines_) {
+ msg.pipelines.add(pipe.toThrift());
+ }
toThrift(msg);
container.addToNodes(msg);
// For the purpose of the BE consider ExchangeNodes to have no children.
@@ -616,6 +637,58 @@ abstract public class PlanNode extends TreeNode<PlanNode> {
public boolean isBlockingNode() { return false; }
/**
+ * Fills in 'pipelines_' with the pipelines that this PlanNode is a member of.
+ *
+ * This is called only for nodes that are not part of the right branch of a
+ * subplan. All nodes in the right branch of a subplan belong to the same
+ * pipeline as the GETNEXT phase of the subplan.
+ */
+ public void computePipelineMembership() {
+ Preconditions.checkState(
+ children_.size() <= 1, "Plan nodes with > 1 child must override");
+ if (children_.size() == 0) {
+ // Leaf node, e.g. SCAN.
+ pipelines_ = Arrays.asList(
+ new PipelineMembership(id_, 0, TExecNodePhase.GETNEXT));
+ return;
+ }
+ children_.get(0).computePipelineMembership();
+ // Default behaviour for simple blocking or streaming nodes.
+ if (isBlockingNode()) {
+ // Executes as root of pipelines that child belongs to and leaf of another
+ // pipeline.
+ pipelines_ = Lists.newArrayList(
+ new PipelineMembership(id_, 0, TExecNodePhase.GETNEXT));
+ for (PipelineMembership childPipeline : children_.get(0).getPipelines()) {
+ if (childPipeline.getPhase() == TExecNodePhase.GETNEXT) {
+ pipelines_.add(new PipelineMembership(
+ childPipeline.getId(), childPipeline.getHeight() + 1, TExecNodePhase.OPEN));
+ }
+ }
+ } else {
+ // Streaming with child, e.g. SELECT. Executes as part of all pipelines the child
+ // belongs to.
+ pipelines_ = Lists.newArrayList();
+ for (PipelineMembership childPipeline : children_.get(0).getPipelines()) {
+ if (childPipeline.getPhase() == TExecNodePhase.GETNEXT) {
+ pipelines_.add(new PipelineMembership(
+ childPipeline.getId(), childPipeline.getHeight() + 1, TExecNodePhase.GETNEXT));
+ }
+ }
+ }
+ }
+
+ /**
+ * Called from SubplanNode to set the pipeline to 'pipelines'.
+ */
+ protected void setPipelinesRecursive(List<PipelineMembership> pipelines) {
+ pipelines_ = pipelines;
+ for (PlanNode child: children_) {
+ child.setPipelinesRecursive(pipelines);
+ }
+ }
+
+ /**
* Compute peak resources consumed when executing this PlanNode, initializing
* 'nodeResourceProfile_'. May only be called after this PlanNode has been placed in
* a PlanFragment because the cost computation is dependent on the enclosing fragment's
http://git-wip-us.apache.org/repos/asf/impala/blob/b7d509d7/fe/src/main/java/org/apache/impala/planner/Planner.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/planner/Planner.java b/fe/src/main/java/org/apache/impala/planner/Planner.java
index 818c526..9d867db 100644
--- a/fe/src/main/java/org/apache/impala/planner/Planner.java
+++ b/fe/src/main/java/org/apache/impala/planner/Planner.java
@@ -375,6 +375,7 @@ public class Planner {
maxPerHostPeakResources = maxPerHostPeakResources.sum(
fragment.getResourceProfile().multiply(fragment.getNumInstancesPerHost(mtDop)));
}
+ planRoots.get(0).computePipelineMembership();
Preconditions.checkState(maxPerHostPeakResources.getMemEstimateBytes() >= 0,
maxPerHostPeakResources.getMemEstimateBytes());
http://git-wip-us.apache.org/repos/asf/impala/blob/b7d509d7/fe/src/main/java/org/apache/impala/planner/SubplanNode.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/planner/SubplanNode.java b/fe/src/main/java/org/apache/impala/planner/SubplanNode.java
index e41290e..a22c397 100644
--- a/fe/src/main/java/org/apache/impala/planner/SubplanNode.java
+++ b/fe/src/main/java/org/apache/impala/planner/SubplanNode.java
@@ -19,12 +19,14 @@ package org.apache.impala.planner;
import org.apache.impala.analysis.Analyzer;
import org.apache.impala.common.InternalException;
+import org.apache.impala.thrift.TExecNodePhase;
import org.apache.impala.thrift.TExplainLevel;
import org.apache.impala.thrift.TPlanNode;
import org.apache.impala.thrift.TPlanNodeType;
import org.apache.impala.thrift.TQueryOptions;
import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
/**
* A SubplanNode evaluates its right child plan tree for every row from its left child,
@@ -116,6 +118,19 @@ public class SubplanNode extends PlanNode {
}
@Override
+ public void computePipelineMembership() {
+ children_.get(0).computePipelineMembership();
+ pipelines_ = Lists.newArrayList();
+ for (PipelineMembership leftPipeline : children_.get(0).getPipelines()) {
+ if (leftPipeline.getPhase() == TExecNodePhase.GETNEXT) {
+ pipelines_.add(new PipelineMembership(
+ leftPipeline.getId(), leftPipeline.getHeight() + 1, TExecNodePhase.GETNEXT));
+ }
+ }
+ children_.get(1).setPipelinesRecursive(pipelines_);
+ }
+
+ @Override
protected String getNodeExplainString(String prefix, String detailPrefix,
TExplainLevel detailLevel) {
StringBuilder output = new StringBuilder();
http://git-wip-us.apache.org/repos/asf/impala/blob/b7d509d7/fe/src/main/java/org/apache/impala/planner/UnionNode.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/planner/UnionNode.java b/fe/src/main/java/org/apache/impala/planner/UnionNode.java
index 52ce508..a8cda0d 100644
--- a/fe/src/main/java/org/apache/impala/planner/UnionNode.java
+++ b/fe/src/main/java/org/apache/impala/planner/UnionNode.java
@@ -26,6 +26,7 @@ import org.apache.impala.analysis.TupleDescriptor;
import org.apache.impala.analysis.TupleId;
import org.apache.impala.analysis.SlotDescriptor;
import org.apache.impala.analysis.SlotRef;
+import org.apache.impala.thrift.TExecNodePhase;
import org.apache.impala.thrift.TExplainLevel;
import org.apache.impala.thrift.TExpr;
import org.apache.impala.thrift.TPlanNode;
@@ -321,4 +322,20 @@ public class UnionNode extends PlanNode {
}
return output.toString();
}
+
+ @Override
+ public void computePipelineMembership() {
+ // The union streams each child's input through, so is part of every pipeline that
+ // its child is a part of.
+ pipelines_ = Lists.newArrayList();
+ for (PlanNode child: children_) {
+ child.computePipelineMembership();
+ for (PipelineMembership childPipeline : child.getPipelines()) {
+ if (childPipeline.getPhase() == TExecNodePhase.GETNEXT) {
+ pipelines_.add(new PipelineMembership(
+ childPipeline.getId(), childPipeline.getHeight() + 1, TExecNodePhase.GETNEXT));
+ }
+ }
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/impala/blob/b7d509d7/testdata/workloads/functional-planner/queries/PlannerTest/constant-folding.test
----------------------------------------------------------------------
diff --git a/testdata/workloads/functional-planner/queries/PlannerTest/constant-folding.test b/testdata/workloads/functional-planner/queries/PlannerTest/constant-folding.test
index 8e73f89..ae2644c 100644
--- a/testdata/workloads/functional-planner/queries/PlannerTest/constant-folding.test
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/constant-folding.test
@@ -12,38 +12,46 @@ PLAN-ROOT SINK
01:SUBPLAN
| mem-estimate=0B mem-reservation=0B thread-reservation=0
| tuple-ids=2,1,0 row-size=52B cardinality=1500000
+| in pipelines: 00(GETNEXT)
|
|--08:NESTED LOOP JOIN [CROSS JOIN]
| | mem-estimate=24B mem-reservation=0B thread-reservation=0
| | tuple-ids=2,1,0 row-size=52B cardinality=100
+| | in pipelines: 00(GETNEXT)
| |
| |--02:SINGULAR ROW SRC
| | parent-subplan=01
| | mem-estimate=0B mem-reservation=0B thread-reservation=0
| | tuple-ids=0 row-size=24B cardinality=1
+| | in pipelines: 00(GETNEXT)
| |
| 04:SUBPLAN
| | mem-estimate=0B mem-reservation=0B thread-reservation=0
| | tuple-ids=2,1 row-size=28B cardinality=100
+| | in pipelines: 00(GETNEXT)
| |
| |--07:NESTED LOOP JOIN [CROSS JOIN]
| | | mem-estimate=24B mem-reservation=0B thread-reservation=0
| | | tuple-ids=2,1 row-size=28B cardinality=10
+| | | in pipelines: 00(GETNEXT)
| | |
| | |--05:SINGULAR ROW SRC
| | | parent-subplan=04
| | | mem-estimate=0B mem-reservation=0B thread-reservation=0
| | | tuple-ids=1 row-size=24B cardinality=1
+| | | in pipelines: 00(GETNEXT)
| | |
| | 06:UNNEST [o.o_lineitems]
| | parent-subplan=04
| | mem-estimate=0B mem-reservation=0B thread-reservation=0
| | tuple-ids=2 row-size=0B cardinality=10
+| | in pipelines: 00(GETNEXT)
| |
| 03:UNNEST [c.c_orders o]
| parent-subplan=01
| mem-estimate=0B mem-reservation=0B thread-reservation=0
| tuple-ids=1 row-size=0B cardinality=10
+| in pipelines: 00(GETNEXT)
|
00:SCAN HDFS [tpch_nested_parquet.customer c]
partitions=1/1 files=4 size=288.98MB
@@ -61,6 +69,7 @@ PLAN-ROOT SINK
parquet dictionary predicates on o_lineitems: 20 + l_linenumber < 0
mem-estimate=264.00MB mem-reservation=16.00MB thread-reservation=1
tuple-ids=0 row-size=24B cardinality=15000
+ in pipelines: 00(GETNEXT)
====
# Test HBase scan node.
select * from functional_hbase.stringids
@@ -82,6 +91,7 @@ PLAN-ROOT SINK
columns: all
mem-estimate=1.00GB mem-reservation=0B thread-reservation=0
tuple-ids=0 row-size=119B cardinality=1
+ in pipelines: 00(GETNEXT)
====
# Test datasource scan node.
select * from functional.alltypes_datasource
@@ -97,6 +107,7 @@ data source predicates: tinyint_col < 256, int_col < 2
predicates: float_col != 0
mem-estimate=1.00GB mem-reservation=0B thread-reservation=0
tuple-ids=0 row-size=116B cardinality=500
+ in pipelines: 00(GETNEXT)
====
# Test aggregation.
select sum(1 + 1 + id)
@@ -117,6 +128,7 @@ PLAN-ROOT SINK
| having: sum(2 + id) <= 10, sum(2 + id) > 1, sum(2 + id) >= 5, 1048576 * count(*) % 2 = 0
| mem-estimate=10.00MB mem-reservation=1.94MB spill-buffer=64.00KB thread-reservation=0
| tuple-ids=1 row-size=17B cardinality=0
+| in pipelines: 01(GETNEXT), 00(OPEN)
|
00:SCAN HDFS [functional.alltypes]
partitions=24/24 files=24 size=478.45KB
@@ -127,6 +139,7 @@ PLAN-ROOT SINK
extrapolated-rows=disabled max-scan-range-rows=310
mem-estimate=128.00MB mem-reservation=32.00KB thread-reservation=1
tuple-ids=0 row-size=20B cardinality=7300
+ in pipelines: 00(GETNEXT)
====
# Test hash join.
select 1 from functional.alltypes a
@@ -147,6 +160,7 @@ PLAN-ROOT SINK
| other predicates: CAST(b.double_col AS DECIMAL(3,2)) > 11.1
| mem-estimate=1.94MB mem-reservation=1.94MB spill-buffer=64.00KB thread-reservation=0
| tuple-ids=0,1N row-size=28B cardinality=7300
+| in pipelines: 00(GETNEXT), 01(OPEN)
|
|--01:SCAN HDFS [functional.alltypes b]
| partitions=24/24 files=24 size=478.45KB
@@ -158,6 +172,7 @@ PLAN-ROOT SINK
| extrapolated-rows=disabled max-scan-range-rows=310
| mem-estimate=128.00MB mem-reservation=32.00KB thread-reservation=1
| tuple-ids=1 row-size=20B cardinality=730
+| in pipelines: 01(GETNEXT)
|
00:SCAN HDFS [functional.alltypes a]
partitions=24/24 files=24 size=478.45KB
@@ -168,6 +183,7 @@ PLAN-ROOT SINK
extrapolated-rows=disabled max-scan-range-rows=310
mem-estimate=128.00MB mem-reservation=32.00KB thread-reservation=1
tuple-ids=0 row-size=8B cardinality=7300
+ in pipelines: 00(GETNEXT)
====
# Test nested-loop join. Same as above but and with a disjunction in the On clause.
# The Where-clause predicate has the lhs/rhs flipped.
@@ -187,6 +203,7 @@ PLAN-ROOT SINK
| predicates: CAST(b.double_col AS DECIMAL(3,2)) > 11.1
| mem-estimate=14.26KB mem-reservation=0B thread-reservation=0
| tuple-ids=0,1N row-size=28B cardinality=7300
+| in pipelines: 00(GETNEXT), 01(OPEN)
|
|--01:SCAN HDFS [functional.alltypes b]
| partitions=24/24 files=24 size=478.45KB
@@ -198,6 +215,7 @@ PLAN-ROOT SINK
| extrapolated-rows=disabled max-scan-range-rows=310
| mem-estimate=128.00MB mem-reservation=32.00KB thread-reservation=1
| tuple-ids=1 row-size=20B cardinality=730
+| in pipelines: 01(GETNEXT)
|
00:SCAN HDFS [functional.alltypes a]
partitions=24/24 files=24 size=478.45KB
@@ -208,6 +226,7 @@ PLAN-ROOT SINK
extrapolated-rows=disabled max-scan-range-rows=310
mem-estimate=128.00MB mem-reservation=32.00KB thread-reservation=1
tuple-ids=0 row-size=8B cardinality=7300
+ in pipelines: 00(GETNEXT)
====
# Test distinct aggregation with grouping.
select sum(distinct 1 + 1 + id)
@@ -226,12 +245,14 @@ PLAN-ROOT SINK
| having: 1048576 * count(*) % 2 = 0
| mem-estimate=10.00MB mem-reservation=1.94MB spill-buffer=64.00KB thread-reservation=0
| tuple-ids=2 row-size=17B cardinality=0
+| in pipelines: 02(GETNEXT), 01(OPEN)
|
01:AGGREGATE
| output: count(*)
| group by: timestamp_col = TIMESTAMP '2016-11-15 00:00:00', 2 + id
| mem-estimate=10.00MB mem-reservation=1.94MB spill-buffer=64.00KB thread-reservation=0
| tuple-ids=1 row-size=17B cardinality=7300
+| in pipelines: 01(GETNEXT), 00(OPEN)
|
00:SCAN HDFS [functional.alltypes]
partitions=24/24 files=24 size=478.45KB
@@ -242,6 +263,7 @@ PLAN-ROOT SINK
extrapolated-rows=disabled max-scan-range-rows=310
mem-estimate=128.00MB mem-reservation=32.00KB thread-reservation=1
tuple-ids=0 row-size=20B cardinality=7300
+ in pipelines: 00(GETNEXT)
====
# Test non-grouping distinct aggregation.
select sum(distinct 1 + 1 + id)
@@ -258,12 +280,14 @@ PLAN-ROOT SINK
| having: 1048576 * zeroifnull(count(*)) % 2 = 0
| mem-estimate=10.00MB mem-reservation=0B spill-buffer=2.00MB thread-reservation=0
| tuple-ids=2 row-size=16B cardinality=0
+| in pipelines: 02(GETNEXT), 01(OPEN)
|
01:AGGREGATE
| output: count(*)
| group by: 2 + id
| mem-estimate=10.00MB mem-reservation=1.94MB spill-buffer=64.00KB thread-reservation=0
| tuple-ids=1 row-size=16B cardinality=7300
+| in pipelines: 01(GETNEXT), 00(OPEN)
|
00:SCAN HDFS [functional.alltypes]
partitions=24/24 files=24 size=478.45KB
@@ -274,6 +298,7 @@ PLAN-ROOT SINK
extrapolated-rows=disabled max-scan-range-rows=310
mem-estimate=128.00MB mem-reservation=32.00KB thread-reservation=1
tuple-ids=0 row-size=4B cardinality=7300
+ in pipelines: 00(GETNEXT)
====
# Test analytic eval node.
select first_value(1 + 1 + int_col - (1 - 1)) over
@@ -293,12 +318,14 @@ PLAN-ROOT SINK
| window: ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
| mem-estimate=4.00MB mem-reservation=4.00MB spill-buffer=2.00MB thread-reservation=0
| tuple-ids=3,2 row-size=61B cardinality=7300
+| in pipelines: 01(GETNEXT)
|
01:SORT
| order by: concat('ab', string_col) ASC NULLS FIRST, greatest(20, bigint_col) ASC
| materialized: concat('ab', string_col), greatest(20, bigint_col)
| mem-estimate=12.00MB mem-reservation=12.00MB spill-buffer=2.00MB thread-reservation=0
| tuple-ids=3 row-size=53B cardinality=7300
+| in pipelines: 01(GETNEXT), 00(OPEN)
|
00:SCAN HDFS [functional.alltypes]
partitions=24/24 files=24 size=478.45KB
@@ -309,6 +336,7 @@ PLAN-ROOT SINK
extrapolated-rows=disabled max-scan-range-rows=310
mem-estimate=128.00MB mem-reservation=32.00KB thread-reservation=1
tuple-ids=0 row-size=29B cardinality=7300
+ in pipelines: 00(GETNEXT)
====
# Test sort node.
select int_col from functional.alltypes
@@ -323,6 +351,7 @@ PLAN-ROOT SINK
| order by: id * 7.5 ASC
| mem-estimate=6.00MB mem-reservation=6.00MB spill-buffer=2.00MB thread-reservation=0
| tuple-ids=1 row-size=8B cardinality=7300
+| in pipelines: 01(GETNEXT), 00(OPEN)
|
00:SCAN HDFS [functional.alltypes]
partitions=24/24 files=24 size=478.45KB
@@ -333,6 +362,7 @@ PLAN-ROOT SINK
extrapolated-rows=disabled max-scan-range-rows=310
mem-estimate=128.00MB mem-reservation=32.00KB thread-reservation=1
tuple-ids=0 row-size=8B cardinality=7300
+ in pipelines: 00(GETNEXT)
====
# Test HDFS table sink.
insert into functional.alltypes (id, int_col) partition(year,month)
@@ -349,6 +379,7 @@ WRITE TO HDFS [functional.alltypes, OVERWRITE=false, PARTITION-KEYS=(CAST(3 + ye
| order by: CAST(3 + year AS INT) ASC NULLS LAST, CAST(month - -1 AS INT) ASC NULLS LAST
| mem-estimate=6.00MB mem-reservation=6.00MB spill-buffer=2.00MB thread-reservation=0
| tuple-ids=1 row-size=16B cardinality=100
+| in pipelines: 01(GETNEXT), 00(OPEN)
|
00:SCAN HDFS [functional.alltypessmall]
partitions=4/4 files=4 size=6.32KB
@@ -359,6 +390,7 @@ WRITE TO HDFS [functional.alltypes, OVERWRITE=false, PARTITION-KEYS=(CAST(3 + ye
extrapolated-rows=disabled max-scan-range-rows=25
mem-estimate=32.00MB mem-reservation=8.00KB thread-reservation=1
tuple-ids=0 row-size=16B cardinality=100
+ in pipelines: 00(GETNEXT)
====
# Constant folding does not work across query blocks.
select sum(id + c3) from
@@ -377,6 +409,7 @@ PLAN-ROOT SINK
| output: sum(id + 10 + 20 + 30)
| mem-estimate=10.00MB mem-reservation=0B spill-buffer=2.00MB thread-reservation=0
| tuple-ids=4 row-size=8B cardinality=1
+| in pipelines: 01(GETNEXT), 00(OPEN)
|
00:SCAN HDFS [functional.alltypes]
partitions=24/24 files=24 size=478.45KB
@@ -388,4 +421,5 @@ PLAN-ROOT SINK
limit: 2
mem-estimate=128.00MB mem-reservation=32.00KB thread-reservation=1
tuple-ids=0 row-size=4B cardinality=2
+ in pipelines: 00(GETNEXT)
====
http://git-wip-us.apache.org/repos/asf/impala/blob/b7d509d7/testdata/workloads/functional-planner/queries/PlannerTest/fk-pk-join-detection.test
----------------------------------------------------------------------
diff --git a/testdata/workloads/functional-planner/queries/PlannerTest/fk-pk-join-detection.test b/testdata/workloads/functional-planner/queries/PlannerTest/fk-pk-join-detection.test
index fdd6792..6de7b49 100644
--- a/testdata/workloads/functional-planner/queries/PlannerTest/fk-pk-join-detection.test
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/fk-pk-join-detection.test
@@ -15,6 +15,7 @@ PLAN-ROOT SINK
| runtime filters: RF000[bloom] <- c_customer_sk
| mem-estimate=8.50MB mem-reservation=8.50MB spill-buffer=512.00KB thread-reservation=0
| tuple-ids=0,1 row-size=355B cardinality=529700
+| in pipelines: 00(GETNEXT), 01(OPEN)
|
|--01:SCAN HDFS [tpcds.customer]
| partitions=1/1 files=1 size=12.60MB
@@ -25,6 +26,7 @@ PLAN-ROOT SINK
| extrapolated-rows=disabled max-scan-range-rows=100000
| mem-estimate=48.00MB mem-reservation=8.00MB thread-reservation=1
| tuple-ids=1 row-size=255B cardinality=16667
+| in pipelines: 01(GETNEXT)
|
00:SCAN HDFS [tpcds.store_sales]
partitions=1824/1824 files=1824 size=346.60MB
@@ -36,6 +38,7 @@ PLAN-ROOT SINK
extrapolated-rows=disabled max-scan-range-rows=130093
mem-estimate=128.00MB mem-reservation=8.00MB thread-reservation=1
tuple-ids=0 row-size=100B cardinality=2880404
+ in pipelines: 00(GETNEXT)
====
# Single-column FK/PK join detection on left outer join. The join cardinality
# is not reduced based on the selectivity of the rhs.
@@ -55,6 +58,7 @@ PLAN-ROOT SINK
| other predicates: c_salutation = 'Mrs.'
| mem-estimate=8.50MB mem-reservation=8.50MB spill-buffer=512.00KB thread-reservation=0
| tuple-ids=0,1N row-size=355B cardinality=2880404
+| in pipelines: 00(GETNEXT), 01(OPEN)
|
|--01:SCAN HDFS [tpcds.customer]
| partitions=1/1 files=1 size=12.60MB
@@ -65,6 +69,7 @@ PLAN-ROOT SINK
| extrapolated-rows=disabled max-scan-range-rows=100000
| mem-estimate=48.00MB mem-reservation=8.00MB thread-reservation=1
| tuple-ids=1 row-size=255B cardinality=16667
+| in pipelines: 01(GETNEXT)
|
00:SCAN HDFS [tpcds.store_sales]
partitions=1824/1824 files=1824 size=346.60MB
@@ -75,6 +80,7 @@ PLAN-ROOT SINK
extrapolated-rows=disabled max-scan-range-rows=130093
mem-estimate=128.00MB mem-reservation=8.00MB thread-reservation=1
tuple-ids=0 row-size=100B cardinality=2880404
+ in pipelines: 00(GETNEXT)
====
# Single-column FK/PK join detection on right outer join. The join cardinality
# is reduced based on the selectivity of the rhs.
@@ -94,6 +100,7 @@ PLAN-ROOT SINK
| runtime filters: RF000[bloom] <- c_customer_sk
| mem-estimate=8.50MB mem-reservation=8.50MB spill-buffer=512.00KB thread-reservation=0
| tuple-ids=0N,1 row-size=355B cardinality=529700
+| in pipelines: 00(GETNEXT), 01(OPEN)
|
|--01:SCAN HDFS [tpcds.customer]
| partitions=1/1 files=1 size=12.60MB
@@ -104,6 +111,7 @@ PLAN-ROOT SINK
| extrapolated-rows=disabled max-scan-range-rows=100000
| mem-estimate=48.00MB mem-reservation=8.00MB thread-reservation=1
| tuple-ids=1 row-size=255B cardinality=16667
+| in pipelines: 01(GETNEXT)
|
00:SCAN HDFS [tpcds.store_sales]
partitions=1824/1824 files=1824 size=346.60MB
@@ -115,6 +123,7 @@ PLAN-ROOT SINK
extrapolated-rows=disabled max-scan-range-rows=130093
mem-estimate=128.00MB mem-reservation=8.00MB thread-reservation=1
tuple-ids=0 row-size=100B cardinality=2880404
+ in pipelines: 00(GETNEXT)
====
# Multi-column FK/PK join detection
select * from
@@ -133,6 +142,7 @@ PLAN-ROOT SINK
| runtime filters: RF000[bloom] <- sr_item_sk, RF001[bloom] <- sr_ticket_number
| mem-estimate=4.75MB mem-reservation=4.75MB spill-buffer=256.00KB thread-reservation=0
| tuple-ids=0,1 row-size=188B cardinality=211838
+| in pipelines: 00(GETNEXT), 01(OPEN)
|
|--01:SCAN HDFS [tpcds.store_returns]
| partitions=1/1 files=1 size=31.19MB
@@ -143,6 +153,7 @@ PLAN-ROOT SINK
| extrapolated-rows=disabled max-scan-range-rows=287514
| mem-estimate=80.00MB mem-reservation=8.00MB thread-reservation=1
| tuple-ids=1 row-size=88B cardinality=28751
+| in pipelines: 01(GETNEXT)
|
00:SCAN HDFS [tpcds.store_sales]
partitions=1824/1824 files=1824 size=346.60MB
@@ -154,6 +165,7 @@ PLAN-ROOT SINK
extrapolated-rows=disabled max-scan-range-rows=130093
mem-estimate=128.00MB mem-reservation=8.00MB thread-reservation=1
tuple-ids=0 row-size=100B cardinality=2880404
+ in pipelines: 00(GETNEXT)
====
# Many-to-many join detection.
select * from
@@ -171,6 +183,7 @@ PLAN-ROOT SINK
| runtime filters: RF000[bloom] <- ws_sold_time_sk
| mem-estimate=108.67MB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
| tuple-ids=0,1 row-size=244B cardinality=44136418
+| in pipelines: 00(GETNEXT), 01(OPEN)
|
|--01:SCAN HDFS [tpcds.web_sales]
| partitions=1/1 files=1 size=140.07MB
@@ -180,6 +193,7 @@ PLAN-ROOT SINK
| extrapolated-rows=disabled max-scan-range-rows=657377
| mem-estimate=160.00MB mem-reservation=8.00MB thread-reservation=1
| tuple-ids=1 row-size=144B cardinality=719384
+| in pipelines: 01(GETNEXT)
|
00:SCAN HDFS [tpcds.store_sales]
partitions=1824/1824 files=1824 size=346.60MB
@@ -191,6 +205,7 @@ PLAN-ROOT SINK
extrapolated-rows=disabled max-scan-range-rows=130093
mem-estimate=128.00MB mem-reservation=8.00MB thread-reservation=1
tuple-ids=0 row-size=100B cardinality=2880404
+ in pipelines: 00(GETNEXT)
====
# PK/PK join is detected as FK/PK.
select * from
@@ -209,6 +224,7 @@ PLAN-ROOT SINK
| runtime filters: RF000[bloom] <- a.d_date_sk
| mem-estimate=17.00MB mem-reservation=17.00MB spill-buffer=1.00MB thread-reservation=0
| tuple-ids=1,0 row-size=606B cardinality=36525
+| in pipelines: 01(GETNEXT), 00(OPEN)
|
|--00:SCAN HDFS [tpcds.date_dim a]
| partitions=1/1 files=1 size=9.84MB
@@ -219,6 +235,7 @@ PLAN-ROOT SINK
| extrapolated-rows=disabled max-scan-range-rows=73049
| mem-estimate=48.00MB mem-reservation=8.00MB thread-reservation=1
| tuple-ids=0 row-size=303B cardinality=36525
+| in pipelines: 00(GETNEXT)
|
01:SCAN HDFS [tpcds.date_dim b]
partitions=1/1 files=1 size=9.84MB
@@ -229,6 +246,7 @@ PLAN-ROOT SINK
extrapolated-rows=disabled max-scan-range-rows=73049
mem-estimate=48.00MB mem-reservation=8.00MB thread-reservation=1
tuple-ids=1 row-size=303B cardinality=73049
+ in pipelines: 01(GETNEXT)
====
# Single query with various join types combined.
select 1 from
@@ -251,6 +269,7 @@ PLAN-ROOT SINK
| runtime filters: RF000[bloom] <- c_current_addr_sk
| mem-estimate=1.94MB mem-reservation=1.94MB spill-buffer=64.00KB thread-reservation=0
| tuple-ids=1,0,3,4,2 row-size=60B cardinality=19358
+| in pipelines: 01(GETNEXT), 02(OPEN)
|
|--02:SCAN HDFS [tpcds.customer]
| partitions=1/1 files=1 size=12.60MB
@@ -260,6 +279,7 @@ PLAN-ROOT SINK
| extrapolated-rows=disabled max-scan-range-rows=100000
| mem-estimate=48.00MB mem-reservation=8.00MB thread-reservation=1
| tuple-ids=2 row-size=4B cardinality=100000
+| in pipelines: 02(GETNEXT)
|
07:HASH JOIN [INNER JOIN]
| hash predicates: sr_returned_date_sk = d2.d_date_sk
@@ -267,6 +287,7 @@ PLAN-ROOT SINK
| runtime filters: RF002[bloom] <- d2.d_date_sk
| mem-estimate=1.94MB mem-reservation=1.94MB spill-buffer=64.00KB thread-reservation=0
| tuple-ids=1,0,3,4 row-size=56B cardinality=8131
+| in pipelines: 01(GETNEXT), 04(OPEN)
|
|--04:SCAN HDFS [tpcds.date_dim d2]
| partitions=1/1 files=1 size=9.84MB
@@ -276,6 +297,7 @@ PLAN-ROOT SINK
| extrapolated-rows=disabled max-scan-range-rows=73049
| mem-estimate=48.00MB mem-reservation=8.00MB thread-reservation=1
| tuple-ids=4 row-size=4B cardinality=73049
+| in pipelines: 04(GETNEXT)
|
06:HASH JOIN [INNER JOIN]
| hash predicates: sr_item_sk = ss_item_sk, sr_ticket_number = ss_ticket_number
@@ -283,6 +305,7 @@ PLAN-ROOT SINK
| runtime filters: RF004[bloom] <- ss_item_sk, RF005[bloom] <- ss_ticket_number
| mem-estimate=1.94MB mem-reservation=1.94MB spill-buffer=64.00KB thread-reservation=0
| tuple-ids=1,0,3 row-size=52B cardinality=8131
+| in pipelines: 01(GETNEXT), 00(OPEN)
|
|--05:HASH JOIN [INNER JOIN]
| | hash predicates: ss_sold_date_sk = d1.d_date_sk
@@ -290,6 +313,7 @@ PLAN-ROOT SINK
| | runtime filters: RF008[bloom] <- d1.d_date_sk
| | mem-estimate=1.94MB mem-reservation=1.94MB spill-buffer=64.00KB thread-reservation=0
| | tuple-ids=0,3 row-size=32B cardinality=11055
+| | in pipelines: 00(GETNEXT), 03(OPEN)
| |
| |--03:SCAN HDFS [tpcds.date_dim d1]
| | partitions=1/1 files=1 size=9.84MB
@@ -300,6 +324,7 @@ PLAN-ROOT SINK
| | extrapolated-rows=disabled max-scan-range-rows=73049
| | mem-estimate=48.00MB mem-reservation=8.00MB thread-reservation=1
| | tuple-ids=3 row-size=8B cardinality=7
+| | in pipelines: 03(GETNEXT)
| |
| 00:SCAN HDFS [tpcds.store_sales]
| partitions=1824/1824 files=1824 size=346.60MB
@@ -311,6 +336,7 @@ PLAN-ROOT SINK
| extrapolated-rows=disabled max-scan-range-rows=130093
| mem-estimate=128.00MB mem-reservation=8.00MB thread-reservation=1
| tuple-ids=0 row-size=24B cardinality=2880404
+| in pipelines: 00(GETNEXT)
|
01:SCAN HDFS [tpcds.store_returns]
partitions=1/1 files=1 size=31.19MB
@@ -321,6 +347,7 @@ PLAN-ROOT SINK
extrapolated-rows=disabled max-scan-range-rows=287514
mem-estimate=80.00MB mem-reservation=8.00MB thread-reservation=1
tuple-ids=1 row-size=20B cardinality=287514
+ in pipelines: 01(GETNEXT)
====
# Assumed FK/PK join becasue of non-trivial equi-join exprs.
select * from
@@ -338,6 +365,7 @@ PLAN-ROOT SINK
| runtime filters: RF000[bloom] <- c_customer_sk / 100
| mem-estimate=34.00MB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
| tuple-ids=0,1 row-size=355B cardinality=2880404
+| in pipelines: 00(GETNEXT), 01(OPEN)
|
|--01:SCAN HDFS [tpcds.customer]
| partitions=1/1 files=1 size=12.60MB
@@ -347,6 +375,7 @@ PLAN-ROOT SINK
| extrapolated-rows=disabled max-scan-range-rows=100000
| mem-estimate=48.00MB mem-reservation=8.00MB thread-reservation=1
| tuple-ids=1 row-size=255B cardinality=100000
+| in pipelines: 01(GETNEXT)
|
00:SCAN HDFS [tpcds.store_sales]
partitions=1824/1824 files=1824 size=346.60MB
@@ -358,6 +387,7 @@ PLAN-ROOT SINK
extrapolated-rows=disabled max-scan-range-rows=130093
mem-estimate=128.00MB mem-reservation=8.00MB thread-reservation=1
tuple-ids=0 row-size=100B cardinality=2880404
+ in pipelines: 00(GETNEXT)
====
# Assumed FK/PK join due to missing stats on the rhs. Join cardinality is equal to
# the lhs cardinality.
@@ -376,6 +406,7 @@ PLAN-ROOT SINK
| runtime filters: RF000[bloom] <- c_customer_sk
| mem-estimate=2.00GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
| tuple-ids=0,1 row-size=8B cardinality=2880404
+| in pipelines: 00(GETNEXT), 01(OPEN)
|
|--01:SCAN HDFS [tpcds_seq_snap.customer]
| partitions=1/1 files=1 size=8.58MB
@@ -385,6 +416,7 @@ PLAN-ROOT SINK
| extrapolated-rows=disabled max-scan-range-rows=unavailable
| mem-estimate=48.00MB mem-reservation=8.00MB thread-reservation=1
| tuple-ids=1 row-size=4B cardinality=unavailable
+| in pipelines: 01(GETNEXT)
|
00:SCAN HDFS [tpcds.store_sales]
partitions=1824/1824 files=1824 size=346.60MB
@@ -396,6 +428,7 @@ PLAN-ROOT SINK
extrapolated-rows=disabled max-scan-range-rows=130093
mem-estimate=128.00MB mem-reservation=8.00MB thread-reservation=1
tuple-ids=0 row-size=4B cardinality=2880404
+ in pipelines: 00(GETNEXT)
====
# Assumed FK/PK join due to missing stats on the lhs. Join cardinality is unknown.
select /* +straight_join */ 1 from
@@ -413,6 +446,7 @@ PLAN-ROOT SINK
| runtime filters: RF000[bloom] <- c_customer_sk
| mem-estimate=1.94MB mem-reservation=1.94MB spill-buffer=64.00KB thread-reservation=0
| tuple-ids=0,1 row-size=8B cardinality=unavailable
+| in pipelines: 00(GETNEXT), 01(OPEN)
|
|--01:SCAN HDFS [tpcds.customer]
| partitions=1/1 files=1 size=12.60MB
@@ -422,6 +456,7 @@ PLAN-ROOT SINK
| extrapolated-rows=disabled max-scan-range-rows=100000
| mem-estimate=48.00MB mem-reservation=8.00MB thread-reservation=1
| tuple-ids=1 row-size=4B cardinality=100000
+| in pipelines: 01(GETNEXT)
|
00:SCAN HDFS [tpcds_seq_snap.store_sales]
partitions=1824/1824 files=1824 size=212.80MB
@@ -433,6 +468,7 @@ PLAN-ROOT SINK
extrapolated-rows=disabled max-scan-range-rows=unavailable
mem-estimate=128.00MB mem-reservation=8.00MB thread-reservation=1
tuple-ids=0 row-size=4B cardinality=unavailable
+ in pipelines: 00(GETNEXT)
====
# Join is detected as many-to-many even though the rhs join columns
# are in a group by on the rhs input.
@@ -452,11 +488,13 @@ PLAN-ROOT SINK
| runtime filters: RF000[bloom] <- ws_sold_time_sk
| mem-estimate=1.94MB mem-reservation=1.94MB spill-buffer=64.00KB thread-reservation=0
| tuple-ids=0,2 row-size=104B cardinality=2440073
+| in pipelines: 00(GETNEXT), 02(OPEN)
|
|--02:AGGREGATE [FINALIZE]
| | group by: ws_sold_time_sk
| | mem-estimate=10.00MB mem-reservation=1.94MB spill-buffer=64.00KB thread-reservation=0
| | tuple-ids=2 row-size=4B cardinality=39771
+| | in pipelines: 02(GETNEXT), 01(OPEN)
| |
| 01:SCAN HDFS [tpcds.web_sales]
| partitions=1/1 files=1 size=140.07MB
@@ -466,6 +504,7 @@ PLAN-ROOT SINK
| extrapolated-rows=disabled max-scan-range-rows=657377
| mem-estimate=160.00MB mem-reservation=8.00MB thread-reservation=1
| tuple-ids=1 row-size=4B cardinality=719384
+| in pipelines: 01(GETNEXT)
|
00:SCAN HDFS [tpcds.store_sales]
partitions=1824/1824 files=1824 size=346.60MB
@@ -477,4 +516,5 @@ PLAN-ROOT SINK
extrapolated-rows=disabled max-scan-range-rows=130093
mem-estimate=128.00MB mem-reservation=8.00MB thread-reservation=1
tuple-ids=0 row-size=100B cardinality=2880404
+ in pipelines: 00(GETNEXT)
====
http://git-wip-us.apache.org/repos/asf/impala/blob/b7d509d7/testdata/workloads/functional-planner/queries/PlannerTest/kudu-selectivity.test
----------------------------------------------------------------------
diff --git a/testdata/workloads/functional-planner/queries/PlannerTest/kudu-selectivity.test b/testdata/workloads/functional-planner/queries/PlannerTest/kudu-selectivity.test
index cf7dfe6..e8c91b0 100644
--- a/testdata/workloads/functional-planner/queries/PlannerTest/kudu-selectivity.test
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/kudu-selectivity.test
@@ -9,6 +9,7 @@ Per-Host Resources: mem-estimate=0B mem-reservation=0B thread-reservation=2
kudu predicates: id = '8600000US00601'
mem-estimate=0B mem-reservation=0B thread-reservation=1
tuple-ids=0 row-size=124B cardinality=1
+ in pipelines: 00(GETNEXT)
---- DISTRIBUTEDPLAN
F01:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1
Per-Host Resources: mem-estimate=0B mem-reservation=0B thread-reservation=1
@@ -18,6 +19,7 @@ Per-Host Resources: mem-estimate=0B mem-reservation=0B thread-reservation=1
01:EXCHANGE [UNPARTITIONED]
mem-estimate=0B mem-reservation=0B thread-reservation=0
tuple-ids=0 row-size=124B cardinality=1
+ in pipelines: 00(GETNEXT)
F00:PLAN FRAGMENT [RANDOM] hosts=3 instances=3
Per-Host Resources: mem-estimate=0B mem-reservation=0B thread-reservation=2
@@ -27,6 +29,7 @@ Per-Host Resources: mem-estimate=0B mem-reservation=0B thread-reservation=2
kudu predicates: id = '8600000US00601'
mem-estimate=0B mem-reservation=0B thread-reservation=1
tuple-ids=0 row-size=124B cardinality=1
+ in pipelines: 00(GETNEXT)
====
# The cardinality from "zip = '2'" should dominate.
select * from functional_kudu.zipcode_incomes where id != '1' and zip = '2'
@@ -41,6 +44,7 @@ Per-Host Resources: mem-estimate=0B mem-reservation=0B thread-reservation=2
kudu predicates: zip = '2'
mem-estimate=0B mem-reservation=0B thread-reservation=1
tuple-ids=0 row-size=124B cardinality=1
+ in pipelines: 00(GETNEXT)
---- DISTRIBUTEDPLAN
F01:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1
Per-Host Resources: mem-estimate=0B mem-reservation=0B thread-reservation=1
@@ -50,6 +54,7 @@ Per-Host Resources: mem-estimate=0B mem-reservation=0B thread-reservation=1
01:EXCHANGE [UNPARTITIONED]
mem-estimate=0B mem-reservation=0B thread-reservation=0
tuple-ids=0 row-size=124B cardinality=1
+ in pipelines: 00(GETNEXT)
F00:PLAN FRAGMENT [RANDOM] hosts=3 instances=3
Per-Host Resources: mem-estimate=0B mem-reservation=0B thread-reservation=2
@@ -60,6 +65,7 @@ Per-Host Resources: mem-estimate=0B mem-reservation=0B thread-reservation=2
kudu predicates: zip = '2'
mem-estimate=0B mem-reservation=0B thread-reservation=1
tuple-ids=0 row-size=124B cardinality=1
+ in pipelines: 00(GETNEXT)
====
select * from functional_kudu.zipcode_incomes where id > '1' and zip > '2'
---- PLAN
@@ -72,6 +78,7 @@ Per-Host Resources: mem-estimate=0B mem-reservation=0B thread-reservation=2
kudu predicates: zip > '2', id > '1'
mem-estimate=0B mem-reservation=0B thread-reservation=1
tuple-ids=0 row-size=124B cardinality=3317
+ in pipelines: 00(GETNEXT)
---- DISTRIBUTEDPLAN
F01:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1
Per-Host Resources: mem-estimate=0B mem-reservation=0B thread-reservation=1
@@ -81,6 +88,7 @@ Per-Host Resources: mem-estimate=0B mem-reservation=0B thread-reservation=1
01:EXCHANGE [UNPARTITIONED]
mem-estimate=0B mem-reservation=0B thread-reservation=0
tuple-ids=0 row-size=124B cardinality=3317
+ in pipelines: 00(GETNEXT)
F00:PLAN FRAGMENT [RANDOM] hosts=3 instances=3
Per-Host Resources: mem-estimate=0B mem-reservation=0B thread-reservation=2
@@ -90,6 +98,7 @@ Per-Host Resources: mem-estimate=0B mem-reservation=0B thread-reservation=2
kudu predicates: zip > '2', id > '1'
mem-estimate=0B mem-reservation=0B thread-reservation=1
tuple-ids=0 row-size=124B cardinality=3317
+ in pipelines: 00(GETNEXT)
====
select * from functional_kudu.zipcode_incomes where id = '1' or id = '2' or zip = '3'
---- PLAN
@@ -102,6 +111,7 @@ Per-Host Resources: mem-estimate=0B mem-reservation=0B thread-reservation=2
predicates: id IN ('1', '2') OR zip = '3'
mem-estimate=0B mem-reservation=0B thread-reservation=1
tuple-ids=0 row-size=124B cardinality=3
+ in pipelines: 00(GETNEXT)
---- DISTRIBUTEDPLAN
F01:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1
Per-Host Resources: mem-estimate=0B mem-reservation=0B thread-reservation=1
@@ -111,6 +121,7 @@ Per-Host Resources: mem-estimate=0B mem-reservation=0B thread-reservation=1
01:EXCHANGE [UNPARTITIONED]
mem-estimate=0B mem-reservation=0B thread-reservation=0
tuple-ids=0 row-size=124B cardinality=3
+ in pipelines: 00(GETNEXT)
F00:PLAN FRAGMENT [RANDOM] hosts=3 instances=3
Per-Host Resources: mem-estimate=0B mem-reservation=0B thread-reservation=2
@@ -120,6 +131,7 @@ Per-Host Resources: mem-estimate=0B mem-reservation=0B thread-reservation=2
predicates: id IN ('1', '2') OR zip = '3'
mem-estimate=0B mem-reservation=0B thread-reservation=1
tuple-ids=0 row-size=124B cardinality=3
+ in pipelines: 00(GETNEXT)
====
# Test alltypes in-list predicates.
select * from functional_kudu.alltypes where
@@ -156,6 +168,7 @@ Per-Host Resources: mem-estimate=0B mem-reservation=0B thread-reservation=2
kudu predicates: double_col IN (0.0), float_col IN (0.0), bigint_col IN (1, 2), int_col IN (1, 2), smallint_col IN (0, 2), string_col IN ('foo', 'foo '), tinyint_col IN (1, 2), bool_col IN (TRUE)
mem-estimate=0B mem-reservation=0B thread-reservation=1
tuple-ids=0 row-size=97B cardinality=5
+ in pipelines: 00(GETNEXT)
====
# Test decimal in-list predicates.
select * from functional_kudu.decimal_tbl where d1 in (1234, 12345);
@@ -169,6 +182,7 @@ Per-Host Resources: mem-estimate=0B mem-reservation=0B thread-reservation=2
kudu predicates: d1 IN (1234, 12345)
mem-estimate=0B mem-reservation=0B thread-reservation=1
tuple-ids=0 row-size=56B cardinality=2
+ in pipelines: 00(GETNEXT)
====
select * from functional_kudu.alltypes where
tinyint_col is not null and
@@ -187,6 +201,7 @@ Per-Host Resources: mem-estimate=0B mem-reservation=0B thread-reservation=2
kudu predicates: smallint_col IS NULL, tinyint_col IS NOT NULL, timestamp_col > TIMESTAMP '1987-05-19 00:00:00.000000010'
mem-estimate=0B mem-reservation=0B thread-reservation=1
tuple-ids=0 row-size=97B cardinality=730
+ in pipelines: 00(GETNEXT)
====
select * from functional_kudu.alltypes where
timestamp_col in (cast('2010-03-01 00:00:00' as timestamp),
@@ -201,6 +216,7 @@ Per-Host Resources: mem-estimate=0B mem-reservation=0B thread-reservation=2
kudu predicates: timestamp_col IN (TIMESTAMP '2010-03-01 00:00:00', TIMESTAMP '2010-03-01 00:01:00')
mem-estimate=0B mem-reservation=0B thread-reservation=1
tuple-ids=0 row-size=97B cardinality=1
+ in pipelines: 00(GETNEXT)
====
select * from functional_kudu.alltypes where
timestamp_col in (cast('2010-03-01 00:00:00' as timestamp),
@@ -216,4 +232,5 @@ Per-Host Resources: mem-estimate=0B mem-reservation=0B thread-reservation=2
predicates: timestamp_col IN (TIMESTAMP '2010-03-01 00:00:00', NULL, TIMESTAMP '2010-03-01 00:01:00')
mem-estimate=0B mem-reservation=0B thread-reservation=1
tuple-ids=0 row-size=97B cardinality=3
+ in pipelines: 00(GETNEXT)
====
http://git-wip-us.apache.org/repos/asf/impala/blob/b7d509d7/testdata/workloads/functional-planner/queries/PlannerTest/max-row-size.test
----------------------------------------------------------------------
diff --git a/testdata/workloads/functional-planner/queries/PlannerTest/max-row-size.test b/testdata/workloads/functional-planner/queries/PlannerTest/max-row-size.test
index c228c0b..f82f1be 100644
--- a/testdata/workloads/functional-planner/queries/PlannerTest/max-row-size.test
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/max-row-size.test
@@ -15,6 +15,7 @@ PLAN-ROOT SINK
04:EXCHANGE [UNPARTITIONED]
| mem-estimate=0B mem-reservation=0B thread-reservation=0
| tuple-ids=0,1 row-size=355B cardinality=150000
+| in pipelines: 00(GETNEXT)
|
F00:PLAN FRAGMENT [RANDOM] hosts=1 instances=1
Per-Host Resources: mem-estimate=41.94MB mem-reservation=33.94MB thread-reservation=2 runtime-filters-memory=1.00MB
@@ -24,10 +25,12 @@ Per-Host Resources: mem-estimate=41.94MB mem-reservation=33.94MB thread-reservat
| runtime filters: RF000[bloom] <- n_nationkey
| mem-estimate=16.94MB mem-reservation=16.94MB spill-buffer=64.00KB thread-reservation=0
| tuple-ids=0,1 row-size=355B cardinality=150000
+| in pipelines: 00(GETNEXT), 01(OPEN)
|
|--03:EXCHANGE [BROADCAST]
| | mem-estimate=0B mem-reservation=0B thread-reservation=0
| | tuple-ids=1 row-size=117B cardinality=25
+| | in pipelines: 01(GETNEXT)
| |
| F01:PLAN FRAGMENT [RANDOM] hosts=1 instances=1
| Per-Host Resources: mem-estimate=16.00MB mem-reservation=32.00KB thread-reservation=2
@@ -39,6 +42,7 @@ Per-Host Resources: mem-estimate=41.94MB mem-reservation=33.94MB thread-reservat
| extrapolated-rows=disabled max-scan-range-rows=25
| mem-estimate=16.00MB mem-reservation=32.00KB thread-reservation=1
| tuple-ids=1 row-size=117B cardinality=25
+| in pipelines: 01(GETNEXT)
|
00:SCAN HDFS [tpch_parquet.customer, RANDOM]
partitions=1/1 files=1 size=12.31MB
@@ -49,6 +53,7 @@ Per-Host Resources: mem-estimate=41.94MB mem-reservation=33.94MB thread-reservat
extrapolated-rows=disabled max-scan-range-rows=150000
mem-estimate=24.00MB mem-reservation=16.00MB thread-reservation=1
tuple-ids=0 row-size=238B cardinality=150000
+ in pipelines: 00(GETNEXT)
====
# Join with large build side.
# Uses default buffer size. Two max-size buffers are required.
@@ -67,6 +72,7 @@ PLAN-ROOT SINK
04:EXCHANGE [UNPARTITIONED]
| mem-estimate=0B mem-reservation=0B thread-reservation=0
| tuple-ids=0,1N row-size=454B cardinality=6001215
+| in pipelines: 00(GETNEXT)
|
F00:PLAN FRAGMENT [RANDOM] hosts=3 instances=3
Per-Host Resources: mem-estimate=380.41MB mem-reservation=86.00MB thread-reservation=2
@@ -75,10 +81,12 @@ Per-Host Resources: mem-estimate=380.41MB mem-reservation=86.00MB thread-reserva
| fk/pk conjuncts: l_orderkey = o_orderkey
| mem-estimate=300.41MB mem-reservation=46.00MB spill-buffer=2.00MB thread-reservation=0
| tuple-ids=0,1N row-size=454B cardinality=6001215
+| in pipelines: 00(GETNEXT), 01(OPEN)
|
|--03:EXCHANGE [BROADCAST]
| | mem-estimate=0B mem-reservation=0B thread-reservation=0
| | tuple-ids=1 row-size=191B cardinality=1500000
+| | in pipelines: 01(GETNEXT)
| |
| F01:PLAN FRAGMENT [RANDOM] hosts=2 instances=2
| Per-Host Resources: mem-estimate=40.00MB mem-reservation=24.00MB thread-reservation=2
@@ -90,6 +98,7 @@ Per-Host Resources: mem-estimate=380.41MB mem-reservation=86.00MB thread-reserva
| extrapolated-rows=disabled max-scan-range-rows=1177135
| mem-estimate=40.00MB mem-reservation=24.00MB thread-reservation=1
| tuple-ids=1 row-size=191B cardinality=1500000
+| in pipelines: 01(GETNEXT)
|
00:SCAN HDFS [tpch_parquet.lineitem, RANDOM]
partitions=1/1 files=3 size=193.71MB
@@ -99,6 +108,7 @@ Per-Host Resources: mem-estimate=380.41MB mem-reservation=86.00MB thread-reserva
extrapolated-rows=disabled max-scan-range-rows=2141802
mem-estimate=80.00MB mem-reservation=40.00MB thread-reservation=1
tuple-ids=0 row-size=263B cardinality=6001215
+ in pipelines: 00(GETNEXT)
====
# Null-aware anti-join with medium build side.
# Uses intermediate buffer size plus two max-size buffers.
@@ -116,6 +126,7 @@ PLAN-ROOT SINK
04:EXCHANGE [UNPARTITIONED]
| mem-estimate=0B mem-reservation=0B thread-reservation=0
| tuple-ids=0 row-size=263B cardinality=6001215
+| in pipelines: 00(GETNEXT)
|
F00:PLAN FRAGMENT [RANDOM] hosts=3 instances=3
Per-Host Resources: mem-estimate=114.00MB mem-reservation=74.00MB thread-reservation=2
@@ -123,10 +134,12 @@ Per-Host Resources: mem-estimate=114.00MB mem-reservation=74.00MB thread-reserva
| hash predicates: l_orderkey = o_orderkey
| mem-estimate=34.00MB mem-reservation=34.00MB spill-buffer=1.00MB thread-reservation=0
| tuple-ids=0 row-size=263B cardinality=6001215
+| in pipelines: 00(GETNEXT), 01(OPEN)
|
|--03:EXCHANGE [BROADCAST]
| | mem-estimate=0B mem-reservation=0B thread-reservation=0
| | tuple-ids=1 row-size=8B cardinality=1500000
+| | in pipelines: 01(GETNEXT)
| |
| F01:PLAN FRAGMENT [RANDOM] hosts=2 instances=2
| Per-Host Resources: mem-estimate=40.00MB mem-reservation=4.00MB thread-reservation=2
@@ -138,6 +151,7 @@ Per-Host Resources: mem-estimate=114.00MB mem-reservation=74.00MB thread-reserva
| extrapolated-rows=disabled max-scan-range-rows=1177135
| mem-estimate=40.00MB mem-reservation=4.00MB thread-reservation=1
| tuple-ids=1 row-size=8B cardinality=1500000
+| in pipelines: 01(GETNEXT)
|
00:SCAN HDFS [tpch_parquet.lineitem, RANDOM]
partitions=1/1 files=3 size=193.71MB
@@ -147,6 +161,7 @@ Per-Host Resources: mem-estimate=114.00MB mem-reservation=74.00MB thread-reserva
extrapolated-rows=disabled max-scan-range-rows=2141802
mem-estimate=80.00MB mem-reservation=40.00MB thread-reservation=1
tuple-ids=0 row-size=263B cardinality=6001215
+ in pipelines: 00(GETNEXT)
====
# Mid NDV aggregation.
# Uses smaller-than-default buffer size. Two max-size buffers are required.
@@ -167,6 +182,7 @@ PLAN-ROOT SINK
08:EXCHANGE [UNPARTITIONED]
| mem-estimate=0B mem-reservation=0B thread-reservation=0
| tuple-ids=2 row-size=33B cardinality=4690314
+| in pipelines: 07(GETNEXT)
|
F03:PLAN FRAGMENT [HASH(l_orderkey,o_orderstatus)] hosts=3 instances=3
Per-Host Resources: mem-estimate=46.00MB mem-reservation=46.00MB thread-reservation=1
@@ -176,10 +192,12 @@ Per-Host Resources: mem-estimate=46.00MB mem-reservation=46.00MB thread-reservat
| having: count(*) = 1
| mem-estimate=46.00MB mem-reservation=46.00MB spill-buffer=2.00MB thread-reservation=0
| tuple-ids=2 row-size=33B cardinality=4690314
+| in pipelines: 07(GETNEXT), 00(OPEN)
|
06:EXCHANGE [HASH(l_orderkey,o_orderstatus)]
| mem-estimate=0B mem-reservation=0B thread-reservation=0
| tuple-ids=2 row-size=33B cardinality=4690314
+| in pipelines: 00(GETNEXT)
|
F02:PLAN FRAGMENT [HASH(l_orderkey)] hosts=3 instances=3
Per-Host Resources: mem-estimate=86.12MB mem-reservation=66.00MB thread-reservation=1 runtime-filters-memory=1.00MB
@@ -188,6 +206,7 @@ Per-Host Resources: mem-estimate=86.12MB mem-reservation=66.00MB thread-reservat
| group by: l_orderkey, o_orderstatus
| mem-estimate=54.12MB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
| tuple-ids=2 row-size=33B cardinality=4690314
+| in pipelines: 00(GETNEXT)
|
02:HASH JOIN [INNER JOIN, PARTITIONED]
| hash predicates: l_orderkey = o_orderkey
@@ -195,10 +214,12 @@ Per-Host Resources: mem-estimate=86.12MB mem-reservation=66.00MB thread-reservat
| runtime filters: RF000[bloom] <- o_orderkey
| mem-estimate=31.00MB mem-reservation=31.00MB spill-buffer=1.00MB thread-reservation=0
| tuple-ids=0,1 row-size=33B cardinality=5757710
+| in pipelines: 00(GETNEXT), 01(OPEN)
|
|--05:EXCHANGE [HASH(o_orderkey)]
| | mem-estimate=0B mem-reservation=0B thread-reservation=0
| | tuple-ids=1 row-size=25B cardinality=1500000
+| | in pipelines: 01(GETNEXT)
| |
| F01:PLAN FRAGMENT [RANDOM] hosts=2 instances=2
| Per-Host Resources: mem-estimate=40.00MB mem-reservation=8.00MB thread-reservation=2
@@ -210,10 +231,12 @@ Per-Host Resources: mem-estimate=86.12MB mem-reservation=66.00MB thread-reservat
| extrapolated-rows=disabled max-scan-range-rows=1177135
| mem-estimate=40.00MB mem-reservation=8.00MB thread-reservation=1
| tuple-ids=1 row-size=25B cardinality=1500000
+| in pipelines: 01(GETNEXT)
|
04:EXCHANGE [HASH(l_orderkey)]
| mem-estimate=0B mem-reservation=0B thread-reservation=0
| tuple-ids=0 row-size=8B cardinality=6001215
+| in pipelines: 00(GETNEXT)
|
F00:PLAN FRAGMENT [RANDOM] hosts=3 instances=3
Per-Host Resources: mem-estimate=81.00MB mem-reservation=5.00MB thread-reservation=2 runtime-filters-memory=1.00MB
@@ -226,6 +249,7 @@ Per-Host Resources: mem-estimate=81.00MB mem-reservation=5.00MB thread-reservati
extrapolated-rows=disabled max-scan-range-rows=2141802
mem-estimate=80.00MB mem-reservation=4.00MB thread-reservation=1
tuple-ids=0 row-size=8B cardinality=6001215
+ in pipelines: 00(GETNEXT)
====
# High NDV aggregation.
# Uses default buffer size. Two max-size buffers are required.
@@ -243,6 +267,7 @@ PLAN-ROOT SINK
04:EXCHANGE [UNPARTITIONED]
| mem-estimate=0B mem-reservation=0B thread-reservation=0
| tuple-ids=1 row-size=263B cardinality=6001215
+| in pipelines: 03(GETNEXT)
|
F01:PLAN FRAGMENT [HASH(tpch_parquet.lineitem.l_orderkey,tpch_parquet.lineitem.l_partkey,tpch_parquet.lineitem.l_suppkey,tpch_parquet.lineitem.l_linenumber,tpch_parquet.lineitem.l_quantity,tpch_parquet.lineitem.l_extendedprice,tpch_parquet.lineitem.l_discount,tpch_parquet.lineitem.l_tax,tpch_parquet.lineitem.l_returnflag,tpch_parquet.lineitem.l_linestatus,tpch_parquet.lineitem.l_shipdate,tpch_parquet.lineitem.l_commitdate,tpch_parquet.lineitem.l_receiptdate,tpch_parquet.lineitem.l_shipinstruct,tpch_parquet.lineitem.l_shipmode,tpch_parquet.lineitem.l_comment)] hosts=3 instances=3
Per-Host Resources: mem-estimate=1.62GB mem-reservation=46.00MB thread-reservation=1
@@ -250,10 +275,12 @@ Per-Host Resources: mem-estimate=1.62GB mem-reservation=46.00MB thread-reservati
| group by: tpch_parquet.lineitem.l_orderkey, tpch_parquet.lineitem.l_partkey, tpch_parquet.lineitem.l_suppkey, tpch_parquet.lineitem.l_linenumber, tpch_parquet.lineitem.l_quantity, tpch_parquet.lineitem.l_extendedprice, tpch_parquet.lineitem.l_discount, tpch_parquet.lineitem.l_tax, tpch_parquet.lineitem.l_returnflag, tpch_parquet.lineitem.l_linestatus, tpch_parquet.lineitem.l_shipdate, tpch_parquet.lineitem.l_commitdate, tpch_parquet.lineitem.l_receiptdate, tpch_parquet.lineitem.l_shipinstruct, tpch_parquet.lineitem.l_shipmode, tpch_parquet.lineitem.l_comment
| mem-estimate=1.62GB mem-reservation=46.00MB spill-buffer=2.00MB thread-reservation=0
| tuple-ids=1 row-size=263B cardinality=6001215
+| in pipelines: 03(GETNEXT), 00(OPEN)
|
02:EXCHANGE [HASH(tpch_parquet.lineitem.l_orderkey,tpch_parquet.lineitem.l_partkey,tpch_parquet.lineitem.l_suppkey,tpch_parquet.lineitem.l_linenumber,tpch_parquet.lineitem.l_quantity,tpch_parquet.lineitem.l_extendedprice,tpch_parquet.lineitem.l_discount,tpch_parquet.lineitem.l_tax,tpch_parquet.lineitem.l_returnflag,tpch_parquet.lineitem.l_linestatus,tpch_parquet.lineitem.l_shipdate,tpch_parquet.lineitem.l_commitdate,tpch_parquet.lineitem.l_receiptdate,tpch_parquet.lineitem.l_shipinstruct,tpch_parquet.lineitem.l_shipmode,tpch_parquet.lineitem.l_comment)]
| mem-estimate=0B mem-reservation=0B thread-reservation=0
| tuple-ids=1 row-size=263B cardinality=6001215
+| in pipelines: 00(GETNEXT)
|
F00:PLAN FRAGMENT [RANDOM] hosts=3 instances=3
Per-Host Resources: mem-estimate=1.69GB mem-reservation=74.00MB thread-reservation=2
@@ -261,6 +288,7 @@ Per-Host Resources: mem-estimate=1.69GB mem-reservation=74.00MB thread-reservati
| group by: tpch_parquet.lineitem.l_orderkey, tpch_parquet.lineitem.l_partkey, tpch_parquet.lineitem.l_suppkey, tpch_parquet.lineitem.l_linenumber, tpch_parquet.lineitem.l_quantity, tpch_parquet.lineitem.l_extendedprice, tpch_parquet.lineitem.l_discount, tpch_parquet.lineitem.l_tax, tpch_parquet.lineitem.l_returnflag, tpch_parquet.lineitem.l_linestatus, tpch_parquet.lineitem.l_shipdate, tpch_parquet.lineitem.l_commitdate, tpch_parquet.lineitem.l_receiptdate, tpch_parquet.lineitem.l_shipinstruct, tpch_parquet.lineitem.l_shipmode, tpch_parquet.lineitem.l_comment
| mem-estimate=1.62GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
| tuple-ids=1 row-size=263B cardinality=6001215
+| in pipelines: 00(GETNEXT)
|
00:SCAN HDFS [tpch_parquet.lineitem, RANDOM]
partitions=1/1 files=3 size=193.71MB
@@ -270,6 +298,7 @@ Per-Host Resources: mem-estimate=1.69GB mem-reservation=74.00MB thread-reservati
extrapolated-rows=disabled max-scan-range-rows=2141802
mem-estimate=80.00MB mem-reservation=40.00MB thread-reservation=1
tuple-ids=0 row-size=263B cardinality=6001215
+ in pipelines: 00(GETNEXT)
====
# High NDV aggregation with string aggregation function.
# Uses default buffer size. Two max-size buffers are required.
@@ -288,6 +317,7 @@ PLAN-ROOT SINK
04:EXCHANGE [UNPARTITIONED]
| mem-estimate=0B mem-reservation=0B thread-reservation=0
| tuple-ids=1 row-size=32B cardinality=6001215
+| in pipelines: 03(GETNEXT)
|
F01:PLAN FRAGMENT [HASH(l_orderkey,l_partkey)] hosts=3 instances=3
Per-Host Resources: mem-estimate=201.46MB mem-reservation=48.00MB thread-reservation=1
@@ -296,10 +326,12 @@ Per-Host Resources: mem-estimate=201.46MB mem-reservation=48.00MB thread-reserva
| group by: l_orderkey, l_partkey
| mem-estimate=201.46MB mem-reservation=48.00MB spill-buffer=2.00MB thread-reservation=0
| tuple-ids=1 row-size=32B cardinality=6001215
+| in pipelines: 03(GETNEXT), 00(OPEN)
|
02:EXCHANGE [HASH(l_orderkey,l_partkey)]
| mem-estimate=0B mem-reservation=0B thread-reservation=0
| tuple-ids=1 row-size=32B cardinality=6001215
+| in pipelines: 00(GETNEXT)
|
F00:PLAN FRAGMENT [RANDOM] hosts=3 instances=3
Per-Host Resources: mem-estimate=281.46MB mem-reservation=50.00MB thread-reservation=2
@@ -308,6 +340,7 @@ Per-Host Resources: mem-estimate=281.46MB mem-reservation=50.00MB thread-reserva
| group by: l_orderkey, l_partkey
| mem-estimate=201.46MB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
| tuple-ids=1 row-size=32B cardinality=6001215
+| in pipelines: 00(GETNEXT)
|
00:SCAN HDFS [tpch_parquet.lineitem, RANDOM]
partitions=1/1 files=3 size=193.71MB
@@ -317,6 +350,7 @@ Per-Host Resources: mem-estimate=281.46MB mem-reservation=50.00MB thread-reserva
extrapolated-rows=disabled max-scan-range-rows=2141802
mem-estimate=80.00MB mem-reservation=16.00MB thread-reservation=1
tuple-ids=0 row-size=33B cardinality=6001215
+ in pipelines: 00(GETNEXT)
====
# Sort + Analytic.
# The size of buffers for these plan nodes is increased to fit max_row_size.
@@ -335,6 +369,7 @@ PLAN-ROOT SINK
04:EXCHANGE [UNPARTITIONED]
| mem-estimate=0B mem-reservation=0B thread-reservation=0
| tuple-ids=3,2 row-size=6B cardinality=7300
+| in pipelines: 01(GETNEXT)
|
F01:PLAN FRAGMENT [HASH(int_col)] hosts=3 instances=3
Per-Host Resources: mem-estimate=40.00MB mem-reservation=40.00MB thread-reservation=1
@@ -343,15 +378,18 @@ Per-Host Resources: mem-estimate=40.00MB mem-reservation=40.00MB thread-reservat
| partition by: int_col
| mem-estimate=16.00MB mem-reservation=16.00MB spill-buffer=8.00MB thread-reservation=0
| tuple-ids=3,2 row-size=6B cardinality=7300
+| in pipelines: 01(GETNEXT)
|
01:SORT
| order by: int_col ASC NULLS FIRST
| mem-estimate=24.00MB mem-reservation=24.00MB spill-buffer=8.00MB thread-reservation=0
| tuple-ids=3 row-size=5B cardinality=7300
+| in pipelines: 01(GETNEXT), 00(OPEN)
|
03:EXCHANGE [HASH(int_col)]
| mem-estimate=0B mem-reservation=0B thread-reservation=0
| tuple-ids=0 row-size=5B cardinality=7300
+| in pipelines: 00(GETNEXT)
|
F00:PLAN FRAGMENT [RANDOM] hosts=3 instances=3
Per-Host Resources: mem-estimate=16.00MB mem-reservation=32.00KB thread-reservation=2
@@ -364,4 +402,5 @@ Per-Host Resources: mem-estimate=16.00MB mem-reservation=32.00KB thread-reservat
extrapolated-rows=disabled max-scan-range-rows=310
mem-estimate=16.00MB mem-reservation=32.00KB thread-reservation=1
tuple-ids=0 row-size=5B cardinality=7300
+ in pipelines: 00(GETNEXT)
====
http://git-wip-us.apache.org/repos/asf/impala/blob/b7d509d7/testdata/workloads/functional-planner/queries/PlannerTest/min-max-runtime-filters.test
----------------------------------------------------------------------
diff --git a/testdata/workloads/functional-planner/queries/PlannerTest/min-max-runtime-filters.test b/testdata/workloads/functional-planner/queries/PlannerTest/min-max-runtime-filters.test
index d638a61..ae8493e 100644
--- a/testdata/workloads/functional-planner/queries/PlannerTest/min-max-runtime-filters.test
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/min-max-runtime-filters.test
@@ -11,6 +11,7 @@ PLAN-ROOT SINK
| output: count(*)
| mem-estimate=10.00MB mem-reservation=0B spill-buffer=2.00MB thread-reservation=0
| tuple-ids=2 row-size=8B cardinality=1
+| in pipelines: 03(GETNEXT), 00(OPEN)
|
02:HASH JOIN [INNER JOIN]
| hash predicates: a.string_col = b.string_col, a.int_col = b.tinyint_col + 1
@@ -18,15 +19,18 @@ PLAN-ROOT SINK
| runtime filters: RF002[min_max] <- b.string_col, RF003[min_max] <- b.tinyint_col + 1
| mem-estimate=1.94MB mem-reservation=1.94MB spill-buffer=64.00KB thread-reservation=0
| tuple-ids=0,1 row-size=39B cardinality=5840
+| in pipelines: 00(GETNEXT), 01(OPEN)
|
|--01:SCAN KUDU [functional_kudu.alltypestiny b]
| mem-estimate=0B mem-reservation=0B thread-reservation=1
| tuple-ids=1 row-size=18B cardinality=8
+| in pipelines: 01(GETNEXT)
|
00:SCAN KUDU [functional_kudu.alltypes a]
runtime filters: RF002[min_max] -> a.string_col, RF003[min_max] -> a.int_col
mem-estimate=0B mem-reservation=0B thread-reservation=1
tuple-ids=0 row-size=21B cardinality=7300
+ in pipelines: 00(GETNEXT)
====
# Filters are not created if the target isn't a bare Kudu column or if 'is (not) distinct'
# is used.
@@ -44,6 +48,7 @@ PLAN-ROOT SINK
| output: count(*)
| mem-estimate=10.00MB mem-reservation=0B spill-buffer=2.00MB thread-reservation=0
| tuple-ids=2 row-size=8B cardinality=1
+| in pipelines: 03(GETNEXT), 00(OPEN)
|
02:HASH JOIN [INNER JOIN]
| hash predicates: a.tinyint_col IS NOT DISTINCT FROM b.tinyint_col, a.int_col + 1 = b.int_col
@@ -51,14 +56,17 @@ PLAN-ROOT SINK
| other predicates: a.string_col IS DISTINCT FROM b.string_col
| mem-estimate=1.94MB mem-reservation=1.94MB spill-buffer=64.00KB thread-reservation=0
| tuple-ids=0,1 row-size=44B cardinality=7300
+| in pipelines: 00(GETNEXT), 01(OPEN)
|
|--01:SCAN KUDU [functional_kudu.alltypestiny b]
| mem-estimate=0B mem-reservation=0B thread-reservation=1
| tuple-ids=1 row-size=22B cardinality=8
+| in pipelines: 01(GETNEXT)
|
00:SCAN KUDU [functional_kudu.alltypes a]
mem-estimate=0B mem-reservation=0B thread-reservation=1
tuple-ids=0 row-size=22B cardinality=7300
+ in pipelines: 00(GETNEXT)
====
# Filters are only assigned when the target expr is cast if its an implicit integer cast.
select count(*) from functional_kudu.alltypes a, functional_kudu.alltypestiny b
@@ -76,6 +84,7 @@ PLAN-ROOT SINK
| output: count(*)
| mem-estimate=10.00MB mem-reservation=0B spill-buffer=2.00MB thread-reservation=0
| tuple-ids=2 row-size=8B cardinality=1
+| in pipelines: 03(GETNEXT), 00(OPEN)
|
02:HASH JOIN [INNER JOIN]
| hash predicates: CAST(a.float_col AS DOUBLE) = b.double_col, CAST(a.int_col AS SMALLINT) = b.smallint_col, a.string_col = b.timestamp_col, a.tinyint_col = b.bigint_col
@@ -83,15 +92,18 @@ PLAN-ROOT SINK
| runtime filters: RF007[min_max] <- b.bigint_col
| mem-estimate=1.94MB mem-reservation=1.94MB spill-buffer=64.00KB thread-reservation=0
| tuple-ids=0,1 row-size=60B cardinality=1460
+| in pipelines: 00(GETNEXT), 01(OPEN)
|
|--01:SCAN KUDU [functional_kudu.alltypestiny b]
| mem-estimate=0B mem-reservation=0B thread-reservation=1
| tuple-ids=1 row-size=34B cardinality=8
+| in pipelines: 01(GETNEXT)
|
00:SCAN KUDU [functional_kudu.alltypes a]
runtime filters: RF007[min_max] -> a.tinyint_col
mem-estimate=0B mem-reservation=0B thread-reservation=1
tuple-ids=0 row-size=26B cardinality=7300
+ in pipelines: 00(GETNEXT)
====
# Query with both Kudu and HDFS filter targets.
select count(*) from functional_kudu.alltypes a, functional_parquet.alltypes b,
@@ -107,6 +119,7 @@ PLAN-ROOT SINK
| output: count(*)
| mem-estimate=10.00MB mem-reservation=0B spill-buffer=2.00MB thread-reservation=0
| tuple-ids=3 row-size=8B cardinality=1
+| in pipelines: 05(GETNEXT), 00(OPEN)
|
04:HASH JOIN [INNER JOIN]
| hash predicates: a.int_col = c.int_col
@@ -114,10 +127,12 @@ PLAN-ROOT SINK
| runtime filters: RF000[bloom] <- c.int_col, RF001[min_max] <- c.int_col
| mem-estimate=1.94MB mem-reservation=1.94MB spill-buffer=64.00KB thread-reservation=0
| tuple-ids=0,1,2 row-size=12B cardinality=5329000
+| in pipelines: 00(GETNEXT), 02(OPEN)
|
|--02:SCAN KUDU [functional_kudu.alltypes c]
| mem-estimate=0B mem-reservation=0B thread-reservation=1
| tuple-ids=2 row-size=4B cardinality=7300
+| in pipelines: 02(GETNEXT)
|
03:HASH JOIN [INNER JOIN]
| hash predicates: a.int_col = b.int_col
@@ -125,9 +140,10 @@ PLAN-ROOT SINK
| runtime filters: RF003[min_max] <- b.int_col
| mem-estimate=2.00GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
| tuple-ids=0,1 row-size=8B cardinality=7300
+| in pipelines: 00(GETNEXT), 01(OPEN)
|
|--01:SCAN HDFS [functional_parquet.alltypes b]
-| partitions=24/24 files=24 size=188.29KB
+| partitions=24/24 files=24 size=188.92KB
| runtime filters: RF000[bloom] -> b.int_col
| stored statistics:
| table: rows=unavailable size=unavailable
@@ -136,9 +152,11 @@ PLAN-ROOT SINK
| extrapolated-rows=disabled max-scan-range-rows=unavailable
| mem-estimate=16.00MB mem-reservation=16.00KB thread-reservation=1
| tuple-ids=1 row-size=4B cardinality=unavailable
+| in pipelines: 01(GETNEXT)
|
00:SCAN KUDU [functional_kudu.alltypes a]
runtime filters: RF001[min_max] -> a.int_col, RF003[min_max] -> a.int_col
mem-estimate=0B mem-reservation=0B thread-reservation=1
tuple-ids=0 row-size=4B cardinality=7300
+ in pipelines: 00(GETNEXT)
====