You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by st...@apache.org on 2023/09/12 07:21:44 UTC
[impala] 02/03: IMPALA-12357: Skip scheduling bloom filter from full-build scan
This is an automated email from the ASF dual-hosted git repository.
stigahuang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit bd2df11709bf1b048e889be058fa758b51b97e76
Author: Riza Suminto <ri...@cloudera.com>
AuthorDate: Wed Aug 2 16:18:47 2023 -0700
IMPALA-12357: Skip scheduling bloom filter from full-build scan
PK-FK join between a dimension table and a fact table is common
occurrences in a query. Such join often does not involve any predicate
filter in the dimension table. Thus, bloom filter value from this kind
of dimension table scan (PK) will most likely to have all values from
the fact table column (FK). It is ineffective to generate this filter
because it is unlikely to reject any rows, especially if the bloom
filter size is large and has high false positive probability (fpp)
estimate.
This patch skip scheduling bloom filter from join node that has this
characteristics:
1. Build side is full table scan (has hard estimates).
2. The build scan does not have any predicate filter nor consume any
runtime filter.
3. The join node is assumed to have PK-FK relationship.
4. The planned bloom filter has resulting fpp estimate higher than
max_filter_error_rate_from_full_scan flag (default to 0.9).
The fourth criteria is an additional control to eliminate based on fpp
threshold because low fpp filter sometimes is still effective in
eliminating rows (i.e., rows with NULL value). Non-bloom filters remain
unchanged as they are relatively lighter to build and evaluate than
bloom filter.
Testing:
- Add testcase in testBloomFilterAssignment
- Pass core tests
- Ran TPC-DS 3TB with following query options:
* RUNTIME_FILTER_MIN_SIZE=8192
* RUNTIME_FILTER_MAX_SIZE=2097152
* MAX_NUM_RUNTIME_FILTERS=50
* RUNTIME_FILTER_WAIT_TIME_MS=10000
19 out of 103 queries show reduction in number of runtime bloom
filters without any notable performance regression.
Change-Id: I494533bc06da84e606cbd1ae1619083333089a5e
Reviewed-on: http://gerrit.cloudera.org:8080/20366
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
be/src/service/fe-support.cc | 12 +
be/src/util/backend-gflag-util.cc | 10 +
common/thrift/BackendGflags.thrift | 2 +
.../java/org/apache/impala/common/TreeNode.java | 10 +
.../org/apache/impala/planner/ExchangeNode.java | 1 +
.../java/org/apache/impala/planner/PlanNode.java | 5 +
.../impala/planner/RuntimeFilterGenerator.java | 201 +++-
.../java/org/apache/impala/planner/ScanNode.java | 6 +
.../java/org/apache/impala/planner/UnionNode.java | 2 +
.../org/apache/impala/service/BackendConfig.java | 4 +
.../java/org/apache/impala/service/FeSupport.java | 15 +
.../PlannerTest/bloom-filter-assignment.test | 1202 ++++++++++++++++++++
12 files changed, 1433 insertions(+), 37 deletions(-)
diff --git a/be/src/service/fe-support.cc b/be/src/service/fe-support.cc
index 89e29c323..498c2e2df 100644
--- a/be/src/service/fe-support.cc
+++ b/be/src/service/fe-support.cc
@@ -652,6 +652,14 @@ Java_org_apache_impala_service_FeSupport_MinLogSpaceForBloomFilter(
return BloomFilter::MinLogSpace(ndv, fpp);
}
+/// Returns the expected false positive rate for the given ndv and log_bufferpool_space.
+extern "C"
+JNIEXPORT jdouble JNICALL
+Java_org_apache_impala_service_FeSupport_FalsePositiveProbForBloomFilter(
+ JNIEnv* env, jclass fe_support_class, jlong ndv, jint log_bufferpool_space) {
+ return BloomFilter::FalsePositiveProb(ndv, log_bufferpool_space);
+}
+
extern "C"
JNIEXPORT jbyteArray JNICALL
Java_org_apache_impala_service_FeSupport_nativeParseDateString(JNIEnv* env,
@@ -753,6 +761,10 @@ static JNINativeMethod native_methods[] = {
const_cast<char*>("MinLogSpaceForBloomFilter"), const_cast<char*>("(JD)I"),
(void*)::Java_org_apache_impala_service_FeSupport_MinLogSpaceForBloomFilter
},
+ {
+ const_cast<char*>("FalsePositiveProbForBloomFilter"), const_cast<char*>("(JI)D"),
+ (void*)::Java_org_apache_impala_service_FeSupport_FalsePositiveProbForBloomFilter
+ },
{
const_cast<char*>("nativeParseDateString"),
const_cast<char*>("(Ljava/lang/String;)[B"),
diff --git a/be/src/util/backend-gflag-util.cc b/be/src/util/backend-gflag-util.cc
index ea09764d4..72fd2471b 100644
--- a/be/src/util/backend-gflag-util.cc
+++ b/be/src/util/backend-gflag-util.cc
@@ -241,6 +241,14 @@ DEFINE_bool_hidden(skip_resource_checking_on_last_executor_group_set, true,
"ensure that query will always get admitted into last executor group set if it does "
"not fit in any other group set.");
+DEFINE_double_hidden(max_filter_error_rate_from_full_scan, 0.9,
+ "(Advance) Skip generating bloom runtime filter that is generated from "
+ "a full build scan and has resulting error rate estimation that is higher than "
+ "this value after filter size limit applied. This config may get ignored if "
+ "target error rate is set with higher value through RUNTIME_FILTER_ERROR_RATE "
+ "query option or max_filter_error_rate backend flag. Setting value less than 0 "
+ "will disable this runtime filter reduction feature.");
+
using strings::Substitute;
namespace impala {
@@ -427,6 +435,8 @@ Status PopulateThriftBackendGflags(TBackendGflags& cfg) {
cfg.__set_iceberg_reload_new_files_threshold(FLAGS_iceberg_reload_new_files_threshold);
cfg.__set_enable_skipping_older_events(FLAGS_enable_skipping_older_events);
cfg.__set_enable_json_scanner(FLAGS_enable_json_scanner);
+ cfg.__set_max_filter_error_rate_from_full_scan(
+ FLAGS_max_filter_error_rate_from_full_scan);
return Status::OK();
}
diff --git a/common/thrift/BackendGflags.thrift b/common/thrift/BackendGflags.thrift
index fa933424f..0ca30ca4a 100644
--- a/common/thrift/BackendGflags.thrift
+++ b/common/thrift/BackendGflags.thrift
@@ -264,4 +264,6 @@ struct TBackendGflags {
116: required bool enable_skipping_older_events;
117: required bool enable_json_scanner
+
+ 118: required double max_filter_error_rate_from_full_scan
}
diff --git a/fe/src/main/java/org/apache/impala/common/TreeNode.java b/fe/src/main/java/org/apache/impala/common/TreeNode.java
index db6f3327c..f10465809 100644
--- a/fe/src/main/java/org/apache/impala/common/TreeNode.java
+++ b/fe/src/main/java/org/apache/impala/common/TreeNode.java
@@ -245,4 +245,14 @@ public abstract class TreeNode<NodeType extends TreeNode<NodeType>> {
visitor.visit((C) this);
for (NodeType p: children_) p.accept(visitor);
}
+
+ /**
+ * A variant of visitor pattern accept method that do post-order traversal instead
+ * of pre-order.
+ */
+ @SuppressWarnings("unchecked")
+ public <C extends TreeNode<NodeType>> void postAccept(Visitor<C> visitor) {
+ for (NodeType p : children_) p.postAccept(visitor);
+ visitor.visit((C) this);
+ }
}
diff --git a/fe/src/main/java/org/apache/impala/planner/ExchangeNode.java b/fe/src/main/java/org/apache/impala/planner/ExchangeNode.java
index 62f2a50ca..e92fa906f 100644
--- a/fe/src/main/java/org/apache/impala/planner/ExchangeNode.java
+++ b/fe/src/main/java/org/apache/impala/planner/ExchangeNode.java
@@ -111,6 +111,7 @@ public class ExchangeNode extends PlanNode {
cardinality_ = capCardinalityAtLimit(children_.get(0).getCardinality());
// Apply the offset correction if there's a valid cardinality
if (cardinality_ > -1) cardinality_ = Math.max(0, cardinality_ - offset_);
+ hasHardEstimates_ = children_.get(0).hasHardEstimates_;
}
/**
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 4fa1209c9..0f662fb81 100644
--- a/fe/src/main/java/org/apache/impala/planner/PlanNode.java
+++ b/fe/src/main/java/org/apache/impala/planner/PlanNode.java
@@ -158,6 +158,11 @@ abstract public class PlanNode extends TreeNode<PlanNode> {
// Gets set correctly in computeProcessingCost().
protected ProcessingCost processingCost_ = ProcessingCost.invalid();
+ // True if this PlanNode and its children has very accurate cardinality estimates,
+ // assuming that the initial stats collection is also accurate and no runtime filter
+ // is involved.
+ protected boolean hasHardEstimates_ = false;
+
protected PlanNode(PlanNodeId id, List<TupleId> tupleIds, String displayName) {
this(id, displayName);
tupleIds_.addAll(tupleIds);
diff --git a/fe/src/main/java/org/apache/impala/planner/RuntimeFilterGenerator.java b/fe/src/main/java/org/apache/impala/planner/RuntimeFilterGenerator.java
index 0e4a6b36e..5034a61ad 100644
--- a/fe/src/main/java/org/apache/impala/planner/RuntimeFilterGenerator.java
+++ b/fe/src/main/java/org/apache/impala/planner/RuntimeFilterGenerator.java
@@ -19,6 +19,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.Comparator;
import java.util.HashMap;
@@ -27,6 +28,7 @@ import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.stream.Collectors;
import org.apache.impala.analysis.Analyzer;
import org.apache.impala.analysis.BinaryPredicate;
@@ -39,15 +41,14 @@ import org.apache.impala.analysis.IsNullPredicate;
import org.apache.impala.analysis.LiteralExpr;
import org.apache.impala.analysis.NullLiteral;
import org.apache.impala.analysis.Predicate;
-import org.apache.impala.analysis.SlotDescriptor;
import org.apache.impala.analysis.SlotId;
import org.apache.impala.analysis.SlotRef;
import org.apache.impala.analysis.TupleDescriptor;
import org.apache.impala.analysis.TupleId;
import org.apache.impala.analysis.TupleIsNullPredicate;
-import org.apache.impala.catalog.FeTable;
import org.apache.impala.catalog.Column;
import org.apache.impala.catalog.FeIcebergTable;
+import org.apache.impala.catalog.FeTable;
import org.apache.impala.catalog.HdfsFileFormat;
import org.apache.impala.catalog.KuduColumn;
import org.apache.impala.catalog.PrimitiveType;
@@ -60,7 +61,6 @@ import org.apache.impala.planner.JoinNode.DistributionMode;
import org.apache.impala.service.BackendConfig;
import org.apache.impala.service.FeSupport;
import org.apache.impala.thrift.TColumnValue;
-import org.apache.impala.thrift.TExplainLevel;
import org.apache.impala.thrift.TQueryOptions;
import org.apache.impala.thrift.TRuntimeFilterDesc;
import org.apache.impala.thrift.TRuntimeFilterMode;
@@ -68,6 +68,7 @@ import org.apache.impala.thrift.TRuntimeFilterTargetDesc;
import org.apache.impala.thrift.TRuntimeFilterType;
import org.apache.impala.util.BitUtil;
import org.apache.impala.util.TColumnValueUtil;
+import org.apache.impala.util.Visitor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -127,6 +128,12 @@ public final class RuntimeFilterGenerator {
private final IdGenerator<RuntimeFilterId> filterIdGenerator =
RuntimeFilterId.createGenerator();
+ // Map between a FragmentId to the height of plan subtree rooted at that FragmentId.
+ // Can also be interpreted as the distance between FragmentId to its furthest leaf
+ // fragment + 1. Leaf fragment has height 1.
+ // Note that RuntimeFilterGenerator operate over a distributed plan tree.
+ private final Map<PlanFragmentId, Integer> fragmentHeight_ = new HashMap<>();
+
/**
* Internal class that encapsulates the max, min and default sizes used for creating
* bloom filter objects, and entry limit for in-list filters.
@@ -153,11 +160,13 @@ public final class RuntimeFilterGenerator {
long maxLimit = tQueryOptions.getRuntime_filter_max_size();
long minBufferSize = BackendConfig.INSTANCE.getMinBufferSize();
maxVal = BitUtil.roundUpToPowerOf2(Math.max(maxLimit, minBufferSize));
+ Preconditions.checkState(maxVal <= MAX_BLOOM_FILTER_SIZE);
long minLimit = tQueryOptions.getRuntime_filter_min_size();
minLimit = Math.max(minLimit, minBufferSize);
// Make sure minVal <= defaultVal <= maxVal
minVal = BitUtil.roundUpToPowerOf2(Math.min(minLimit, maxVal));
+ Preconditions.checkState(minVal >= MIN_BLOOM_FILTER_SIZE);
long defaultValue = tQueryOptions.getRuntime_bloom_filter_size();
defaultValue = Math.max(defaultValue, minVal);
@@ -220,6 +229,8 @@ public final class RuntimeFilterGenerator {
private long filterSizeBytes_ = 0;
// Size of the filter (in Bytes) before applying min/max filter sizes.
private long originalFilterSizeBytes_ = 0;
+ // False positive probability of this filter. Only set for bloom filter.
+ private double est_fpp_ = 0;
// If true, the filter is produced by a broadcast join and there is at least one
// destination scan node which is in the same fragment as the join; set in
// DistributedPlanner.createHashJoinFragment().
@@ -236,6 +247,10 @@ public final class RuntimeFilterGenerator {
// If set, indicates that the filter is targeted for Kudu scan node with source
// timestamp truncation.
private boolean isTimestampTruncation_ = false;
+ // The level of this runtime filter.
+ // Runtime filter level is defined as the height of build side subtree of the
+ // join node that produce this filter.
+ private int level_ = 1;
/**
* Internal representation of a runtime filter target.
@@ -313,13 +328,13 @@ public final class RuntimeFilterGenerator {
@Override
public String toString() {
StringBuilder output = new StringBuilder();
- return output.append("Target Id: " + node.getId() + " ")
- .append("Target expr: " + expr.debugString() + " ")
- .append("Partition columns: " + isBoundByPartitionColumns)
- .append("Is column stored in data files: " + isColumnInDataFile)
- .append("Is local: " + isLocalTarget)
- .append("lowValue: " + (lowValue != null ? lowValue.toString() : -1))
- .append("highValue: " + (highValue != null ? highValue.toString() : -1))
+ return output.append("Target Id: " + node.getId())
+ .append(" Target expr: " + expr.debugString())
+ .append(" Partition columns: " + isBoundByPartitionColumns)
+ .append(" Is column stored in data files: " + isColumnInDataFile)
+ .append(" Is local: " + isLocalTarget)
+ .append(" lowValue: " + (lowValue != null ? lowValue.toString() : -1))
+ .append(" highValue: " + (highValue != null ? highValue.toString() : -1))
.toString();
}
}
@@ -327,7 +342,7 @@ public final class RuntimeFilterGenerator {
private RuntimeFilter(RuntimeFilterId filterId, JoinNode filterSrcNode, Expr srcExpr,
Expr origTargetExpr, Operator exprCmpOp, Map<TupleId, List<SlotId>> targetSlots,
TRuntimeFilterType type, FilterSizeLimits filterSizeLimits,
- boolean isTimestampTruncation) {
+ boolean isTimestampTruncation, int level) {
id_ = filterId;
src_ = filterSrcNode;
srcExpr_ = srcExpr;
@@ -336,6 +351,7 @@ public final class RuntimeFilterGenerator {
targetSlotsByTid_ = targetSlots;
type_ = type;
isTimestampTruncation_ = isTimestampTruncation;
+ level_ = level;
computeNdvEstimate();
calculateFilterSize(filterSizeLimits);
}
@@ -387,7 +403,7 @@ public final class RuntimeFilterGenerator {
public static RuntimeFilter create(IdGenerator<RuntimeFilterId> idGen,
Analyzer analyzer, Expr joinPredicate, JoinNode filterSrcNode,
TRuntimeFilterType type, FilterSizeLimits filterSizeLimits,
- boolean isTimestampTruncation) {
+ boolean isTimestampTruncation, int level) {
Preconditions.checkNotNull(idGen);
Preconditions.checkNotNull(joinPredicate);
Preconditions.checkNotNull(filterSrcNode);
@@ -522,7 +538,7 @@ public final class RuntimeFilterGenerator {
}
return new RuntimeFilter(idGen.getNextId(), filterSrcNode, srcExpr, targetExpr,
normalizedJoinConjunct.getOp(), targetSlots, type, filterSizeLimits,
- isTimestampTruncation);
+ isTimestampTruncation, level);
}
/**
@@ -715,6 +731,9 @@ public final class RuntimeFilterGenerator {
filterSizeBytes_ = originalFilterSizeBytes_;
filterSizeBytes_ = Math.max(filterSizeBytes_, filterSizeLimits.minVal);
filterSizeBytes_ = Math.min(filterSizeBytes_, filterSizeLimits.maxVal);
+ int actualLogFilterSize = (int) (Math.log(filterSizeBytes_) / Math.log(2));
+ est_fpp_ =
+ FeSupport.GetFalsePositiveProbForBloomFilter(ndvEstimate_, actualLogFilterSize);
}
/**
@@ -728,18 +747,21 @@ public final class RuntimeFilterGenerator {
public String debugString() {
StringBuilder output = new StringBuilder();
- return output.append("FilterID: " + id_ + " ")
- .append("Type: " + type_ + " ")
- .append("Source: " + src_.getId() + " ")
- .append("SrcExpr: " + getSrcExpr().debugString() + " ")
- .append("Target(s): ")
- .append(Joiner.on(", ").join(targets_) + " ")
- .append("Selectivity: " + getSelectivity() + " ")
- .append("Build key NDV: " + buildKeyNdv_ + " ")
- .append("NDV estimate " + ndvEstimate_ + " ")
- .append("Filter size (bytes): " + filterSizeBytes_ + " ")
- .append("Original filter size (bytes): " + originalFilterSizeBytes_ + " ")
- .toString();
+ output.append("FilterID: " + id_)
+ .append(" Type: " + type_)
+ .append(" Source: " + src_.getId())
+ .append(" SrcExpr: " + getSrcExpr().debugString())
+ .append(" Target(s): " + Joiner.on(", ").join(targets_))
+ .append(" Selectivity: " + getSelectivity())
+ .append(" Build key NDV: " + buildKeyNdv_)
+ .append(" NDV estimate " + ndvEstimate_)
+ .append(" Filter size (bytes): " + filterSizeBytes_)
+ .append(" Original filter size (bytes): " + originalFilterSizeBytes_)
+ .append(" Level: " + level_);
+ if (type_ == TRuntimeFilterType.BLOOM) {
+ output.append(" Est fpp: " + est_fpp_);
+ }
+ return output.toString();
}
}
@@ -812,6 +834,92 @@ public final class RuntimeFilterGenerator {
}
}
+ /**
+ * Visitor class over PlanNode tree to check whether the build hand side of join is
+ * eligible for runtime filter pruning based on full scan assumption.
+ */
+ private class BuildVisitor implements Visitor<PlanNode> {
+ private boolean hasIncomingFilter_ = false;
+
+ @Override
+ public void visit(PlanNode a) {
+ if (!isEligibleForPrunning() || !(a instanceof ScanNode)) return;
+
+ ScanNode scan = (ScanNode) a;
+ TupleId sourceTid = scan.getTupleIds().get(0);
+ List<RuntimeFilter> incomingFilters = runtimeFiltersByTid_.get(sourceTid);
+ hasIncomingFilter_ |= (incomingFilters != null && !incomingFilters.isEmpty());
+ }
+
+ public boolean isEligibleForPrunning() { return !hasIncomingFilter_; }
+
+ public void reset() { hasIncomingFilter_ = false; }
+ }
+
+ /**
+ * Remove filters that are unlikely to be useful out of allFilters list.
+ */
+ private void reduceFilters(Collection<RuntimeFilter> allFilters) {
+ if (BackendConfig.INSTANCE.getMaxFilterErrorRateFromFullScan() < 0) return;
+
+ // Filter out bloom filter with fpp > max_filter_error_rate_from_full_scan that
+ // comes from full table scan without filter predicate or incoming runtime filter.
+ // Only do this optimization for filter level 1 for simplicity.
+ boolean hasEliminatedFilter = true;
+ final double maxAllowedFpp =
+ Math.max(BackendConfig.INSTANCE.getMaxFilterErrorRateFromFullScan(),
+ filterSizeLimits_.targetFpp);
+ List<RuntimeFilter> highFppFilters =
+ allFilters.stream()
+ .filter(f
+ -> f.getType() == TRuntimeFilterType.BLOOM
+ && f.level_ <= 1
+ // Build hand node has hard cardinality estimates
+ // (no reduction before runtime filter addition).
+ && f.src_.getChild(1).hasHardEstimates_
+ // At least assumed to have fk/pk relationship.
+ && f.src_.fkPkEqJoinConjuncts_ != null
+ // TODO: FK/PK alone may not be enough to justify eliminating the
+ // filter because it could end up eliminating a filter that is
+ // effective because it filters a lot of NULLs. Therefore, fpp
+ // threshold check is added here to only consider the high fpp one.
+ // This last check can be removed once IMPALA-2743 is implemented
+ // to cover the NULL filtering.
+ && f.est_fpp_ > maxAllowedFpp)
+ .collect(Collectors.toList());
+
+ // As we eliminate a runtime filter in one iteration, there might be another runtime
+ // filter that becomes eligible for removal. Loop this routine until it does eliminate
+ // any other runtime filter.
+ while (!highFppFilters.isEmpty() && hasEliminatedFilter) {
+ hasEliminatedFilter = false;
+ List<RuntimeFilter> skippedFilters = new ArrayList<>();
+ BuildVisitor buildVisitor = new BuildVisitor();
+ for (RuntimeFilter filter : highFppFilters) {
+ // Check if the build side of filter.src_ has a filtering scan or not.
+ buildVisitor.reset();
+ filter.src_.getChild(1).accept(buildVisitor);
+ if (buildVisitor.isEligibleForPrunning()) {
+ // Build side of filter.src_ is all full scan.
+ // Skip this filter and remove it from all of its target TupleIds in
+ // runtimeFiltersByTid_ map.
+ hasEliminatedFilter = true;
+ skippedFilters.add(filter);
+ for (RuntimeFilter.RuntimeFilterTarget target : filter.getTargets()) {
+ TupleId targetTid = target.node.getTupleIds().get(0);
+ runtimeFiltersByTid_.get(targetTid).remove(filter);
+ }
+ }
+ }
+
+ // Remove all skippedFilters from highFppFilters and allFilters for next iteration.
+ for (RuntimeFilter filter : skippedFilters) {
+ highFppFilters.remove(filter);
+ allFilters.remove(filter);
+ }
+ }
+ }
+
/**
* Returns a list of all the registered runtime filters, ordered by filter ID.
*/
@@ -828,9 +936,21 @@ public final class RuntimeFilterGenerator {
}
}
);
+ reduceFilters(resultList);
return resultList;
}
+ private void generateFilters(PlannerContext ctx, PlanNode root) {
+ root.getFragment().postAccept(f -> {
+ int height = 0;
+ for (PlanFragment child : f.getChildren()) {
+ height = Math.max(height, fragmentHeight_.get(child.getId()));
+ }
+ fragmentHeight_.put(((PlanFragment) f).getId(), height + 1);
+ });
+ generateFiltersRecursive(ctx, root);
+ }
+
/**
* Generates the runtime filters for a query by recursively traversing the distributed
* plan tree rooted at 'root'. In the top-down traversal of the plan tree, candidate
@@ -838,9 +958,18 @@ public final class RuntimeFilterGenerator {
* In the bottom-up traversal of the plan tree, the filters are assigned to destination
* (scan) nodes. Filters that cannot be assigned to a scan node are discarded.
*/
- private void generateFilters(PlannerContext ctx, PlanNode root) {
+ private void generateFiltersRecursive(PlannerContext ctx, PlanNode root) {
if (root instanceof HashJoinNode || root instanceof NestedLoopJoinNode) {
JoinNode joinNode = (JoinNode) root;
+
+ // Determine level of filter produced by root.
+ PlanNode buildHandNode = root.getChild(1);
+ while (buildHandNode.getFragment() == root.getFragment()
+ && buildHandNode.hasChild(0)) {
+ buildHandNode = buildHandNode.getChild(0);
+ }
+ int level = fragmentHeight_.get(buildHandNode.getFragment().getId());
+
List<Expr> joinConjuncts = new ArrayList<>();
if (!joinNode.getJoinOp().isLeftOuterJoin()
&& !joinNode.getJoinOp().isFullOuterJoin()
@@ -858,10 +987,9 @@ public final class RuntimeFilterGenerator {
for (TRuntimeFilterType filterType : TRuntimeFilterType.values()) {
if (!enabledRuntimeFilterTypes.contains(filterType)) continue;
for (Expr conjunct : joinConjuncts) {
- RuntimeFilter filter =
- RuntimeFilter.create(filterIdGenerator, ctx.getRootAnalyzer(), conjunct,
- joinNode, filterType, filterSizeLimits_,
- /* isTimestampTruncation */ false);
+ RuntimeFilter filter = RuntimeFilter.create(filterIdGenerator,
+ ctx.getRootAnalyzer(), conjunct, joinNode, filterType, filterSizeLimits_,
+ /* isTimestampTruncation */ false, level);
if (filter != null) {
registerRuntimeFilter(filter);
filters.add(filter);
@@ -872,27 +1000,26 @@ public final class RuntimeFilterGenerator {
&& Predicate.isEquivalencePredicate(conjunct)
&& conjunct.getChild(0).getType().isTimestamp()
&& conjunct.getChild(1).getType().isTimestamp()) {
- RuntimeFilter filter2 =
- RuntimeFilter.create(filterIdGenerator, ctx.getRootAnalyzer(), conjunct,
- joinNode, filterType, filterSizeLimits_,
- /* isTimestampTruncation */ true);
+ RuntimeFilter filter2 = RuntimeFilter.create(filterIdGenerator,
+ ctx.getRootAnalyzer(), conjunct, joinNode, filterType, filterSizeLimits_,
+ /* isTimestampTruncation */ true, level);
if (filter2 == null) continue;
registerRuntimeFilter(filter2);
filters.add(filter2);
}
}
}
- generateFilters(ctx, root.getChild(0));
+ generateFiltersRecursive(ctx, root.getChild(0));
// Finalize every runtime filter of that join. This is to ensure that we don't
// assign a filter to a scan node from the right subtree of joinNode or ancestor
// join nodes in case we don't find a destination node in the left subtree.
for (RuntimeFilter runtimeFilter: filters) finalizeRuntimeFilter(runtimeFilter);
- generateFilters(ctx, root.getChild(1));
+ generateFiltersRecursive(ctx, root.getChild(1));
} else if (root instanceof ScanNode) {
assignRuntimeFilters(ctx, (ScanNode) root);
} else {
for (PlanNode childNode: root.getChildren()) {
- generateFilters(ctx, childNode);
+ generateFiltersRecursive(ctx, childNode);
}
}
}
diff --git a/fe/src/main/java/org/apache/impala/planner/ScanNode.java b/fe/src/main/java/org/apache/impala/planner/ScanNode.java
index 7ab07bbe9..a2bcda486 100644
--- a/fe/src/main/java/org/apache/impala/planner/ScanNode.java
+++ b/fe/src/main/java/org/apache/impala/planner/ScanNode.java
@@ -89,6 +89,12 @@ abstract public class ScanNode extends PlanNode {
desc_ = desc;
}
+ @Override
+ public void computeStats(Analyzer analyzer) {
+ super.computeStats(analyzer);
+ hasHardEstimates_ = !hasScanConjuncts() && !isAccessingCollectionType();
+ }
+
public TupleDescriptor getTupleDesc() { return desc_; }
@Override
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 a83a4ecdc..30636ad2e 100644
--- a/fe/src/main/java/org/apache/impala/planner/UnionNode.java
+++ b/fe/src/main/java/org/apache/impala/planner/UnionNode.java
@@ -116,6 +116,7 @@ public class UnionNode extends PlanNode {
super.computeStats(analyzer);
long totalChildCardinality = 0;
boolean haveChildWithCardinality = false;
+ hasHardEstimates_ = true;
for (PlanNode child: children_) {
// ignore missing child cardinality info in the hope it won't matter enough
// to change the planning outcome
@@ -129,6 +130,7 @@ public class UnionNode extends PlanNode {
// subsets of each other, i.e. not just partly overlapping.
numNodes_ = Math.max(child.getNumNodes(), numNodes_);
numInstances_ = Math.max(child.getNumInstances(), numInstances_);
+ hasHardEstimates_ &= child.hasHardEstimates_;
}
// Consider estimate valid if we have at least one child with known cardinality, or
// only constant values.
diff --git a/fe/src/main/java/org/apache/impala/service/BackendConfig.java b/fe/src/main/java/org/apache/impala/service/BackendConfig.java
index 93025fee3..980e7b016 100644
--- a/fe/src/main/java/org/apache/impala/service/BackendConfig.java
+++ b/fe/src/main/java/org/apache/impala/service/BackendConfig.java
@@ -432,4 +432,8 @@ public class BackendConfig {
public boolean isJsonScannerEnabled() {
return backendCfg_.enable_json_scanner;
}
+
+ public double getMaxFilterErrorRateFromFullScan() {
+ return backendCfg_.max_filter_error_rate_from_full_scan;
+ }
}
diff --git a/fe/src/main/java/org/apache/impala/service/FeSupport.java b/fe/src/main/java/org/apache/impala/service/FeSupport.java
index 7e9d08cea..e8ea8430a 100644
--- a/fe/src/main/java/org/apache/impala/service/FeSupport.java
+++ b/fe/src/main/java/org/apache/impala/service/FeSupport.java
@@ -132,6 +132,8 @@ public class FeSupport {
byte[] queryOptions);
public native static int MinLogSpaceForBloomFilter(long ndv, double fpp);
+ public native static double FalsePositiveProbForBloomFilter(
+ long ndv, int logBufferpoolSpace);
// Parses date string, verifies if it is valid and returns the resulting
// TParseDateStringResult object. Different date string variations are accepted.
@@ -436,6 +438,19 @@ public class FeSupport {
return MinLogSpaceForBloomFilter(ndv, fpp);
}
+ /**
+ * Returns the expected false positive rate for the given ndv and logBufferpoolSpace.
+ */
+ public static double GetFalsePositiveProbForBloomFilter(
+ long ndv, int logBufferpoolSpace) {
+ try {
+ return FalsePositiveProbForBloomFilter(ndv, logBufferpoolSpace);
+ } catch (UnsatisfiedLinkError e) {
+ loadLibrary();
+ }
+ return FalsePositiveProbForBloomFilter(ndv, logBufferpoolSpace);
+ }
+
public static byte[] GetPartialCatalogObject(byte[] thriftReq)
throws InternalException {
try {
diff --git a/testdata/workloads/functional-planner/queries/PlannerTest/bloom-filter-assignment.test b/testdata/workloads/functional-planner/queries/PlannerTest/bloom-filter-assignment.test
index c141b52cc..a8f648bcd 100644
--- a/testdata/workloads/functional-planner/queries/PlannerTest/bloom-filter-assignment.test
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/bloom-filter-assignment.test
@@ -540,3 +540,1205 @@ PLAN-ROOT SINK
tuple-ids=0 row-size=16B cardinality=12.88K
in pipelines: 00(GETNEXT)
====
+# IMPALA-12357: filter size 512KB can achieve fpp lower than 0.9.
+select STRAIGHT_JOIN count(*) from
+ (select l_orderkey from tpch_parquet.lineitem) a
+ join (select o_orderkey from tpch_parquet.orders) l1 on a.l_orderkey = l1.o_orderkey
+ join (select o_orderkey from tpch_parquet.orders) l2 on l1.o_orderkey = l2.o_orderkey
+ join (select o_orderkey from tpch_parquet.orders) l3 on l2.o_orderkey = l3.o_orderkey;
+---- QUERYOPTIONS
+RUNTIME_FILTER_MAX_SIZE=512KB
+EXPLAIN_LEVEL=2
+---- PLAN
+F00:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1
+| Per-Host Resources: mem-estimate=379.83MB mem-reservation=119.50MB thread-reservation=5 runtime-filters-memory=1.50MB
+PLAN-ROOT SINK
+| output exprs: count(*)
+| mem-estimate=0B mem-reservation=0B thread-reservation=0
+|
+07:AGGREGATE [FINALIZE]
+| output: count(*)
+| mem-estimate=16.00KB mem-reservation=0B spill-buffer=2.00MB thread-reservation=0
+| tuple-ids=8 row-size=8B cardinality=1
+| in pipelines: 07(GETNEXT), 00(OPEN)
+|
+06:HASH JOIN [INNER JOIN]
+| hash predicates: o_orderkey = o_orderkey
+| fk/pk conjuncts: o_orderkey = o_orderkey
+| runtime filters: RF000[bloom] <- o_orderkey
+| mem-estimate=59.44MB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| tuple-ids=0,2,4,6 row-size=32B cardinality=1
+| in pipelines: 00(GETNEXT), 03(OPEN)
+|
+|--03:SCAN HDFS [tpch_parquet.orders]
+| HDFS partitions=1/1 files=2 size=54.21MB
+| stored statistics:
+| table: rows=1.50M size=54.21MB
+| columns: all
+| extrapolated-rows=disabled max-scan-range-rows=1.18M
+| mem-estimate=40.00MB mem-reservation=4.00MB thread-reservation=1
+| tuple-ids=6 row-size=8B cardinality=1.50M
+| in pipelines: 03(GETNEXT)
+|
+05:HASH JOIN [INNER JOIN]
+| hash predicates: o_orderkey = o_orderkey
+| fk/pk conjuncts: o_orderkey = o_orderkey
+| runtime filters: RF002[bloom] <- o_orderkey
+| mem-estimate=59.44MB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| tuple-ids=0,2,4 row-size=24B cardinality=1
+| in pipelines: 00(GETNEXT), 02(OPEN)
+|
+|--02:SCAN HDFS [tpch_parquet.orders]
+| HDFS partitions=1/1 files=2 size=54.21MB
+| runtime filters: RF000[bloom] -> o_orderkey
+| stored statistics:
+| table: rows=1.50M size=54.21MB
+| columns: all
+| extrapolated-rows=disabled max-scan-range-rows=1.18M
+| mem-estimate=40.00MB mem-reservation=4.00MB thread-reservation=1
+| tuple-ids=4 row-size=8B cardinality=1.50M
+| in pipelines: 02(GETNEXT)
+|
+04:HASH JOIN [INNER JOIN]
+| hash predicates: l_orderkey = o_orderkey
+| fk/pk conjuncts: l_orderkey = o_orderkey
+| runtime filters: RF004[bloom] <- o_orderkey
+| mem-estimate=59.44MB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| tuple-ids=0,2 row-size=16B cardinality=1
+| in pipelines: 00(GETNEXT), 01(OPEN)
+|
+|--01:SCAN HDFS [tpch_parquet.orders]
+| HDFS partitions=1/1 files=2 size=54.21MB
+| runtime filters: RF000[bloom] -> tpch_parquet.orders.o_orderkey, RF002[bloom] -> o_orderkey
+| stored statistics:
+| table: rows=1.50M size=54.21MB
+| columns: all
+| extrapolated-rows=disabled max-scan-range-rows=1.18M
+| mem-estimate=40.00MB mem-reservation=4.00MB thread-reservation=1
+| tuple-ids=2 row-size=8B cardinality=1.50M
+| in pipelines: 01(GETNEXT)
+|
+00:SCAN HDFS [tpch_parquet.lineitem]
+ HDFS partitions=1/1 files=3 size=194.00MB
+ runtime filters: RF000[bloom] -> tpch_parquet.lineitem.l_orderkey, RF002[bloom] -> tpch_parquet.lineitem.l_orderkey, RF004[bloom] -> l_orderkey
+ stored statistics:
+ table: rows=6.00M size=194.00MB
+ columns: all
+ extrapolated-rows=disabled max-scan-range-rows=2.14M
+ mem-estimate=80.00MB mem-reservation=4.00MB thread-reservation=1
+ tuple-ids=0 row-size=8B cardinality=6.00M
+ in pipelines: 00(GETNEXT)
+---- DISTRIBUTEDPLAN
+F04:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1
+| Per-Host Resources: mem-estimate=32.00KB mem-reservation=0B thread-reservation=1
+PLAN-ROOT SINK
+| output exprs: count(*)
+| mem-estimate=0B mem-reservation=0B thread-reservation=0
+|
+12:AGGREGATE [FINALIZE]
+| output: count:merge(*)
+| mem-estimate=16.00KB mem-reservation=0B spill-buffer=2.00MB thread-reservation=0
+| tuple-ids=8 row-size=8B cardinality=1
+| in pipelines: 12(GETNEXT), 07(OPEN)
+|
+11:EXCHANGE [UNPARTITIONED]
+| mem-estimate=16.00KB mem-reservation=0B thread-reservation=0
+| tuple-ids=8 row-size=8B cardinality=1
+| in pipelines: 07(GETNEXT)
+|
+F00:PLAN FRAGMENT [RANDOM] hosts=3 instances=3
+Per-Host Resources: mem-estimate=289.90MB mem-reservation=107.50MB thread-reservation=2 runtime-filters-memory=1.50MB
+07:AGGREGATE
+| output: count(*)
+| mem-estimate=16.00KB mem-reservation=0B spill-buffer=2.00MB thread-reservation=0
+| tuple-ids=8 row-size=8B cardinality=1
+| in pipelines: 07(GETNEXT), 00(OPEN)
+|
+06:HASH JOIN [INNER JOIN, BROADCAST]
+| hash predicates: o_orderkey = o_orderkey
+| fk/pk conjuncts: o_orderkey = o_orderkey
+| runtime filters: RF000[bloom] <- o_orderkey
+| mem-estimate=59.44MB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| tuple-ids=0,2,4,6 row-size=32B cardinality=5.76M
+| in pipelines: 00(GETNEXT), 03(OPEN)
+|
+|--10:EXCHANGE [BROADCAST]
+| | mem-estimate=10.02MB mem-reservation=0B thread-reservation=0
+| | tuple-ids=6 row-size=8B cardinality=1.50M
+| | in pipelines: 03(GETNEXT)
+| |
+| F03:PLAN FRAGMENT [RANDOM] hosts=2 instances=2
+| Per-Host Resources: mem-estimate=40.05MB mem-reservation=4.00MB thread-reservation=2
+| 03:SCAN HDFS [tpch_parquet.orders, RANDOM]
+| HDFS partitions=1/1 files=2 size=54.21MB
+| stored statistics:
+| table: rows=1.50M size=54.21MB
+| columns: all
+| extrapolated-rows=disabled max-scan-range-rows=1.18M
+| mem-estimate=40.00MB mem-reservation=4.00MB thread-reservation=1
+| tuple-ids=6 row-size=8B cardinality=1.50M
+| in pipelines: 03(GETNEXT)
+|
+05:HASH JOIN [INNER JOIN, BROADCAST]
+| hash predicates: o_orderkey = o_orderkey
+| fk/pk conjuncts: o_orderkey = o_orderkey
+| runtime filters: RF002[bloom] <- o_orderkey
+| mem-estimate=59.44MB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| tuple-ids=0,2,4 row-size=24B cardinality=5.76M
+| in pipelines: 00(GETNEXT), 02(OPEN)
+|
+|--09:EXCHANGE [BROADCAST]
+| | mem-estimate=10.02MB mem-reservation=0B thread-reservation=0
+| | tuple-ids=4 row-size=8B cardinality=1.50M
+| | in pipelines: 02(GETNEXT)
+| |
+| F02:PLAN FRAGMENT [RANDOM] hosts=2 instances=2
+| Per-Host Resources: mem-estimate=40.55MB mem-reservation=4.50MB thread-reservation=2 runtime-filters-memory=512.00KB
+| 02:SCAN HDFS [tpch_parquet.orders, RANDOM]
+| HDFS partitions=1/1 files=2 size=54.21MB
+| runtime filters: RF000[bloom] -> o_orderkey
+| stored statistics:
+| table: rows=1.50M size=54.21MB
+| columns: all
+| extrapolated-rows=disabled max-scan-range-rows=1.18M
+| mem-estimate=40.00MB mem-reservation=4.00MB thread-reservation=1
+| tuple-ids=4 row-size=8B cardinality=1.50M
+| in pipelines: 02(GETNEXT)
+|
+04:HASH JOIN [INNER JOIN, BROADCAST]
+| hash predicates: l_orderkey = o_orderkey
+| fk/pk conjuncts: l_orderkey = o_orderkey
+| runtime filters: RF004[bloom] <- o_orderkey
+| mem-estimate=59.44MB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| tuple-ids=0,2 row-size=16B cardinality=5.76M
+| in pipelines: 00(GETNEXT), 01(OPEN)
+|
+|--08:EXCHANGE [BROADCAST]
+| | mem-estimate=10.02MB mem-reservation=0B thread-reservation=0
+| | tuple-ids=2 row-size=8B cardinality=1.50M
+| | in pipelines: 01(GETNEXT)
+| |
+| F01:PLAN FRAGMENT [RANDOM] hosts=2 instances=2
+| Per-Host Resources: mem-estimate=41.05MB mem-reservation=5.00MB thread-reservation=2 runtime-filters-memory=1.00MB
+| 01:SCAN HDFS [tpch_parquet.orders, RANDOM]
+| HDFS partitions=1/1 files=2 size=54.21MB
+| runtime filters: RF000[bloom] -> tpch_parquet.orders.o_orderkey, RF002[bloom] -> o_orderkey
+| stored statistics:
+| table: rows=1.50M size=54.21MB
+| columns: all
+| extrapolated-rows=disabled max-scan-range-rows=1.18M
+| mem-estimate=40.00MB mem-reservation=4.00MB thread-reservation=1
+| tuple-ids=2 row-size=8B cardinality=1.50M
+| in pipelines: 01(GETNEXT)
+|
+00:SCAN HDFS [tpch_parquet.lineitem, RANDOM]
+ HDFS partitions=1/1 files=3 size=194.00MB
+ runtime filters: RF000[bloom] -> tpch_parquet.lineitem.l_orderkey, RF002[bloom] -> tpch_parquet.lineitem.l_orderkey, RF004[bloom] -> l_orderkey
+ stored statistics:
+ table: rows=6.00M size=194.00MB
+ columns: all
+ extrapolated-rows=disabled max-scan-range-rows=2.14M
+ mem-estimate=80.00MB mem-reservation=4.00MB thread-reservation=1
+ tuple-ids=0 row-size=8B cardinality=6.00M
+ in pipelines: 00(GETNEXT)
+====
+# IMPALA-12357: filter size < 512KB can not achieve fpp lower than 0.9 and will be skipped.
+select STRAIGHT_JOIN count(*) from
+ (select l_orderkey from tpch_parquet.lineitem) a
+ join (select o_orderkey from tpch_parquet.orders) l1 on a.l_orderkey = l1.o_orderkey
+ join (select o_orderkey from tpch_parquet.orders) l2 on l1.o_orderkey = l2.o_orderkey
+ join (select o_orderkey from tpch_parquet.orders) l3 on l2.o_orderkey = l3.o_orderkey;
+---- QUERYOPTIONS
+RUNTIME_FILTER_MAX_SIZE=256KB
+EXPLAIN_LEVEL=2
+---- PLAN
+F00:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1
+| Per-Host Resources: mem-estimate=378.33MB mem-reservation=118.00MB thread-reservation=5
+PLAN-ROOT SINK
+| output exprs: count(*)
+| mem-estimate=0B mem-reservation=0B thread-reservation=0
+|
+07:AGGREGATE [FINALIZE]
+| output: count(*)
+| mem-estimate=16.00KB mem-reservation=0B spill-buffer=2.00MB thread-reservation=0
+| tuple-ids=8 row-size=8B cardinality=1
+| in pipelines: 07(GETNEXT), 00(OPEN)
+|
+06:HASH JOIN [INNER JOIN]
+| hash predicates: o_orderkey = o_orderkey
+| fk/pk conjuncts: o_orderkey = o_orderkey
+| mem-estimate=59.44MB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| tuple-ids=0,2,4,6 row-size=32B cardinality=1
+| in pipelines: 00(GETNEXT), 03(OPEN)
+|
+|--03:SCAN HDFS [tpch_parquet.orders]
+| HDFS partitions=1/1 files=2 size=54.21MB
+| stored statistics:
+| table: rows=1.50M size=54.21MB
+| columns: all
+| extrapolated-rows=disabled max-scan-range-rows=1.18M
+| mem-estimate=40.00MB mem-reservation=4.00MB thread-reservation=1
+| tuple-ids=6 row-size=8B cardinality=1.50M
+| in pipelines: 03(GETNEXT)
+|
+05:HASH JOIN [INNER JOIN]
+| hash predicates: o_orderkey = o_orderkey
+| fk/pk conjuncts: o_orderkey = o_orderkey
+| mem-estimate=59.44MB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| tuple-ids=0,2,4 row-size=24B cardinality=1
+| in pipelines: 00(GETNEXT), 02(OPEN)
+|
+|--02:SCAN HDFS [tpch_parquet.orders]
+| HDFS partitions=1/1 files=2 size=54.21MB
+| stored statistics:
+| table: rows=1.50M size=54.21MB
+| columns: all
+| extrapolated-rows=disabled max-scan-range-rows=1.18M
+| mem-estimate=40.00MB mem-reservation=4.00MB thread-reservation=1
+| tuple-ids=4 row-size=8B cardinality=1.50M
+| in pipelines: 02(GETNEXT)
+|
+04:HASH JOIN [INNER JOIN]
+| hash predicates: l_orderkey = o_orderkey
+| fk/pk conjuncts: l_orderkey = o_orderkey
+| mem-estimate=59.44MB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| tuple-ids=0,2 row-size=16B cardinality=1
+| in pipelines: 00(GETNEXT), 01(OPEN)
+|
+|--01:SCAN HDFS [tpch_parquet.orders]
+| HDFS partitions=1/1 files=2 size=54.21MB
+| stored statistics:
+| table: rows=1.50M size=54.21MB
+| columns: all
+| extrapolated-rows=disabled max-scan-range-rows=1.18M
+| mem-estimate=40.00MB mem-reservation=4.00MB thread-reservation=1
+| tuple-ids=2 row-size=8B cardinality=1.50M
+| in pipelines: 01(GETNEXT)
+|
+00:SCAN HDFS [tpch_parquet.lineitem]
+ HDFS partitions=1/1 files=3 size=194.00MB
+ stored statistics:
+ table: rows=6.00M size=194.00MB
+ columns: all
+ extrapolated-rows=disabled max-scan-range-rows=2.14M
+ mem-estimate=80.00MB mem-reservation=4.00MB thread-reservation=1
+ tuple-ids=0 row-size=8B cardinality=6.00M
+ in pipelines: 00(GETNEXT)
+---- DISTRIBUTEDPLAN
+F04:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1
+| Per-Host Resources: mem-estimate=32.00KB mem-reservation=0B thread-reservation=1
+PLAN-ROOT SINK
+| output exprs: count(*)
+| mem-estimate=0B mem-reservation=0B thread-reservation=0
+|
+12:AGGREGATE [FINALIZE]
+| output: count:merge(*)
+| mem-estimate=16.00KB mem-reservation=0B spill-buffer=2.00MB thread-reservation=0
+| tuple-ids=8 row-size=8B cardinality=1
+| in pipelines: 12(GETNEXT), 07(OPEN)
+|
+11:EXCHANGE [UNPARTITIONED]
+| mem-estimate=16.00KB mem-reservation=0B thread-reservation=0
+| tuple-ids=8 row-size=8B cardinality=1
+| in pipelines: 07(GETNEXT)
+|
+F00:PLAN FRAGMENT [RANDOM] hosts=3 instances=3
+Per-Host Resources: mem-estimate=288.40MB mem-reservation=106.00MB thread-reservation=2
+07:AGGREGATE
+| output: count(*)
+| mem-estimate=16.00KB mem-reservation=0B spill-buffer=2.00MB thread-reservation=0
+| tuple-ids=8 row-size=8B cardinality=1
+| in pipelines: 07(GETNEXT), 00(OPEN)
+|
+06:HASH JOIN [INNER JOIN, BROADCAST]
+| hash predicates: o_orderkey = o_orderkey
+| fk/pk conjuncts: o_orderkey = o_orderkey
+| mem-estimate=59.44MB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| tuple-ids=0,2,4,6 row-size=32B cardinality=5.76M
+| in pipelines: 00(GETNEXT), 03(OPEN)
+|
+|--10:EXCHANGE [BROADCAST]
+| | mem-estimate=10.02MB mem-reservation=0B thread-reservation=0
+| | tuple-ids=6 row-size=8B cardinality=1.50M
+| | in pipelines: 03(GETNEXT)
+| |
+| F03:PLAN FRAGMENT [RANDOM] hosts=2 instances=2
+| Per-Host Resources: mem-estimate=40.05MB mem-reservation=4.00MB thread-reservation=2
+| 03:SCAN HDFS [tpch_parquet.orders, RANDOM]
+| HDFS partitions=1/1 files=2 size=54.21MB
+| stored statistics:
+| table: rows=1.50M size=54.21MB
+| columns: all
+| extrapolated-rows=disabled max-scan-range-rows=1.18M
+| mem-estimate=40.00MB mem-reservation=4.00MB thread-reservation=1
+| tuple-ids=6 row-size=8B cardinality=1.50M
+| in pipelines: 03(GETNEXT)
+|
+05:HASH JOIN [INNER JOIN, BROADCAST]
+| hash predicates: o_orderkey = o_orderkey
+| fk/pk conjuncts: o_orderkey = o_orderkey
+| mem-estimate=59.44MB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| tuple-ids=0,2,4 row-size=24B cardinality=5.76M
+| in pipelines: 00(GETNEXT), 02(OPEN)
+|
+|--09:EXCHANGE [BROADCAST]
+| | mem-estimate=10.02MB mem-reservation=0B thread-reservation=0
+| | tuple-ids=4 row-size=8B cardinality=1.50M
+| | in pipelines: 02(GETNEXT)
+| |
+| F02:PLAN FRAGMENT [RANDOM] hosts=2 instances=2
+| Per-Host Resources: mem-estimate=40.05MB mem-reservation=4.00MB thread-reservation=2
+| 02:SCAN HDFS [tpch_parquet.orders, RANDOM]
+| HDFS partitions=1/1 files=2 size=54.21MB
+| stored statistics:
+| table: rows=1.50M size=54.21MB
+| columns: all
+| extrapolated-rows=disabled max-scan-range-rows=1.18M
+| mem-estimate=40.00MB mem-reservation=4.00MB thread-reservation=1
+| tuple-ids=4 row-size=8B cardinality=1.50M
+| in pipelines: 02(GETNEXT)
+|
+04:HASH JOIN [INNER JOIN, BROADCAST]
+| hash predicates: l_orderkey = o_orderkey
+| fk/pk conjuncts: l_orderkey = o_orderkey
+| mem-estimate=59.44MB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| tuple-ids=0,2 row-size=16B cardinality=5.76M
+| in pipelines: 00(GETNEXT), 01(OPEN)
+|
+|--08:EXCHANGE [BROADCAST]
+| | mem-estimate=10.02MB mem-reservation=0B thread-reservation=0
+| | tuple-ids=2 row-size=8B cardinality=1.50M
+| | in pipelines: 01(GETNEXT)
+| |
+| F01:PLAN FRAGMENT [RANDOM] hosts=2 instances=2
+| Per-Host Resources: mem-estimate=40.05MB mem-reservation=4.00MB thread-reservation=2
+| 01:SCAN HDFS [tpch_parquet.orders, RANDOM]
+| HDFS partitions=1/1 files=2 size=54.21MB
+| stored statistics:
+| table: rows=1.50M size=54.21MB
+| columns: all
+| extrapolated-rows=disabled max-scan-range-rows=1.18M
+| mem-estimate=40.00MB mem-reservation=4.00MB thread-reservation=1
+| tuple-ids=2 row-size=8B cardinality=1.50M
+| in pipelines: 01(GETNEXT)
+|
+00:SCAN HDFS [tpch_parquet.lineitem, RANDOM]
+ HDFS partitions=1/1 files=3 size=194.00MB
+ stored statistics:
+ table: rows=6.00M size=194.00MB
+ columns: all
+ extrapolated-rows=disabled max-scan-range-rows=2.14M
+ mem-estimate=80.00MB mem-reservation=4.00MB thread-reservation=1
+ tuple-ids=0 row-size=8B cardinality=6.00M
+ in pipelines: 00(GETNEXT)
+====
+# IMPALA-12357: no filter elimination because no fk/pk conjuncts detected.
+select STRAIGHT_JOIN count(*) from
+ (select l_orderkey from tpch_parquet.lineitem a LIMIT 1) a
+ join (select l_orderkey from tpch_parquet.lineitem) l1 on a.l_orderkey = l1.l_orderkey
+ join (select l_orderkey from tpch_parquet.lineitem) l2 on l1.l_orderkey = l2.l_orderkey
+ join (select l_orderkey from tpch_parquet.lineitem) l3 on l2.l_orderkey = l3.l_orderkey;
+---- QUERYOPTIONS
+RUNTIME_FILTER_MAX_SIZE=256KB
+EXPLAIN_LEVEL=2
+---- PLAN
+F00:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1
+| Per-Host Resources: mem-estimate=1.21GB mem-reservation=118.75MB thread-reservation=5 runtime-filters-memory=768.00KB
+PLAN-ROOT SINK
+| output exprs: count(*)
+| mem-estimate=0B mem-reservation=0B thread-reservation=0
+|
+07:AGGREGATE [FINALIZE]
+| output: count(*)
+| mem-estimate=16.00KB mem-reservation=0B spill-buffer=2.00MB thread-reservation=0
+| tuple-ids=8 row-size=8B cardinality=1
+| in pipelines: 07(GETNEXT), 00(OPEN)
+|
+06:HASH JOIN [INNER JOIN]
+| hash predicates: l_orderkey = l_orderkey
+| fk/pk conjuncts: none
+| runtime filters: RF000[bloom] <- l_orderkey
+| mem-estimate=305.50MB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| tuple-ids=0,2,4,6 row-size=32B cardinality=58
+| in pipelines: 00(GETNEXT), 03(OPEN)
+|
+|--03:SCAN HDFS [tpch_parquet.lineitem]
+| HDFS partitions=1/1 files=3 size=194.00MB
+| stored statistics:
+| table: rows=6.00M size=194.00MB
+| columns: all
+| extrapolated-rows=disabled max-scan-range-rows=2.14M
+| mem-estimate=80.00MB mem-reservation=4.00MB thread-reservation=1
+| tuple-ids=6 row-size=8B cardinality=6.00M
+| in pipelines: 03(GETNEXT)
+|
+05:HASH JOIN [INNER JOIN]
+| hash predicates: l_orderkey = l_orderkey
+| fk/pk conjuncts: none
+| runtime filters: RF002[bloom] <- l_orderkey
+| mem-estimate=305.50MB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| tuple-ids=0,2,4 row-size=24B cardinality=15
+| in pipelines: 00(GETNEXT), 02(OPEN)
+|
+|--02:SCAN HDFS [tpch_parquet.lineitem]
+| HDFS partitions=1/1 files=3 size=194.00MB
+| runtime filters: RF000[bloom] -> l_orderkey
+| stored statistics:
+| table: rows=6.00M size=194.00MB
+| columns: all
+| extrapolated-rows=disabled max-scan-range-rows=2.14M
+| mem-estimate=80.00MB mem-reservation=4.00MB thread-reservation=1
+| tuple-ids=4 row-size=8B cardinality=6.00M
+| in pipelines: 02(GETNEXT)
+|
+04:HASH JOIN [INNER JOIN]
+| hash predicates: l_orderkey = l_orderkey
+| fk/pk conjuncts: none
+| runtime filters: RF004[bloom] <- l_orderkey
+| mem-estimate=305.50MB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| tuple-ids=0,2 row-size=16B cardinality=4
+| in pipelines: 00(GETNEXT), 01(OPEN)
+|
+|--01:SCAN HDFS [tpch_parquet.lineitem]
+| HDFS partitions=1/1 files=3 size=194.00MB
+| runtime filters: RF000[bloom] -> tpch_parquet.lineitem.l_orderkey, RF002[bloom] -> l_orderkey
+| stored statistics:
+| table: rows=6.00M size=194.00MB
+| columns: all
+| extrapolated-rows=disabled max-scan-range-rows=2.14M
+| mem-estimate=80.00MB mem-reservation=4.00MB thread-reservation=1
+| tuple-ids=2 row-size=8B cardinality=6.00M
+| in pipelines: 01(GETNEXT)
+|
+00:SCAN HDFS [tpch_parquet.lineitem a]
+ HDFS partitions=1/1 files=3 size=194.00MB
+ runtime filters: RF004[bloom] -> l_orderkey
+ stored statistics:
+ table: rows=6.00M size=194.00MB
+ columns: all
+ extrapolated-rows=disabled max-scan-range-rows=2.14M
+ limit: 1
+ mem-estimate=80.00MB mem-reservation=4.00MB thread-reservation=1
+ tuple-ids=0 row-size=8B cardinality=1
+ in pipelines: 00(GETNEXT)
+---- DISTRIBUTEDPLAN
+F01:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1
+| Per-Host Resources: mem-estimate=947.37MB mem-reservation=102.75MB thread-reservation=1 runtime-filters-memory=768.00KB
+PLAN-ROOT SINK
+| output exprs: count(*)
+| mem-estimate=0B mem-reservation=0B thread-reservation=0
+|
+07:AGGREGATE [FINALIZE]
+| output: count(*)
+| mem-estimate=16.00KB mem-reservation=0B spill-buffer=2.00MB thread-reservation=0
+| tuple-ids=8 row-size=8B cardinality=1
+| in pipelines: 07(GETNEXT), 00(OPEN)
+|
+06:HASH JOIN [INNER JOIN, BROADCAST]
+| hash predicates: l_orderkey = l_orderkey
+| fk/pk conjuncts: none
+| runtime filters: RF000[bloom] <- l_orderkey
+| mem-estimate=305.50MB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| tuple-ids=0,2,4,6 row-size=32B cardinality=58
+| in pipelines: 00(GETNEXT), 03(OPEN)
+|
+|--11:EXCHANGE [UNPARTITIONED]
+| | mem-estimate=10.04MB mem-reservation=0B thread-reservation=0
+| | tuple-ids=6 row-size=8B cardinality=6.00M
+| | in pipelines: 03(GETNEXT)
+| |
+| F04:PLAN FRAGMENT [RANDOM] hosts=3 instances=3
+| Per-Host Resources: mem-estimate=80.05MB mem-reservation=4.00MB thread-reservation=2
+| 03:SCAN HDFS [tpch_parquet.lineitem, RANDOM]
+| HDFS partitions=1/1 files=3 size=194.00MB
+| stored statistics:
+| table: rows=6.00M size=194.00MB
+| columns: all
+| extrapolated-rows=disabled max-scan-range-rows=2.14M
+| mem-estimate=80.00MB mem-reservation=4.00MB thread-reservation=1
+| tuple-ids=6 row-size=8B cardinality=6.00M
+| in pipelines: 03(GETNEXT)
+|
+05:HASH JOIN [INNER JOIN, BROADCAST]
+| hash predicates: l_orderkey = l_orderkey
+| fk/pk conjuncts: none
+| runtime filters: RF002[bloom] <- l_orderkey
+| mem-estimate=305.50MB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| tuple-ids=0,2,4 row-size=24B cardinality=15
+| in pipelines: 00(GETNEXT), 02(OPEN)
+|
+|--10:EXCHANGE [UNPARTITIONED]
+| | mem-estimate=10.04MB mem-reservation=0B thread-reservation=0
+| | tuple-ids=4 row-size=8B cardinality=6.00M
+| | in pipelines: 02(GETNEXT)
+| |
+| F03:PLAN FRAGMENT [RANDOM] hosts=3 instances=3
+| Per-Host Resources: mem-estimate=80.30MB mem-reservation=4.25MB thread-reservation=2 runtime-filters-memory=256.00KB
+| 02:SCAN HDFS [tpch_parquet.lineitem, RANDOM]
+| HDFS partitions=1/1 files=3 size=194.00MB
+| runtime filters: RF000[bloom] -> l_orderkey
+| stored statistics:
+| table: rows=6.00M size=194.00MB
+| columns: all
+| extrapolated-rows=disabled max-scan-range-rows=2.14M
+| mem-estimate=80.00MB mem-reservation=4.00MB thread-reservation=1
+| tuple-ids=4 row-size=8B cardinality=6.00M
+| in pipelines: 02(GETNEXT)
+|
+04:HASH JOIN [INNER JOIN, BROADCAST]
+| hash predicates: l_orderkey = l_orderkey
+| fk/pk conjuncts: none
+| runtime filters: RF004[bloom] <- l_orderkey
+| mem-estimate=305.50MB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| tuple-ids=0,2 row-size=16B cardinality=4
+| in pipelines: 00(GETNEXT), 01(OPEN)
+|
+|--09:EXCHANGE [UNPARTITIONED]
+| | mem-estimate=10.04MB mem-reservation=0B thread-reservation=0
+| | tuple-ids=2 row-size=8B cardinality=6.00M
+| | in pipelines: 01(GETNEXT)
+| |
+| F02:PLAN FRAGMENT [RANDOM] hosts=3 instances=3
+| Per-Host Resources: mem-estimate=80.55MB mem-reservation=4.50MB thread-reservation=2 runtime-filters-memory=512.00KB
+| 01:SCAN HDFS [tpch_parquet.lineitem, RANDOM]
+| HDFS partitions=1/1 files=3 size=194.00MB
+| runtime filters: RF000[bloom] -> tpch_parquet.lineitem.l_orderkey, RF002[bloom] -> l_orderkey
+| stored statistics:
+| table: rows=6.00M size=194.00MB
+| columns: all
+| extrapolated-rows=disabled max-scan-range-rows=2.14M
+| mem-estimate=80.00MB mem-reservation=4.00MB thread-reservation=1
+| tuple-ids=2 row-size=8B cardinality=6.00M
+| in pipelines: 01(GETNEXT)
+|
+08:EXCHANGE [UNPARTITIONED]
+| limit: 1
+| mem-estimate=16.00KB mem-reservation=0B thread-reservation=0
+| tuple-ids=0 row-size=8B cardinality=1
+| in pipelines: 00(GETNEXT)
+|
+F00:PLAN FRAGMENT [RANDOM] hosts=3 instances=3
+Per-Host Resources: mem-estimate=80.30MB mem-reservation=4.25MB thread-reservation=2 runtime-filters-memory=256.00KB
+00:SCAN HDFS [tpch_parquet.lineitem a, RANDOM]
+ HDFS partitions=1/1 files=3 size=194.00MB
+ runtime filters: RF004[bloom] -> l_orderkey
+ stored statistics:
+ table: rows=6.00M size=194.00MB
+ columns: all
+ extrapolated-rows=disabled max-scan-range-rows=2.14M
+ limit: 1
+ mem-estimate=80.00MB mem-reservation=4.00MB thread-reservation=1
+ tuple-ids=0 row-size=8B cardinality=1
+ in pipelines: 00(GETNEXT)
+====
+# IMPALA-12357: RF004 still created because 01:SCAN has filter predicate.
+select STRAIGHT_JOIN count(*) from
+ (select l_orderkey from tpch_parquet.lineitem) a
+ join (select o_orderkey, o_custkey from tpch_parquet.orders) l1 on a.l_orderkey = l1.o_orderkey
+ join (select o_orderkey from tpch_parquet.orders) l2 on l1.o_orderkey = l2.o_orderkey
+ join (select o_orderkey from tpch_parquet.orders) l3 on l2.o_orderkey = l3.o_orderkey
+where l1.o_custkey < 1000;
+---- QUERYOPTIONS
+RUNTIME_FILTER_MAX_SIZE=256KB
+EXPLAIN_LEVEL=2
+---- PLAN
+F00:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1
+| Per-Host Resources: mem-estimate=367.64MB mem-reservation=96.75MB thread-reservation=5 runtime-filters-memory=256.00KB
+PLAN-ROOT SINK
+| output exprs: count(*)
+| mem-estimate=0B mem-reservation=0B thread-reservation=0
+|
+07:AGGREGATE [FINALIZE]
+| output: count(*)
+| mem-estimate=16.00KB mem-reservation=0B spill-buffer=2.00MB thread-reservation=0
+| tuple-ids=8 row-size=8B cardinality=1
+| in pipelines: 07(GETNEXT), 00(OPEN)
+|
+06:HASH JOIN [INNER JOIN]
+| hash predicates: o_orderkey = o_orderkey
+| fk/pk conjuncts: o_orderkey = o_orderkey
+| mem-estimate=59.44MB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| tuple-ids=0,2,4,6 row-size=40B cardinality=1
+| in pipelines: 00(GETNEXT), 03(OPEN)
+|
+|--03:SCAN HDFS [tpch_parquet.orders]
+| HDFS partitions=1/1 files=2 size=54.21MB
+| stored statistics:
+| table: rows=1.50M size=54.21MB
+| columns: all
+| extrapolated-rows=disabled max-scan-range-rows=1.18M
+| mem-estimate=40.00MB mem-reservation=4.00MB thread-reservation=1
+| tuple-ids=6 row-size=8B cardinality=1.50M
+| in pipelines: 03(GETNEXT)
+|
+05:HASH JOIN [INNER JOIN]
+| hash predicates: o_orderkey = o_orderkey
+| fk/pk conjuncts: o_orderkey = o_orderkey
+| mem-estimate=59.44MB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| tuple-ids=0,2,4 row-size=32B cardinality=1
+| in pipelines: 00(GETNEXT), 02(OPEN)
+|
+|--02:SCAN HDFS [tpch_parquet.orders]
+| HDFS partitions=1/1 files=2 size=54.21MB
+| stored statistics:
+| table: rows=1.50M size=54.21MB
+| columns: all
+| extrapolated-rows=disabled max-scan-range-rows=1.18M
+| mem-estimate=40.00MB mem-reservation=4.00MB thread-reservation=1
+| tuple-ids=4 row-size=8B cardinality=1.50M
+| in pipelines: 02(GETNEXT)
+|
+04:HASH JOIN [INNER JOIN]
+| hash predicates: l_orderkey = o_orderkey
+| fk/pk conjuncts: l_orderkey = o_orderkey
+| runtime filters: RF004[bloom] <- o_orderkey
+| mem-estimate=8.50MB mem-reservation=8.50MB spill-buffer=512.00KB thread-reservation=0
+| tuple-ids=0,2 row-size=24B cardinality=1
+| in pipelines: 00(GETNEXT), 01(OPEN)
+|
+|--01:SCAN HDFS [tpch_parquet.orders]
+| HDFS partitions=1/1 files=2 size=54.21MB
+| predicates: tpch_parquet.orders.o_custkey < CAST(1000 AS BIGINT)
+| stored statistics:
+| table: rows=1.50M size=54.21MB
+| columns: all
+| extrapolated-rows=disabled max-scan-range-rows=1.18M
+| parquet statistics predicates: tpch_parquet.orders.o_custkey < CAST(1000 AS BIGINT)
+| parquet dictionary predicates: tpch_parquet.orders.o_custkey < CAST(1000 AS BIGINT)
+| mem-estimate=80.00MB mem-reservation=8.00MB thread-reservation=1
+| tuple-ids=2 row-size=16B cardinality=150.00K
+| in pipelines: 01(GETNEXT)
+|
+00:SCAN HDFS [tpch_parquet.lineitem]
+ HDFS partitions=1/1 files=3 size=194.00MB
+ runtime filters: RF004[bloom] -> l_orderkey
+ stored statistics:
+ table: rows=6.00M size=194.00MB
+ columns: all
+ extrapolated-rows=disabled max-scan-range-rows=2.14M
+ mem-estimate=80.00MB mem-reservation=4.00MB thread-reservation=1
+ tuple-ids=0 row-size=8B cardinality=6.00M
+ in pipelines: 00(GETNEXT)
+---- DISTRIBUTEDPLAN
+F05:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1
+| Per-Host Resources: mem-estimate=32.00KB mem-reservation=0B thread-reservation=1
+PLAN-ROOT SINK
+| output exprs: count(*)
+| mem-estimate=0B mem-reservation=0B thread-reservation=0
+|
+13:AGGREGATE [FINALIZE]
+| output: count:merge(*)
+| mem-estimate=16.00KB mem-reservation=0B spill-buffer=2.00MB thread-reservation=0
+| tuple-ids=8 row-size=8B cardinality=1
+| in pipelines: 13(GETNEXT), 07(OPEN)
+|
+12:EXCHANGE [UNPARTITIONED]
+| mem-estimate=16.00KB mem-reservation=0B thread-reservation=0
+| tuple-ids=8 row-size=8B cardinality=1
+| in pipelines: 07(GETNEXT)
+|
+F03:PLAN FRAGMENT [HASH(o_orderkey)] hosts=3 instances=3
+Per-Host Resources: mem-estimate=83.98MB mem-reservation=68.00MB thread-reservation=1
+07:AGGREGATE
+| output: count(*)
+| mem-estimate=16.00KB mem-reservation=0B spill-buffer=2.00MB thread-reservation=0
+| tuple-ids=8 row-size=8B cardinality=1
+| in pipelines: 07(GETNEXT), 00(OPEN)
+|
+06:HASH JOIN [INNER JOIN, PARTITIONED]
+| hash predicates: o_orderkey = o_orderkey
+| fk/pk conjuncts: o_orderkey = o_orderkey
+| mem-estimate=34.00MB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| tuple-ids=0,2,4,6 row-size=40B cardinality=575.77K
+| in pipelines: 00(GETNEXT), 03(OPEN)
+|
+|--11:EXCHANGE [HASH(o_orderkey)]
+| | mem-estimate=5.75MB mem-reservation=0B thread-reservation=0
+| | tuple-ids=6 row-size=8B cardinality=1.50M
+| | in pipelines: 03(GETNEXT)
+| |
+| F04:PLAN FRAGMENT [RANDOM] hosts=2 instances=2
+| Per-Host Resources: mem-estimate=40.14MB mem-reservation=4.00MB thread-reservation=2
+| 03:SCAN HDFS [tpch_parquet.orders, RANDOM]
+| HDFS partitions=1/1 files=2 size=54.21MB
+| stored statistics:
+| table: rows=1.50M size=54.21MB
+| columns: all
+| extrapolated-rows=disabled max-scan-range-rows=1.18M
+| mem-estimate=40.00MB mem-reservation=4.00MB thread-reservation=1
+| tuple-ids=6 row-size=8B cardinality=1.50M
+| in pipelines: 03(GETNEXT)
+|
+05:HASH JOIN [INNER JOIN, PARTITIONED]
+| hash predicates: o_orderkey = o_orderkey
+| fk/pk conjuncts: o_orderkey = o_orderkey
+| mem-estimate=34.00MB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| tuple-ids=0,2,4 row-size=32B cardinality=575.77K
+| in pipelines: 00(GETNEXT), 02(OPEN)
+|
+|--10:EXCHANGE [HASH(o_orderkey)]
+| | mem-estimate=5.75MB mem-reservation=0B thread-reservation=0
+| | tuple-ids=4 row-size=8B cardinality=1.50M
+| | in pipelines: 02(GETNEXT)
+| |
+| F02:PLAN FRAGMENT [RANDOM] hosts=2 instances=2
+| Per-Host Resources: mem-estimate=40.14MB mem-reservation=4.00MB thread-reservation=2
+| 02:SCAN HDFS [tpch_parquet.orders, RANDOM]
+| HDFS partitions=1/1 files=2 size=54.21MB
+| stored statistics:
+| table: rows=1.50M size=54.21MB
+| columns: all
+| extrapolated-rows=disabled max-scan-range-rows=1.18M
+| mem-estimate=40.00MB mem-reservation=4.00MB thread-reservation=1
+| tuple-ids=4 row-size=8B cardinality=1.50M
+| in pipelines: 02(GETNEXT)
+|
+09:EXCHANGE [HASH(o_orderkey)]
+| mem-estimate=4.49MB mem-reservation=0B thread-reservation=0
+| tuple-ids=0,2 row-size=24B cardinality=575.77K
+| in pipelines: 00(GETNEXT)
+|
+F00:PLAN FRAGMENT [RANDOM] hosts=3 instances=3
+Per-Host Resources: mem-estimate=91.08MB mem-reservation=12.75MB thread-reservation=2 runtime-filters-memory=256.00KB
+04:HASH JOIN [INNER JOIN, BROADCAST]
+| hash predicates: l_orderkey = o_orderkey
+| fk/pk conjuncts: l_orderkey = o_orderkey
+| runtime filters: RF004[bloom] <- o_orderkey
+| mem-estimate=8.50MB mem-reservation=8.50MB spill-buffer=512.00KB thread-reservation=0
+| tuple-ids=0,2 row-size=24B cardinality=575.77K
+| in pipelines: 00(GETNEXT), 01(OPEN)
+|
+|--08:EXCHANGE [BROADCAST]
+| | mem-estimate=2.33MB mem-reservation=0B thread-reservation=0
+| | tuple-ids=2 row-size=16B cardinality=150.00K
+| | in pipelines: 01(GETNEXT)
+| |
+| F01:PLAN FRAGMENT [RANDOM] hosts=2 instances=2
+| Per-Host Resources: mem-estimate=80.08MB mem-reservation=8.00MB thread-reservation=2
+| 01:SCAN HDFS [tpch_parquet.orders, RANDOM]
+| HDFS partitions=1/1 files=2 size=54.21MB
+| predicates: tpch_parquet.orders.o_custkey < CAST(1000 AS BIGINT)
+| stored statistics:
+| table: rows=1.50M size=54.21MB
+| columns: all
+| extrapolated-rows=disabled max-scan-range-rows=1.18M
+| parquet statistics predicates: tpch_parquet.orders.o_custkey < CAST(1000 AS BIGINT)
+| parquet dictionary predicates: tpch_parquet.orders.o_custkey < CAST(1000 AS BIGINT)
+| mem-estimate=80.00MB mem-reservation=8.00MB thread-reservation=1
+| tuple-ids=2 row-size=16B cardinality=150.00K
+| in pipelines: 01(GETNEXT)
+|
+00:SCAN HDFS [tpch_parquet.lineitem, RANDOM]
+ HDFS partitions=1/1 files=3 size=194.00MB
+ runtime filters: RF004[bloom] -> l_orderkey
+ stored statistics:
+ table: rows=6.00M size=194.00MB
+ columns: all
+ extrapolated-rows=disabled max-scan-range-rows=2.14M
+ mem-estimate=80.00MB mem-reservation=4.00MB thread-reservation=1
+ tuple-ids=0 row-size=8B cardinality=6.00M
+ in pipelines: 00(GETNEXT)
+====
+# IMPALA-12357: build fragment with union still eligible for filter elimination.
+select STRAIGHT_JOIN count(*) from
+ (select l_orderkey from tpch_parquet.lineitem) a
+ join (
+ select o_orderkey from tpch_parquet.orders
+ union all
+ select o_orderkey from tpch_parquet.orders
+ ) l1 on a.l_orderkey = l1.o_orderkey;
+---- QUERYOPTIONS
+RUNTIME_FILTER_MAX_SIZE=256KB
+EXPLAIN_LEVEL=2
+---- PLAN
+F00:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1
+| Per-Host Resources: mem-estimate=262.03MB mem-reservation=42.25MB thread-reservation=3
+PLAN-ROOT SINK
+| output exprs: count(*)
+| mem-estimate=0B mem-reservation=0B thread-reservation=0
+|
+05:AGGREGATE [FINALIZE]
+| output: count(*)
+| mem-estimate=16.00KB mem-reservation=0B spill-buffer=2.00MB thread-reservation=0
+| tuple-ids=6 row-size=8B cardinality=1
+| in pipelines: 05(GETNEXT), 00(OPEN)
+|
+04:HASH JOIN [INNER JOIN]
+| hash predicates: l_orderkey = o_orderkey
+| fk/pk conjuncts: assumed fk/pk
+| mem-estimate=141.78MB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| tuple-ids=0,4 row-size=16B cardinality=2
+| in pipelines: 00(GETNEXT), 02(OPEN), 03(OPEN)
+|
+|--01:UNION
+| | pass-through-operands: all
+| | mem-estimate=0B mem-reservation=0B thread-reservation=0
+| | tuple-ids=4 row-size=8B cardinality=3.00M
+| | in pipelines: 02(GETNEXT), 03(GETNEXT)
+| |
+| |--03:SCAN HDFS [tpch_parquet.orders]
+| | HDFS partitions=1/1 files=2 size=54.21MB
+| | stored statistics:
+| | table: rows=1.50M size=54.21MB
+| | columns: all
+| | extrapolated-rows=disabled max-scan-range-rows=1.18M
+| | mem-estimate=40.00MB mem-reservation=4.00MB thread-reservation=1
+| | tuple-ids=3 row-size=8B cardinality=1.50M
+| | in pipelines: 03(GETNEXT)
+| |
+| 02:SCAN HDFS [tpch_parquet.orders]
+| HDFS partitions=1/1 files=2 size=54.21MB
+| stored statistics:
+| table: rows=1.50M size=54.21MB
+| columns: all
+| extrapolated-rows=disabled max-scan-range-rows=1.18M
+| mem-estimate=40.00MB mem-reservation=4.00MB thread-reservation=1
+| tuple-ids=2 row-size=8B cardinality=1.50M
+| in pipelines: 02(GETNEXT)
+|
+00:SCAN HDFS [tpch_parquet.lineitem]
+ HDFS partitions=1/1 files=3 size=194.00MB
+ stored statistics:
+ table: rows=6.00M size=194.00MB
+ columns: all
+ extrapolated-rows=disabled max-scan-range-rows=2.14M
+ mem-estimate=80.00MB mem-reservation=4.00MB thread-reservation=1
+ tuple-ids=0 row-size=8B cardinality=6.00M
+ in pipelines: 00(GETNEXT)
+---- DISTRIBUTEDPLAN
+F05:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1
+| Per-Host Resources: mem-estimate=32.00KB mem-reservation=0B thread-reservation=1
+PLAN-ROOT SINK
+| output exprs: count(*)
+| mem-estimate=0B mem-reservation=0B thread-reservation=0
+|
+09:AGGREGATE [FINALIZE]
+| output: count:merge(*)
+| mem-estimate=16.00KB mem-reservation=0B spill-buffer=2.00MB thread-reservation=0
+| tuple-ids=6 row-size=8B cardinality=1
+| in pipelines: 09(GETNEXT), 05(OPEN)
+|
+08:EXCHANGE [UNPARTITIONED]
+| mem-estimate=16.00KB mem-reservation=0B thread-reservation=0
+| tuple-ids=6 row-size=8B cardinality=1
+| in pipelines: 05(GETNEXT)
+|
+F04:PLAN FRAGMENT [HASH(l_orderkey)] hosts=3 instances=3
+Per-Host Resources: mem-estimate=67.32MB mem-reservation=34.00MB thread-reservation=1
+05:AGGREGATE
+| output: count(*)
+| mem-estimate=16.00KB mem-reservation=0B spill-buffer=2.00MB thread-reservation=0
+| tuple-ids=6 row-size=8B cardinality=1
+| in pipelines: 05(GETNEXT), 00(OPEN)
+|
+04:HASH JOIN [INNER JOIN, PARTITIONED]
+| hash predicates: l_orderkey = o_orderkey
+| fk/pk conjuncts: assumed fk/pk
+| mem-estimate=47.26MB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| tuple-ids=0,4 row-size=16B cardinality=11.52M
+| in pipelines: 00(GETNEXT), 02(OPEN), 03(OPEN)
+|
+|--07:EXCHANGE [HASH(o_orderkey)]
+| | mem-estimate=10.02MB mem-reservation=0B thread-reservation=0
+| | tuple-ids=4 row-size=8B cardinality=3.00M
+| | in pipelines: 02(GETNEXT), 03(GETNEXT)
+| |
+| F03:PLAN FRAGMENT [RANDOM] hosts=2 instances=2
+| Per-Host Resources: mem-estimate=40.14MB mem-reservation=4.00MB thread-reservation=2
+| 01:UNION
+| | pass-through-operands: all
+| | mem-estimate=0B mem-reservation=0B thread-reservation=0
+| | tuple-ids=4 row-size=8B cardinality=3.00M
+| | in pipelines: 02(GETNEXT), 03(GETNEXT)
+| |
+| |--03:SCAN HDFS [tpch_parquet.orders, RANDOM]
+| | HDFS partitions=1/1 files=2 size=54.21MB
+| | stored statistics:
+| | table: rows=1.50M size=54.21MB
+| | columns: all
+| | extrapolated-rows=disabled max-scan-range-rows=1.18M
+| | mem-estimate=40.00MB mem-reservation=4.00MB thread-reservation=1
+| | tuple-ids=3 row-size=8B cardinality=1.50M
+| | in pipelines: 03(GETNEXT)
+| |
+| 02:SCAN HDFS [tpch_parquet.orders, RANDOM]
+| HDFS partitions=1/1 files=2 size=54.21MB
+| stored statistics:
+| table: rows=1.50M size=54.21MB
+| columns: all
+| extrapolated-rows=disabled max-scan-range-rows=1.18M
+| mem-estimate=40.00MB mem-reservation=4.00MB thread-reservation=1
+| tuple-ids=2 row-size=8B cardinality=1.50M
+| in pipelines: 02(GETNEXT)
+|
+06:EXCHANGE [HASH(l_orderkey)]
+| mem-estimate=10.04MB mem-reservation=0B thread-reservation=0
+| tuple-ids=0 row-size=8B cardinality=6.00M
+| in pipelines: 00(GETNEXT)
+|
+F00:PLAN FRAGMENT [RANDOM] hosts=3 instances=3
+Per-Host Resources: mem-estimate=80.14MB mem-reservation=4.00MB thread-reservation=2
+00:SCAN HDFS [tpch_parquet.lineitem, RANDOM]
+ HDFS partitions=1/1 files=3 size=194.00MB
+ stored statistics:
+ table: rows=6.00M size=194.00MB
+ columns: all
+ extrapolated-rows=disabled max-scan-range-rows=2.14M
+ mem-estimate=80.00MB mem-reservation=4.00MB thread-reservation=1
+ tuple-ids=0 row-size=8B cardinality=6.00M
+ in pipelines: 00(GETNEXT)
+====
+# IMPALA-12357: build fragment with plan node other than scan or union is not eligible
+# for filter elimination.
+select STRAIGHT_JOIN count(*) from
+ (select l_orderkey from tpch_parquet.lineitem) a
+ join (select distinct o_orderkey from tpch_parquet.orders) l1 on a.l_orderkey = l1.o_orderkey;
+---- QUERYOPTIONS
+RUNTIME_FILTER_MAX_SIZE=256KB
+EXPLAIN_LEVEL=2
+---- PLAN
+F00:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1
+| Per-Host Resources: mem-estimate=173.69MB mem-reservation=72.25MB thread-reservation=3 runtime-filters-memory=256.00KB
+PLAN-ROOT SINK
+| output exprs: count(*)
+| mem-estimate=0B mem-reservation=0B thread-reservation=0
+|
+04:AGGREGATE [FINALIZE]
+| output: count(*)
+| mem-estimate=16.00KB mem-reservation=0B spill-buffer=2.00MB thread-reservation=0
+| tuple-ids=5 row-size=8B cardinality=1
+| in pipelines: 04(GETNEXT), 00(OPEN)
+|
+03:HASH JOIN [INNER JOIN]
+| hash predicates: l_orderkey = o_orderkey
+| fk/pk conjuncts: l_orderkey = o_orderkey
+| runtime filters: RF000[bloom] <- o_orderkey
+| mem-estimate=59.44MB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| tuple-ids=0,3 row-size=16B cardinality=5.76M
+| in pipelines: 00(GETNEXT), 02(OPEN)
+|
+|--02:AGGREGATE [FINALIZE]
+| | group by: o_orderkey
+| | mem-estimate=34.00MB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| | tuple-ids=3 row-size=8B cardinality=1.50M
+| | in pipelines: 02(GETNEXT), 01(OPEN)
+| |
+| 01:SCAN HDFS [tpch_parquet.orders]
+| HDFS partitions=1/1 files=2 size=54.21MB
+| stored statistics:
+| table: rows=1.50M size=54.21MB
+| columns: all
+| extrapolated-rows=disabled max-scan-range-rows=1.18M
+| mem-estimate=40.00MB mem-reservation=4.00MB thread-reservation=1
+| tuple-ids=2 row-size=8B cardinality=1.50M
+| in pipelines: 01(GETNEXT)
+|
+00:SCAN HDFS [tpch_parquet.lineitem]
+ HDFS partitions=1/1 files=3 size=194.00MB
+ runtime filters: RF000[bloom] -> l_orderkey
+ stored statistics:
+ table: rows=6.00M size=194.00MB
+ columns: all
+ extrapolated-rows=disabled max-scan-range-rows=2.14M
+ mem-estimate=80.00MB mem-reservation=4.00MB thread-reservation=1
+ tuple-ids=0 row-size=8B cardinality=6.00M
+ in pipelines: 00(GETNEXT)
+---- DISTRIBUTEDPLAN
+F03:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1
+| Per-Host Resources: mem-estimate=32.00KB mem-reservation=0B thread-reservation=1
+PLAN-ROOT SINK
+| output exprs: count(*)
+| mem-estimate=0B mem-reservation=0B thread-reservation=0
+|
+09:AGGREGATE [FINALIZE]
+| output: count:merge(*)
+| mem-estimate=16.00KB mem-reservation=0B spill-buffer=2.00MB thread-reservation=0
+| tuple-ids=5 row-size=8B cardinality=1
+| in pipelines: 09(GETNEXT), 04(OPEN)
+|
+08:EXCHANGE [UNPARTITIONED]
+| mem-estimate=16.00KB mem-reservation=0B thread-reservation=0
+| tuple-ids=5 row-size=8B cardinality=1
+| in pipelines: 04(GETNEXT)
+|
+F02:PLAN FRAGMENT [HASH(o_orderkey)] hosts=3 instances=3
+Per-Host Resources: mem-estimate=61.29MB mem-reservation=51.25MB thread-reservation=1 runtime-filters-memory=256.00KB
+04:AGGREGATE
+| output: count(*)
+| mem-estimate=16.00KB mem-reservation=0B spill-buffer=2.00MB thread-reservation=0
+| tuple-ids=5 row-size=8B cardinality=1
+| in pipelines: 04(GETNEXT), 00(OPEN)
+|
+03:HASH JOIN [INNER JOIN, PARTITIONED]
+| hash predicates: l_orderkey = o_orderkey
+| fk/pk conjuncts: l_orderkey = o_orderkey
+| runtime filters: RF000[bloom] <- o_orderkey
+| mem-estimate=34.00MB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| tuple-ids=0,3 row-size=16B cardinality=5.76M
+| in pipelines: 00(GETNEXT), 06(OPEN)
+|
+|--06:AGGREGATE [FINALIZE]
+| | group by: o_orderkey
+| | mem-estimate=17.00MB mem-reservation=17.00MB spill-buffer=1.00MB thread-reservation=0
+| | tuple-ids=3 row-size=8B cardinality=1.50M
+| | in pipelines: 06(GETNEXT), 01(OPEN)
+| |
+| 05:EXCHANGE [HASH(o_orderkey)]
+| | mem-estimate=5.75MB mem-reservation=0B thread-reservation=0
+| | tuple-ids=3 row-size=8B cardinality=1.50M
+| | in pipelines: 01(GETNEXT)
+| |
+| F01:PLAN FRAGMENT [RANDOM] hosts=2 instances=2
+| Per-Host Resources: mem-estimate=57.14MB mem-reservation=21.00MB thread-reservation=2
+| 02:AGGREGATE [STREAMING]
+| | group by: o_orderkey
+| | mem-estimate=17.00MB mem-reservation=17.00MB spill-buffer=1.00MB thread-reservation=0
+| | tuple-ids=3 row-size=8B cardinality=1.50M
+| | in pipelines: 01(GETNEXT)
+| |
+| 01:SCAN HDFS [tpch_parquet.orders, RANDOM]
+| HDFS partitions=1/1 files=2 size=54.21MB
+| stored statistics:
+| table: rows=1.50M size=54.21MB
+| columns: all
+| extrapolated-rows=disabled max-scan-range-rows=1.18M
+| mem-estimate=40.00MB mem-reservation=4.00MB thread-reservation=1
+| tuple-ids=2 row-size=8B cardinality=1.50M
+| in pipelines: 01(GETNEXT)
+|
+07:EXCHANGE [HASH(l_orderkey)]
+| mem-estimate=10.04MB mem-reservation=0B thread-reservation=0
+| tuple-ids=0 row-size=8B cardinality=6.00M
+| in pipelines: 00(GETNEXT)
+|
+F00:PLAN FRAGMENT [RANDOM] hosts=3 instances=3
+Per-Host Resources: mem-estimate=80.39MB mem-reservation=4.25MB thread-reservation=2 runtime-filters-memory=256.00KB
+00:SCAN HDFS [tpch_parquet.lineitem, RANDOM]
+ HDFS partitions=1/1 files=3 size=194.00MB
+ runtime filters: RF000[bloom] -> l_orderkey
+ stored statistics:
+ table: rows=6.00M size=194.00MB
+ columns: all
+ extrapolated-rows=disabled max-scan-range-rows=2.14M
+ mem-estimate=80.00MB mem-reservation=4.00MB thread-reservation=1
+ tuple-ids=0 row-size=8B cardinality=6.00M
+ in pipelines: 00(GETNEXT)
+====
+# IMPALA-12357: build fragment scanning collection type is not eligible
+# for filter elimination.
+select STRAIGHT_JOIN count(*) from
+ (select l_orderkey from tpch_parquet.lineitem) a
+ join (select item from functional_parquet.complextypestbl_medium.int_array_array.item) l1 on a.l_orderkey = l1.item;
+---- QUERYOPTIONS
+RUNTIME_FILTER_MAX_SIZE=256KB
+EXPLAIN_LEVEL=2
+---- PLAN
+F00:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1
+| Per-Host Resources: mem-estimate=980.77MB mem-reservation=38.75MB thread-reservation=3 runtime-filters-memory=256.00KB
+PLAN-ROOT SINK
+| output exprs: count(*)
+| mem-estimate=0B mem-reservation=0B thread-reservation=0
+|
+03:AGGREGATE [FINALIZE]
+| output: count(*)
+| mem-estimate=16.00KB mem-reservation=0B spill-buffer=2.00MB thread-reservation=0
+| tuple-ids=4 row-size=8B cardinality=1
+| in pipelines: 03(GETNEXT), 00(OPEN)
+|
+02:HASH JOIN [INNER JOIN]
+| hash predicates: l_orderkey = item
+| fk/pk conjuncts: assumed fk/pk
+| runtime filters: RF000[bloom] <- item
+| mem-estimate=884.52MB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| tuple-ids=0,2 row-size=12B cardinality=6.00M
+| in pipelines: 00(GETNEXT), 01(OPEN)
+|
+|--01:SCAN HDFS [functional_parquet.complextypestbl_medium.int_array_array.item]
+| HDFS partitions=1/1 files=1 size=480.14KB
+| stored statistics:
+| table: rows=unavailable size=480.14KB
+| columns: all
+| extrapolated-rows=disabled max-scan-range-rows=unavailable
+| mem-estimate=16.00MB mem-reservation=512.00KB thread-reservation=1
+| tuple-ids=2 row-size=4B cardinality=30.54M
+| in pipelines: 01(GETNEXT)
+|
+00:SCAN HDFS [tpch_parquet.lineitem]
+ HDFS partitions=1/1 files=3 size=194.00MB
+ runtime filters: RF000[bloom] -> l_orderkey
+ stored statistics:
+ table: rows=6.00M size=194.00MB
+ columns: all
+ extrapolated-rows=disabled max-scan-range-rows=2.14M
+ mem-estimate=80.00MB mem-reservation=4.00MB thread-reservation=1
+ tuple-ids=0 row-size=8B cardinality=6.00M
+ in pipelines: 00(GETNEXT)
+---- DISTRIBUTEDPLAN
+F03:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1
+| Per-Host Resources: mem-estimate=32.00KB mem-reservation=0B thread-reservation=1
+PLAN-ROOT SINK
+| output exprs: count(*)
+| mem-estimate=0B mem-reservation=0B thread-reservation=0
+|
+07:AGGREGATE [FINALIZE]
+| output: count:merge(*)
+| mem-estimate=16.00KB mem-reservation=0B spill-buffer=2.00MB thread-reservation=0
+| tuple-ids=4 row-size=8B cardinality=1
+| in pipelines: 07(GETNEXT), 03(OPEN)
+|
+06:EXCHANGE [UNPARTITIONED]
+| mem-estimate=16.00KB mem-reservation=0B thread-reservation=0
+| tuple-ids=4 row-size=8B cardinality=1
+| in pipelines: 03(GETNEXT)
+|
+F02:PLAN FRAGMENT [HASH(l_orderkey)] hosts=3 instances=3
+Per-Host Resources: mem-estimate=315.13MB mem-reservation=34.25MB thread-reservation=1 runtime-filters-memory=256.00KB
+03:AGGREGATE
+| output: count(*)
+| mem-estimate=16.00KB mem-reservation=0B spill-buffer=2.00MB thread-reservation=0
+| tuple-ids=4 row-size=8B cardinality=1
+| in pipelines: 03(GETNEXT), 00(OPEN)
+|
+02:HASH JOIN [INNER JOIN, PARTITIONED]
+| hash predicates: l_orderkey = item
+| fk/pk conjuncts: assumed fk/pk
+| runtime filters: RF000[bloom] <- item
+| mem-estimate=294.84MB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| tuple-ids=0,2 row-size=12B cardinality=6.00M
+| in pipelines: 00(GETNEXT), 01(OPEN)
+|
+|--05:EXCHANGE [HASH(item)]
+| | mem-estimate=10.01MB mem-reservation=0B thread-reservation=0
+| | tuple-ids=2 row-size=4B cardinality=30.54M
+| | in pipelines: 01(GETNEXT)
+| |
+| F01:PLAN FRAGMENT [RANDOM] hosts=1 instances=1
+| Per-Host Resources: mem-estimate=16.09MB mem-reservation=512.00KB thread-reservation=2
+| 01:SCAN HDFS [functional_parquet.complextypestbl_medium.int_array_array.item, RANDOM]
+| HDFS partitions=1/1 files=1 size=480.14KB
+| stored statistics:
+| table: rows=unavailable size=480.14KB
+| columns: all
+| extrapolated-rows=disabled max-scan-range-rows=unavailable
+| mem-estimate=16.00MB mem-reservation=512.00KB thread-reservation=1
+| tuple-ids=2 row-size=4B cardinality=30.54M
+| in pipelines: 01(GETNEXT)
+|
+04:EXCHANGE [HASH(l_orderkey)]
+| mem-estimate=10.04MB mem-reservation=0B thread-reservation=0
+| tuple-ids=0 row-size=8B cardinality=6.00M
+| in pipelines: 00(GETNEXT)
+|
+F00:PLAN FRAGMENT [RANDOM] hosts=3 instances=3
+Per-Host Resources: mem-estimate=80.39MB mem-reservation=4.25MB thread-reservation=2 runtime-filters-memory=256.00KB
+00:SCAN HDFS [tpch_parquet.lineitem, RANDOM]
+ HDFS partitions=1/1 files=3 size=194.00MB
+ runtime filters: RF000[bloom] -> l_orderkey
+ stored statistics:
+ table: rows=6.00M size=194.00MB
+ columns: all
+ extrapolated-rows=disabled max-scan-range-rows=2.14M
+ mem-estimate=80.00MB mem-reservation=4.00MB thread-reservation=1
+ tuple-ids=0 row-size=8B cardinality=6.00M
+ in pipelines: 00(GETNEXT)
+====