This is an automated email from the ASF dual-hosted git repository. michaelsmith pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/impala.git
commit 7a94adbc3076881f6e7043a429a7294a04bd7519 Author: Riza Suminto <[email protected]> AuthorDate: Wed Jun 7 18:16:34 2023 -0700 IMPALA-12192: Fix scaling bug in scan fragment IMPALA-12091 has a bug where scan fragment parallelism will always be limited solely by the ScanNode cost. If ScanNode is colocated with other query node operators that have higher processing costs, Planner will not scale it up beyond what is allowed by the ScanNode cost. This patch fixes the problem in two aspects. The first is to allow a scan fragment to scale up higher as long as it is within the total fragment cost and the number of effective scan ranges. The second is to add missing Math.max() in CostingSegment.java which causes lower fragment parallelism even when the total fragment cost is high. IMPALA-10287 optimization is re-enabled to reduce regression in TPC-DS Q78. Ideally, the broadcast vs partitioned costing formula during distributed planning should not rely on numInstance. But enabling this optimization ensures consistent query plan shape when comparing against MT_DOP plan. This optimization can still be disabled by specifying USE_DOP_FOR_COSTING=false. This patch also does some cleanup including: - Fix "max-parallelism" value in explain string. - Make a constant in ScanNode.rowMaterializationCost() into a backend flag named scan_range_cost_factor for experimental purposes. - Replace all references to ProcessingCost.isComputeCost() to queryOptions.isCompute_processing_cost() directly. - Add Precondition in PlanFragment.getNumInstances() to verify that the fragment's num instance is not modified anymore after the costing algorithm finish. Testing: - Manually run TPCDS Q84 over tpcds10_parquet and confirm that the leftmost scan fragment parallelism is raised from 12 (before the patch) to 18 (after the patch). - Add test in PlannerTest.testProcessingCost that reproduces the issue. - Update compute stats test in test_executor_groups.py to maintain test assertion. - Pass core tests. Change-Id: I7010f6c3bc48ae3f74e8db98a83f645b6c157226 Reviewed-on: http://gerrit.cloudera.org:8080/20024 Reviewed-by: Impala Public Jenkins <[email protected]> Tested-by: Impala Public Jenkins <[email protected]> --- be/src/util/backend-gflag-util.cc | 8 + common/thrift/BackendGflags.thrift | 2 + .../org/apache/impala/planner/CostingSegment.java | 5 +- .../java/org/apache/impala/planner/DataSink.java | 2 +- .../apache/impala/planner/DistributedPlanner.java | 16 +- .../org/apache/impala/planner/HBaseScanNode.java | 4 +- .../org/apache/impala/planner/HdfsScanNode.java | 4 +- .../org/apache/impala/planner/KuduScanNode.java | 4 +- .../org/apache/impala/planner/PlanFragment.java | 138 +++-- .../java/org/apache/impala/planner/PlanNode.java | 4 +- .../java/org/apache/impala/planner/Planner.java | 2 +- .../org/apache/impala/planner/ProcessingCost.java | 4 - .../java/org/apache/impala/planner/ScanNode.java | 53 +- .../org/apache/impala/service/BackendConfig.java | 4 + .../java/org/apache/impala/service/Frontend.java | 3 +- .../queries/PlannerTest/tpcds-processing-cost.test | 673 +++++++++++++-------- tests/custom_cluster/test_executor_groups.py | 68 ++- 17 files changed, 585 insertions(+), 409 deletions(-) diff --git a/be/src/util/backend-gflag-util.cc b/be/src/util/backend-gflag-util.cc index ec5f317ec..eda30258c 100644 --- a/be/src/util/backend-gflag-util.cc +++ b/be/src/util/backend-gflag-util.cc @@ -224,6 +224,13 @@ DEFINE_int64_hidden(min_processing_per_thread, 10000000, "number of cores in selected executor group, MT_DOP, or PROCESSING_COST_MIN_THREAD " "query option. Must be a positive integer. Default to 10M."); +// TODO: Benchmark and tune this config with an optimal value. +DEFINE_double_hidden(scan_range_cost_factor, 0.005, + "(Advance) Cost factor associated with processing one scan range. Combined with " + "min_processing_per_thread flag, this flag define the cost to process one scan range " + "as (scan_range_cost_factor * min_processing_per_thread). Default to 0.005, which " + "roughly means that one scan node instance will handle at most 200 scan ranges."); + DEFINE_bool_hidden(skip_resource_checking_on_last_executor_group_set, true, "(Advance) If true, memory and cpu resource checking will be skipped when a query " "is being planned against the last (largest) executor group set. Setting true will " @@ -411,6 +418,7 @@ Status PopulateThriftBackendGflags(TBackendGflags& cfg) { FLAGS_skip_resource_checking_on_last_executor_group_set); cfg.__set_file_metadata_reload_properties(FLAGS_file_metadata_reload_properties); cfg.__set_thrift_rpc_max_message_size(FLAGS_thrift_rpc_max_message_size); + cfg.__set_scan_range_cost_factor(FLAGS_scan_range_cost_factor); return Status::OK(); } diff --git a/common/thrift/BackendGflags.thrift b/common/thrift/BackendGflags.thrift index 878733ec2..77e7420c0 100644 --- a/common/thrift/BackendGflags.thrift +++ b/common/thrift/BackendGflags.thrift @@ -254,4 +254,6 @@ struct TBackendGflags { 111: required i32 thrift_rpc_max_message_size 112: required string file_metadata_reload_properties + + 113: required double scan_range_cost_factor } diff --git a/fe/src/main/java/org/apache/impala/planner/CostingSegment.java b/fe/src/main/java/org/apache/impala/planner/CostingSegment.java index 8c79a6511..27ea37300 100644 --- a/fe/src/main/java/org/apache/impala/planner/CostingSegment.java +++ b/fe/src/main/java/org/apache/impala/planner/CostingSegment.java @@ -190,10 +190,7 @@ public class CostingSegment extends TreeNode<CostingSegment> { ProcessingCost.tryAdjustConsumerParallelism( nodeStepCount, minParallelism, maxParallelism, producerCost, cost_); - newParallelism = cost_.getNumInstancesExpected(); - Preconditions.checkState(newParallelism >= minParallelism, - "originalParallelism=" + originalParallelism + ". newParallelism=" - + newParallelism + " < minParallelism=" + minParallelism); + newParallelism = Math.max(newParallelism, cost_.getNumInstancesExpected()); Preconditions.checkState(newParallelism <= maxParallelism, "originalParallelism=" + originalParallelism + ". newParallelism=" + newParallelism + " > maxParallelism=" + maxParallelism); diff --git a/fe/src/main/java/org/apache/impala/planner/DataSink.java b/fe/src/main/java/org/apache/impala/planner/DataSink.java index 4b665b4c4..2cda96a02 100644 --- a/fe/src/main/java/org/apache/impala/planner/DataSink.java +++ b/fe/src/main/java/org/apache/impala/planner/DataSink.java @@ -60,7 +60,7 @@ public abstract class DataSink { if (explainLevel.ordinal() >= TExplainLevel.EXTENDED.ordinal()) { output.append(detailPrefix); output.append(resourceProfile_.getExplainString()); - if (ProcessingCost.isComputeCost(queryOptions)) { + if (queryOptions.isCompute_processing_cost()) { // Show processing cost total. output.append(" cost="); if (processingCost_.isValid()) { diff --git a/fe/src/main/java/org/apache/impala/planner/DistributedPlanner.java b/fe/src/main/java/org/apache/impala/planner/DistributedPlanner.java index a6845bb15..f2aa836bf 100644 --- a/fe/src/main/java/org/apache/impala/planner/DistributedPlanner.java +++ b/fe/src/main/java/org/apache/impala/planner/DistributedPlanner.java @@ -192,7 +192,7 @@ public class DistributedPlanner { PlanFragment inputFragment, InsertStmt insertStmt, Analyzer analyzer, List<PlanFragment> fragments) throws ImpalaException { - boolean isComputeCost = ProcessingCost.isComputeCost(analyzer.getQueryOptions()); + boolean isComputeCost = analyzer.getQueryOptions().isCompute_processing_cost(); boolean enforce_hdfs_writer_limit = insertStmt.getTargetTable() instanceof FeFsTable && (analyzer.getQueryOptions().getMax_fs_writers() > 0 || isComputeCost); @@ -271,9 +271,8 @@ public class DistributedPlanner { // to help estimate the scan parallelism. int maxScanThread = 1; for (HdfsScanNode scanNode : hdfsScanNodes) { - long totalScanRange = scanNode.getEffectiveNumScanRanges(); - ProcessingCost scanCost = scanNode.computeScanProcessingCost( - analyzer.getQueryOptions(), totalScanRange); + ProcessingCost scanCost = + scanNode.computeScanProcessingCost(analyzer.getQueryOptions()); maxScanThread = Math.max( maxScanThread, scanCost.getNumInstanceMax(inputFragment.getNumNodes())); } @@ -571,8 +570,7 @@ public class DistributedPlanner { // because it must be broadcast once per instance. long dataPayload = rhsDataSize * leftChildNodes; long hashTblBuildCost = dataPayload; - if (mt_dop > 1 && ctx_.getQueryOptions().use_dop_for_costing - && !ProcessingCost.isComputeCost(ctx_.getQueryOptions())) { + if (mt_dop > 1 && ctx_.getQueryOptions().use_dop_for_costing) { // In the broadcast join a single thread per node is building the hash // table of size N compared to the partition case where m threads are // building hash tables of size N/m each (assuming uniform distribution). @@ -582,9 +580,9 @@ public class DistributedPlanner { // growth (a tunable parameter). We use the sqrt to model a non-linear // function since the slowdown with broadcast is not exactly linear. // TODO: more analysis is needed to establish an accurate correlation. - // TODO: this cost calculation is disabled if COMPUTE_PROCESSING_COST=true - // since num instances might change after plan created during - // Planner.computeProcessingCost() later. Need to find way to enable this back. + // TODO: revisit this calculation if COMPUTE_PROCESSING_COST=true. + // Num instances might change during Planner.computeProcessingCost(), + // later after parallel plan created. PlanNode leftPlanRoot = leftChildFragment.getPlanRoot(); int actual_dop = leftPlanRoot.getNumInstances()/leftPlanRoot.getNumNodes(); hashTblBuildCost *= (long) (ctx_.getQueryOptions().broadcast_to_partition_factor diff --git a/fe/src/main/java/org/apache/impala/planner/HBaseScanNode.java b/fe/src/main/java/org/apache/impala/planner/HBaseScanNode.java index 4db626000..96460b14b 100644 --- a/fe/src/main/java/org/apache/impala/planner/HBaseScanNode.java +++ b/fe/src/main/java/org/apache/impala/planner/HBaseScanNode.java @@ -656,9 +656,7 @@ public class HBaseScanNode extends ScanNode { @Override public void computeProcessingCost(TQueryOptions queryOptions) { - Preconditions.checkNotNull(scanRangeSpecs_); - processingCost_ = - computeScanProcessingCost(queryOptions, scanRangeSpecs_.getConcrete_rangesSize()); + processingCost_ = computeScanProcessingCost(queryOptions); } @Override diff --git a/fe/src/main/java/org/apache/impala/planner/HdfsScanNode.java b/fe/src/main/java/org/apache/impala/planner/HdfsScanNode.java index 55904d3f6..2e1a87d88 100644 --- a/fe/src/main/java/org/apache/impala/planner/HdfsScanNode.java +++ b/fe/src/main/java/org/apache/impala/planner/HdfsScanNode.java @@ -2112,8 +2112,7 @@ public class HdfsScanNode extends ScanNode { @Override public void computeProcessingCost(TQueryOptions queryOptions) { Preconditions.checkNotNull(scanRangeSpecs_); - processingCost_ = - computeScanProcessingCost(queryOptions, getEffectiveNumScanRanges()); + processingCost_ = computeScanProcessingCost(queryOptions); } @Override @@ -2496,6 +2495,7 @@ public class HdfsScanNode extends ScanNode { * Return the number of scan ranges when considering MAX_SCAN_RANGE_LENGTH option. * computeScanRangeLocations() must be called before calling this. */ + @Override public long getEffectiveNumScanRanges() { Preconditions.checkNotNull(scanRangeSpecs_); Preconditions.checkState( diff --git a/fe/src/main/java/org/apache/impala/planner/KuduScanNode.java b/fe/src/main/java/org/apache/impala/planner/KuduScanNode.java index e774f114b..d2ec18c91 100644 --- a/fe/src/main/java/org/apache/impala/planner/KuduScanNode.java +++ b/fe/src/main/java/org/apache/impala/planner/KuduScanNode.java @@ -397,9 +397,7 @@ public class KuduScanNode extends ScanNode { @Override public void computeProcessingCost(TQueryOptions queryOptions) { - Preconditions.checkNotNull(scanRangeSpecs_); - processingCost_ = - computeScanProcessingCost(queryOptions, scanRangeSpecs_.getConcrete_rangesSize()); + processingCost_ = computeScanProcessingCost(queryOptions); } @Override 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 33206e500..4231e649c 100644 --- a/fe/src/main/java/org/apache/impala/planner/PlanFragment.java +++ b/fe/src/main/java/org/apache/impala/planner/PlanFragment.java @@ -144,9 +144,11 @@ public class PlanFragment extends TreeNode<PlanFragment> { // processing cost of this fragment. Set in computeCostingSegment(). private CostingSegment rootSegment_; - // Maximum allowed parallelism based on minimum processing load per fragment. + // Maximum allowed parallelism of this fragment before caping it against + // MAX_FRAGMENT_INSTANCES_PER_NODE query option and Analyzer.getAvailableCoresPerNode(). + // It is displayed in explain string of query profile to assist with query tuning. // Set in adjustToMaxParallelism(). - private int costBasedMaxParallelism_ = -1; + private int maxParallelism_ = -1; // An adjusted number of instance based on ProcessingCost calculation. // A positive value implies that the instance count has been adjusted, either through @@ -536,6 +538,15 @@ public class PlanFragment extends TreeNode<PlanFragment> { } else if (sink_ instanceof HdfsTableSink) { return ((HdfsTableSink)sink_).getNumInstances(); } else { + if (originalInstanceCount_ > -1) { + Preconditions.checkState(hasAdjustedInstanceCount()); + Preconditions.checkState( + planRoot_.getNumInstances() == getAdjustedInstanceCount(), + "Instance count of " + getId() + " with plan root " + + planRoot_.getDisplayLabel() + " (" + planRoot_.getNumInstances() + + ") does not follow cost based adjustment (" + getAdjustedInstanceCount() + + ")!"); + } return planRoot_.getNumInstances(); } } @@ -661,7 +672,7 @@ public class PlanFragment extends TreeNode<PlanFragment> { */ public String getFragmentHeaderString(String firstLinePrefix, String detailPrefix, TQueryOptions queryOptions, TExplainLevel explainLevel) { - boolean isComputeCost = ProcessingCost.isComputeCost(queryOptions); + boolean isComputeCost = queryOptions.isCompute_processing_cost(); boolean useMTFragment = Planner.useMTFragment(queryOptions); StringBuilder builder = new StringBuilder(); builder.append(String.format("%s%s:PLAN FRAGMENT [%s]", firstLinePrefix, @@ -729,8 +740,8 @@ public class PlanFragment extends TreeNode<PlanFragment> { // Print processing cost. builder.append(detailPrefix); builder.append("max-parallelism="); - if (costBasedMaxParallelism_ > 0) { - builder.append(costBasedMaxParallelism_); + if (maxParallelism_ > 0) { + builder.append(maxParallelism_); } else { builder.append(getAdjustedInstanceCount()); } @@ -1044,111 +1055,120 @@ public class PlanFragment extends TreeNode<PlanFragment> { int maxThreadAllowed = IntMath.saturatedMultiply(maxThreadPerNode, getNumNodes()); boolean canTryLower = true; - // Compute maximum allowed parallelism. - int maxParallelism = getNumInstances(); + // Compute selectedParallelism as the maximum allowed parallelism. + int selectedParallelism = getNumInstances(); if (isFixedParallelism_) { - maxParallelism = getAdjustedInstanceCount(); + selectedParallelism = getAdjustedInstanceCount(); canTryLower = false; } else if ((sink_ instanceof JoinBuildSink) && !((JoinBuildSink) sink_).isShared()) { // This is a non-shared (PARTITIONED) join build fragment. // Parallelism of this fragment is equal to its parent parallelism. Preconditions.checkState(parentParallelism > 0); - if (LOG.isTraceEnabled() && maxParallelism != parentParallelism) { - logCountAdjustmentTrace(maxParallelism, parentParallelism, + if (LOG.isTraceEnabled() && selectedParallelism != parentParallelism) { + logCountAdjustmentTrace(selectedParallelism, parentParallelism, "Partitioned join build fragment follow parent's parallelism."); } - maxParallelism = parentParallelism; + selectedParallelism = parentParallelism; canTryLower = false; // no need to compute effective parallelism anymore. } else { + int costBasedMaxParallelism = Math.max(nodeStepCount, getCostBasedMaxParallelism()); + if (hasUnionNode()) { // We set parallelism of union fragment as a max between its input fragments and // its collocated ScanNode's expected parallelism // (see Scheduler::CreateCollocatedAndScanInstances()). // We skip any join builder child fragment here because their parallelism // is not adjusted yet. - maxParallelism = 1; + maxParallelism_ = 1; for (PlanFragment child : getChildren()) { if (child.getSink() instanceof JoinBuildSink) continue; Preconditions.checkState(child.hasAdjustedInstanceCount()); - maxParallelism = Math.max(maxParallelism, child.getAdjustedInstanceCount()); + maxParallelism_ = Math.max(maxParallelism_, child.getAdjustedInstanceCount()); } List<ScanNode> scanNodes = Lists.newArrayList(); collectPlanNodes(Predicates.instanceOf(ScanNode.class), scanNodes); - for (ScanNode scanNode : scanNodes) { - // Increase maxParallelism following ScanNode with largest effective scan range - // count. - maxParallelism = - Math.max(maxParallelism, scanNode.getMaxScannerThreads(nodeStepCount)); + if (!scanNodes.isEmpty()) { + // The existence of scan node may justify an increase of parallelism for this + // union fargment, but it should be capped at costBasedMaxParallelism. + long maxRangesPerScanNode = 1; + for (ScanNode scanNode : scanNodes) { + maxRangesPerScanNode = + Math.max(maxRangesPerScanNode, scanNode.getEffectiveNumScanRanges()); + } + maxParallelism_ = Math.max(maxParallelism_, + (int) Math.min(costBasedMaxParallelism, maxRangesPerScanNode)); + } + + if (maxParallelism_ > maxThreadAllowed) { + selectedParallelism = maxThreadAllowed; + if (LOG.isTraceEnabled()) { + logCountAdjustmentTrace( + getNumInstances(), selectedParallelism, "Follow maxThreadPerNode."); + } + } else { + selectedParallelism = maxParallelism_; + if (LOG.isTraceEnabled()) { + logCountAdjustmentTrace(getNumInstances(), selectedParallelism, + "Follow minimum work per thread."); + } } canTryLower = false; } else { // This is an interior fragment or fragment with single scan node. - // We calculate maxParallelism, minParallelism, and costBasedMaxParallelism across + // We calculate maxParallelism_, minParallelism, and selectedParallelism across // all executors in the selected executor group. - maxParallelism = maxThreadAllowed; + maxParallelism_ = costBasedMaxParallelism; - // Bound maxParallelism by ScanNode's effective scan range count if this fragment - // has ScanNode. + // Bound maxParallelism_ by ScanNode's effective scan range count if + // this fragment has ScanNode. List<ScanNode> scanNodes = Lists.newArrayList(); collectPlanNodes(Predicates.instanceOf(ScanNode.class), scanNodes); if (!scanNodes.isEmpty()) { Preconditions.checkState(scanNodes.size() == 1); ScanNode scanNode = scanNodes.get(0); - int maxScannerThreads = scanNode.getMaxScannerThreads(nodeStepCount); - if (nodeStepCount == getNumNodes()) { - Preconditions.checkState(maxScannerThreads >= getNumNodes(), - "nodeStepCount=" + nodeStepCount + " getNumNodes=" + getNumNodes() - + " maxScannerThreads=" + maxScannerThreads); - } else { - // This fragment parallelism is limited by its scan range count. - Preconditions.checkState(nodeStepCount == 1); - Preconditions.checkState(maxScannerThreads <= getNumNodes(), - "nodeStepCount=" + nodeStepCount + " getNumNodes=" + getNumNodes() - + " maxScannerThreads=" + maxScannerThreads); - } - maxParallelism = Math.min(maxParallelism, maxScannerThreads); - // Prevent caller from lowering parallelism if fragment has ScanNode. + maxParallelism_ = + (int) Math.min(maxParallelism_, scanNode.getEffectiveNumScanRanges()); + + // Prevent caller from lowering parallelism if fragment has ScanNode + // because there is no child fragment to compare with. canTryLower = false; } int minParallelism = Math.min( - maxParallelism, IntMath.saturatedMultiply(minThreadPerNode, getNumNodes())); - int costBasedMaxParallelism_ = - Math.max(nodeStepCount, getCostBasedMaxParallelism()); + maxThreadAllowed, IntMath.saturatedMultiply(minThreadPerNode, getNumNodes())); - if (costBasedMaxParallelism_ < maxParallelism) { - if (costBasedMaxParallelism_ < minParallelism) { - maxParallelism = minParallelism; + if (maxParallelism_ > maxThreadAllowed) { + selectedParallelism = maxThreadAllowed; + if (LOG.isTraceEnabled()) { + logCountAdjustmentTrace( + getNumInstances(), selectedParallelism, "Follow maxThreadPerNode."); + } + } else { + if (maxParallelism_ < minParallelism && scanNodes.isEmpty()) { + maxParallelism_ = minParallelism; canTryLower = false; if (LOG.isTraceEnabled()) { logCountAdjustmentTrace( - getNumInstances(), maxParallelism, "Follow minThreadPerNode."); + getNumInstances(), maxParallelism_, "Follow minThreadPerNode."); } - } else { - maxParallelism = costBasedMaxParallelism_; - if (LOG.isTraceEnabled()) { - logCountAdjustmentTrace( - getNumInstances(), maxParallelism, "Follow minimum work per thread."); - } - } - } else { - if (LOG.isTraceEnabled()) { + } else if (LOG.isTraceEnabled()) { logCountAdjustmentTrace( - getNumInstances(), maxParallelism, "Follow maxThreadPerNode."); + getNumInstances(), maxParallelism_, "Follow minimum work per thread."); } + selectedParallelism = maxParallelism_; } } } - // Validate that maxParallelism does not exceed maxThreadAllowed. - // maxParallelism can be lower than minThreadPerNode, ie., in the case of plan root - // sink (only 1 per query) or scan with very few scan ranges, so this does not + // Validate that selectedParallelism does not exceed maxThreadAllowed. + // selectedParallelism can be lower than minThreadPerNode, ie., in the case of plan + // root sink (only 1 per query) or scan with very few scan ranges, so this does not // validate against minThreadPerNode. - Preconditions.checkState(maxParallelism <= maxThreadAllowed); + Preconditions.checkState(selectedParallelism <= maxThreadAllowed); - // Initialize this fragment's parallelism to the maxParallelism. - setAdjustedInstanceCount(maxParallelism); + // Initialize this fragment's parallelism to the selectedParallelism. + setAdjustedInstanceCount(selectedParallelism); return canTryLower; } 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 1e3923394..4fa1209c9 100644 --- a/fe/src/main/java/org/apache/impala/planner/PlanNode.java +++ b/fe/src/main/java/org/apache/impala/planner/PlanNode.java @@ -347,7 +347,7 @@ abstract public class PlanNode extends TreeNode<PlanNode> { expBuilder.append(nodeResourceProfile_.getExplainString()); expBuilder.append("\n"); - if (ProcessingCost.isComputeCost(queryOptions) && processingCost_.isValid() + if (queryOptions.isCompute_processing_cost() && processingCost_.isValid() && detailLevel.ordinal() >= TExplainLevel.VERBOSE.ordinal()) { // Print processing cost. expBuilder.append(processingCost_.getExplainString(detailPrefix, false)); @@ -373,7 +373,7 @@ abstract public class PlanNode extends TreeNode<PlanNode> { .append(PrintUtils.printBytes(Math.round(avgRowSize_))) .append(" cardinality=") .append(PrintUtils.printEstCardinality(cardinality_)); - if (ProcessingCost.isComputeCost(queryOptions)) { + if (queryOptions.isCompute_processing_cost()) { // Show processing cost total. expBuilder.append(" cost="); if (processingCost_.isValid()) { 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 0975b4b86..ac178eb60 100644 --- a/fe/src/main/java/org/apache/impala/planner/Planner.java +++ b/fe/src/main/java/org/apache/impala/planner/Planner.java @@ -491,7 +491,7 @@ public class Planner { public static void computeProcessingCost( List<PlanFragment> planRoots, TQueryExecRequest request, PlannerContext planCtx) { Analyzer rootAnalyzer = planCtx.getRootAnalyzer(); - if (!ProcessingCost.isComputeCost(rootAnalyzer.getQueryOptions())) { + if (!rootAnalyzer.getQueryOptions().isCompute_processing_cost()) { request.setCores_required(-1); return; } diff --git a/fe/src/main/java/org/apache/impala/planner/ProcessingCost.java b/fe/src/main/java/org/apache/impala/planner/ProcessingCost.java index 691a12c1b..010affcba 100644 --- a/fe/src/main/java/org/apache/impala/planner/ProcessingCost.java +++ b/fe/src/main/java/org/apache/impala/planner/ProcessingCost.java @@ -96,10 +96,6 @@ public abstract class ProcessingCost implements Cloneable { return processingCost; } - public static boolean isComputeCost(TQueryOptions queryOptions) { - return queryOptions.isCompute_processing_cost(); - } - /** * Merge multiple ProcessingCost into a single new ProcessingCost. * <p> 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 353beaa4e..7ab07bbe9 100644 --- a/fe/src/main/java/org/apache/impala/planner/ScanNode.java +++ b/fe/src/main/java/org/apache/impala/planner/ScanNode.java @@ -45,6 +45,7 @@ import org.apache.impala.util.ExprUtil; import com.google.common.base.Joiner; import com.google.common.base.MoreObjects; import com.google.common.base.Preconditions; +import com.google.common.math.IntMath; import com.google.common.math.LongMath; /** @@ -343,28 +344,25 @@ abstract public class ScanNode extends PlanNode { } /** - * Given effectiveScanRangeCount, compute processing cost of this scan node. + * Compute processing cost of this scan node. * <p> * This method does not mutate any state of the scan node object, including * the processingCost_ field. Caller must set processingCost_ themself with * the return value of this method. */ - protected ProcessingCost computeScanProcessingCost( - TQueryOptions queryOptions, long effectiveScanRangeCount) { - ProcessingCost cardinalityBasedCost = ProcessingCost.basicCost(getDisplayLabel(), - getInputCardinality(), ExprUtil.computeExprsTotalCost(conjuncts_), - rowMaterializationCost(effectiveScanRangeCount)); - - int maxScannerThreads = (int) Math.min(effectiveScanRangeCount, Integer.MAX_VALUE); - if (ProcessingCost.isComputeCost(queryOptions)) { - // maxThread calculation below intentionally does not include core count from - // executor group config. This is to allow scan fragment parallelism to scale - // regardless of the core count limit. - int maxThreads = Math.max(queryOptions.getProcessing_cost_min_threads(), - queryOptions.getMax_fragment_instances_per_node()); - maxScannerThreads = (int) Math.min( - maxScannerThreads, LongMath.saturatedMultiply(getNumNodes(), maxThreads)); - } + protected ProcessingCost computeScanProcessingCost(TQueryOptions queryOptions) { + Preconditions.checkArgument(queryOptions.isCompute_processing_cost()); + ProcessingCost cardinalityBasedCost = + ProcessingCost.basicCost(getDisplayLabel(), getInputCardinality(), + ExprUtil.computeExprsTotalCost(conjuncts_), rowMaterializationCost()); + + // maxThread calculation below intentionally does not include core count from + // executor group config. This is to allow scan fragment parallelism to scale + // regardless of the core count limit. + int maxThreadsPerNode = Math.max(queryOptions.getProcessing_cost_min_threads(), + queryOptions.getMax_fragment_instances_per_node()); + int maxScannerThreads = (int) Math.min(getEffectiveNumScanRanges(), + IntMath.saturatedMultiply(getNumNodes(), maxThreadsPerNode)); if (getInputCardinality() == 0) { Preconditions.checkState(cardinalityBasedCost.getTotalCost() == 0, @@ -391,19 +389,18 @@ abstract public class ScanNode extends PlanNode { /** * Estimate per-row cost as 1 per 1KB row size plus - * (0.5% * min_processing_per_thread) for every scan ranges. + * (scan_range_cost_factor * min_processing_per_thread) for every scan ranges. * <p> * This reflects deserialization/copy cost per row and scan range open cost. - * (0.5% * min_processing_per_thread) for every scan ranges roughly means that one scan - * node instance will handle at most 200 scan ranges. */ - private float rowMaterializationCost(long effectiveScanRangeCount) { + private float rowMaterializationCost() { float perRowCost = getAvgRowSize() / 1024; if (getInputCardinality() <= 0) return perRowCost; - float perScanRangeCost = BackendConfig.INSTANCE.getMinProcessingPerThread() * 0.005f - / getInputCardinality() * effectiveScanRangeCount; - return perRowCost + perScanRangeCost; + float scanRangeCostPerRow = BackendConfig.INSTANCE.getMinProcessingPerThread() + * BackendConfig.INSTANCE.getScanRangeCostFactor() / getInputCardinality() + * getEffectiveNumScanRanges(); + return perRowCost + scanRangeCostPerRow; } /** @@ -420,10 +417,8 @@ abstract public class ScanNode extends PlanNode { public ExprSubstitutionMap getOptimizedAggSmap() { return optimizedAggSmap_; } - /** - * Return maximum number of scanner thread, rounded up to next multiple of numNodes. - */ - protected int getMaxScannerThreads(int numNodes) { - return processingCost_.getNumInstanceMax(numNodes); + protected long getEffectiveNumScanRanges() { + Preconditions.checkNotNull(scanRangeSpecs_); + return scanRangeSpecs_.getConcrete_rangesSize(); } } 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 68c136ceb..607910568 100644 --- a/fe/src/main/java/org/apache/impala/service/BackendConfig.java +++ b/fe/src/main/java/org/apache/impala/service/BackendConfig.java @@ -407,4 +407,8 @@ public class BackendConfig { public void setFileMetadataReloadProperties(String newPropertiesConfig) { backendCfg_.file_metadata_reload_properties = newPropertiesConfig; } + + public float getScanRangeCostFactor() { + return (float) backendCfg_.scan_range_cost_factor; + } } diff --git a/fe/src/main/java/org/apache/impala/service/Frontend.java b/fe/src/main/java/org/apache/impala/service/Frontend.java index f10762c50..ea406cae5 100644 --- a/fe/src/main/java/org/apache/impala/service/Frontend.java +++ b/fe/src/main/java/org/apache/impala/service/Frontend.java @@ -148,7 +148,6 @@ import org.apache.impala.hooks.QueryEventHookManager; import org.apache.impala.planner.HdfsScanNode; import org.apache.impala.planner.PlanFragment; import org.apache.impala.planner.Planner; -import org.apache.impala.planner.ProcessingCost; import org.apache.impala.planner.ScanNode; import org.apache.impala.thrift.TAlterDbParams; import org.apache.impala.thrift.TBackendGflags; @@ -2028,7 +2027,7 @@ public class Frontend { // Capture the current state. planCtx.compilationState_.captureState(); - boolean isComputeCost = ProcessingCost.isComputeCost(queryOptions); + boolean isComputeCost = queryOptions.isCompute_processing_cost(); if (isComputeCost) { FrontendProfile.getCurrent().setToCounter(CPU_COUNT_DIVISOR, TUnit.DOUBLE_VALUE, diff --git a/testdata/workloads/functional-planner/queries/PlannerTest/tpcds-processing-cost.test b/testdata/workloads/functional-planner/queries/PlannerTest/tpcds-processing-cost.test index 934c0621f..e4ed15684 100644 --- a/testdata/workloads/functional-planner/queries/PlannerTest/tpcds-processing-cost.test +++ b/testdata/workloads/functional-planner/queries/PlannerTest/tpcds-processing-cost.test @@ -270,11 +270,11 @@ with cross_items as order by channel,i_brand_id,i_class_id,i_category_id LIMIT 100 ---- PARALLELPLANS -Max Per-Host Resource Reservation: Memory=668.62MB Threads=163 -Per-Host Resource Estimates: Memory=2.34GB +Max Per-Host Resource Reservation: Memory=698.06MB Threads=166 +Per-Host Resource Estimates: Memory=2.41GB F80:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1 | Per-Instance Resources: mem-estimate=4.03MB mem-reservation=4.00MB thread-reservation=1 -| max-parallelism=1 segment-costs=[606] cpu-comparison-result=157 [max(1 (self) vs 157 (sum children))] +| max-parallelism=1 segment-costs=[606] cpu-comparison-result=166 [max(1 (self) vs 166 (sum children))] PLAN-ROOT SINK | output exprs: CASE valid_tid(104,105,106,107,108) WHEN 104 THEN channel WHEN 105 THEN channel WHEN 106 THEN channel WHEN 107 THEN channel WHEN 108 THEN NULL END, CASE valid_tid(104,105,106,107,108) WHEN 104 THEN i_brand_id WHEN 105 THEN i_brand_id WHEN 106 THEN i_brand_id WHEN 107 THEN NULL WHEN 108 THEN NULL END, CASE valid_tid(104,105,106,107,108) WHEN 104 THEN i_class_id WHEN 105 THEN i_class_id WHEN 106 THEN NULL WHEN 107 THEN NULL WHEN 108 THEN NULL END, CASE valid_tid(104,105,10 [...] | mem-estimate=4.00MB mem-reservation=4.00MB spill-buffer=2.00MB thread-reservation=0 cost=600 @@ -288,7 +288,7 @@ PLAN-ROOT SINK | F79:PLAN FRAGMENT [HASH(CASE valid_tid(104,105,106,107,108) WHEN 104 THEN murmur_hash(channel) WHEN 105 THEN murmur_hash(channel) WHEN 106 THEN murmur_hash(channel) WHEN 107 THEN murmur_hash(channel) WHEN 108 THEN murmur_hash(NULL) END,CASE valid_tid(104,105,106,107,108) WHEN 104 THEN murmur_hash(i_brand_id) WHEN 105 THEN murmur_hash(i_brand_id) WHEN 106 THEN murmur_hash(i_brand_id) WHEN 107 THEN murmur_hash(NULL) WHEN 108 THEN murmur_hash(NULL) END,CASE valid_tid(104,105,106,107,108) WH [...] Per-Instance Resources: mem-estimate=70.05MB mem-reservation=30.44MB thread-reservation=1 -max-parallelism=6 segment-costs=[2028749, 2270779, 400, 6] cpu-comparison-result=157 [max(6 (self) vs 157 (sum children))] +max-parallelism=6 segment-costs=[2028749, 2270779, 400, 6] cpu-comparison-result=166 [max(6 (self) vs 166 (sum children))] 129:TOP-N [LIMIT=100] | order by: CASE valid_tid(104,105,106,107,108) WHEN 104 THEN channel WHEN 105 THEN channel WHEN 106 THEN channel WHEN 107 THEN channel WHEN 108 THEN NULL END ASC, CASE valid_tid(104,105,106,107,108) WHEN 104 THEN i_brand_id WHEN 105 THEN i_brand_id WHEN 106 THEN i_brand_id WHEN 107 THEN NULL WHEN 108 THEN NULL END ASC, CASE valid_tid(104,105,106,107,108) WHEN 104 THEN i_class_id WHEN 105 THEN i_class_id WHEN 106 THEN NULL WHEN 107 THEN NULL WHEN 108 THEN NULL END ASC, CASE valid_tid(10 [...] | mem-estimate=4.69KB mem-reservation=0B thread-reservation=0 @@ -329,7 +329,7 @@ max-parallelism=6 segment-costs=[2028749, 2270779, 400, 6] cpu-comparison-result | F78:PLAN FRAGMENT [RANDOM] hosts=3 instances=12 (adjusted from 48) Per-Instance Resources: mem-estimate=68.27MB mem-reservation=25.94MB thread-reservation=1 -max-parallelism=12 segment-costs=[749813, 429878, 215944, 13294224, 82367] cpu-comparison-result=157 [max(12 (self) vs 157 (sum children))] +max-parallelism=12 segment-costs=[749813, 429878, 215944, 13294224, 82367] cpu-comparison-result=166 [max(12 (self) vs 166 (sum children))] 127:AGGREGATE [STREAMING] | Class 0 | output: sum(sales), sum(number_sales) @@ -364,7 +364,7 @@ max-parallelism=12 segment-costs=[749813, 429878, 215944, 13294224, 82367] cpu-c | | | |--F89:PLAN FRAGMENT [RANDOM] hosts=3 instances=3 | | | Per-Instance Resources: mem-estimate=16.02KB mem-reservation=0B thread-reservation=1 -| | | max-parallelism=3 segment-costs=[3] cpu-comparison-result=12 [max(3 (self) vs 12 (sum children))] +| | | max-parallelism=3 segment-costs=[3] cpu-comparison-result=15 [max(3 (self) vs 15 (sum children))] | | JOIN BUILD | | | join-table-id=08 plan-id=09 cohort-id=01 | | | mem-estimate=16B mem-reservation=0B thread-reservation=0 cost=0 @@ -376,7 +376,7 @@ max-parallelism=12 segment-costs=[749813, 429878, 215944, 13294224, 82367] cpu-c | | | | | F77:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1 | | Per-Instance Resources: mem-estimate=96.00KB mem-reservation=0B thread-reservation=1 -| | max-parallelism=2 segment-costs=[2, 1] cpu-comparison-result=12 [max(2 (self) vs 12 (sum children))] +| | max-parallelism=2 segment-costs=[2, 1] cpu-comparison-result=15 [max(2 (self) vs 15 (sum children))] | | 212:AGGREGATE [FINALIZE] | | | output: avg:merge(quantity * list_price) | | | mem-estimate=16.00KB mem-reservation=0B spill-buffer=2.00MB thread-reservation=0 @@ -388,9 +388,9 @@ max-parallelism=12 segment-costs=[749813, 429878, 215944, 13294224, 82367] cpu-c | | | tuple-ids=98 row-size=16B cardinality=1 cost=1 | | | in pipelines: 125(GETNEXT) | | | -| | F76:PLAN FRAGMENT [RANDOM] hosts=3 instances=12 (adjusted from 48) +| | F76:PLAN FRAGMENT [RANDOM] hosts=3 instances=15 (adjusted from 48) | | Per-Instance Resources: mem-estimate=16.02MB mem-reservation=4.00MB thread-reservation=1 -| | max-parallelism=12 segment-costs=[131839765, 1] cpu-comparison-result=12 [max(12 (self) vs 12 (sum children))] +| | max-parallelism=15 segment-costs=[131839765, 1] cpu-comparison-result=15 [max(15 (self) vs 12 (sum children))] | | 125:AGGREGATE | | | output: avg(CAST(quantity AS DECIMAL(10,0)) * list_price) | | | mem-estimate=16.00KB mem-reservation=0B spill-buffer=2.00MB thread-reservation=0 @@ -411,13 +411,13 @@ max-parallelism=12 segment-costs=[749813, 429878, 215944, 13294224, 82367] cpu-c | | | | in pipelines: 122(GETNEXT), 123(OPEN) | | | | | | | |--F92:PLAN FRAGMENT [RANDOM] hosts=3 instances=3 -| | | | | Per-Instance Resources: mem-estimate=7.82MB mem-reservation=7.75MB thread-reservation=1 +| | | | | Per-Instance Resources: mem-estimate=9.75MB mem-reservation=9.69MB thread-reservation=1 | | | | | max-parallelism=3 segment-costs=[7563] | | | | JOIN BUILD | | | | | join-table-id=11 plan-id=12 cohort-id=04 | | | | | build expressions: d_date_sk | | | | | runtime filters: RF107[min_max] <- d_date_sk -| | | | | mem-estimate=7.75MB mem-reservation=7.75MB spill-buffer=64.00KB thread-reservation=0 cost=7305 +| | | | | mem-estimate=9.69MB mem-reservation=9.69MB spill-buffer=64.00KB thread-reservation=0 cost=7305 | | | | | | | | | 210:EXCHANGE [BROADCAST] | | | | | mem-estimate=69.07KB mem-reservation=0B thread-reservation=0 @@ -460,13 +460,13 @@ max-parallelism=12 segment-costs=[749813, 429878, 215944, 13294224, 82367] cpu-c | | | | in pipelines: 119(GETNEXT), 120(OPEN) | | | | | | | |--F91:PLAN FRAGMENT [RANDOM] hosts=3 instances=3 -| | | | | Per-Instance Resources: mem-estimate=7.82MB mem-reservation=7.75MB thread-reservation=1 +| | | | | Per-Instance Resources: mem-estimate=9.75MB mem-reservation=9.69MB thread-reservation=1 | | | | | max-parallelism=3 segment-costs=[7563] | | | | JOIN BUILD | | | | | join-table-id=10 plan-id=11 cohort-id=04 | | | | | build expressions: d_date_sk | | | | | runtime filters: RF105[min_max] <- d_date_sk -| | | | | mem-estimate=7.75MB mem-reservation=7.75MB spill-buffer=64.00KB thread-reservation=0 cost=7305 +| | | | | mem-estimate=9.69MB mem-reservation=9.69MB spill-buffer=64.00KB thread-reservation=0 cost=7305 | | | | | | | | | 209:EXCHANGE [BROADCAST] | | | | | mem-estimate=69.07KB mem-reservation=0B thread-reservation=0 @@ -509,12 +509,12 @@ max-parallelism=12 segment-costs=[749813, 429878, 215944, 13294224, 82367] cpu-c | | | in pipelines: 116(GETNEXT), 117(OPEN) | | | | | |--F90:PLAN FRAGMENT [RANDOM] hosts=3 instances=3 -| | | | Per-Instance Resources: mem-estimate=7.82MB mem-reservation=7.75MB thread-reservation=1 +| | | | Per-Instance Resources: mem-estimate=9.75MB mem-reservation=9.69MB thread-reservation=1 | | | | max-parallelism=3 segment-costs=[7563] | | | JOIN BUILD | | | | join-table-id=09 plan-id=10 cohort-id=04 | | | | build expressions: d_date_sk -| | | | mem-estimate=7.75MB mem-reservation=7.75MB spill-buffer=64.00KB thread-reservation=0 cost=7305 +| | | | mem-estimate=9.69MB mem-reservation=9.69MB spill-buffer=64.00KB thread-reservation=0 cost=7305 | | | | | | | 208:EXCHANGE [BROADCAST] | | | | mem-estimate=69.07KB mem-reservation=0B thread-reservation=0 @@ -1103,7 +1103,7 @@ max-parallelism=12 segment-costs=[749813, 429878, 215944, 13294224, 82367] cpu-c | | | |--F85:PLAN FRAGMENT [RANDOM] hosts=3 instances=3 | | | Per-Instance Resources: mem-estimate=16.02KB mem-reservation=0B thread-reservation=1 -| | | max-parallelism=3 segment-costs=[3] cpu-comparison-result=12 [max(3 (self) vs 12 (sum children))] +| | | max-parallelism=3 segment-costs=[3] cpu-comparison-result=15 [max(3 (self) vs 15 (sum children))] | | JOIN BUILD | | | join-table-id=04 plan-id=05 cohort-id=01 | | | mem-estimate=16B mem-reservation=0B thread-reservation=0 cost=0 @@ -1115,7 +1115,7 @@ max-parallelism=12 segment-costs=[749813, 429878, 215944, 13294224, 82367] cpu-c | | | | | F51:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1 | | Per-Instance Resources: mem-estimate=96.00KB mem-reservation=0B thread-reservation=1 -| | max-parallelism=2 segment-costs=[2, 1] cpu-comparison-result=12 [max(2 (self) vs 12 (sum children))] +| | max-parallelism=2 segment-costs=[2, 1] cpu-comparison-result=15 [max(2 (self) vs 15 (sum children))] | | 184:AGGREGATE [FINALIZE] | | | output: avg:merge(quantity * list_price) | | | mem-estimate=16.00KB mem-reservation=0B spill-buffer=2.00MB thread-reservation=0 @@ -1127,9 +1127,9 @@ max-parallelism=12 segment-costs=[749813, 429878, 215944, 13294224, 82367] cpu-c | | | tuple-ids=64 row-size=16B cardinality=1 cost=1 | | | in pipelines: 83(GETNEXT) | | | -| | F50:PLAN FRAGMENT [RANDOM] hosts=3 instances=12 (adjusted from 48) +| | F50:PLAN FRAGMENT [RANDOM] hosts=3 instances=15 (adjusted from 48) | | Per-Instance Resources: mem-estimate=16.02MB mem-reservation=4.00MB thread-reservation=1 -| | max-parallelism=12 segment-costs=[131839765, 1] cpu-comparison-result=12 [max(12 (self) vs 12 (sum children))] +| | max-parallelism=15 segment-costs=[131839765, 1] cpu-comparison-result=15 [max(15 (self) vs 12 (sum children))] | | 83:AGGREGATE | | | output: avg(CAST(quantity AS DECIMAL(10,0)) * list_price) | | | mem-estimate=16.00KB mem-reservation=0B spill-buffer=2.00MB thread-reservation=0 @@ -1150,13 +1150,13 @@ max-parallelism=12 segment-costs=[749813, 429878, 215944, 13294224, 82367] cpu-c | | | | in pipelines: 80(GETNEXT), 81(OPEN) | | | | | | | |--F88:PLAN FRAGMENT [RANDOM] hosts=3 instances=3 -| | | | | Per-Instance Resources: mem-estimate=7.82MB mem-reservation=7.75MB thread-reservation=1 +| | | | | Per-Instance Resources: mem-estimate=9.75MB mem-reservation=9.69MB thread-reservation=1 | | | | | max-parallelism=3 segment-costs=[7563] | | | | JOIN BUILD | | | | | join-table-id=07 plan-id=08 cohort-id=03 | | | | | build expressions: d_date_sk | | | | | runtime filters: RF071[min_max] <- d_date_sk -| | | | | mem-estimate=7.75MB mem-reservation=7.75MB spill-buffer=64.00KB thread-reservation=0 cost=7305 +| | | | | mem-estimate=9.69MB mem-reservation=9.69MB spill-buffer=64.00KB thread-reservation=0 cost=7305 | | | | | | | | | 182:EXCHANGE [BROADCAST] | | | | | mem-estimate=69.07KB mem-reservation=0B thread-reservation=0 @@ -1199,13 +1199,13 @@ max-parallelism=12 segment-costs=[749813, 429878, 215944, 13294224, 82367] cpu-c | | | | in pipelines: 77(GETNEXT), 78(OPEN) | | | | | | | |--F87:PLAN FRAGMENT [RANDOM] hosts=3 instances=3 -| | | | | Per-Instance Resources: mem-estimate=7.82MB mem-reservation=7.75MB thread-reservation=1 +| | | | | Per-Instance Resources: mem-estimate=9.75MB mem-reservation=9.69MB thread-reservation=1 | | | | | max-parallelism=3 segment-costs=[7563] | | | | JOIN BUILD | | | | | join-table-id=06 plan-id=07 cohort-id=03 | | | | | build expressions: d_date_sk | | | | | runtime filters: RF069[min_max] <- d_date_sk -| | | | | mem-estimate=7.75MB mem-reservation=7.75MB spill-buffer=64.00KB thread-reservation=0 cost=7305 +| | | | | mem-estimate=9.69MB mem-reservation=9.69MB spill-buffer=64.00KB thread-reservation=0 cost=7305 | | | | | | | | | 181:EXCHANGE [BROADCAST] | | | | | mem-estimate=69.07KB mem-reservation=0B thread-reservation=0 @@ -1248,12 +1248,12 @@ max-parallelism=12 segment-costs=[749813, 429878, 215944, 13294224, 82367] cpu-c | | | in pipelines: 74(GETNEXT), 75(OPEN) | | | | | |--F86:PLAN FRAGMENT [RANDOM] hosts=3 instances=3 -| | | | Per-Instance Resources: mem-estimate=7.82MB mem-reservation=7.75MB thread-reservation=1 +| | | | Per-Instance Resources: mem-estimate=9.75MB mem-reservation=9.69MB thread-reservation=1 | | | | max-parallelism=3 segment-costs=[7563] | | | JOIN BUILD | | | | join-table-id=05 plan-id=06 cohort-id=03 | | | | build expressions: d_date_sk -| | | | mem-estimate=7.75MB mem-reservation=7.75MB spill-buffer=64.00KB thread-reservation=0 cost=7305 +| | | | mem-estimate=9.69MB mem-reservation=9.69MB spill-buffer=64.00KB thread-reservation=0 cost=7305 | | | | | | | 180:EXCHANGE [BROADCAST] | | | | mem-estimate=69.07KB mem-reservation=0B thread-reservation=0 @@ -1842,7 +1842,7 @@ max-parallelism=12 segment-costs=[749813, 429878, 215944, 13294224, 82367] cpu-c | |--F81:PLAN FRAGMENT [RANDOM] hosts=3 instances=3 | | Per-Instance Resources: mem-estimate=16.02KB mem-reservation=0B thread-reservation=1 -| | max-parallelism=3 segment-costs=[3] cpu-comparison-result=12 [max(3 (self) vs 12 (sum children))] +| | max-parallelism=3 segment-costs=[3] cpu-comparison-result=15 [max(3 (self) vs 15 (sum children))] | JOIN BUILD | | join-table-id=00 plan-id=01 cohort-id=01 | | mem-estimate=16B mem-reservation=0B thread-reservation=0 cost=0 @@ -1854,7 +1854,7 @@ max-parallelism=12 segment-costs=[749813, 429878, 215944, 13294224, 82367] cpu-c | | | F25:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1 | Per-Instance Resources: mem-estimate=96.00KB mem-reservation=0B thread-reservation=1 -| max-parallelism=2 segment-costs=[2, 1] cpu-comparison-result=12 [max(2 (self) vs 12 (sum children))] +| max-parallelism=2 segment-costs=[2, 1] cpu-comparison-result=15 [max(2 (self) vs 15 (sum children))] | 156:AGGREGATE [FINALIZE] | | output: avg:merge(quantity * list_price) | | mem-estimate=16.00KB mem-reservation=0B spill-buffer=2.00MB thread-reservation=0 @@ -1866,9 +1866,9 @@ max-parallelism=12 segment-costs=[749813, 429878, 215944, 13294224, 82367] cpu-c | | tuple-ids=30 row-size=16B cardinality=1 cost=1 | | in pipelines: 41(GETNEXT) | | -| F24:PLAN FRAGMENT [RANDOM] hosts=3 instances=12 (adjusted from 48) +| F24:PLAN FRAGMENT [RANDOM] hosts=3 instances=15 (adjusted from 48) | Per-Instance Resources: mem-estimate=16.02MB mem-reservation=4.00MB thread-reservation=1 -| max-parallelism=12 segment-costs=[131839765, 1] cpu-comparison-result=12 [max(12 (self) vs 12 (sum children))] +| max-parallelism=15 segment-costs=[131839765, 1] cpu-comparison-result=15 [max(15 (self) vs 12 (sum children))] | 41:AGGREGATE | | output: avg(CAST(quantity AS DECIMAL(10,0)) * list_price) | | mem-estimate=16.00KB mem-reservation=0B spill-buffer=2.00MB thread-reservation=0 @@ -1889,13 +1889,13 @@ max-parallelism=12 segment-costs=[749813, 429878, 215944, 13294224, 82367] cpu-c | | | in pipelines: 38(GETNEXT), 39(OPEN) | | | | | |--F84:PLAN FRAGMENT [RANDOM] hosts=3 instances=3 -| | | | Per-Instance Resources: mem-estimate=7.82MB mem-reservation=7.75MB thread-reservation=1 +| | | | Per-Instance Resources: mem-estimate=9.75MB mem-reservation=9.69MB thread-reservation=1 | | | | max-parallelism=3 segment-costs=[7563] | | | JOIN BUILD | | | | join-table-id=03 plan-id=04 cohort-id=02 | | | | build expressions: d_date_sk | | | | runtime filters: RF035[min_max] <- d_date_sk -| | | | mem-estimate=7.75MB mem-reservation=7.75MB spill-buffer=64.00KB thread-reservation=0 cost=7305 +| | | | mem-estimate=9.69MB mem-reservation=9.69MB spill-buffer=64.00KB thread-reservation=0 cost=7305 | | | | | | | 154:EXCHANGE [BROADCAST] | | | | mem-estimate=69.07KB mem-reservation=0B thread-reservation=0 @@ -1938,13 +1938,13 @@ max-parallelism=12 segment-costs=[749813, 429878, 215944, 13294224, 82367] cpu-c | | | in pipelines: 35(GETNEXT), 36(OPEN) | | | | | |--F83:PLAN FRAGMENT [RANDOM] hosts=3 instances=3 -| | | | Per-Instance Resources: mem-estimate=7.82MB mem-reservation=7.75MB thread-reservation=1 +| | | | Per-Instance Resources: mem-estimate=9.75MB mem-reservation=9.69MB thread-reservation=1 | | | | max-parallelism=3 segment-costs=[7563] | | | JOIN BUILD | | | | join-table-id=02 plan-id=03 cohort-id=02 | | | | build expressions: d_date_sk | | | | runtime filters: RF033[min_max] <- d_date_sk -| | | | mem-estimate=7.75MB mem-reservation=7.75MB spill-buffer=64.00KB thread-reservation=0 cost=7305 +| | | | mem-estimate=9.69MB mem-reservation=9.69MB spill-buffer=64.00KB thread-reservation=0 cost=7305 | | | | | | | 153:EXCHANGE [BROADCAST] | | | | mem-estimate=69.07KB mem-reservation=0B thread-reservation=0 @@ -1987,12 +1987,12 @@ max-parallelism=12 segment-costs=[749813, 429878, 215944, 13294224, 82367] cpu-c | | in pipelines: 32(GETNEXT), 33(OPEN) | | | |--F82:PLAN FRAGMENT [RANDOM] hosts=3 instances=3 -| | | Per-Instance Resources: mem-estimate=7.82MB mem-reservation=7.75MB thread-reservation=1 +| | | Per-Instance Resources: mem-estimate=9.75MB mem-reservation=9.69MB thread-reservation=1 | | | max-parallelism=3 segment-costs=[7563] | | JOIN BUILD | | | join-table-id=01 plan-id=02 cohort-id=02 | | | build expressions: d_date_sk -| | | mem-estimate=7.75MB mem-reservation=7.75MB spill-buffer=64.00KB thread-reservation=0 cost=7305 +| | | mem-estimate=9.69MB mem-reservation=9.69MB spill-buffer=64.00KB thread-reservation=0 cost=7305 | | | | | 152:EXCHANGE [BROADCAST] | | | mem-estimate=69.07KB mem-reservation=0B thread-reservation=0 @@ -2707,11 +2707,11 @@ ORDER BY this_year.channel, this_year.i_category_id LIMIT 100 ---- PARALLELPLANS -Max Per-Host Resource Reservation: Memory=427.88MB Threads=119 -Per-Host Resource Estimates: Memory=1.54GB +Max Per-Host Resource Reservation: Memory=447.50MB Threads=121 +Per-Host Resource Estimates: Memory=1.58GB F56:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1 | Per-Instance Resources: mem-estimate=4.02MB mem-reservation=4.00MB thread-reservation=1 -| max-parallelism=1 segment-costs=[61] cpu-comparison-result=112 [max(1 (self) vs 112 (sum children))] +| max-parallelism=1 segment-costs=[61] cpu-comparison-result=118 [max(1 (self) vs 118 (sum children))] PLAN-ROOT SINK | output exprs: channel, i_brand_id, i_class_id, i_category_id, sales, number_sales, channel, i_brand_id, i_class_id, i_category_id, sales, number_sales | mem-estimate=4.00MB mem-reservation=4.00MB spill-buffer=2.00MB thread-reservation=0 cost=60 @@ -2725,7 +2725,7 @@ PLAN-ROOT SINK | F19:PLAN FRAGMENT [HASH(i_brand_id,i_class_id,i_category_id)] hosts=3 instances=6 (adjusted from 48) Per-Instance Resources: mem-estimate=10.13MB mem-reservation=1.94MB thread-reservation=1 -max-parallelism=6 segment-costs=[1366, 833, 1] cpu-comparison-result=112 [max(6 (self) vs 112 (sum children))] +max-parallelism=6 segment-costs=[1366, 833, 1] cpu-comparison-result=118 [max(6 (self) vs 118 (sum children))] 91:TOP-N [LIMIT=100] | order by: channel ASC, i_brand_id ASC, i_class_id ASC, i_category_id ASC | mem-estimate=480B mem-reservation=0B thread-reservation=0 @@ -2742,7 +2742,7 @@ max-parallelism=6 segment-costs=[1366, 833, 1] cpu-comparison-result=112 [max(6 | |--F57:PLAN FRAGMENT [HASH(i_brand_id,i_class_id,i_category_id)] hosts=3 instances=6 (adjusted from 48) | | Per-Instance Resources: mem-estimate=14.94MB mem-reservation=6.88MB thread-reservation=1 runtime-filters-memory=3.00MB -| | max-parallelism=6 segment-costs=[1366, 813] cpu-comparison-result=56 [max(6 (self) vs 56 (sum children))] +| | max-parallelism=6 segment-costs=[1366, 813] cpu-comparison-result=59 [max(6 (self) vs 59 (sum children))] | JOIN BUILD | | join-table-id=00 plan-id=01 cohort-id=01 | | build expressions: i_brand_id, i_category_id, i_class_id @@ -2758,7 +2758,7 @@ max-parallelism=6 segment-costs=[1366, 833, 1] cpu-comparison-result=112 [max(6 | | | |--F58:PLAN FRAGMENT [HASH(i_brand_id,i_class_id,i_category_id)] hosts=3 instances=3 | | | Per-Instance Resources: mem-estimate=16.02KB mem-reservation=0B thread-reservation=1 -| | | max-parallelism=3 segment-costs=[3] cpu-comparison-result=12 [max(3 (self) vs 12 (sum children))] +| | | max-parallelism=3 segment-costs=[3] cpu-comparison-result=15 [max(3 (self) vs 15 (sum children))] | | JOIN BUILD | | | join-table-id=01 plan-id=02 cohort-id=02 | | | mem-estimate=16B mem-reservation=0B thread-reservation=0 cost=0 @@ -2770,7 +2770,7 @@ max-parallelism=6 segment-costs=[1366, 833, 1] cpu-comparison-result=112 [max(6 | | | | | F55:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1 | | Per-Instance Resources: mem-estimate=96.00KB mem-reservation=0B thread-reservation=1 -| | max-parallelism=2 segment-costs=[2, 1] cpu-comparison-result=12 [max(2 (self) vs 12 (sum children))] +| | max-parallelism=2 segment-costs=[2, 1] cpu-comparison-result=15 [max(2 (self) vs 15 (sum children))] | | 150:AGGREGATE [FINALIZE] | | | output: avg:merge(quantity * list_price) | | | mem-estimate=16.00KB mem-reservation=0B spill-buffer=2.00MB thread-reservation=0 @@ -2782,9 +2782,9 @@ max-parallelism=6 segment-costs=[1366, 833, 1] cpu-comparison-result=112 [max(6 | | | tuple-ids=69 row-size=16B cardinality=1 cost=1 | | | in pipelines: 88(GETNEXT) | | | -| | F54:PLAN FRAGMENT [RANDOM] hosts=3 instances=12 (adjusted from 48) +| | F54:PLAN FRAGMENT [RANDOM] hosts=3 instances=15 (adjusted from 48) | | Per-Instance Resources: mem-estimate=16.02MB mem-reservation=4.00MB thread-reservation=1 -| | max-parallelism=12 segment-costs=[131839765, 1] cpu-comparison-result=12 [max(12 (self) vs 12 (sum children))] +| | max-parallelism=15 segment-costs=[131839765, 1] cpu-comparison-result=15 [max(15 (self) vs 12 (sum children))] | | 88:AGGREGATE | | | output: avg(CAST(quantity AS DECIMAL(10,0)) * list_price) | | | mem-estimate=16.00KB mem-reservation=0B spill-buffer=2.00MB thread-reservation=0 @@ -2805,13 +2805,13 @@ max-parallelism=6 segment-costs=[1366, 833, 1] cpu-comparison-result=112 [max(6 | | | | in pipelines: 85(GETNEXT), 86(OPEN) | | | | | | | |--F61:PLAN FRAGMENT [RANDOM] hosts=3 instances=3 -| | | | | Per-Instance Resources: mem-estimate=7.82MB mem-reservation=7.75MB thread-reservation=1 +| | | | | Per-Instance Resources: mem-estimate=9.75MB mem-reservation=9.69MB thread-reservation=1 | | | | | max-parallelism=3 segment-costs=[7563] | | | | JOIN BUILD | | | | | join-table-id=04 plan-id=05 cohort-id=03 | | | | | build expressions: d_date_sk | | | | | runtime filters: RF081[min_max] <- d_date_sk -| | | | | mem-estimate=7.75MB mem-reservation=7.75MB spill-buffer=64.00KB thread-reservation=0 cost=7305 +| | | | | mem-estimate=9.69MB mem-reservation=9.69MB spill-buffer=64.00KB thread-reservation=0 cost=7305 | | | | | | | | | 148:EXCHANGE [BROADCAST] | | | | | mem-estimate=69.07KB mem-reservation=0B thread-reservation=0 @@ -2854,13 +2854,13 @@ max-parallelism=6 segment-costs=[1366, 833, 1] cpu-comparison-result=112 [max(6 | | | | in pipelines: 82(GETNEXT), 83(OPEN) | | | | | | | |--F60:PLAN FRAGMENT [RANDOM] hosts=3 instances=3 -| | | | | Per-Instance Resources: mem-estimate=7.82MB mem-reservation=7.75MB thread-reservation=1 +| | | | | Per-Instance Resources: mem-estimate=9.75MB mem-reservation=9.69MB thread-reservation=1 | | | | | max-parallelism=3 segment-costs=[7563] | | | | JOIN BUILD | | | | | join-table-id=03 plan-id=04 cohort-id=03 | | | | | build expressions: d_date_sk | | | | | runtime filters: RF079[min_max] <- d_date_sk -| | | | | mem-estimate=7.75MB mem-reservation=7.75MB spill-buffer=64.00KB thread-reservation=0 cost=7305 +| | | | | mem-estimate=9.69MB mem-reservation=9.69MB spill-buffer=64.00KB thread-reservation=0 cost=7305 | | | | | | | | | 147:EXCHANGE [BROADCAST] | | | | | mem-estimate=69.07KB mem-reservation=0B thread-reservation=0 @@ -2903,12 +2903,12 @@ max-parallelism=6 segment-costs=[1366, 833, 1] cpu-comparison-result=112 [max(6 | | | in pipelines: 79(GETNEXT), 80(OPEN) | | | | | |--F59:PLAN FRAGMENT [RANDOM] hosts=3 instances=3 -| | | | Per-Instance Resources: mem-estimate=7.82MB mem-reservation=7.75MB thread-reservation=1 +| | | | Per-Instance Resources: mem-estimate=9.75MB mem-reservation=9.69MB thread-reservation=1 | | | | max-parallelism=3 segment-costs=[7563] | | | JOIN BUILD | | | | join-table-id=02 plan-id=03 cohort-id=03 | | | | build expressions: d_date_sk -| | | | mem-estimate=7.75MB mem-reservation=7.75MB spill-buffer=64.00KB thread-reservation=0 cost=7305 +| | | | mem-estimate=9.69MB mem-reservation=9.69MB spill-buffer=64.00KB thread-reservation=0 cost=7305 | | | | | | | 146:EXCHANGE [BROADCAST] | | | | mem-estimate=69.07KB mem-reservation=0B thread-reservation=0 @@ -3548,7 +3548,7 @@ max-parallelism=6 segment-costs=[1366, 833, 1] cpu-comparison-result=112 [max(6 | |--F75:PLAN FRAGMENT [HASH(i_brand_id,i_class_id,i_category_id)] hosts=3 instances=3 | | Per-Instance Resources: mem-estimate=16.02KB mem-reservation=0B thread-reservation=1 -| | max-parallelism=3 segment-costs=[3] cpu-comparison-result=12 [max(3 (self) vs 12 (sum children))] +| | max-parallelism=3 segment-costs=[3] cpu-comparison-result=15 [max(3 (self) vs 15 (sum children))] | JOIN BUILD | | join-table-id=18 plan-id=19 cohort-id=01 | | mem-estimate=16B mem-reservation=0B thread-reservation=0 cost=0 @@ -3560,7 +3560,7 @@ max-parallelism=6 segment-costs=[1366, 833, 1] cpu-comparison-result=112 [max(6 | | | F27:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1 | Per-Instance Resources: mem-estimate=96.00KB mem-reservation=0B thread-reservation=1 -| max-parallelism=2 segment-costs=[2, 1] cpu-comparison-result=12 [max(2 (self) vs 12 (sum children))] +| max-parallelism=2 segment-costs=[2, 1] cpu-comparison-result=15 [max(2 (self) vs 15 (sum children))] | 120:AGGREGATE [FINALIZE] | | output: avg:merge(quantity * list_price) | | mem-estimate=16.00KB mem-reservation=0B spill-buffer=2.00MB thread-reservation=0 @@ -3572,9 +3572,9 @@ max-parallelism=6 segment-costs=[1366, 833, 1] cpu-comparison-result=112 [max(6 | | tuple-ids=32 row-size=16B cardinality=1 cost=1 | | in pipelines: 43(GETNEXT) | | -| F26:PLAN FRAGMENT [RANDOM] hosts=3 instances=12 (adjusted from 48) +| F26:PLAN FRAGMENT [RANDOM] hosts=3 instances=15 (adjusted from 48) | Per-Instance Resources: mem-estimate=16.02MB mem-reservation=4.00MB thread-reservation=1 -| max-parallelism=12 segment-costs=[131839765, 1] cpu-comparison-result=12 [max(12 (self) vs 12 (sum children))] +| max-parallelism=15 segment-costs=[131839765, 1] cpu-comparison-result=15 [max(15 (self) vs 12 (sum children))] | 43:AGGREGATE | | output: avg(CAST(quantity AS DECIMAL(10,0)) * list_price) | | mem-estimate=16.00KB mem-reservation=0B spill-buffer=2.00MB thread-reservation=0 @@ -3595,13 +3595,13 @@ max-parallelism=6 segment-costs=[1366, 833, 1] cpu-comparison-result=112 [max(6 | | | in pipelines: 40(GETNEXT), 41(OPEN) | | | | | |--F78:PLAN FRAGMENT [RANDOM] hosts=3 instances=3 -| | | | Per-Instance Resources: mem-estimate=7.82MB mem-reservation=7.75MB thread-reservation=1 +| | | | Per-Instance Resources: mem-estimate=9.75MB mem-reservation=9.69MB thread-reservation=1 | | | | max-parallelism=3 segment-costs=[7563] | | | JOIN BUILD | | | | join-table-id=21 plan-id=22 cohort-id=07 | | | | build expressions: d_date_sk | | | | runtime filters: RF043[min_max] <- d_date_sk -| | | | mem-estimate=7.75MB mem-reservation=7.75MB spill-buffer=64.00KB thread-reservation=0 cost=7305 +| | | | mem-estimate=9.69MB mem-reservation=9.69MB spill-buffer=64.00KB thread-reservation=0 cost=7305 | | | | | | | 118:EXCHANGE [BROADCAST] | | | | mem-estimate=69.07KB mem-reservation=0B thread-reservation=0 @@ -3644,13 +3644,13 @@ max-parallelism=6 segment-costs=[1366, 833, 1] cpu-comparison-result=112 [max(6 | | | in pipelines: 37(GETNEXT), 38(OPEN) | | | | | |--F77:PLAN FRAGMENT [RANDOM] hosts=3 instances=3 -| | | | Per-Instance Resources: mem-estimate=7.82MB mem-reservation=7.75MB thread-reservation=1 +| | | | Per-Instance Resources: mem-estimate=9.75MB mem-reservation=9.69MB thread-reservation=1 | | | | max-parallelism=3 segment-costs=[7563] | | | JOIN BUILD | | | | join-table-id=20 plan-id=21 cohort-id=07 | | | | build expressions: d_date_sk | | | | runtime filters: RF041[min_max] <- d_date_sk -| | | | mem-estimate=7.75MB mem-reservation=7.75MB spill-buffer=64.00KB thread-reservation=0 cost=7305 +| | | | mem-estimate=9.69MB mem-reservation=9.69MB spill-buffer=64.00KB thread-reservation=0 cost=7305 | | | | | | | 117:EXCHANGE [BROADCAST] | | | | mem-estimate=69.07KB mem-reservation=0B thread-reservation=0 @@ -3693,12 +3693,12 @@ max-parallelism=6 segment-costs=[1366, 833, 1] cpu-comparison-result=112 [max(6 | | in pipelines: 34(GETNEXT), 35(OPEN) | | | |--F76:PLAN FRAGMENT [RANDOM] hosts=3 instances=3 -| | | Per-Instance Resources: mem-estimate=7.82MB mem-reservation=7.75MB thread-reservation=1 +| | | Per-Instance Resources: mem-estimate=9.75MB mem-reservation=9.69MB thread-reservation=1 | | | max-parallelism=3 segment-costs=[7563] | | JOIN BUILD | | | join-table-id=19 plan-id=20 cohort-id=07 | | | build expressions: d_date_sk -| | | mem-estimate=7.75MB mem-reservation=7.75MB spill-buffer=64.00KB thread-reservation=0 cost=7305 +| | | mem-estimate=9.69MB mem-reservation=9.69MB spill-buffer=64.00KB thread-reservation=0 cost=7305 | | | | | 116:EXCHANGE [BROADCAST] | | | mem-estimate=69.07KB mem-reservation=0B thread-reservation=0 @@ -5403,83 +5403,83 @@ ORDER BY c_last_name, sales LIMIT 100; ---- PARALLELPLANS -Max Per-Host Resource Reservation: Memory=663.12MB Threads=83 -Per-Host Resource Estimates: Memory=1.49GB -F35:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1 +Max Per-Host Resource Reservation: Memory=661.38MB Threads=89 +Per-Host Resource Estimates: Memory=1.52GB +F37:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1 | Per-Instance Resources: mem-estimate=4.03MB mem-reservation=4.00MB thread-reservation=1 | max-parallelism=1 segment-costs=[306] cpu-comparison-result=75 [max(1 (self) vs 75 (sum children))] PLAN-ROOT SINK | output exprs: c_last_name, c_first_name, sales | mem-estimate=4.00MB mem-reservation=4.00MB spill-buffer=2.00MB thread-reservation=0 cost=300 | -96:MERGING-EXCHANGE [UNPARTITIONED] +98:MERGING-EXCHANGE [UNPARTITIONED] | order by: c_last_name ASC, c_first_name ASC, sales ASC | limit: 100 | mem-estimate=34.49KB mem-reservation=0B thread-reservation=0 | tuple-ids=44 row-size=52B cardinality=100 cost=6 | in pipelines: 53(GETNEXT) | -F34:PLAN FRAGMENT [RANDOM] hosts=3 instances=6 (adjusted from 48) +F36:PLAN FRAGMENT [RANDOM] hosts=3 instances=6 (adjusted from 48) Per-Instance Resources: mem-estimate=10.51MB mem-reservation=1.94MB thread-reservation=1 max-parallelism=6 segment-costs=[32154, 29405, 300, 6] cpu-comparison-result=75 [max(6 (self) vs 75 (sum children))] 53:TOP-N [LIMIT=100] | order by: c_last_name ASC, c_first_name ASC, sales ASC | mem-estimate=5.08KB mem-reservation=0B thread-reservation=0 | tuple-ids=44 row-size=52B cardinality=100 cost=300 -| in pipelines: 53(GETNEXT), 74(OPEN), 95(OPEN) +| in pipelines: 53(GETNEXT), 75(OPEN), 97(OPEN) | 00:UNION | pass-through-operands: all | mem-estimate=0B mem-reservation=0B thread-reservation=0 | tuple-ids=42 row-size=52B cardinality=20.15K cost=0 -| in pipelines: 74(GETNEXT), 95(GETNEXT) +| in pipelines: 75(GETNEXT), 97(GETNEXT) | -|--95:AGGREGATE [FINALIZE] +|--97:AGGREGATE [FINALIZE] | | output: sum:merge(ws_quantity * ws_list_price) | | group by: c_last_name, c_first_name | | mem-estimate=10.00MB mem-reservation=1.94MB spill-buffer=64.00KB thread-reservation=0 | | tuple-ids=41 row-size=52B cardinality=9.63K cost=28878 -| | in pipelines: 95(GETNEXT), 78(OPEN) +| | in pipelines: 97(GETNEXT), 79(OPEN) | | -| 94:EXCHANGE [HASH(c_last_name,c_first_name)] +| 96:EXCHANGE [HASH(c_last_name,c_first_name)] | | mem-estimate=498.62KB mem-reservation=0B thread-reservation=0 | | tuple-ids=41 row-size=52B cardinality=9.63K cost=527 -| | in pipelines: 78(GETNEXT) +| | in pipelines: 79(GETNEXT) | | -| F20:PLAN FRAGMENT [HASH(itemdesc,i_item_sk,d_date)] hosts=3 instances=6 (adjusted from 48) -| Per-Instance Resources: mem-estimate=54.63MB mem-reservation=36.00MB thread-reservation=1 -| max-parallelism=6 segment-costs=[9542330, 264332, 527] cpu-comparison-result=36 [max(6 (self) vs 36 (sum children))] +| F34:PLAN FRAGMENT [HASH(i_item_sk)] hosts=3 instances=6 (adjusted from 48) +| Per-Instance Resources: mem-estimate=15.37MB mem-reservation=2.00MB thread-reservation=1 +| max-parallelism=6 segment-costs=[276749, 527] cpu-comparison-result=36 [max(6 (self) vs 36 (sum children))] | 52:AGGREGATE [STREAMING] | | output: sum(CAST(ws_quantity AS DECIMAL(10,0)) * ws_list_price) | | group by: c_last_name, c_first_name | | mem-estimate=10.00MB mem-reservation=2.00MB spill-buffer=64.00KB thread-reservation=0 | | tuple-ids=41 row-size=52B cardinality=9.63K cost=28878 -| | in pipelines: 78(GETNEXT) +| | in pipelines: 79(GETNEXT) | | -| 51:HASH JOIN [INNER JOIN, BROADCAST] +| 51:HASH JOIN [INNER JOIN, PARTITIONED] | | hash-table-id=10 | | hash predicates: i_item_sk = ws_item_sk | | fk/pk conjuncts: i_item_sk = ws_item_sk -| | mem-estimate=0B mem-reservation=0B spill-buffer=128.00KB thread-reservation=0 +| | mem-estimate=0B mem-reservation=0B spill-buffer=64.00KB thread-reservation=0 | | tuple-ids=28,21,23,39,22 row-size=162B cardinality=9.63K cost=235454 -| | in pipelines: 78(GETNEXT), 27(OPEN) +| | in pipelines: 79(GETNEXT), 27(OPEN) | | -| |--F46:PLAN FRAGMENT [HASH(itemdesc,i_item_sk,d_date)] hosts=3 instances=3 -| | | Per-Instance Resources: mem-estimate=8.28MB mem-reservation=6.75MB thread-reservation=1 runtime-filters-memory=1.00MB -| | | max-parallelism=3 segment-costs=[13235] cpu-comparison-result=24 [max(9 (self) vs 24 (sum children))] +| |--F48:PLAN FRAGMENT [HASH(i_item_sk)] hosts=3 instances=6 (adjusted from 48) +| | | Per-Instance Resources: mem-estimate=3.95MB mem-reservation=2.94MB thread-reservation=1 runtime-filters-memory=1.00MB +| | | max-parallelism=6 segment-costs=[10829] cpu-comparison-result=24 [max(12 (self) vs 24 (sum children))] | | JOIN BUILD | | | join-table-id=10 plan-id=11 cohort-id=05 | | | build expressions: ws_item_sk | | | runtime filters: RF018[bloom] <- ws_item_sk, RF019[min_max] <- ws_item_sk -| | | mem-estimate=5.75MB mem-reservation=5.75MB spill-buffer=128.00KB thread-reservation=0 cost=9626 +| | | mem-estimate=1.94MB mem-reservation=1.94MB spill-buffer=64.00KB thread-reservation=0 cost=9626 | | | -| | 93:EXCHANGE [BROADCAST] -| | | mem-estimate=1.53MB mem-reservation=0B thread-reservation=0 -| | | tuple-ids=21,23,39,22 row-size=112B cardinality=9.63K cost=3609 +| | 95:EXCHANGE [HASH(ws_item_sk)] +| | | mem-estimate=1.01MB mem-reservation=0B thread-reservation=0 +| | | tuple-ids=21,23,39,22 row-size=112B cardinality=9.63K cost=1203 | | | in pipelines: 27(GETNEXT) | | | -| | F32:PLAN FRAGMENT [HASH(ws_bill_customer_sk)] hosts=2 instances=4 (adjusted from 2) -| | Per-Instance Resources: mem-estimate=1018.27KB mem-reservation=0B thread-reservation=1 +| | F33:PLAN FRAGMENT [HASH(ws_bill_customer_sk)] hosts=2 instances=4 (adjusted from 2) +| | Per-Instance Resources: mem-estimate=3.49MB mem-reservation=0B thread-reservation=1 | | max-parallelism=4 segment-costs=[11619] | | 50:HASH JOIN [INNER JOIN, PARTITIONED] | | | hash-table-id=11 @@ -5489,7 +5489,7 @@ max-parallelism=6 segment-costs=[32154, 29405, 300, 6] cpu-comparison-result=75 | | | tuple-ids=21,23,39,22 row-size=112B cardinality=9.63K cost=9626 | | | in pipelines: 27(GETNEXT), 28(OPEN) | | | -| | |--F47:PLAN FRAGMENT [HASH(ws_bill_customer_sk)] hosts=2 instances=4 (adjusted from 2) +| | |--F49:PLAN FRAGMENT [HASH(ws_bill_customer_sk)] hosts=2 instances=4 (adjusted from 2) | | | | Per-Instance Resources: mem-estimate=6.73MB mem-reservation=2.88MB thread-reservation=1 | | | | max-parallelism=4 segment-costs=[104294] | | | JOIN BUILD @@ -5498,12 +5498,12 @@ max-parallelism=6 segment-costs=[32154, 29405, 300, 6] cpu-comparison-result=75 | | | | runtime filters: RF025[min_max] <- customer.c_customer_sk | | | | mem-estimate=2.88MB mem-reservation=2.88MB spill-buffer=128.00KB thread-reservation=0 cost=100000 | | | | -| | | 92:EXCHANGE [HASH(customer.c_customer_sk)] +| | | 93:EXCHANGE [HASH(customer.c_customer_sk)] | | | | mem-estimate=3.85MB mem-reservation=0B thread-reservation=0 | | | | tuple-ids=22 row-size=40B cardinality=100.00K cost=4294 | | | | in pipelines: 28(GETNEXT) | | | | -| | | F31:PLAN FRAGMENT [RANDOM] hosts=1 instances=1 +| | | F32:PLAN FRAGMENT [RANDOM] hosts=1 instances=1 | | | Per-Instance Resources: mem-estimate=16.69MB mem-reservation=1.00MB thread-reservation=1 | | | max-parallelism=1 segment-costs=[58197] | | | 28:SCAN HDFS [tpcds_parquet.customer, RANDOM] @@ -5516,12 +5516,12 @@ max-parallelism=6 segment-costs=[32154, 29405, 300, 6] cpu-comparison-result=75 | | | tuple-ids=22 row-size=40B cardinality=100.00K cost=53903 | | | in pipelines: 28(GETNEXT) | | | -| | 91:EXCHANGE [HASH(ws_bill_customer_sk)] +| | 92:EXCHANGE [HASH(ws_bill_customer_sk)] | | | mem-estimate=506.41KB mem-reservation=0B thread-reservation=0 | | | tuple-ids=21,23,39 row-size=72B cardinality=9.63K cost=790 | | | in pipelines: 27(GETNEXT) | | | -| | F21:PLAN FRAGMENT [RANDOM] hosts=2 instances=2 +| | F22:PLAN FRAGMENT [RANDOM] hosts=2 instances=2 | | Per-Host Shared Resources: mem-estimate=2.00MB mem-reservation=2.00MB thread-reservation=0 runtime-filters-memory=2.00MB | | Per-Instance Resources: mem-estimate=17.31MB mem-reservation=8.00MB thread-reservation=1 | | max-parallelism=2 segment-costs=[879889] @@ -5531,9 +5531,9 @@ max-parallelism=6 segment-costs=[32154, 29405, 300, 6] cpu-comparison-result=75 | | | fk/pk conjuncts: ws_bill_customer_sk = c_customer_sk | | | mem-estimate=0B mem-reservation=0B spill-buffer=64.00KB thread-reservation=0 | | | tuple-ids=21,23,39 row-size=72B cardinality=9.63K cost=42854 -| | | in pipelines: 27(GETNEXT), 89(OPEN) +| | | in pipelines: 27(GETNEXT), 90(OPEN) | | | -| | |--F48:PLAN FRAGMENT [RANDOM] hosts=2 instances=2 +| | |--F50:PLAN FRAGMENT [RANDOM] hosts=2 instances=2 | | | | Per-Instance Resources: mem-estimate=3.52MB mem-reservation=2.94MB thread-reservation=1 runtime-filters-memory=1.00MB | | | | max-parallelism=2 segment-costs=[10782] cpu-comparison-result=16 [max(2 (self) vs 16 (sum children))] | | | JOIN BUILD @@ -5542,28 +5542,28 @@ max-parallelism=6 segment-costs=[32154, 29405, 300, 6] cpu-comparison-result=75 | | | | runtime filters: RF026[bloom] <- c_customer_sk, RF027[min_max] <- c_customer_sk | | | | mem-estimate=1.94MB mem-reservation=1.94MB spill-buffer=64.00KB thread-reservation=0 cost=10000 | | | | -| | | 90:EXCHANGE [BROADCAST] +| | | 91:EXCHANGE [BROADCAST] | | | | mem-estimate=591.56KB mem-reservation=0B thread-reservation=0 | | | | tuple-ids=39 row-size=36B cardinality=10.00K cost=782 -| | | | in pipelines: 89(GETNEXT) +| | | | in pipelines: 90(GETNEXT) | | | | -| | | F30:PLAN FRAGMENT [HASH(c_customer_sk)] hosts=3 instances=6 (adjusted from 48) +| | | F31:PLAN FRAGMENT [HASH(c_customer_sk)] hosts=3 instances=6 (adjusted from 48) | | | Per-Instance Resources: mem-estimate=11.61MB mem-reservation=1.94MB thread-reservation=1 | | | max-parallelism=6 segment-costs=[303907, 391] cpu-comparison-result=16 [max(6 (self) vs 16 (sum children))] -| | | 89:AGGREGATE [FINALIZE] +| | | 90:AGGREGATE [FINALIZE] | | | | output: sum:merge(ss_quantity * ss_sales_price), max:merge(tpcds_cmax) | | | | group by: c_customer_sk | | | | having: sum(ss_quantity * ss_sales_price) > CAST(0.500000 AS DECIMAL(10,6)) * max(tpcds_cmax) | | | | mem-estimate=10.00MB mem-reservation=1.94MB spill-buffer=64.00KB thread-reservation=0 | | | | tuple-ids=39 row-size=36B cardinality=10.00K cost=300000 -| | | | in pipelines: 89(GETNEXT), 36(OPEN) +| | | | in pipelines: 90(GETNEXT), 36(OPEN) | | | | -| | | 88:EXCHANGE [HASH(c_customer_sk)] +| | | 89:EXCHANGE [HASH(c_customer_sk)] | | | | mem-estimate=1.61MB mem-reservation=0B thread-reservation=0 | | | | tuple-ids=39 row-size=36B cardinality=100.00K cost=3907 | | | | in pipelines: 36(GETNEXT) | | | | -| | | F23:PLAN FRAGMENT [RANDOM] hosts=3 instances=12 (adjusted from 48) +| | | F24:PLAN FRAGMENT [RANDOM] hosts=3 instances=12 (adjusted from 48) | | | Per-Instance Resources: mem-estimate=26.94MB mem-reservation=10.00MB thread-reservation=1 | | | max-parallelism=12 segment-costs=[102755368, 3907] cpu-comparison-result=16 [max(12 (self) vs 16 (sum children))] | | | 47:AGGREGATE [STREAMING] @@ -5577,56 +5577,56 @@ max-parallelism=6 segment-costs=[32154, 29405, 300, 6] cpu-comparison-result=75 | | | | join table id: 13 | | | | mem-estimate=0B mem-reservation=0B thread-reservation=0 | | | | tuple-ids=30,31,37 row-size=32B cardinality=2.88M cost=0 -| | | | in pipelines: 36(GETNEXT), 86(OPEN) +| | | | in pipelines: 36(GETNEXT), 87(OPEN) | | | | -| | | |--F49:PLAN FRAGMENT [RANDOM] hosts=3 instances=3 +| | | |--F51:PLAN FRAGMENT [RANDOM] hosts=3 instances=3 | | | | | Per-Instance Resources: mem-estimate=16.02KB mem-reservation=0B thread-reservation=1 | | | | | max-parallelism=3 segment-costs=[3] cpu-comparison-result=12 [max(3 (self) vs 12 (sum children))] | | | | JOIN BUILD | | | | | join-table-id=13 plan-id=14 cohort-id=07 | | | | | mem-estimate=16B mem-reservation=0B thread-reservation=0 cost=0 | | | | | -| | | | 87:EXCHANGE [BROADCAST] +| | | | 88:EXCHANGE [BROADCAST] | | | | | mem-estimate=16.00KB mem-reservation=0B thread-reservation=0 | | | | | tuple-ids=37 row-size=16B cardinality=1 cost=3 -| | | | | in pipelines: 86(GETNEXT) +| | | | | in pipelines: 87(GETNEXT) | | | | | -| | | | F29:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1 +| | | | F30:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1 | | | | Per-Instance Resources: mem-estimate=96.00KB mem-reservation=0B thread-reservation=1 | | | | max-parallelism=2 segment-costs=[2, 1] cpu-comparison-result=12 [max(2 (self) vs 12 (sum children))] -| | | | 86:AGGREGATE [FINALIZE] +| | | | 87:AGGREGATE [FINALIZE] | | | | | output: max:merge(csales) | | | | | mem-estimate=16.00KB mem-reservation=0B spill-buffer=2.00MB thread-reservation=0 | | | | | tuple-ids=37 row-size=16B cardinality=1 cost=1 -| | | | | in pipelines: 86(GETNEXT), 44(OPEN) +| | | | | in pipelines: 87(GETNEXT), 44(OPEN) | | | | | -| | | | 85:EXCHANGE [UNPARTITIONED] +| | | | 86:EXCHANGE [UNPARTITIONED] | | | | | mem-estimate=16.00KB mem-reservation=0B thread-reservation=0 | | | | | tuple-ids=37 row-size=16B cardinality=1 cost=1 | | | | | in pipelines: 44(GETNEXT) | | | | | -| | | | F28:PLAN FRAGMENT [HASH(c_customer_sk)] hosts=3 instances=6 (adjusted from 48) +| | | | F29:PLAN FRAGMENT [HASH(c_customer_sk)] hosts=3 instances=6 (adjusted from 48) | | | | Per-Instance Resources: mem-estimate=10.92MB mem-reservation=1.94MB thread-reservation=1 | | | | max-parallelism=6 segment-costs=[202344, 100000, 1] cpu-comparison-result=12 [max(6 (self) vs 12 (sum children))] | | | | 44:AGGREGATE | | | | | output: max(sum(ss_quantity * ss_sales_price)) | | | | | mem-estimate=16.00KB mem-reservation=0B spill-buffer=2.00MB thread-reservation=0 | | | | | tuple-ids=37 row-size=16B cardinality=1 cost=100000 -| | | | | in pipelines: 44(GETNEXT), 84(OPEN) +| | | | | in pipelines: 44(GETNEXT), 85(OPEN) | | | | | -| | | | 84:AGGREGATE [FINALIZE] +| | | | 85:AGGREGATE [FINALIZE] | | | | | output: sum:merge(ss_quantity * ss_sales_price) | | | | | group by: c_customer_sk | | | | | mem-estimate=10.00MB mem-reservation=1.94MB spill-buffer=64.00KB thread-reservation=0 | | | | | tuple-ids=35 row-size=20B cardinality=100.00K cost=200000 -| | | | | in pipelines: 84(GETNEXT), 38(OPEN) +| | | | | in pipelines: 85(GETNEXT), 38(OPEN) | | | | | -| | | | 83:EXCHANGE [HASH(c_customer_sk)] +| | | | 84:EXCHANGE [HASH(c_customer_sk)] | | | | | mem-estimate=939.04KB mem-reservation=0B thread-reservation=0 | | | | | tuple-ids=35 row-size=20B cardinality=100.00K cost=2344 | | | | | in pipelines: 38(GETNEXT) | | | | | -| | | | F25:PLAN FRAGMENT [RANDOM] hosts=3 instances=12 (adjusted from 48) +| | | | F26:PLAN FRAGMENT [RANDOM] hosts=3 instances=12 (adjusted from 48) | | | | Per-Host Shared Resources: mem-estimate=1.00MB mem-reservation=1.00MB thread-reservation=0 runtime-filters-memory=1.00MB | | | | Per-Instance Resources: mem-estimate=26.56MB mem-reservation=6.00MB thread-reservation=1 | | | | max-parallelism=12 segment-costs=[101189027, 2344] cpu-comparison-result=12 [max(12 (self) vs 8 (sum children))] @@ -5645,7 +5645,7 @@ max-parallelism=6 segment-costs=[32154, 29405, 300, 6] cpu-comparison-result=75 | | | | | tuple-ids=32,34,33 row-size=28B cardinality=2.35M cost=2354541 | | | | | in pipelines: 38(GETNEXT), 39(OPEN) | | | | | -| | | | |--F50:PLAN FRAGMENT [RANDOM] hosts=3 instances=3 +| | | | |--F52:PLAN FRAGMENT [RANDOM] hosts=3 instances=3 | | | | | | Per-Instance Resources: mem-estimate=19.39MB mem-reservation=19.00MB thread-reservation=1 | | | | | | max-parallelism=3 segment-costs=[102346] | | | | | JOIN BUILD @@ -5654,12 +5654,12 @@ max-parallelism=6 segment-costs=[32154, 29405, 300, 6] cpu-comparison-result=75 | | | | | | runtime filters: RF033[min_max] <- c_customer_sk | | | | | | mem-estimate=19.00MB mem-reservation=19.00MB spill-buffer=256.00KB thread-reservation=0 cost=100000 | | | | | | -| | | | | 82:EXCHANGE [BROADCAST] +| | | | | 83:EXCHANGE [BROADCAST] | | | | | | mem-estimate=398.62KB mem-reservation=0B thread-reservation=0 | | | | | | tuple-ids=33 row-size=4B cardinality=100.00K cost=2346 | | | | | | in pipelines: 39(GETNEXT) | | | | | | -| | | | | F27:PLAN FRAGMENT [RANDOM] hosts=1 instances=1 +| | | | | F28:PLAN FRAGMENT [RANDOM] hosts=1 instances=1 | | | | | Per-Instance Resources: mem-estimate=16.03MB mem-reservation=512.00KB thread-reservation=1 | | | | | max-parallelism=1 segment-costs=[51173] | | | | | 39:SCAN HDFS [tpcds_parquet.customer, RANDOM] @@ -5680,7 +5680,7 @@ max-parallelism=6 segment-costs=[32154, 29405, 300, 6] cpu-comparison-result=75 | | | | | tuple-ids=32,34 row-size=24B cardinality=2.35M cost=2880404 | | | | | in pipelines: 38(GETNEXT), 40(OPEN) | | | | | -| | | | |--F51:PLAN FRAGMENT [RANDOM] hosts=3 instances=3 +| | | | |--F53:PLAN FRAGMENT [RANDOM] hosts=3 instances=3 | | | | | | Per-Instance Resources: mem-estimate=8.77MB mem-reservation=8.75MB thread-reservation=1 runtime-filters-memory=1.00MB | | | | | | max-parallelism=3 segment-costs=[1545] | | | | | JOIN BUILD @@ -5689,12 +5689,12 @@ max-parallelism=6 segment-costs=[32154, 29405, 300, 6] cpu-comparison-result=75 | | | | | | runtime filters: RF034[bloom] <- d_date_sk | | | | | | mem-estimate=7.75MB mem-reservation=7.75MB spill-buffer=64.00KB thread-reservation=0 cost=1491 | | | | | | -| | | | | 81:EXCHANGE [BROADCAST] +| | | | | 82:EXCHANGE [BROADCAST] | | | | | | mem-estimate=23.65KB mem-reservation=0B thread-reservation=0 | | | | | | tuple-ids=34 row-size=8B cardinality=1.49K cost=54 | | | | | | in pipelines: 40(GETNEXT) | | | | | | -| | | | | F26:PLAN FRAGMENT [RANDOM] hosts=1 instances=1 +| | | | | F27:PLAN FRAGMENT [RANDOM] hosts=1 instances=1 | | | | | Per-Instance Resources: mem-estimate=16.05MB mem-reservation=512.00KB thread-reservation=1 | | | | | max-parallelism=1 segment-costs=[123638] | | | | | 40:SCAN HDFS [tpcds_parquet.date_dim, RANDOM] @@ -5730,7 +5730,7 @@ max-parallelism=6 segment-costs=[32154, 29405, 300, 6] cpu-comparison-result=75 | | | | tuple-ids=30,31 row-size=16B cardinality=2.88M cost=2880404 | | | | in pipelines: 36(GETNEXT), 37(OPEN) | | | | -| | | |--F52:PLAN FRAGMENT [RANDOM] hosts=3 instances=3 +| | | |--F54:PLAN FRAGMENT [RANDOM] hosts=3 instances=3 | | | | | Per-Instance Resources: mem-estimate=19.39MB mem-reservation=19.00MB thread-reservation=1 | | | | | max-parallelism=3 segment-costs=[102346] | | | | JOIN BUILD @@ -5739,12 +5739,12 @@ max-parallelism=6 segment-costs=[32154, 29405, 300, 6] cpu-comparison-result=75 | | | | | runtime filters: RF031[min_max] <- c_customer_sk | | | | | mem-estimate=19.00MB mem-reservation=19.00MB spill-buffer=256.00KB thread-reservation=0 cost=100000 | | | | | -| | | | 80:EXCHANGE [BROADCAST] +| | | | 81:EXCHANGE [BROADCAST] | | | | | mem-estimate=398.62KB mem-reservation=0B thread-reservation=0 | | | | | tuple-ids=31 row-size=4B cardinality=100.00K cost=2346 | | | | | in pipelines: 37(GETNEXT) | | | | | -| | | | F24:PLAN FRAGMENT [RANDOM] hosts=1 instances=1 +| | | | F25:PLAN FRAGMENT [RANDOM] hosts=1 instances=1 | | | | Per-Instance Resources: mem-estimate=16.03MB mem-reservation=512.00KB thread-reservation=1 | | | | max-parallelism=1 segment-costs=[51173] | | | | 37:SCAN HDFS [tpcds_parquet.customer, RANDOM] @@ -5778,7 +5778,7 @@ max-parallelism=6 segment-costs=[32154, 29405, 300, 6] cpu-comparison-result=75 | | | tuple-ids=21,23 row-size=36B cardinality=42.85K cost=719384 | | | in pipelines: 27(GETNEXT), 29(OPEN) | | | -| | |--F53:PLAN FRAGMENT [RANDOM] hosts=2 instances=2 +| | |--F55:PLAN FRAGMENT [RANDOM] hosts=2 instances=2 | | | | Per-Instance Resources: mem-estimate=2.95MB mem-reservation=2.94MB thread-reservation=1 runtime-filters-memory=1.00MB | | | | max-parallelism=2 segment-costs=[112] | | | JOIN BUILD @@ -5787,12 +5787,12 @@ max-parallelism=6 segment-costs=[32154, 29405, 300, 6] cpu-comparison-result=75 | | | | runtime filters: RF028[bloom] <- d_date_sk, RF029[min_max] <- d_date_sk | | | | mem-estimate=1.94MB mem-reservation=1.94MB spill-buffer=64.00KB thread-reservation=0 cost=108 | | | | -| | | 79:EXCHANGE [BROADCAST] +| | | 80:EXCHANGE [BROADCAST] | | | | mem-estimate=16.00KB mem-reservation=0B thread-reservation=0 | | | | tuple-ids=23 row-size=12B cardinality=108 cost=4 | | | | in pipelines: 29(GETNEXT) | | | | -| | | F22:PLAN FRAGMENT [RANDOM] hosts=1 instances=1 +| | | F23:PLAN FRAGMENT [RANDOM] hosts=1 instances=1 | | | Per-Instance Resources: mem-estimate=16.06MB mem-reservation=512.00KB thread-reservation=1 | | | max-parallelism=1 segment-costs=[196957] | | | 29:SCAN HDFS [tpcds_parquet.date_dim, RANDOM] @@ -5819,20 +5819,28 @@ max-parallelism=6 segment-costs=[32154, 29405, 300, 6] cpu-comparison-result=75 | | tuple-ids=21 row-size=24B cardinality=719.38K cost=116861 | | in pipelines: 27(GETNEXT) | | -| 78:AGGREGATE [FINALIZE] +| 94:EXCHANGE [HASH(i_item_sk)] +| | mem-estimate=4.06MB mem-reservation=0B thread-reservation=0 +| | tuple-ids=28 row-size=50B cardinality=235.45K cost=12417 +| | in pipelines: 79(GETNEXT) +| | +| F21:PLAN FRAGMENT [HASH(itemdesc,i_item_sk,d_date)] hosts=3 instances=6 (adjusted from 48) +| Per-Instance Resources: mem-estimate=44.63MB mem-reservation=34.00MB thread-reservation=1 +| max-parallelism=6 segment-costs=[9542330, 12417] cpu-comparison-result=12 [max(6 (self) vs 12 (sum children))] +| 79:AGGREGATE [FINALIZE] | | output: count:merge(*) | | group by: itemdesc, i_item_sk, d_date | | having: count(*) > CAST(4 AS BIGINT) | | mem-estimate=34.00MB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0 | | tuple-ids=28 row-size=50B cardinality=235.45K cost=9418164 -| | in pipelines: 78(GETNEXT), 30(OPEN) +| | in pipelines: 79(GETNEXT), 30(OPEN) | | -| 77:EXCHANGE [HASH(itemdesc,i_item_sk,d_date)] +| 78:EXCHANGE [HASH(itemdesc,i_item_sk,d_date)] | | mem-estimate=10.63MB mem-reservation=0B thread-reservation=0 | | tuple-ids=28 row-size=50B cardinality=2.35M cost=124166 | | in pipelines: 30(GETNEXT) | | -| F17:PLAN FRAGMENT [RANDOM] hosts=3 instances=12 (adjusted from 48) +| F18:PLAN FRAGMENT [RANDOM] hosts=3 instances=12 (adjusted from 48) | Per-Host Shared Resources: mem-estimate=2.00MB mem-reservation=2.00MB thread-reservation=0 runtime-filters-memory=2.00MB | Per-Instance Resources: mem-estimate=34.27MB mem-reservation=17.50MB thread-reservation=1 | max-parallelism=12 segment-costs=[105886861, 124166] cpu-comparison-result=12 [max(12 (self) vs 8 (sum children))] @@ -5851,7 +5859,7 @@ max-parallelism=6 segment-costs=[32154, 29405, 300, 6] cpu-comparison-result=75 | | tuple-ids=24,25,26 row-size=162B cardinality=2.35M cost=2354541 | | in pipelines: 30(GETNEXT), 32(OPEN) | | -| |--F54:PLAN FRAGMENT [RANDOM] hosts=3 instances=3 +| |--F56:PLAN FRAGMENT [RANDOM] hosts=3 instances=3 | | | Per-Instance Resources: mem-estimate=21.19MB mem-reservation=19.00MB thread-reservation=1 | | | max-parallelism=3 segment-costs=[24564] | | JOIN BUILD @@ -5860,12 +5868,12 @@ max-parallelism=6 segment-costs=[32154, 29405, 300, 6] cpu-comparison-result=75 | | | runtime filters: RF021[min_max] <- tpcds_parquet.item.i_item_sk | | | mem-estimate=19.00MB mem-reservation=19.00MB spill-buffer=256.00KB thread-reservation=0 cost=18000 | | | -| | 76:EXCHANGE [BROADCAST] +| | 77:EXCHANGE [BROADCAST] | | | mem-estimate=2.19MB mem-reservation=0B thread-reservation=0 | | | tuple-ids=26 row-size=120B cardinality=18.00K cost=6564 | | | in pipelines: 32(GETNEXT) | | | -| | F19:PLAN FRAGMENT [RANDOM] hosts=1 instances=1 +| | F20:PLAN FRAGMENT [RANDOM] hosts=1 instances=1 | | Per-Host Shared Resources: mem-estimate=1.00MB mem-reservation=1.00MB thread-reservation=0 runtime-filters-memory=1.00MB | | Per-Instance Resources: mem-estimate=16.49MB mem-reservation=2.00MB thread-reservation=1 | | max-parallelism=1 segment-costs=[54306] @@ -5888,7 +5896,7 @@ max-parallelism=6 segment-costs=[32154, 29405, 300, 6] cpu-comparison-result=75 | | tuple-ids=24,25 row-size=42B cardinality=2.35M cost=2880404 | | in pipelines: 30(GETNEXT), 31(OPEN) | | -| |--F55:PLAN FRAGMENT [RANDOM] hosts=3 instances=3 +| |--F57:PLAN FRAGMENT [RANDOM] hosts=3 instances=3 | | | Per-Instance Resources: mem-estimate=8.83MB mem-reservation=8.75MB thread-reservation=1 runtime-filters-memory=1.00MB | | | max-parallelism=3 segment-costs=[1641] | | JOIN BUILD @@ -5897,12 +5905,12 @@ max-parallelism=6 segment-costs=[32154, 29405, 300, 6] cpu-comparison-result=75 | | | runtime filters: RF022[bloom] <- d_date_sk | | | mem-estimate=7.75MB mem-reservation=7.75MB spill-buffer=64.00KB thread-reservation=0 cost=1491 | | | -| | 75:EXCHANGE [BROADCAST] +| | 76:EXCHANGE [BROADCAST] | | | mem-estimate=77.68KB mem-reservation=0B thread-reservation=0 | | | tuple-ids=25 row-size=30B cardinality=1.49K cost=150 | | | in pipelines: 31(GETNEXT) | | | -| | F18:PLAN FRAGMENT [RANDOM] hosts=1 instances=1 +| | F19:PLAN FRAGMENT [RANDOM] hosts=1 instances=1 | | Per-Instance Resources: mem-estimate=16.13MB mem-reservation=2.00MB thread-reservation=1 | | max-parallelism=1 segment-costs=[125240] | | 31:SCAN HDFS [tpcds_parquet.date_dim, RANDOM] @@ -5930,21 +5938,21 @@ max-parallelism=6 segment-costs=[32154, 29405, 300, 6] cpu-comparison-result=75 | tuple-ids=24 row-size=12B cardinality=2.88M cost=91233752 | in pipelines: 30(GETNEXT) | -74:AGGREGATE [FINALIZE] +75:AGGREGATE [FINALIZE] | output: sum:merge(cs_quantity * cs_list_price) | group by: c_last_name, c_first_name | mem-estimate=10.00MB mem-reservation=1.94MB spill-buffer=64.00KB thread-reservation=0 | tuple-ids=20 row-size=52B cardinality=10.53K cost=31578 -| in pipelines: 74(GETNEXT), 57(OPEN) +| in pipelines: 75(GETNEXT), 57(OPEN) | -73:EXCHANGE [HASH(c_last_name,c_first_name)] +74:EXCHANGE [HASH(c_last_name,c_first_name)] | mem-estimate=513.84KB mem-reservation=0B thread-reservation=0 | tuple-ids=20 row-size=52B cardinality=10.53K cost=576 | in pipelines: 57(GETNEXT) | -F03:PLAN FRAGMENT [HASH(itemdesc,i_item_sk,d_date)] hosts=3 instances=6 (adjusted from 48) -Per-Instance Resources: mem-estimate=54.63MB mem-reservation=36.00MB thread-reservation=1 -max-parallelism=6 segment-costs=[9542330, 267032, 576] cpu-comparison-result=39 [max(6 (self) vs 39 (sum children))] +F16:PLAN FRAGMENT [HASH(i_item_sk)] hosts=3 instances=6 (adjusted from 48) +Per-Instance Resources: mem-estimate=15.37MB mem-reservation=2.00MB thread-reservation=1 +max-parallelism=6 segment-costs=[279449, 576] cpu-comparison-result=39 [max(6 (self) vs 39 (sum children))] 26:AGGREGATE [STREAMING] | output: sum(CAST(cs_quantity AS DECIMAL(10,0)) * cs_list_price) | group by: c_last_name, c_first_name @@ -5952,30 +5960,30 @@ max-parallelism=6 segment-costs=[9542330, 267032, 576] cpu-comparison-result=39 | tuple-ids=20 row-size=52B cardinality=10.53K cost=31578 | in pipelines: 57(GETNEXT) | -25:HASH JOIN [INNER JOIN, BROADCAST] +25:HASH JOIN [INNER JOIN, PARTITIONED] | hash-table-id=00 | hash predicates: i_item_sk = cs_item_sk | fk/pk conjuncts: i_item_sk = cs_item_sk -| mem-estimate=0B mem-reservation=0B spill-buffer=128.00KB thread-reservation=0 +| mem-estimate=0B mem-reservation=0B spill-buffer=64.00KB thread-reservation=0 | tuple-ids=7,0,2,18,1 row-size=162B cardinality=10.53K cost=235454 | in pipelines: 57(GETNEXT), 01(OPEN) | -|--F36:PLAN FRAGMENT [HASH(itemdesc,i_item_sk,d_date)] hosts=3 instances=3 -| | Per-Instance Resources: mem-estimate=8.62MB mem-reservation=6.75MB thread-reservation=1 runtime-filters-memory=1.00MB -| | max-parallelism=3 segment-costs=[14474] cpu-comparison-result=27 [max(12 (self) vs 27 (sum children))] +|--F38:PLAN FRAGMENT [HASH(i_item_sk)] hosts=3 instances=6 (adjusted from 48) +| | Per-Instance Resources: mem-estimate=4.06MB mem-reservation=2.94MB thread-reservation=1 runtime-filters-memory=1.00MB +| | max-parallelism=6 segment-costs=[11842] cpu-comparison-result=27 [max(15 (self) vs 27 (sum children))] | JOIN BUILD | | join-table-id=00 plan-id=01 cohort-id=01 | | build expressions: cs_item_sk | | runtime filters: RF000[bloom] <- cs_item_sk, RF001[min_max] <- cs_item_sk -| | mem-estimate=5.75MB mem-reservation=5.75MB spill-buffer=128.00KB thread-reservation=0 cost=10526 +| | mem-estimate=1.94MB mem-reservation=1.94MB spill-buffer=64.00KB thread-reservation=0 cost=10526 | | -| 72:EXCHANGE [BROADCAST] -| | mem-estimate=1.87MB mem-reservation=0B thread-reservation=0 -| | tuple-ids=0,2,18,1 row-size=112B cardinality=10.53K cost=3948 +| 73:EXCHANGE [HASH(cs_item_sk)] +| | mem-estimate=1.12MB mem-reservation=0B thread-reservation=0 +| | tuple-ids=0,2,18,1 row-size=112B cardinality=10.53K cost=1316 | | in pipelines: 01(GETNEXT) | | | F15:PLAN FRAGMENT [HASH(cs_bill_customer_sk)] hosts=3 instances=6 (adjusted from 3) -| Per-Instance Resources: mem-estimate=1010.56KB mem-reservation=0B thread-reservation=1 +| Per-Instance Resources: mem-estimate=3.49MB mem-reservation=0B thread-reservation=1 | max-parallelism=6 segment-costs=[12706] | 24:HASH JOIN [INNER JOIN, PARTITIONED] | | hash-table-id=01 @@ -5985,7 +5993,7 @@ max-parallelism=6 segment-costs=[9542330, 267032, 576] cpu-comparison-result=39 | | tuple-ids=0,2,18,1 row-size=112B cardinality=10.53K cost=10526 | | in pipelines: 01(GETNEXT), 02(OPEN) | | -| |--F37:PLAN FRAGMENT [HASH(cs_bill_customer_sk)] hosts=3 instances=6 (adjusted from 3) +| |--F39:PLAN FRAGMENT [HASH(cs_bill_customer_sk)] hosts=3 instances=6 (adjusted from 3) | | | Per-Instance Resources: mem-estimate=6.73MB mem-reservation=2.88MB thread-reservation=1 | | | max-parallelism=6 segment-costs=[104294] | | JOIN BUILD @@ -6029,7 +6037,7 @@ max-parallelism=6 segment-costs=[9542330, 267032, 576] cpu-comparison-result=39 | | tuple-ids=0,2,18 row-size=72B cardinality=10.53K cost=85309 | | in pipelines: 01(GETNEXT), 68(OPEN) | | -| |--F38:PLAN FRAGMENT [RANDOM] hosts=3 instances=3 +| |--F40:PLAN FRAGMENT [RANDOM] hosts=3 instances=3 | | | Per-Instance Resources: mem-estimate=3.52MB mem-reservation=2.94MB thread-reservation=1 runtime-filters-memory=1.00MB | | | max-parallelism=3 segment-costs=[11173] cpu-comparison-result=16 [max(3 (self) vs 16 (sum children))] | | JOIN BUILD @@ -6075,7 +6083,7 @@ max-parallelism=6 segment-costs=[9542330, 267032, 576] cpu-comparison-result=39 | | | tuple-ids=9,10,16 row-size=32B cardinality=2.88M cost=0 | | | in pipelines: 10(GETNEXT), 65(OPEN) | | | -| | |--F39:PLAN FRAGMENT [RANDOM] hosts=3 instances=3 +| | |--F41:PLAN FRAGMENT [RANDOM] hosts=3 instances=3 | | | | Per-Instance Resources: mem-estimate=16.02KB mem-reservation=0B thread-reservation=1 | | | | max-parallelism=3 segment-costs=[3] cpu-comparison-result=12 [max(3 (self) vs 12 (sum children))] | | | JOIN BUILD @@ -6141,7 +6149,7 @@ max-parallelism=6 segment-costs=[9542330, 267032, 576] cpu-comparison-result=39 | | | | tuple-ids=11,13,12 row-size=28B cardinality=2.35M cost=2354541 | | | | in pipelines: 12(GETNEXT), 13(OPEN) | | | | -| | | |--F40:PLAN FRAGMENT [RANDOM] hosts=3 instances=3 +| | | |--F42:PLAN FRAGMENT [RANDOM] hosts=3 instances=3 | | | | | Per-Instance Resources: mem-estimate=19.39MB mem-reservation=19.00MB thread-reservation=1 | | | | | max-parallelism=3 segment-costs=[102346] | | | | JOIN BUILD @@ -6176,7 +6184,7 @@ max-parallelism=6 segment-costs=[9542330, 267032, 576] cpu-comparison-result=39 | | | | tuple-ids=11,13 row-size=24B cardinality=2.35M cost=2880404 | | | | in pipelines: 12(GETNEXT), 14(OPEN) | | | | -| | | |--F41:PLAN FRAGMENT [RANDOM] hosts=3 instances=3 +| | | |--F43:PLAN FRAGMENT [RANDOM] hosts=3 instances=3 | | | | | Per-Instance Resources: mem-estimate=8.77MB mem-reservation=8.75MB thread-reservation=1 runtime-filters-memory=1.00MB | | | | | max-parallelism=3 segment-costs=[1545] | | | | JOIN BUILD @@ -6226,7 +6234,7 @@ max-parallelism=6 segment-costs=[9542330, 267032, 576] cpu-comparison-result=39 | | | tuple-ids=9,10 row-size=16B cardinality=2.88M cost=2880404 | | | in pipelines: 10(GETNEXT), 11(OPEN) | | | -| | |--F42:PLAN FRAGMENT [RANDOM] hosts=3 instances=3 +| | |--F44:PLAN FRAGMENT [RANDOM] hosts=3 instances=3 | | | | Per-Instance Resources: mem-estimate=19.39MB mem-reservation=19.00MB thread-reservation=1 | | | | max-parallelism=3 segment-costs=[102346] | | | JOIN BUILD @@ -6274,7 +6282,7 @@ max-parallelism=6 segment-costs=[9542330, 267032, 576] cpu-comparison-result=39 | | tuple-ids=0,2 row-size=36B cardinality=85.31K cost=1441548 | | in pipelines: 01(GETNEXT), 03(OPEN) | | -| |--F43:PLAN FRAGMENT [RANDOM] hosts=3 instances=3 +| |--F45:PLAN FRAGMENT [RANDOM] hosts=3 instances=3 | | | Per-Instance Resources: mem-estimate=2.95MB mem-reservation=2.94MB thread-reservation=1 runtime-filters-memory=1.00MB | | | max-parallelism=3 segment-costs=[114] | | JOIN BUILD @@ -6315,6 +6323,14 @@ max-parallelism=6 segment-costs=[9542330, 267032, 576] cpu-comparison-result=39 | tuple-ids=0 row-size=24B cardinality=1.44M cost=183787 | in pipelines: 01(GETNEXT) | +72:EXCHANGE [HASH(i_item_sk)] +| mem-estimate=4.06MB mem-reservation=0B thread-reservation=0 +| tuple-ids=7 row-size=50B cardinality=235.45K cost=12417 +| in pipelines: 57(GETNEXT) +| +F03:PLAN FRAGMENT [HASH(itemdesc,i_item_sk,d_date)] hosts=3 instances=6 (adjusted from 48) +Per-Instance Resources: mem-estimate=44.63MB mem-reservation=34.00MB thread-reservation=1 +max-parallelism=6 segment-costs=[9542330, 12417] cpu-comparison-result=12 [max(6 (self) vs 12 (sum children))] 57:AGGREGATE [FINALIZE] | output: count:merge(*) | group by: itemdesc, i_item_sk, d_date @@ -6347,7 +6363,7 @@ max-parallelism=12 segment-costs=[105886861, 124166] cpu-comparison-result=12 [m | tuple-ids=3,4,5 row-size=162B cardinality=2.35M cost=2354541 | in pipelines: 04(GETNEXT), 06(OPEN) | -|--F44:PLAN FRAGMENT [RANDOM] hosts=3 instances=3 +|--F46:PLAN FRAGMENT [RANDOM] hosts=3 instances=3 | | Per-Instance Resources: mem-estimate=21.19MB mem-reservation=19.00MB thread-reservation=1 | | max-parallelism=3 segment-costs=[24564] | JOIN BUILD @@ -6384,7 +6400,7 @@ max-parallelism=12 segment-costs=[105886861, 124166] cpu-comparison-result=12 [m | tuple-ids=3,4 row-size=42B cardinality=2.35M cost=2880404 | in pipelines: 04(GETNEXT), 05(OPEN) | -|--F45:PLAN FRAGMENT [RANDOM] hosts=3 instances=3 +|--F47:PLAN FRAGMENT [RANDOM] hosts=3 instances=3 | | Per-Instance Resources: mem-estimate=8.83MB mem-reservation=8.75MB thread-reservation=1 runtime-filters-memory=1.00MB | | max-parallelism=3 segment-costs=[1641] | JOIN BUILD @@ -6530,11 +6546,11 @@ ORDER BY 1, LIMIT 100; ---- PARALLELPLANS -Max Per-Host Resource Reservation: Memory=168.06MB Threads=25 -Per-Host Resource Estimates: Memory=416MB -F15:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1 +Max Per-Host Resource Reservation: Memory=152.56MB Threads=28 +Per-Host Resource Estimates: Memory=393MB +F16:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1 | Per-Instance Resources: mem-estimate=24.84MB mem-reservation=15.94MB thread-reservation=1 -| max-parallelism=1 segment-costs=[8891, 17696, 35376, 119504, 400, 500] cpu-comparison-result=26 [max(1 (self) vs 26 (sum children))] +| max-parallelism=1 segment-costs=[8891, 17696, 35376, 119504, 400, 500] cpu-comparison-result=32 [max(1 (self) vs 32 (sum children))] PLAN-ROOT SINK | output exprs: channel, item, return_ratio, return_rank, currency_rank | mem-estimate=4.00MB mem-reservation=4.00MB spill-buffer=2.00MB thread-reservation=0 cost=500 @@ -6585,38 +6601,37 @@ PLAN-ROOT SINK | | tuple-ids=56,53 row-size=80B cardinality=17.05K cost=17055 | | in pipelines: 29(GETNEXT) | | -| 50:MERGING-EXCHANGE [UNPARTITIONED] +| 51:MERGING-EXCHANGE [UNPARTITIONED] | | order by: (CAST(sum(coalesce(sr.sr_return_quantity, 0)) AS DECIMAL(15,4)) / CAST(sum(coalesce(sts.ss_quantity, 0)) AS DECIMAL(15,4))) ASC | | mem-estimate=855.73KB mem-reservation=0B thread-reservation=0 | | tuple-ids=56 row-size=72B cardinality=17.05K cost=1266 | | in pipelines: 29(GETNEXT) | | -| F13:PLAN FRAGMENT [HASH(sts.ss_item_sk)] hosts=3 instances=6 (adjusted from 48) +| F14:PLAN FRAGMENT [HASH(sts.ss_item_sk)] hosts=3 instances=6 (adjusted from 48) | Per-Instance Resources: mem-estimate=16.00MB mem-reservation=7.94MB thread-reservation=1 -| max-parallelism=6 segment-costs=[86275, 17055, 1266] cpu-comparison-result=12 [max(6 (self) vs 12 (sum children))] +| max-parallelism=6 segment-costs=[86275, 17055, 1266] cpu-comparison-result=18 [max(6 (self) vs 18 (sum children))] | 29:SORT | | order by: (CAST(sum(coalesce(sr.sr_return_quantity, 0)) AS DECIMAL(15,4)) / CAST(sum(coalesce(sts.ss_quantity, 0)) AS DECIMAL(15,4))) ASC | | materialized: (CAST(sum(coalesce(sr.sr_return_quantity, 0)) AS DECIMAL(15,4)) / CAST(sum(coalesce(sts.ss_quantity, 0)) AS DECIMAL(15,4))) | | mem-estimate=6.00MB mem-reservation=6.00MB spill-buffer=2.00MB thread-reservation=0 | | tuple-ids=56 row-size=72B cardinality=17.05K cost=17055 -| | in pipelines: 29(GETNEXT), 49(OPEN) +| | in pipelines: 29(GETNEXT), 50(OPEN) | | -| 49:AGGREGATE [FINALIZE] +| 50:AGGREGATE [FINALIZE] | | output: sum:merge(coalesce(sr.sr_return_quantity, 0)), sum:merge(coalesce(sts.ss_quantity, 0)), sum:merge(coalesce(sr.sr_return_amt, 0)), sum:merge(coalesce(sts.ss_net_paid, 0)) | | group by: sts.ss_item_sk | | mem-estimate=10.00MB mem-reservation=1.94MB spill-buffer=64.00KB thread-reservation=0 | | tuple-ids=19 row-size=56B cardinality=17.05K cost=85275 -| | in pipelines: 49(GETNEXT), 23(OPEN) +| | in pipelines: 50(GETNEXT), 23(OPEN) | | -| 48:EXCHANGE [HASH(sts.ss_item_sk)] -| | mem-estimate=1.01MB mem-reservation=0B thread-reservation=0 +| 49:EXCHANGE [HASH(sts.ss_item_sk)] +| | mem-estimate=670.90KB mem-reservation=0B thread-reservation=0 | | tuple-ids=19 row-size=56B cardinality=17.05K cost=1000 | | in pipelines: 23(GETNEXT) | | -| F10:PLAN FRAGMENT [RANDOM] hosts=3 instances=12 (adjusted from 48) -| Per-Host Shared Resources: mem-estimate=1.00MB mem-reservation=1.00MB thread-reservation=0 runtime-filters-memory=1.00MB -| Per-Instance Resources: mem-estimate=27.41MB mem-reservation=6.00MB thread-reservation=1 -| max-parallelism=12 segment-costs=[100880619, 1000] cpu-comparison-result=12 [max(12 (self) vs 8 (sum children))] +| F12:PLAN FRAGMENT [HASH(sts.ss_item_sk,sts.ss_ticket_number)] hosts=3 instances=6 (adjusted from 48) +| Per-Instance Resources: mem-estimate=14.76MB mem-reservation=2.00MB thread-reservation=1 +| max-parallelism=6 segment-costs=[959522, 1000] cpu-comparison-result=18 [max(18 (self) vs 11 (sum children))] | 28:AGGREGATE [STREAMING] | | output: sum(CAST(coalesce(sr.sr_return_quantity, CAST(0 AS INT)) AS BIGINT)), sum(CAST(coalesce(sts.ss_quantity, CAST(0 AS INT)) AS BIGINT)), sum(coalesce(sr.sr_return_amt, CAST(0 AS DECIMAL(7,2)))), sum(coalesce(sts.ss_net_paid, CAST(0 AS DECIMAL(7,2)))) | | group by: sts.ss_item_sk @@ -6632,21 +6647,21 @@ PLAN-ROOT SINK | | tuple-ids=16,17N,18 row-size=68B cardinality=17.05K cost=288040 | | in pipelines: 23(GETNEXT), 25(OPEN) | | -| |--F20:PLAN FRAGMENT [RANDOM] hosts=3 instances=3 -| | | Per-Instance Resources: mem-estimate=8.77MB mem-reservation=8.75MB thread-reservation=1 runtime-filters-memory=1.00MB +| |--F21:PLAN FRAGMENT [HASH(sts.ss_item_sk,sts.ss_ticket_number)] hosts=3 instances=3 +| | | Per-Instance Resources: mem-estimate=4.89MB mem-reservation=4.88MB thread-reservation=1 runtime-filters-memory=1.00MB | | | max-parallelism=3 segment-costs=[114] | | JOIN BUILD | | | join-table-id=04 plan-id=05 cohort-id=03 | | | build expressions: d_date_sk | | | runtime filters: RF004[bloom] <- d_date_sk -| | | mem-estimate=7.75MB mem-reservation=7.75MB spill-buffer=64.00KB thread-reservation=0 cost=108 +| | | mem-estimate=3.88MB mem-reservation=3.88MB spill-buffer=64.00KB thread-reservation=0 cost=108 | | | -| | 47:EXCHANGE [BROADCAST] +| | 48:EXCHANGE [BROADCAST] | | | mem-estimate=16.00KB mem-reservation=0B thread-reservation=0 | | | tuple-ids=18 row-size=12B cardinality=108 cost=6 | | | in pipelines: 25(GETNEXT) | | | -| | F12:PLAN FRAGMENT [RANDOM] hosts=1 instances=1 +| | F13:PLAN FRAGMENT [RANDOM] hosts=1 instances=1 | | Per-Instance Resources: mem-estimate=16.06MB mem-reservation=512.00KB thread-reservation=1 | | max-parallelism=1 segment-costs=[196957] | | 25:SCAN HDFS [tpcds_parquet.date_dim, RANDOM] @@ -6662,30 +6677,30 @@ PLAN-ROOT SINK | | tuple-ids=18 row-size=12B cardinality=108 cost=196955 | | in pipelines: 25(GETNEXT) | | -| 26:HASH JOIN [LEFT OUTER JOIN, BROADCAST] +| 26:HASH JOIN [LEFT OUTER JOIN, PARTITIONED] | | hash-table-id=05 | | hash predicates: sts.ss_item_sk = sr.sr_item_sk, sts.ss_ticket_number = sr.sr_ticket_number | | fk/pk conjuncts: sts.ss_item_sk = sr.sr_item_sk, sts.ss_ticket_number = sr.sr_ticket_number | | other predicates: sr.sr_return_amt > CAST(10000 AS DECIMAL(5,0)) -| | mem-estimate=0B mem-reservation=0B spill-buffer=128.00KB thread-reservation=0 +| | mem-estimate=0B mem-reservation=0B spill-buffer=64.00KB thread-reservation=0 | | tuple-ids=16,17N row-size=56B cardinality=288.04K cost=576080 | | in pipelines: 23(GETNEXT), 24(OPEN) | | -| |--F21:PLAN FRAGMENT [RANDOM] hosts=3 instances=3 -| | | Per-Instance Resources: mem-estimate=12.19MB mem-reservation=11.50MB thread-reservation=1 -| | | max-parallelism=3 segment-costs=[59863] +| |--F22:PLAN FRAGMENT [HASH(sts.ss_item_sk,sts.ss_ticket_number)] hosts=3 instances=6 (adjusted from 48) +| | | Per-Instance Resources: mem-estimate=2.62MB mem-reservation=1.94MB thread-reservation=1 +| | | max-parallelism=6 segment-costs=[58289] | | JOIN BUILD | | | join-table-id=05 plan-id=06 cohort-id=03 | | | build expressions: sr.sr_item_sk, sr.sr_ticket_number -| | | mem-estimate=11.50MB mem-reservation=11.50MB spill-buffer=128.00KB thread-reservation=0 cost=57502 +| | | mem-estimate=1.94MB mem-reservation=1.94MB spill-buffer=64.00KB thread-reservation=0 cost=57502 | | | -| | 46:EXCHANGE [BROADCAST] +| | 47:EXCHANGE [HASH(sr.sr_item_sk,sr.sr_ticket_number)] | | | mem-estimate=701.85KB mem-reservation=0B thread-reservation=0 -| | | tuple-ids=17 row-size=24B cardinality=28.75K cost=2361 +| | | tuple-ids=17 row-size=24B cardinality=28.75K cost=787 | | | in pipelines: 24(GETNEXT) | | | | | F11:PLAN FRAGMENT [RANDOM] hosts=1 instances=1 -| | Per-Instance Resources: mem-estimate=16.11MB mem-reservation=4.00MB thread-reservation=1 +| | Per-Instance Resources: mem-estimate=16.66MB mem-reservation=4.00MB thread-reservation=1 | | max-parallelism=1 segment-costs=[345040] | | 24:SCAN HDFS [tpcds_parquet.store_returns sr, RANDOM] | | HDFS partitions=1/1 files=1 size=15.43MB @@ -6700,6 +6715,15 @@ PLAN-ROOT SINK | | tuple-ids=17 row-size=24B cardinality=28.75K cost=344253 | | in pipelines: 24(GETNEXT) | | +| 46:EXCHANGE [HASH(sts.ss_item_sk,sts.ss_ticket_number)] +| | mem-estimate=3.35MB mem-reservation=0B thread-reservation=0 +| | tuple-ids=16 row-size=32B cardinality=288.04K cost=10127 +| | in pipelines: 23(GETNEXT) +| | +| F10:PLAN FRAGMENT [RANDOM] hosts=3 instances=12 (adjusted from 48) +| Per-Host Shared Resources: mem-estimate=1.00MB mem-reservation=1.00MB thread-reservation=0 runtime-filters-memory=1.00MB +| Per-Instance Resources: mem-estimate=16.84MB mem-reservation=4.00MB thread-reservation=1 +| max-parallelism=12 segment-costs=[99941351] | 23:SCAN HDFS [tpcds_parquet.store_sales sts, RANDOM] | HDFS partitions=1824/1824 files=1824 size=199.44MB | predicates: sts.ss_net_paid > CAST(0 AS DECIMAL(3,0)), sts.ss_net_profit > CAST(1 AS DECIMAL(3,0)), sts.ss_quantity > CAST(0 AS INT) @@ -6791,7 +6815,7 @@ PLAN-ROOT SINK | | tuple-ids=8,9N,10 row-size=68B cardinality=8.53K cost=144155 | | in pipelines: 12(GETNEXT), 14(OPEN) | | -| |--F18:PLAN FRAGMENT [RANDOM] hosts=3 instances=3 +| |--F19:PLAN FRAGMENT [RANDOM] hosts=3 instances=3 | | | Per-Instance Resources: mem-estimate=2.95MB mem-reservation=2.94MB thread-reservation=1 runtime-filters-memory=1.00MB | | | max-parallelism=3 segment-costs=[114] | | JOIN BUILD @@ -6830,7 +6854,7 @@ PLAN-ROOT SINK | | tuple-ids=8,9N row-size=56B cardinality=144.16K cost=288310 | | in pipelines: 12(GETNEXT), 13(OPEN) | | -| |--F19:PLAN FRAGMENT [RANDOM] hosts=3 instances=3 +| |--F20:PLAN FRAGMENT [RANDOM] hosts=3 instances=3 | | | Per-Instance Resources: mem-estimate=2.29MB mem-reservation=1.94MB thread-reservation=1 | | | max-parallelism=3 segment-costs=[29996] | | JOIN BUILD @@ -6949,7 +6973,7 @@ max-parallelism=2 segment-costs=[2517877, 252] cpu-comparison-result=6 [max(2 (s | tuple-ids=0,1N,2 row-size=68B cardinality=4.29K cost=71938 | in pipelines: 01(GETNEXT), 03(OPEN) | -|--F16:PLAN FRAGMENT [RANDOM] hosts=2 instances=2 +|--F17:PLAN FRAGMENT [RANDOM] hosts=2 instances=2 | | Per-Instance Resources: mem-estimate=2.95MB mem-reservation=2.94MB thread-reservation=1 runtime-filters-memory=1.00MB | | max-parallelism=2 segment-costs=[112] | JOIN BUILD @@ -6988,7 +7012,7 @@ max-parallelism=2 segment-costs=[2517877, 252] cpu-comparison-result=6 [max(2 (s | tuple-ids=0,1N row-size=56B cardinality=71.94K cost=143876 | in pipelines: 01(GETNEXT), 02(OPEN) | -|--F17:PLAN FRAGMENT [RANDOM] hosts=2 instances=2 +|--F18:PLAN FRAGMENT [RANDOM] hosts=2 instances=2 | | Per-Instance Resources: mem-estimate=2.13MB mem-reservation=1.94MB thread-reservation=1 | | max-parallelism=2 segment-costs=[14746] | JOIN BUILD @@ -7089,51 +7113,51 @@ ORDER BY channel, LIMIT 100; ---- PARALLELPLANS -Max Per-Host Resource Reservation: Memory=101.94MB Threads=26 -Per-Host Resource Estimates: Memory=328MB -F13:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1 +Max Per-Host Resource Reservation: Memory=73.69MB Threads=30 +Per-Host Resource Estimates: Memory=315MB +F15:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1 | Per-Instance Resources: mem-estimate=4.05MB mem-reservation=4.00MB thread-reservation=1 -| max-parallelism=1 segment-costs=[708] cpu-comparison-result=32 [max(1 (self) vs 32 (sum children))] +| max-parallelism=1 segment-costs=[708] cpu-comparison-result=33 [max(1 (self) vs 33 (sum children))] PLAN-ROOT SINK | output exprs: channel, col_name, d_year, d_qoy, i_category, count(*), sum(ext_sales_price) | mem-estimate=4.00MB mem-reservation=4.00MB spill-buffer=2.00MB thread-reservation=0 cost=700 | -28:MERGING-EXCHANGE [UNPARTITIONED] +30:MERGING-EXCHANGE [UNPARTITIONED] | order by: channel ASC, col_name ASC, d_year ASC, d_qoy ASC, i_category ASC | limit: 100 | mem-estimate=48.05KB mem-reservation=0B thread-reservation=0 | tuple-ids=12 row-size=74B cardinality=100 cost=8 | in pipelines: 17(GETNEXT) | -F12:PLAN FRAGMENT [HASH(channel,col_name,d_year,d_qoy,i_category)] hosts=3 instances=6 (adjusted from 48) -Per-Instance Resources: mem-estimate=12.57MB mem-reservation=1.94MB thread-reservation=1 -max-parallelism=6 segment-costs=[499288, 500, 8] cpu-comparison-result=32 [max(6 (self) vs 32 (sum children))] +F14:PLAN FRAGMENT [HASH(channel,col_name,d_year,d_qoy,i_category)] hosts=3 instances=6 (adjusted from 48) +Per-Instance Resources: mem-estimate=12.11MB mem-reservation=1.94MB thread-reservation=1 +max-parallelism=6 segment-costs=[499288, 500, 8] cpu-comparison-result=33 [max(6 (self) vs 33 (sum children))] 17:TOP-N [LIMIT=100] | order by: channel ASC, col_name ASC, d_year ASC, d_qoy ASC, i_category ASC | mem-estimate=7.22KB mem-reservation=0B thread-reservation=0 | tuple-ids=12 row-size=74B cardinality=100 cost=500 -| in pipelines: 17(GETNEXT), 27(OPEN) +| in pipelines: 17(GETNEXT), 29(OPEN) | -27:AGGREGATE [FINALIZE] +29:AGGREGATE [FINALIZE] | output: count:merge(*), sum:merge(ext_sales_price) | group by: channel, col_name, d_year, d_qoy, i_category | mem-estimate=10.00MB mem-reservation=1.94MB spill-buffer=64.00KB thread-reservation=0 | tuple-ids=11 row-size=74B cardinality=70.56K cost=493920 -| in pipelines: 27(GETNEXT), 01(OPEN), 07(OPEN), 11(OPEN) +| in pipelines: 29(GETNEXT), 01(OPEN), 07(OPEN), 11(OPEN) | -26:EXCHANGE [HASH(channel,col_name,d_year,d_qoy,i_category)] -| mem-estimate=2.57MB mem-reservation=0B thread-reservation=0 +28:EXCHANGE [HASH(channel,col_name,d_year,d_qoy,i_category)] +| mem-estimate=2.11MB mem-reservation=0B thread-reservation=0 | tuple-ids=11 row-size=74B cardinality=70.56K cost=5368 | in pipelines: 01(GETNEXT), 07(GETNEXT), 11(GETNEXT) | -F11:PLAN FRAGMENT [RANDOM] hosts=3 instances=12 (adjusted from 48) -Per-Host Shared Resources: mem-estimate=3.00MB mem-reservation=3.00MB thread-reservation=0 runtime-filters-memory=3.00MB -Per-Instance Resources: mem-estimate=27.83MB mem-reservation=3.00MB thread-reservation=1 -max-parallelism=12 segment-costs=[97906925, 5368] cpu-comparison-result=32 [max(21 (self) vs 32 (sum children))] +F13:PLAN FRAGMENT [RANDOM] hosts=3 instances=6 (adjusted from 48) +Per-Host Shared Resources: mem-estimate=1.00MB mem-reservation=1.00MB thread-reservation=0 runtime-filters-memory=1.00MB +Per-Instance Resources: mem-estimate=27.83MB mem-reservation=3.25MB thread-reservation=1 +max-parallelism=6 segment-costs=[3647079, 5368] cpu-comparison-result=33 [max(33 (self) vs 32 (sum children))] 16:AGGREGATE [STREAMING] | output: count(*), sum(ext_sales_price) | group by: channel, col_name, d_year, d_qoy, i_category -| mem-estimate=10.00MB mem-reservation=2.00MB spill-buffer=64.00KB thread-reservation=0 +| mem-estimate=10.00MB mem-reservation=3.00MB spill-buffer=128.00KB thread-reservation=0 | tuple-ids=11 row-size=74B cardinality=70.56K cost=961604 | in pipelines: 01(GETNEXT), 07(GETNEXT), 11(GETNEXT) | @@ -7143,29 +7167,29 @@ max-parallelism=12 segment-costs=[97906925, 5368] cpu-comparison-result=32 [max( | in pipelines: 01(GETNEXT), 07(GETNEXT), 11(GETNEXT) | |--15:HASH JOIN [INNER JOIN, PARTITIONED] -| | hash-table-id=04 +| | hash-table-id=03 | | hash predicates: cs_item_sk = i_item_sk | | fk/pk conjuncts: cs_item_sk = i_item_sk | | mem-estimate=0B mem-reservation=0B spill-buffer=64.00KB thread-reservation=0 | | tuple-ids=6,8,7 row-size=58B cardinality=7.17K cost=7165 | | in pipelines: 11(GETNEXT), 12(OPEN) | | -| |--F18:PLAN FRAGMENT [RANDOM] hosts=3 instances=12 (adjusted from 48) +| |--F19:PLAN FRAGMENT [RANDOM] hosts=3 instances=6 (adjusted from 48) | | | Per-Instance Resources: mem-estimate=3.41MB mem-reservation=2.94MB thread-reservation=1 runtime-filters-memory=1.00MB -| | | max-parallelism=12 segment-costs=[18526] +| | | max-parallelism=6 segment-costs=[18526] | | JOIN BUILD -| | | join-table-id=04 plan-id=05 cohort-id=01 +| | | join-table-id=03 plan-id=04 cohort-id=01 | | | build expressions: i_item_sk | | | runtime filters: RF008[bloom] <- i_item_sk, RF009[min_max] <- i_item_sk | | | mem-estimate=1.94MB mem-reservation=1.94MB spill-buffer=64.00KB thread-reservation=0 cost=18000 | | | -| | 25:EXCHANGE [HASH(i_item_sk)] +| | 27:EXCHANGE [HASH(i_item_sk)] | | | mem-estimate=485.07KB mem-reservation=0B thread-reservation=0 | | | tuple-ids=7 row-size=26B cardinality=18.00K cost=526 | | | in pipelines: 12(GETNEXT) | | | -| | F09:PLAN FRAGMENT [RANDOM] hosts=1 instances=1 -| | Per-Instance Resources: mem-estimate=17.40MB mem-reservation=256.00KB thread-reservation=1 +| | F11:PLAN FRAGMENT [RANDOM] hosts=1 instances=1 +| | Per-Instance Resources: mem-estimate=16.70MB mem-reservation=256.00KB thread-reservation=1 | | max-parallelism=1 segment-costs=[50982] | | 12:SCAN HDFS [tpcds_parquet.item, RANDOM] | | HDFS partitions=1/1 files=1 size=1.73MB @@ -7177,13 +7201,13 @@ max-parallelism=12 segment-costs=[97906925, 5368] cpu-comparison-result=32 [max( | | tuple-ids=7 row-size=26B cardinality=18.00K cost=50456 | | in pipelines: 12(GETNEXT) | | -| 24:EXCHANGE [HASH(cs_item_sk)] +| 26:EXCHANGE [HASH(cs_item_sk)] | | mem-estimate=314.63KB mem-reservation=0B thread-reservation=0 | | tuple-ids=6,8 row-size=32B cardinality=7.17K cost=280 | | in pipelines: 11(GETNEXT) | | -| F08:PLAN FRAGMENT [HASH(cs_sold_date_sk)] hosts=3 instances=6 (adjusted from 3) -| Per-Instance Resources: mem-estimate=1.99MB mem-reservation=0B thread-reservation=1 +| F10:PLAN FRAGMENT [HASH(cs_sold_date_sk)] hosts=3 instances=6 (adjusted from 3) +| Per-Instance Resources: mem-estimate=1.05MB mem-reservation=0B thread-reservation=1 | max-parallelism=6 segment-costs=[7613] | 14:HASH JOIN [INNER JOIN, PARTITIONED] | | hash-table-id=05 @@ -7193,7 +7217,7 @@ max-parallelism=12 segment-costs=[97906925, 5368] cpu-comparison-result=32 [max( | | tuple-ids=6,8 row-size=32B cardinality=7.17K cost=7165 | | in pipelines: 11(GETNEXT), 13(OPEN) | | -| |--F19:PLAN FRAGMENT [HASH(cs_sold_date_sk)] hosts=3 instances=6 (adjusted from 3) +| |--F21:PLAN FRAGMENT [HASH(cs_sold_date_sk)] hosts=3 instances=6 (adjusted from 3) | | | Per-Instance Resources: mem-estimate=3.79MB mem-reservation=2.94MB thread-reservation=1 runtime-filters-memory=1.00MB | | | max-parallelism=6 segment-costs=[74191] | | JOIN BUILD @@ -7202,12 +7226,12 @@ max-parallelism=12 segment-costs=[97906925, 5368] cpu-comparison-result=32 [max( | | | runtime filters: RF010[bloom] <- d_date_sk, RF011[min_max] <- d_date_sk | | | mem-estimate=1.94MB mem-reservation=1.94MB spill-buffer=64.00KB thread-reservation=0 cost=73049 | | | -| | 23:EXCHANGE [HASH(d_date_sk)] +| | 25:EXCHANGE [HASH(d_date_sk)] | | | mem-estimate=872.04KB mem-reservation=0B thread-reservation=0 | | | tuple-ids=8 row-size=12B cardinality=73.05K cost=1142 | | | in pipelines: 13(GETNEXT) | | | -| | F07:PLAN FRAGMENT [RANDOM] hosts=1 instances=1 +| | F09:PLAN FRAGMENT [RANDOM] hosts=1 instances=1 | | Per-Instance Resources: mem-estimate=16.38MB mem-reservation=512.00KB thread-reservation=1 | | max-parallelism=1 segment-costs=[51999] | | 13:SCAN HDFS [tpcds_parquet.date_dim, RANDOM] @@ -7220,12 +7244,12 @@ max-parallelism=12 segment-costs=[97906925, 5368] cpu-comparison-result=32 [max( | | tuple-ids=8 row-size=12B cardinality=73.05K cost=50857 | | in pipelines: 13(GETNEXT) | | -| 22:EXCHANGE [HASH(cs_sold_date_sk)] +| 24:EXCHANGE [HASH(cs_sold_date_sk)] | | mem-estimate=118.65KB mem-reservation=0B thread-reservation=0 | | tuple-ids=6 row-size=20B cardinality=7.17K cost=168 | | in pipelines: 11(GETNEXT) | | -| F06:PLAN FRAGMENT [RANDOM] hosts=3 instances=3 +| F08:PLAN FRAGMENT [RANDOM] hosts=3 instances=3 | Per-Host Shared Resources: mem-estimate=2.00MB mem-reservation=2.00MB thread-reservation=0 runtime-filters-memory=2.00MB | Per-Instance Resources: mem-estimate=16.56MB mem-reservation=8.00MB thread-reservation=1 | max-parallelism=3 segment-costs=[1619872] @@ -7242,54 +7266,54 @@ max-parallelism=12 segment-costs=[97906925, 5368] cpu-comparison-result=32 [max( | in pipelines: 11(GETNEXT) | |--10:HASH JOIN [INNER JOIN, BROADCAST] -| | hash-table-id=02 +| | hash-table-id=01 | | hash predicates: i_item_sk = ws_item_sk | | fk/pk conjuncts: i_item_sk = ws_item_sk | | mem-estimate=0B mem-reservation=0B spill-buffer=64.00KB thread-reservation=0 | | tuple-ids=4,5,3 row-size=58B cardinality=173 cost=18000 | | in pipelines: 07(GETNEXT), 08(OPEN) | | -| |--F16:PLAN FRAGMENT [RANDOM] hosts=3 instances=3 -| | | Per-Instance Resources: mem-estimate=8.77MB mem-reservation=8.75MB thread-reservation=1 runtime-filters-memory=1.00MB +| |--F17:PLAN FRAGMENT [RANDOM] hosts=3 instances=3 +| | | Per-Instance Resources: mem-estimate=4.89MB mem-reservation=4.88MB thread-reservation=1 runtime-filters-memory=1.00MB | | | max-parallelism=3 segment-costs=[194] cpu-comparison-result=4 [max(4 (self) vs 3 (sum children))] | | JOIN BUILD -| | | join-table-id=02 plan-id=03 cohort-id=01 +| | | join-table-id=01 plan-id=02 cohort-id=01 | | | build expressions: ws_item_sk | | | runtime filters: RF004[bloom] <- ws_item_sk, RF005[min_max] <- ws_item_sk -| | | mem-estimate=7.75MB mem-reservation=7.75MB spill-buffer=64.00KB thread-reservation=0 cost=173 +| | | mem-estimate=3.88MB mem-reservation=3.88MB spill-buffer=64.00KB thread-reservation=0 cost=173 | | | -| | 21:EXCHANGE [BROADCAST] +| | 23:EXCHANGE [BROADCAST] | | | mem-estimate=16.00KB mem-reservation=0B thread-reservation=0 | | | tuple-ids=5,3 row-size=32B cardinality=173 cost=21 | | | in pipelines: 08(GETNEXT) | | | -| | F04:PLAN FRAGMENT [RANDOM] hosts=1 instances=1 +| | F06:PLAN FRAGMENT [RANDOM] hosts=1 instances=1 | | Per-Host Shared Resources: mem-estimate=1.00MB mem-reservation=1.00MB thread-reservation=0 runtime-filters-memory=1.00MB | | Per-Instance Resources: mem-estimate=16.16MB mem-reservation=512.00KB thread-reservation=1 | | max-parallelism=1 segment-costs=[123913] | | 09:HASH JOIN [INNER JOIN, BROADCAST] -| | | hash-table-id=03 +| | | hash-table-id=02 | | | hash predicates: d_date_sk = ws_sold_date_sk | | | fk/pk conjuncts: none | | | mem-estimate=0B mem-reservation=0B spill-buffer=64.00KB thread-reservation=0 | | | tuple-ids=5,3 row-size=32B cardinality=173 cost=73049 | | | in pipelines: 08(GETNEXT), 06(OPEN) | | | -| | |--F17:PLAN FRAGMENT [RANDOM] hosts=1 instances=1 +| | |--F18:PLAN FRAGMENT [RANDOM] hosts=1 instances=1 | | | | Per-Instance Resources: mem-estimate=2.95MB mem-reservation=2.94MB thread-reservation=1 runtime-filters-memory=1.00MB | | | | max-parallelism=1 segment-costs=[178] | | | JOIN BUILD -| | | | join-table-id=03 plan-id=04 cohort-id=02 +| | | | join-table-id=02 plan-id=03 cohort-id=02 | | | | build expressions: ws_sold_date_sk | | | | runtime filters: RF006[bloom] <- ws_sold_date_sk, RF007[min_max] <- ws_sold_date_sk | | | | mem-estimate=1.94MB mem-reservation=1.94MB spill-buffer=64.00KB thread-reservation=0 cost=173 | | | | -| | | 20:EXCHANGE [BROADCAST] +| | | 22:EXCHANGE [BROADCAST] | | | | mem-estimate=16.00KB mem-reservation=0B thread-reservation=0 | | | | tuple-ids=3 row-size=20B cardinality=173 cost=5 | | | | in pipelines: 06(GETNEXT) | | | | -| | | F05:PLAN FRAGMENT [RANDOM] hosts=2 instances=2 +| | | F07:PLAN FRAGMENT [RANDOM] hosts=2 instances=2 | | | Per-Instance Resources: mem-estimate=16.09MB mem-reservation=8.00MB thread-reservation=1 | | | max-parallelism=2 segment-costs=[833440] | | | 06:SCAN HDFS [tpcds_parquet.web_sales, RANDOM] @@ -7325,30 +7349,30 @@ max-parallelism=12 segment-costs=[97906925, 5368] cpu-comparison-result=32 [max( | tuple-ids=4 row-size=26B cardinality=18.00K cost=50456 | in pipelines: 07(GETNEXT) | -05:HASH JOIN [INNER JOIN, BROADCAST] +05:HASH JOIN [INNER JOIN, PARTITIONED] | hash-table-id=00 | hash predicates: ss_sold_date_sk = d_date_sk | fk/pk conjuncts: ss_sold_date_sk = d_date_sk -| mem-estimate=0B mem-reservation=0B spill-buffer=256.00KB thread-reservation=0 +| mem-estimate=0B mem-reservation=0B spill-buffer=64.00KB thread-reservation=0 | tuple-ids=0,1,2 row-size=58B cardinality=130.03K cost=130034 | in pipelines: 01(GETNEXT), 03(OPEN) | -|--F14:PLAN FRAGMENT [RANDOM] hosts=3 instances=3 -| | Per-Instance Resources: mem-estimate=20.85MB mem-reservation=20.00MB thread-reservation=1 runtime-filters-memory=1.00MB -| | max-parallelism=3 segment-costs=[76475] +|--F16:PLAN FRAGMENT [RANDOM] hosts=3 instances=6 (adjusted from 48) +| | Per-Instance Resources: mem-estimate=3.79MB mem-reservation=2.94MB thread-reservation=1 runtime-filters-memory=1.00MB +| | max-parallelism=6 segment-costs=[74191] | JOIN BUILD | | join-table-id=00 plan-id=01 cohort-id=01 | | build expressions: d_date_sk | | runtime filters: RF000[bloom] <- d_date_sk -| | mem-estimate=19.00MB mem-reservation=19.00MB spill-buffer=256.00KB thread-reservation=0 cost=73049 +| | mem-estimate=1.94MB mem-reservation=1.94MB spill-buffer=64.00KB thread-reservation=0 cost=73049 | | -| 19:EXCHANGE [BROADCAST] +| 21:EXCHANGE [HASH(d_date_sk)] | | mem-estimate=872.04KB mem-reservation=0B thread-reservation=0 -| | tuple-ids=2 row-size=12B cardinality=73.05K cost=3426 +| | tuple-ids=2 row-size=12B cardinality=73.05K cost=1142 | | in pipelines: 03(GETNEXT) | | -| F02:PLAN FRAGMENT [RANDOM] hosts=1 instances=1 -| Per-Instance Resources: mem-estimate=16.06MB mem-reservation=512.00KB thread-reservation=1 +| F03:PLAN FRAGMENT [RANDOM] hosts=1 instances=1 +| Per-Instance Resources: mem-estimate=16.38MB mem-reservation=512.00KB thread-reservation=1 | max-parallelism=1 segment-costs=[51999] | 03:SCAN HDFS [tpcds_parquet.date_dim, RANDOM] | HDFS partitions=1/1 files=1 size=2.15MB @@ -7360,30 +7384,38 @@ max-parallelism=12 segment-costs=[97906925, 5368] cpu-comparison-result=32 [max( | tuple-ids=2 row-size=12B cardinality=73.05K cost=50857 | in pipelines: 03(GETNEXT) | -04:HASH JOIN [INNER JOIN, BROADCAST] -| hash-table-id=01 +20:EXCHANGE [HASH(ss_sold_date_sk)] +| mem-estimate=2.21MB mem-reservation=0B thread-reservation=0 +| tuple-ids=0,1 row-size=46B cardinality=130.03K cost=6844 +| in pipelines: 01(GETNEXT) +| +F02:PLAN FRAGMENT [HASH(ss_item_sk)] hosts=3 instances=6 (adjusted from 48) +Per-Instance Resources: mem-estimate=2.37MB mem-reservation=0B thread-reservation=1 +max-parallelism=6 segment-costs=[139926] +04:HASH JOIN [INNER JOIN, PARTITIONED] +| hash-table-id=04 | hash predicates: ss_item_sk = i_item_sk | fk/pk conjuncts: ss_item_sk = i_item_sk | mem-estimate=0B mem-reservation=0B spill-buffer=64.00KB thread-reservation=0 | tuple-ids=0,1 row-size=46B cardinality=130.03K cost=130034 | in pipelines: 01(GETNEXT), 02(OPEN) | -|--F15:PLAN FRAGMENT [RANDOM] hosts=3 instances=3 -| | Per-Instance Resources: mem-estimate=9.22MB mem-reservation=8.75MB thread-reservation=1 runtime-filters-memory=1.00MB -| | max-parallelism=3 segment-costs=[19578] +|--F20:PLAN FRAGMENT [HASH(ss_item_sk)] hosts=3 instances=6 (adjusted from 48) +| | Per-Instance Resources: mem-estimate=3.41MB mem-reservation=2.94MB thread-reservation=1 runtime-filters-memory=1.00MB +| | max-parallelism=6 segment-costs=[18526] | JOIN BUILD -| | join-table-id=01 plan-id=02 cohort-id=01 +| | join-table-id=04 plan-id=05 cohort-id=01 | | build expressions: i_item_sk | | runtime filters: RF002[bloom] <- i_item_sk, RF003[min_max] <- i_item_sk -| | mem-estimate=7.75MB mem-reservation=7.75MB spill-buffer=64.00KB thread-reservation=0 cost=18000 +| | mem-estimate=1.94MB mem-reservation=1.94MB spill-buffer=64.00KB thread-reservation=0 cost=18000 | | -| 18:EXCHANGE [BROADCAST] +| 19:EXCHANGE [HASH(i_item_sk)] | | mem-estimate=485.07KB mem-reservation=0B thread-reservation=0 -| | tuple-ids=1 row-size=26B cardinality=18.00K cost=1578 +| | tuple-ids=1 row-size=26B cardinality=18.00K cost=526 | | in pipelines: 02(GETNEXT) | | | F01:PLAN FRAGMENT [RANDOM] hosts=1 instances=1 -| Per-Instance Resources: mem-estimate=16.12MB mem-reservation=256.00KB thread-reservation=1 +| Per-Instance Resources: mem-estimate=16.70MB mem-reservation=256.00KB thread-reservation=1 | max-parallelism=1 segment-costs=[50982] | 02:SCAN HDFS [tpcds_parquet.item, RANDOM] | HDFS partitions=1/1 files=1 size=1.73MB @@ -7395,6 +7427,15 @@ max-parallelism=12 segment-costs=[97906925, 5368] cpu-comparison-result=32 [max( | tuple-ids=1 row-size=26B cardinality=18.00K cost=50456 | in pipelines: 02(GETNEXT) | +18:EXCHANGE [HASH(ss_item_sk)] +| mem-estimate=1.11MB mem-reservation=0B thread-reservation=0 +| tuple-ids=0 row-size=20B cardinality=130.03K cost=3048 +| in pipelines: 01(GETNEXT) +| +F00:PLAN FRAGMENT [RANDOM] hosts=3 instances=12 (adjusted from 48) +Per-Host Shared Resources: mem-estimate=2.00MB mem-reservation=2.00MB thread-reservation=0 runtime-filters-memory=2.00MB +Per-Instance Resources: mem-estimate=16.56MB mem-reservation=1.00MB thread-reservation=1 +max-parallelism=12 segment-costs=[94139704] 01:SCAN HDFS [tpcds_parquet.store_sales, RANDOM] HDFS partitions=1824/1824 files=1824 size=199.44MB predicates: ss_store_sk IS NULL @@ -8720,3 +8761,113 @@ max-parallelism=12 segment-costs=[101912983, 1] cpu-comparison-result=20 [max(12 tuple-ids=0 row-size=36B cardinality=2.88M cost=91301264 in pipelines: 01(GETNEXT) ==== +# IMPALA-12192: Test that scan fragment parallelism can scale beyond the scan node cost. +# In this query, preagg node colocated in the scan fragment will have higher cost than +# the scan node, and increase the parallelism of the scan fragment. +select + ss_net_paid_inc_tax, + ss_ext_list_price, + ss_net_profit, + ss_net_paid, + ss_ext_sales_price, + sum(ss_store_sk) +from store_sales +group by rollup ( + ss_net_paid_inc_tax, + ss_ext_list_price, + ss_net_profit, + ss_net_paid, + ss_ext_sales_price) +limit 100 +---- PARALLELPLANS +Max Per-Host Resource Reservation: Memory=1.76GB Threads=11 +Per-Host Resource Estimates: Memory=2.28GB +F02:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1 +| Per-Instance Resources: mem-estimate=4.03MB mem-reservation=4.00MB thread-reservation=1 +| max-parallelism=1 segment-costs=[604] cpu-comparison-result=21 [max(1 (self) vs 21 (sum children))] +PLAN-ROOT SINK +| output exprs: CASE valid_tid(1,2,3,4,5,6) WHEN 1 THEN ss_net_paid_inc_tax WHEN 2 THEN ss_net_paid_inc_tax WHEN 3 THEN ss_net_paid_inc_tax WHEN 4 THEN ss_net_paid_inc_tax WHEN 5 THEN ss_net_paid_inc_tax WHEN 6 THEN NULL END, CASE valid_tid(1,2,3,4,5,6) WHEN 1 THEN ss_ext_list_price WHEN 2 THEN ss_ext_list_price WHEN 3 THEN ss_ext_list_price WHEN 4 THEN ss_ext_list_price WHEN 5 THEN NULL WHEN 6 THEN NULL END, CASE valid_tid(1,2,3,4,5,6) WHEN 1 THEN ss_net_profit WHEN 2 THEN ss_net_profi [...] +| mem-estimate=4.00MB mem-reservation=4.00MB spill-buffer=2.00MB thread-reservation=0 cost=600 +| +05:EXCHANGE [UNPARTITIONED] +| limit: 100 +| mem-estimate=32.68KB mem-reservation=0B thread-reservation=0 +| tuple-ids=7 row-size=32B cardinality=100 cost=4 +| in pipelines: 02(GETNEXT) +| +F01:PLAN FRAGMENT [HASH(CASE valid_tid(1,2,3,4,5,6) WHEN 1 THEN murmur_hash(ss_net_paid_inc_tax) WHEN 2 THEN murmur_hash(ss_net_paid_inc_tax) WHEN 3 THEN murmur_hash(ss_net_paid_inc_tax) WHEN 4 THEN murmur_hash(ss_net_paid_inc_tax) WHEN 5 THEN murmur_hash(ss_net_paid_inc_tax) WHEN 6 THEN murmur_hash(NULL) END,CASE valid_tid(1,2,3,4,5,6) WHEN 1 THEN murmur_hash(ss_ext_list_price) WHEN 2 THEN murmur_hash(ss_ext_list_price) WHEN 3 THEN murmur_hash(ss_ext_list_price) WHEN 4 THEN murmur_hash( [...] +Per-Instance Resources: mem-estimate=303.35MB mem-reservation=188.94MB thread-reservation=1 +max-parallelism=9 segment-costs=[75109407, 84972256, 4] cpu-comparison-result=21 [max(9 (self) vs 21 (sum children))] +02:AGGREGATE [FINALIZE] +| output: aggif(valid_tid(1,2,3,4,5,6) IN (CAST(1 AS INT), CAST(2 AS INT), CAST(3 AS INT), CAST(4 AS INT), CAST(5 AS INT), CAST(6 AS INT)), CASE valid_tid(1,2,3,4,5,6) WHEN CAST(1 AS INT) THEN sum(ss_store_sk) WHEN CAST(2 AS INT) THEN sum(ss_store_sk) WHEN CAST(3 AS INT) THEN sum(ss_store_sk) WHEN CAST(4 AS INT) THEN sum(ss_store_sk) WHEN CAST(5 AS INT) THEN sum(ss_store_sk) WHEN CAST(6 AS INT) THEN sum(ss_store_sk) END) +| group by: CASE valid_tid(1,2,3,4,5,6) WHEN CAST(1 AS INT) THEN ss_net_paid_inc_tax WHEN CAST(2 AS INT) THEN ss_net_paid_inc_tax WHEN CAST(3 AS INT) THEN ss_net_paid_inc_tax WHEN CAST(4 AS INT) THEN ss_net_paid_inc_tax WHEN CAST(5 AS INT) THEN ss_net_paid_inc_tax WHEN CAST(6 AS INT) THEN NULL END, CASE valid_tid(1,2,3,4,5,6) WHEN CAST(1 AS INT) THEN ss_ext_list_price WHEN CAST(2 AS INT) THEN ss_ext_list_price WHEN CAST(3 AS INT) THEN ss_ext_list_price WHEN CAST(4 AS INT) THEN ss_ext_li [...] +| limit: 100 +| mem-estimate=56.60MB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0 +| tuple-ids=7 row-size=32B cardinality=100 cost=84972256 +| in pipelines: 02(GETNEXT), 04(OPEN) +| +04:AGGREGATE [FINALIZE] +| Class 0 +| output: sum:merge(ss_store_sk) +| group by: ss_net_paid_inc_tax, ss_ext_list_price, ss_net_profit, ss_net_paid, ss_ext_sales_price +| Class 1 +| output: sum:merge(ss_store_sk) +| group by: ss_net_paid_inc_tax, ss_ext_list_price, ss_net_profit, ss_net_paid, NULL +| Class 2 +| output: sum:merge(ss_store_sk) +| group by: ss_net_paid_inc_tax, ss_ext_list_price, ss_net_profit, NULL, NULL +| Class 3 +| output: sum:merge(ss_store_sk) +| group by: ss_net_paid_inc_tax, ss_ext_list_price, NULL, NULL, NULL +| Class 4 +| output: sum:merge(ss_store_sk) +| group by: ss_net_paid_inc_tax, NULL, NULL, NULL, NULL +| Class 5 +| output: sum:merge(ss_store_sk) +| group by: NULL, NULL, NULL, NULL, NULL +| mem-estimate=246.76MB mem-reservation=154.94MB thread-reservation=0 +| tuple-ids=1N,2N,3N,4N,5N,6N row-size=168B cardinality=12.14M cost=72833364 +| in pipelines: 04(GETNEXT), 00(OPEN) +| +03:EXCHANGE [HASH(CASE valid_tid(1,2,3,4,5,6) WHEN 1 THEN murmur_hash(ss_net_paid_inc_tax) WHEN 2 THEN murmur_hash(ss_net_paid_inc_tax) WHEN 3 THEN murmur_hash(ss_net_paid_inc_tax) WHEN 4 THEN murmur_hash(ss_net_paid_inc_tax) WHEN 5 THEN murmur_hash(ss_net_paid_inc_tax) WHEN 6 THEN murmur_hash(NULL) END,CASE valid_tid(1,2,3,4,5,6) WHEN 1 THEN murmur_hash(ss_ext_list_price) WHEN 2 THEN murmur_hash(ss_ext_list_price) WHEN 3 THEN murmur_hash(ss_ext_list_price) WHEN 4 THEN murmur_hash(ss_ext [...] +| mem-estimate=13.94MB mem-reservation=0B thread-reservation=0 +| tuple-ids=1N,2N,3N,4N,5N,6N row-size=168B cardinality=12.14M cost=2276043 +| in pipelines: 00(GETNEXT) +| +F00:PLAN FRAGMENT [RANDOM] hosts=3 instances=21 (adjusted from 48) +Per-Instance Resources: mem-estimate=202.75MB mem-reservation=176.00MB thread-reservation=1 +max-parallelism=21 segment-costs=[194962048, 2276043] +01:AGGREGATE [STREAMING] +| Class 0 +| output: sum(CAST(ss_store_sk AS BIGINT)) +| group by: ss_net_paid_inc_tax, ss_ext_list_price, ss_net_profit, ss_net_paid, ss_ext_sales_price +| Class 1 +| output: sum(CAST(ss_store_sk AS BIGINT)) +| group by: ss_net_paid_inc_tax, ss_ext_list_price, ss_net_profit, ss_net_paid, NULL +| Class 2 +| output: sum(CAST(ss_store_sk AS BIGINT)) +| group by: ss_net_paid_inc_tax, ss_ext_list_price, ss_net_profit, NULL, NULL +| Class 3 +| output: sum(CAST(ss_store_sk AS BIGINT)) +| group by: ss_net_paid_inc_tax, ss_ext_list_price, NULL, NULL, NULL +| Class 4 +| output: sum(CAST(ss_store_sk AS BIGINT)) +| group by: ss_net_paid_inc_tax, NULL, NULL, NULL, NULL +| Class 5 +| output: sum(CAST(ss_store_sk AS BIGINT)) +| group by: NULL, NULL, NULL, NULL, NULL +| mem-estimate=180.00MB mem-reservation=172.00MB thread-reservation=0 +| tuple-ids=1N,2N,3N,4N,5N,6N row-size=168B cardinality=12.14M cost=103694544 +| in pipelines: 00(GETNEXT) +| +00:SCAN HDFS [tpcds_parquet.store_sales, RANDOM] + HDFS partitions=1824/1824 files=1824 size=199.46MB + stored statistics: + table: rows=2.88M size=199.46MB + partitions: 1824/1824 rows=2.88M + columns: all + extrapolated-rows=disabled max-scan-range-rows=130.09K + mem-estimate=16.00MB mem-reservation=4.00MB thread-reservation=0 + tuple-ids=0 row-size=24B cardinality=2.88M cost=91267504 + in pipelines: 00(GETNEXT) +==== diff --git a/tests/custom_cluster/test_executor_groups.py b/tests/custom_cluster/test_executor_groups.py index d919ee8e0..c42c06c15 100644 --- a/tests/custom_cluster/test_executor_groups.py +++ b/tests/custom_cluster/test_executor_groups.py @@ -42,9 +42,6 @@ CPU_TEST_QUERY = "select * from tpcds_parquet.store_sales where ss_item_sk = 1 l GROUPING_TEST_QUERY = ("select ss_item_sk from tpcds_parquet.store_sales" " group by (ss_item_sk) order by ss_item_sk limit 10") -# A query to test behavior of child queries. -COMPUTE_STATS_QUERY = "COMPUTE STATS tpcds_parquet.store_sales" - DEFAULT_RESOURCE_POOL = "default-pool" @@ -936,25 +933,50 @@ class TestExecutorGroups(CustomClusterTestSuite): ["Executor Group: root.small-group", "EffectiveParallelism: 11", "ExecutorGroupsConsidered: 2"]) - # Test that REQUEST_POOL will override executor group selection - self._set_query_options({ - 'MT_DOP': '0', - 'REQUEST_POOL': 'root.large'}) - self._run_query_and_verify_profile(CPU_TEST_QUERY, - ["Executor Group: root.large-group", - ("Verdict: query option REQUEST_POOL=root.large is set. " - "Memory and cpu limit checking is skipped."), - "EffectiveParallelism: 13", "ExecutorGroupsConsidered: 1"]) + # Unset MT_DOP + self._set_query_options({'MT_DOP': '0'}) + + # Create small table based on tpcds_parquet.store_sales that will be used later + # for COMPUTE STATS test. + self._run_query_and_verify_profile( + ("create table {0}.{1} partitioned by (ss_sold_date_sk) as " + "select ss_sold_time_sk, ss_net_paid_inc_tax, ss_net_profit, ss_sold_date_sk " + "from tpcds_parquet.store_sales where ss_sold_date_sk < 2452184" + ).format(unique_database, "store_sales_subset"), + ["Executor Group: root.small", "ExecutorGroupsConsidered: 2", + "Verdict: Match", "CpuAsk: 10"]) + + compute_stats_query = ("compute stats {0}.{1}").format( + unique_database, "store_sales_subset") + + # Test that child queries unset REQUEST_POOL that was set by Frontend planner for + # parent query. One child queries should run in root.small, and another one in + # root.large. + self._verify_total_admitted_queries("root.small", 3) + self._verify_total_admitted_queries("root.large", 1) + self._run_query_and_verify_profile(compute_stats_query, + ["ExecutorGroupsConsidered: 1", + "Verdict: Assign to first group because query is not auto-scalable"], + ["Executor Group:"]) + self._verify_total_admitted_queries("root.small", 4) + self._verify_total_admitted_queries("root.large", 2) # Test that child queries follow REQUEST_POOL that was set by client. # Two child queries should all run in root.large. - self._verify_total_admitted_queries("root.large", 2) - self._run_query_and_verify_profile(COMPUTE_STATS_QUERY, + self._set_query_options({'REQUEST_POOL': 'root.large'}) + self._run_query_and_verify_profile(compute_stats_query, ["ExecutorGroupsConsidered: 1", "Verdict: Assign to first group because query is not auto-scalable"], ["Executor Group:"]) self._verify_total_admitted_queries("root.large", 4) + # Test that REQUEST_POOL will override executor group selection + self._run_query_and_verify_profile(CPU_TEST_QUERY, + ["Executor Group: root.large-group", + ("Verdict: query option REQUEST_POOL=root.large is set. " + "Memory and cpu limit checking is skipped."), + "EffectiveParallelism: 13", "ExecutorGroupsConsidered: 1"]) + # Test setting REQUEST_POOL and disabling COMPUTE_PROCESSING_COST self._set_query_options({ 'COMPUTE_PROCESSING_COST': 'false', @@ -971,18 +993,6 @@ class TestExecutorGroups(CustomClusterTestSuite): 'REQUEST_POOL': '', 'COMPUTE_PROCESSING_COST': 'true'}) - # Test that child queries unset REQUEST_POOL that was set by Frontend planner for - # parent query. One child queries should run in root.small, and another one in - # root.large. - self._verify_total_admitted_queries("root.small", 2) - self._verify_total_admitted_queries("root.large", 5) - self._run_query_and_verify_profile(COMPUTE_STATS_QUERY, - ["ExecutorGroupsConsidered: 1", - "Verdict: Assign to first group because query is not auto-scalable"], - ["Executor Group:"]) - self._verify_total_admitted_queries("root.small", 3) - self._verify_total_admitted_queries("root.large", 6) - # Test that GROUPING_TEST_QUERY will get assigned to the large group. self._run_query_and_verify_profile(GROUPING_TEST_QUERY, ["Executor Group: root.large-group", "ExecutorGroupsConsidered: 3", @@ -1153,7 +1163,7 @@ class TestExecutorGroups(CustomClusterTestSuite): self._verify_query_num_for_resource_pool("root.small", 4) self._verify_query_num_for_resource_pool("root.tiny", 4) self._verify_query_num_for_resource_pool("root.large", 12) - self._verify_total_admitted_queries("root.small", 4) + self._verify_total_admitted_queries("root.small", 5) self._verify_total_admitted_queries("root.tiny", 6) self._verify_total_admitted_queries("root.large", 16) @@ -1223,14 +1233,14 @@ class TestExecutorGroups(CustomClusterTestSuite): self._set_query_options({'COMPUTE_PROCESSING_COST': 'true'}) self._run_query_and_verify_profile(GROUPING_TEST_QUERY, ["Executor Group: root.large-group", "ExecutorGroupsConsidered: 3", - "Verdict: Match", "CpuAsk: 12"]) + "Verdict: Match", "CpuAsk: 15"]) # Test that high_scan_cost_query will get assigned to the large group. high_scan_cost_query = ("SELECT ss_item_sk FROM tpcds_parquet.store_sales " "WHERE ss_item_sk < 1000000 GROUP BY ss_item_sk LIMIT 10") self._run_query_and_verify_profile(high_scan_cost_query, ["Executor Group: root.large-group", "ExecutorGroupsConsidered: 3", - "Verdict: Match", "CpuAsk: 15"]) + "Verdict: Match", "CpuAsk: 18"]) # Test that high_scan_cost_query will get assigned to the small group # if MAX_FRAGMENT_INSTANCES_PER_NODE is limited to 1.
