You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by bo...@apache.org on 2025/03/13 21:17:38 UTC
(impala) 01/02: IMPALA-13333: Limit memory estimation if PlanNode can spill
This is an automated email from the ASF dual-hosted git repository.
boroknagyz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 7a5adc15ca0c3c374967a82d98675eacc1826916
Author: Riza Suminto <ri...@cloudera.com>
AuthorDate: Thu Sep 5 15:38:43 2024 -0700
IMPALA-13333: Limit memory estimation if PlanNode can spill
SortNode, AggregationNode, and HashJoinNode (the build side) can spill
to disk. However, their memory estimation does not consider this
capability and assumes it will hold all rows in memory. This causes
memory overestimation if cardinality is also overestimated. In reality,
the whole query execution in a single host is often subject to much
lower memory upper-bound and not allowed to exceed it.
This upper-bound is dictated by, but not limited to:
- MEM_LIMIT
- MEM_LIMIT_COORDINATORS
- MEM_LIMIT_EXECUTORS
- MAX_MEM_ESTIMATE_FOR_ADMISSION
- impala.admission-control.max-query-mem-limit.<pool_name>
from admission control.
This patch adds SpillableOperator interface that defines an alternative
of either PlanNode.computeNodeResourceProfile() or
DataSink.computeResourceProfile() if a lower memory upper-bound can be
reasoned about from configs mentioned above. This interface is applied
to SortNode, AggregationNode, HashJoinNode, and JoinBuildSink.
The in-memory vs spill-to-disk bias is controlled through
MEM_ESTIMATE_SCALE_FOR_SPILLING_OPERATOR option. A scale between
[0.0,1.0] to control estimate peak memory of query operator that has
spill-to-disk capabilities. Setting value closer to 1.0 will make
Planner bias towards keeping as much rows as possible in memory, while
setting value closer to 0.0 will make Planner bias towards spilling rows
to disk under memory pressure. Note that lowering
MEM_ESTIMATE_SCALE_FOR_SPILLING_OPERATOR can make query that previously
rejected by Admission Controller becomes admittable, but also may
spill-to-disk more and have higher risk to exhaust scratch space more
than before.
There are some caveats on this memory bounding patch:
- It checks if spill-to-disk is enabled in the coordinator, but
individual backend executors might not have it configured. Mismatch of
spill-to-disk configs between the coordinator and backend executor,
however, is rare and can be considered as misconfiguration.
- It does not check the actual total scratch space available to
spill-to-disk. However, query execution will be forced to spill anyway
if memory usage exceeds the three memory configs above. Raising
MEM_LIMIT / MEM_LIMIT_EXECUTORS option can help increase memory
estimation and increase the likelihood for the query to get assigned
to a larger executor group set, which usually has a bigger total
scratch space.
- The memory bound is divided evenly among all instances of a fragment
kind in a single host. But in theory, they should be able to share
and grow their memory usage independently beyond memory estimate as
long max memory reservation is not set.
- This does not consider other memory-related configs such as
clamp_query_mem_limit_backend_mem_limit or disable_pool_mem_limits
flag. But the admission controller will still enforce them if set.
Testing:
- Pass FE and custom cluster tests with core exploration.
Change-Id: I290c4e889d4ab9e921e356f0f55a9c8b11d0854e
Reviewed-on: http://gerrit.cloudera.org:8080/21762
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Riza Suminto <ri...@cloudera.com>
---
be/src/scheduling/schedule-state.cc | 1 +
be/src/service/query-options-test.cc | 1 +
be/src/service/query-options.cc | 7 +
be/src/service/query-options.h | 4 +-
common/thrift/ImpalaService.thrift | 17 +++
common/thrift/Query.thrift | 3 +
.../java/org/apache/impala/analysis/Analyzer.java | 77 ++++++++++
.../org/apache/impala/planner/AggregationNode.java | 51 +++++--
.../apache/impala/planner/DistributedPlanner.java | 18 ++-
.../org/apache/impala/planner/HashJoinNode.java | 50 +++++--
.../org/apache/impala/planner/JoinBuildSink.java | 20 ++-
.../org/apache/impala/planner/PlanFragment.java | 32 ++++-
.../java/org/apache/impala/planner/PlanNode.java | 16 +++
.../org/apache/impala/planner/ResourceProfile.java | 59 +++++---
.../impala/planner/ResourceProfileBuilder.java | 32 ++++-
.../java/org/apache/impala/planner/SortNode.java | 43 +++++-
.../apache/impala/planner/SpillableOperator.java | 39 +++++
.../java/org/apache/impala/service/Frontend.java | 3 +-
.../apache/impala/util/JniRequestPoolService.java | 1 -
.../org/apache/impala/util/RequestPoolService.java | 3 +
.../org/apache/impala/planner/PlannerTestBase.java | 24 +---
.../impala/planner/TpcdsCpuCostPlannerTest.java | 154 ++++++++++++++++++--
fe/src/test/resources/llama-site-3-groups.xml | 4 +-
.../tpcds_cpu_cost/tpcds-ddl-iceberg.test | 24 ++--
.../tpcds_cpu_cost/tpcds-ddl-parquet.test | 24 ++--
.../PlannerTest/tpcds_cpu_cost/tpcds-q22.test | 6 +-
.../PlannerTest/tpcds_cpu_cost/tpcds-q23a.test | 8 +-
.../PlannerTest/tpcds_cpu_cost/tpcds-q23b.test | 8 +-
.../PlannerTest/tpcds_cpu_cost/tpcds-q38.test | 10 +-
.../PlannerTest/tpcds_cpu_cost/tpcds-q51.test | 24 ++--
.../PlannerTest/tpcds_cpu_cost/tpcds-q57.test | 6 +-
...tpcds-q64.test => tpcds-q64-mem_limit-10g.test} | 32 ++---
...test => tpcds-q64-mem_limit_executors-20g.test} | 32 ++---
.../PlannerTest/tpcds_cpu_cost/tpcds-q64.test | 8 +-
...tpcds-q67.test => tpcds-q67-mem_limit-10g.test} | 26 ++--
...test => tpcds-q67-mem_limit_executors-20g.test} | 26 ++--
.../PlannerTest/tpcds_cpu_cost/tpcds-q67.test | 18 +--
.../PlannerTest/tpcds_cpu_cost/tpcds-q72.test | 28 ++--
.../PlannerTest/tpcds_cpu_cost/tpcds-q87.test | 10 +-
.../PlannerTest/tpcds_cpu_cost/tpcds-q95.test | 8 +-
...tpcds-q97.test => tpcds-q97-mem_limit-10g.test} | 30 ++--
...test => tpcds-q97-mem_limit_executors-20g.test} | 10 +-
.../PlannerTest/tpcds_cpu_cost/tpcds-q97.test | 8 +-
tests/custom_cluster/test_executor_groups.py | 158 +++++++++++++++++++--
44 files changed, 884 insertions(+), 279 deletions(-)
diff --git a/be/src/scheduling/schedule-state.cc b/be/src/scheduling/schedule-state.cc
index 2651ddab7..dd878f89a 100644
--- a/be/src/scheduling/schedule-state.cc
+++ b/be/src/scheduling/schedule-state.cc
@@ -295,6 +295,7 @@ void ScheduleState::CompareMinCoordinatorMemToAdmit(
}
}
+// Please update Analyzer.getMaxMemLimitPerHost() in Analyzer.java if updating this.
void ScheduleState::UpdateMemoryRequirements(const TPoolConfig& pool_cfg,
int64_t coord_mem_limit_admission, int64_t executor_mem_limit_admission) {
// If the min_query_mem_limit and max_query_mem_limit are not set in the pool config
diff --git a/be/src/service/query-options-test.cc b/be/src/service/query-options-test.cc
index 3be966ed8..ba70a9cf5 100644
--- a/be/src/service/query-options-test.cc
+++ b/be/src/service/query-options-test.cc
@@ -324,6 +324,7 @@ TEST(QueryOptions, SetFractionalOptions) {
{MAKE_OPTIONDEF(join_selectivity_correlation_factor), true},
{MAKE_OPTIONDEF(agg_mem_correlation_factor), true},
{MAKE_OPTIONDEF(runtime_filter_cardinality_reduction_scale), true},
+ {MAKE_OPTIONDEF(mem_estimate_scale_for_spilling_operator), true},
};
for (const auto& test_case : case_set) {
const OptionDef<double>& option_def = test_case.first;
diff --git a/be/src/service/query-options.cc b/be/src/service/query-options.cc
index 834b87765..151316917 100644
--- a/be/src/service/query-options.cc
+++ b/be/src/service/query-options.cc
@@ -1351,6 +1351,13 @@ Status impala::SetQueryOption(TImpalaQueryOptions::type option, const string& va
query_options->__set_skip_unneeded_updates_col_limit(int32_t_val);
break;
}
+ case TImpalaQueryOptions::MEM_ESTIMATE_SCALE_FOR_SPILLING_OPERATOR: {
+ double double_val = 0.0f;
+ RETURN_IF_ERROR(QueryOptionParser::ParseAndCheckInclusiveRange<double>(
+ option, value, 0.0, 1.0, &double_val));
+ query_options->__set_mem_estimate_scale_for_spilling_operator(double_val);
+ break;
+ }
default:
string key = to_string(option);
if (IsRemovedQueryOption(key)) {
diff --git a/be/src/service/query-options.h b/be/src/service/query-options.h
index 77c969822..6b4973a30 100644
--- a/be/src/service/query-options.h
+++ b/be/src/service/query-options.h
@@ -51,7 +51,7 @@ typedef std::unordered_map<string, beeswax::TQueryOptionLevel::type>
// plus one. Thus, the second argument to the DCHECK has to be updated every
// time we add or remove a query option to/from the enum TImpalaQueryOptions.
constexpr unsigned NUM_QUERY_OPTIONS =
- TImpalaQueryOptions::SKIP_UNNEEDED_UPDATES_COL_LIMIT + 1;
+ TImpalaQueryOptions::MEM_ESTIMATE_SCALE_FOR_SPILLING_OPERATOR + 1;
#define QUERY_OPTS_TABLE \
DCHECK_EQ(_TImpalaQueryOptions_VALUES_TO_NAMES.size(), NUM_QUERY_OPTIONS); \
REMOVED_QUERY_OPT_FN(abort_on_default_limit_exceeded, ABORT_ON_DEFAULT_LIMIT_EXCEEDED) \
@@ -372,6 +372,8 @@ constexpr unsigned NUM_QUERY_OPTIONS =
TQueryOptionLevel::ADVANCED) \
QUERY_OPT_FN(skip_unneeded_updates_col_limit, \
SKIP_UNNEEDED_UPDATES_COL_LIMIT, TQueryOptionLevel::ADVANCED) \
+ QUERY_OPT_FN(mem_estimate_scale_for_spilling_operator, \
+ MEM_ESTIMATE_SCALE_FOR_SPILLING_OPERATOR, TQueryOptionLevel::DEVELOPMENT) \
;
/// Enforce practical limits on some query options to avoid undesired query state.
diff --git a/common/thrift/ImpalaService.thrift b/common/thrift/ImpalaService.thrift
index 41a602f2b..1d400b0b9 100644
--- a/common/thrift/ImpalaService.thrift
+++ b/common/thrift/ImpalaService.thrift
@@ -1005,6 +1005,23 @@ enum TImpalaQueryOptions {
// given value, then no filtering is done.
// Set it non-positive to turn this feature off completely.
SKIP_UNNEEDED_UPDATES_COL_LIMIT = 189
+
+ // A scale between (0.0, 1.0] (not including 0.0) to control estimate peak memory of
+ // query operators that have spill-to-disk capabilities. Setting value closer to 1.0
+ // will make Planner bias towards keeping as much rows as possible in memory, while
+ // setting value closer to 0.0 will make Planner bias towards spilling rows to disk
+ // under memory pressure.
+ // Resulting memory estimate will never be less than minimum memory required for
+ // a query operator to start execution. MEM_LIMIT, MEM_LIMIT_COORDINATORS,
+ // MEM_LIMIT_EXECUTORS, per-pool memory clamping configuration, and
+ // MAX_MEM_ESTIMATE_FOR_ADMISSION will be considered into peak memory estimation if any
+ // of them is set.
+ // Setting to 0.0 or unsetting scratch_dirs flag will revert Planner to old behavior
+ // where it assumes that all rows must be held in memory, regardless of any memory
+ // limit configurations mentioned above. It is recommended to set this query option
+ // to either 1.0 (enabled) or 0.0 (disabled) and only set value in between for
+ // experimental purpose.
+ MEM_ESTIMATE_SCALE_FOR_SPILLING_OPERATOR = 190
}
// The summary of a DML statement.
diff --git a/common/thrift/Query.thrift b/common/thrift/Query.thrift
index bda38fa9c..f4087874c 100644
--- a/common/thrift/Query.thrift
+++ b/common/thrift/Query.thrift
@@ -771,6 +771,9 @@ struct TQueryOptions {
// See comment in ImpalaService.thrift
190: optional i32 skip_unneeded_updates_col_limit = 10
+
+ // See comment in ImpalaService.thrift
+ 191: optional double mem_estimate_scale_for_spilling_operator = 0.0
}
// Impala currently has three types of sessions: Beeswax, HiveServer2 and external
diff --git a/fe/src/main/java/org/apache/impala/analysis/Analyzer.java b/fe/src/main/java/org/apache/impala/analysis/Analyzer.java
index 7ba4a488e..26d235c46 100644
--- a/fe/src/main/java/org/apache/impala/analysis/Analyzer.java
+++ b/fe/src/main/java/org/apache/impala/analysis/Analyzer.java
@@ -113,9 +113,11 @@ import org.apache.impala.service.FeSupport;
import org.apache.impala.thrift.QueryConstants;
import org.apache.impala.thrift.TAccessEvent;
import org.apache.impala.thrift.TCatalogObjectType;
+import org.apache.impala.thrift.TExecutorGroupSet;
import org.apache.impala.thrift.TImpalaQueryOptions;
import org.apache.impala.thrift.TLineageGraph;
import org.apache.impala.thrift.TNetworkAddress;
+import org.apache.impala.thrift.TPoolConfig;
import org.apache.impala.thrift.TQueryCtx;
import org.apache.impala.thrift.TQueryOptions;
import org.apache.impala.util.AcidUtils;
@@ -128,6 +130,7 @@ import org.apache.impala.util.IntIterator;
import org.apache.impala.util.KuduUtil;
import org.apache.impala.util.ListMap;
import org.apache.impala.util.MetaStoreUtil;
+import org.apache.impala.util.RequestPoolService;
import org.apache.impala.util.TSessionStateUtil;
import org.apache.kudu.client.KuduClient;
import org.github.jamm.CannotAccessFieldException;
@@ -626,6 +629,16 @@ public class Analyzer {
// Set by Frontend.java.
private int availableCoresPerNode_ = -1;
+ // The following properties are parsed from TExecutorGroupSet selected by Frontend.
+ // They are set by Frontend.java via setPoolMemLimit().
+ //
+ // Value of impala.admission-control.clamp-mem-limit-query-option.<pool_name>.
+ private boolean poolHasMemLimitClamp_ = false;
+ // Value of impala.admission-control.max-query-mem-limit.<pool_name>,
+ // which is the maximum memory limit of destination request pool for this query in
+ // bytes. Value 0 or less means unknown.
+ private long poolMaxMemLimit_ = -1;
+
// Cache of KuduTables opened for this query. (map from table name to kudu table)
// This cache prevent multiple openTable calls for a given table in the same query.
public final Map<String, org.apache.kudu.client.KuduTable> kuduTables =
@@ -746,6 +759,70 @@ public class Analyzer {
Math.min(QueryConstants.MAX_FRAGMENT_INSTANCES_PER_NODE, x);
}
+ /**
+ * Return maximum memory limit per executor host, in bytes unit, that will be enforced
+ * by the Admission Controller. PlanNodes can use this information to lower
+ * its memory estimates since Admission Controller will not allow query
+ * to grow its memory usage in single executor host beyond the value returned here.
+ * This method is maintained to match order in ScheduleState::UpdateMemoryRequirements.
+ */
+ public long getMaxMemLimitPerHost(boolean isCoordinatorFragment) {
+ TQueryOptions opts = getQueryOptions();
+
+ if (opts.isSetMem_limit() && opts.getMem_limit() > 0) {
+ // MEM_LIMIT
+ long mem_limit = opts.getMem_limit();
+ if (globalState_.poolHasMemLimitClamp_ && globalState_.poolMaxMemLimit_ > 0) {
+ // clamp-mem-limit-query-option is True and max-query-mem-limit is set.
+ // max-query-mem-limit take precedence over MEM_LIMIT if it is lower.
+ mem_limit = Math.min(mem_limit, globalState_.poolMaxMemLimit_);
+ }
+ return mem_limit;
+ } else if (isCoordinatorFragment && opts.isSetMem_limit_coordinators()
+ && opts.getMem_limit_coordinators() > 0) {
+ // MEM_LIMIT_COORDINATORS
+ return opts.getMem_limit_coordinators();
+ } else if (!isCoordinatorFragment && opts.isSetMem_limit_executors()
+ && opts.getMem_limit_executors() > 0) {
+ // MEM_LIMIT_EXECUTORS
+ return opts.getMem_limit_executors();
+ } else if (globalState_.poolMaxMemLimit_ > 0) {
+ // No MEM_LIMIT* option is set and max-query-mem-limit is set.
+ // Return the max-query-mem-limit regardless of clamp-mem-limit-query-option value.
+ return globalState_.poolMaxMemLimit_;
+ } else if (opts.isSetMax_mem_estimate_for_admission()
+ && opts.getMax_mem_estimate_for_admission() > 0) {
+ // MAX_MEM_ESTIMATE_FOR_ADMISSION
+ return opts.getMax_mem_estimate_for_admission();
+ }
+ return Long.MAX_VALUE;
+ }
+
+ /**
+ * Update 'poolHasMemLimitClamp_' and 'poolMaxMemLimit_' for this query if TPoolConfig
+ * for a given 'groupSet' has
+ * impala.admission-control.clamp-mem-limit-query-option.<pool_name> and
+ * impala.admission-control.max-query-mem-limit.<pool_name> set in llama file.
+ */
+ public void setPoolMemLimit(TExecutorGroupSet groupSet) {
+ Preconditions.checkNotNull(groupSet);
+ // Only respect max_mem_limit from groupSet if RequestPoolService is configured
+ // (fair_scheduler_allocation_path flag is not empty) and
+ // impala.admission-control.clamp-mem-limit-query-option.<pool_name> is True and
+ // impala.admission-control.max-query-mem-limit.<pool_name> is set.
+ RequestPoolService poolService = RequestPoolService.getInstance();
+ if (poolService != null) {
+ TPoolConfig poolConfig =
+ poolService.getPoolConfig(groupSet.getExec_group_name_prefix());
+ if (poolConfig.isSetClamp_mem_limit_query_option()) {
+ globalState_.poolHasMemLimitClamp_ = poolConfig.isClamp_mem_limit_query_option();
+ }
+ if (poolConfig.isSetMax_query_mem_limit()) {
+ globalState_.poolMaxMemLimit_ = poolConfig.getMax_query_mem_limit();
+ }
+ }
+ }
+
public int getMinParallelismPerNode() {
if (getQueryOptions().isCompute_processing_cost()) {
return getQueryOptions().getProcessing_cost_min_threads();
diff --git a/fe/src/main/java/org/apache/impala/planner/AggregationNode.java b/fe/src/main/java/org/apache/impala/planner/AggregationNode.java
index ceb2ffe08..cea564663 100644
--- a/fe/src/main/java/org/apache/impala/planner/AggregationNode.java
+++ b/fe/src/main/java/org/apache/impala/planner/AggregationNode.java
@@ -63,7 +63,7 @@ import com.google.common.collect.Lists;
* Aggregation computation.
*
*/
-public class AggregationNode extends PlanNode {
+public class AggregationNode extends PlanNode implements SpillableOperator {
private static final Logger LOG = LoggerFactory.getLogger(AggregationNode.class);
// Default per-instance memory requirement used if no valid stats are available.
@@ -914,6 +914,16 @@ public class AggregationNode extends PlanNode {
@Override
public void computeNodeResourceProfile(TQueryOptions queryOptions) {
+ computeResourceProfileIfSpill(queryOptions, Long.MAX_VALUE);
+ }
+
+ @Override
+ public void computeResourceProfileIfSpill(
+ TQueryOptions queryOptions, long maxMemoryEstimatePerInstance) {
+ boolean biasToSpill = shouldComputeResourcesWithSpill(queryOptions);
+ if (!biasToSpill) {
+ Preconditions.checkArgument(maxMemoryEstimatePerInstance == Long.MAX_VALUE);
+ }
resourceProfiles_ = Lists.newArrayListWithCapacity(aggInfos_.size());
resourceProfiles_.clear();
// IMPALA-2945: Behavior change if estimatePreaggDuplicate is true.
@@ -929,18 +939,38 @@ public class AggregationNode extends PlanNode {
inputCardinality =
smallestValidCardinality(inputCardinality, aggClassOutputCardinality);
}
- resourceProfiles_.add(
- computeAggClassResourceProfile(queryOptions, aggInfo, inputCardinality));
+ resourceProfiles_.add(computeAggClassResourceProfile(
+ queryOptions, aggInfo, inputCardinality, maxMemoryEstimatePerInstance));
aggIdx++;
}
+ ResourceProfile totalResource = ResourceProfile.noReservation(0);
if (aggInfos_.size() == 1) {
- nodeResourceProfile_ = resourceProfiles_.get(0);
+ totalResource = resourceProfiles_.get(0);
} else {
- nodeResourceProfile_ = ResourceProfile.noReservation(0);
for (ResourceProfile aggProfile : resourceProfiles_) {
- nodeResourceProfile_ = nodeResourceProfile_.sum(aggProfile);
+ totalResource = totalResource.sum(aggProfile);
}
}
+
+ if (aggInfos_.size() > 1 && biasToSpill) {
+ // Total mem estimate might still exceed maxMemoryEstimatePerInstance.
+ // Do another capping over the sum ResourceProfile, but maintain total mem estimate
+ // higher than total min mem reservation.
+ ResourceProfileBuilder builder =
+ new ResourceProfileBuilder()
+ .setMemEstimateBytes(totalResource.getMemEstimateBytes())
+ .setMinMemReservationBytes(totalResource.getMinMemReservationBytes())
+ .setThreadReservation(totalResource.getThreadReservation())
+ .setMemEstimateScale(
+ queryOptions.getMem_estimate_scale_for_spilling_operator(),
+ maxMemoryEstimatePerInstance);
+ if (totalResource.getMaxMemReservationBytes() > 0) {
+ builder.setMaxMemReservationBytes(totalResource.getMaxMemReservationBytes());
+ }
+ nodeResourceProfile_ = builder.build();
+ } else {
+ nodeResourceProfile_ = totalResource;
+ }
}
private long estimatePerInstanceDataBytes(
@@ -989,8 +1019,8 @@ public class AggregationNode extends PlanNode {
return perInstanceDataBytes;
}
- private ResourceProfile computeAggClassResourceProfile(
- TQueryOptions queryOptions, AggregateInfo aggInfo, long inputCardinality) {
+ private ResourceProfile computeAggClassResourceProfile(TQueryOptions queryOptions,
+ AggregateInfo aggInfo, long inputCardinality, long maxMemoryEstimatePerInstance) {
Preconditions.checkNotNull(
fragment_, "PlanNode must be placed into a fragment before calling this method.");
long perInstanceCardinality =
@@ -1107,6 +1137,11 @@ public class AggregationNode extends PlanNode {
builder.setMemEstimateBytes(
Math.min(perInstanceMemEstimate, maxReservationBytes));
}
+ if (shouldComputeResourcesWithSpill(queryOptions)) {
+ builder.setMemEstimateScale(
+ queryOptions.getMem_estimate_scale_for_spilling_operator(),
+ maxMemoryEstimatePerInstance);
+ }
return builder.build();
}
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 eb17be65e..a9824908e 100644
--- a/fe/src/main/java/org/apache/impala/planner/DistributedPlanner.java
+++ b/fe/src/main/java/org/apache/impala/planner/DistributedPlanner.java
@@ -729,16 +729,20 @@ public class DistributedPlanner {
ctx_.getQueryOptions().getDefault_join_distribution_mode());
}
- // Decide the distribution mode based on the estimated costs, the mem limit and
- // the broadcast bytes limit. The last value is a safety check to ensure we
- // don't broadcast very large inputs (for example in case the broadcast cost was
- // not computed correctly and the query mem limit has not been set or set too high)
+ // Decide the distribution mode based on the estimated costs, the mem_limit /
+ // mem_limit_executors option, the max-query-mem-limit config of the request pool
+ // (if exist), and the broadcast bytes limit.
+ // The last value is a safety check to ensure we don't broadcast very large inputs
+ // (for example in case the broadcast cost was not computed correctly and the query
+ // mem limit has not been set or set too high).
+ // mem_limit_coordinators option is not considered because it is unknown if the
+ // resulting HashJoin fragment is the final plan's root or not.
long htSize = Math.round(rhsDataSize * PlannerContext.HASH_TBL_SPACE_OVERHEAD);
- long memLimit = ctx_.getQueryOptions().mem_limit;
+ long memLimit = ctx_.getRootAnalyzer().getMaxMemLimitPerHost(false);
long broadcast_bytes_limit = ctx_.getQueryOptions().getBroadcast_bytes_limit();
- if (broadcastCost <= partitionCost && (memLimit == 0 || htSize <= memLimit) &&
- (broadcast_bytes_limit == 0 || htSize <= broadcast_bytes_limit)) {
+ if (broadcastCost <= partitionCost && htSize <= memLimit
+ && (broadcast_bytes_limit == 0 || htSize <= broadcast_bytes_limit)) {
return DistributionMode.BROADCAST;
}
// Partitioned was cheaper or the broadcast HT would not fit within the mem limit.
diff --git a/fe/src/main/java/org/apache/impala/planner/HashJoinNode.java b/fe/src/main/java/org/apache/impala/planner/HashJoinNode.java
index 96cae6785..dc418fc58 100644
--- a/fe/src/main/java/org/apache/impala/planner/HashJoinNode.java
+++ b/fe/src/main/java/org/apache/impala/planner/HashJoinNode.java
@@ -37,7 +37,6 @@ import org.apache.impala.thrift.TPlanNode;
import org.apache.impala.thrift.TPlanNodeType;
import org.apache.impala.thrift.TQueryOptions;
import org.apache.impala.util.BitUtil;
-import org.apache.impala.util.ExprUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -50,7 +49,7 @@ import com.google.common.base.Preconditions;
* plan generated for a table ref. Typically, that is the right child, but due to join
* inversion (for outer/semi/cross joins) it could also be the left child.
*/
-public class HashJoinNode extends JoinNode {
+public class HashJoinNode extends JoinNode implements SpillableOperator {
private final static Logger LOG = LoggerFactory.getLogger(HashJoinNode.class);
// Coefficients for estimating hash join CPU processing cost. Derived from
@@ -231,14 +230,38 @@ public class HashJoinNode extends JoinNode {
return output.toString();
}
+ @Override
+ public void computeNodeResourceProfile(TQueryOptions queryOptions) {
+ computeResourceProfileIfSpill(queryOptions, Long.MAX_VALUE);
+ }
+
+ @Override
+ public void computeResourceProfileIfSpill(
+ TQueryOptions queryOptions, long maxMemoryEstimatePerInstance) {
+ Pair<ResourceProfile, ResourceProfile> profiles =
+ computeJoinResourceProfile(queryOptions, maxMemoryEstimatePerInstance);
+ if (hasSeparateBuild()) {
+ // All build resource consumption is accounted for in the separate builder.
+ nodeResourceProfile_ = profiles.first;
+ } else {
+ // Both build and profile resources are accounted for in the node.
+ nodeResourceProfile_ = profiles.first.combine(profiles.second);
+ }
+ }
+
+ @Override
+ public Pair<ResourceProfile, ResourceProfile> computeJoinResourceProfile(
+ TQueryOptions queryOptions) {
+ return computeJoinResourceProfile(queryOptions, Long.MAX_VALUE);
+ }
+
/**
* Helper method to compute the resource requirements for the join that can be
* called from the builder or the join node. Returns a pair of the probe
* resource requirements and the build resource requirements.
*/
- @Override
public Pair<ResourceProfile, ResourceProfile> computeJoinResourceProfile(
- TQueryOptions queryOptions) {
+ TQueryOptions queryOptions, long maxMemoryEstimatePerInstance) {
long perBuildInstanceMemEstimate;
long perBuildInstanceDataBytes;
int numInstances = fragment_.getNumInstances();
@@ -326,11 +349,20 @@ public class HashJoinNode extends JoinNode {
.setMinMemReservationBytes(perInstanceProbeMinMemReservation)
.setSpillableBufferBytes(bufferSize)
.setMaxRowBufferBytes(maxRowBufferSize).build();
- ResourceProfile buildProfile = new ResourceProfileBuilder()
- .setMemEstimateBytes(perBuildInstanceMemEstimate)
- .setMinMemReservationBytes(perInstanceBuildMinMemReservation)
- .setSpillableBufferBytes(bufferSize)
- .setMaxRowBufferBytes(maxRowBufferSize).build();
+ ResourceProfileBuilder buildProfileBuilder =
+ new ResourceProfileBuilder()
+ .setMemEstimateBytes(perBuildInstanceMemEstimate)
+ .setMinMemReservationBytes(perInstanceBuildMinMemReservation)
+ .setSpillableBufferBytes(bufferSize)
+ .setMaxRowBufferBytes(maxRowBufferSize);
+ if (shouldComputeResourcesWithSpill(queryOptions)) {
+ buildProfileBuilder.setMemEstimateScale(
+ queryOptions.getMem_estimate_scale_for_spilling_operator(),
+ maxMemoryEstimatePerInstance);
+ } else {
+ Preconditions.checkArgument(maxMemoryEstimatePerInstance == Long.MAX_VALUE);
+ }
+ ResourceProfile buildProfile = buildProfileBuilder.build();
return Pair.create(probeProfile, buildProfile);
}
diff --git a/fe/src/main/java/org/apache/impala/planner/JoinBuildSink.java b/fe/src/main/java/org/apache/impala/planner/JoinBuildSink.java
index 2b13c1879..4136dfda7 100644
--- a/fe/src/main/java/org/apache/impala/planner/JoinBuildSink.java
+++ b/fe/src/main/java/org/apache/impala/planner/JoinBuildSink.java
@@ -29,8 +29,6 @@ import org.apache.impala.thrift.TDataSinkType;
import org.apache.impala.thrift.TExplainLevel;
import org.apache.impala.thrift.TJoinBuildSink;
import org.apache.impala.thrift.TQueryOptions;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import com.google.common.base.Preconditions;
@@ -39,9 +37,7 @@ import static org.apache.impala.analysis.ToSqlOptions.DEFAULT;
/**
* Sink to materialize the build side of a join.
*/
-public class JoinBuildSink extends DataSink {
- private final static Logger LOG = LoggerFactory.getLogger(JoinBuildSink.class);
-
+public class JoinBuildSink extends DataSink implements SpillableOperator {
// id of join's build-side table assigned during planning
private final JoinTableId joinTableId_;
@@ -168,6 +164,20 @@ public class JoinBuildSink extends DataSink {
resourceProfile_ = joinNode_.computeJoinResourceProfile(queryOptions).second;
}
+ @Override
+ public void computeResourceProfileIfSpill(
+ TQueryOptions queryOptions, long maxMemoryEstimatePerInstance) {
+ if (joinNode_ instanceof HashJoinNode) {
+ // Only HashJoinNode implements SpillableOperator.
+ resourceProfile_ =
+ ((HashJoinNode) joinNode_)
+ .computeJoinResourceProfile(queryOptions, maxMemoryEstimatePerInstance)
+ .second;
+ } else {
+ computeResourceProfile(queryOptions);
+ }
+ }
+
@Override
public void collectExprs(List<Expr> exprs) {
exprs.addAll(buildExprs_);
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 8021e62cd..4ca2c7f6c 100644
--- a/fe/src/main/java/org/apache/impala/planner/PlanFragment.java
+++ b/fe/src/main/java/org/apache/impala/planner/PlanFragment.java
@@ -464,8 +464,21 @@ public class PlanFragment extends TreeNode<PlanFragment> {
private void computeRuntimeFilterResources(Analyzer analyzer) {
Map<RuntimeFilterId, RuntimeFilter> consumedFilters = new HashMap<>();
Map<RuntimeFilterId, RuntimeFilter> producedFilters = new HashMap<>();
+ TQueryOptions queryOptions = analyzer.getQueryOptions();
+ boolean biasToSpill = PlanNode.shouldComputeResourcesWithSpill(queryOptions);
+ long maxMemPerInstance = Long.MAX_VALUE;
+ if (biasToSpill) {
+ boolean isCoordinator = (sink_ instanceof PlanRootSink);
+ maxMemPerInstance = analyzer.getMaxMemLimitPerHost(isCoordinator)
+ / getNumInstancesPerHost(queryOptions);
+ }
// Visit all sinks and nodes to identify filters produced or consumed by fragment.
- sink_.computeResourceProfile(analyzer.getQueryOptions());
+ if (biasToSpill && sink_ instanceof SpillableOperator) {
+ ((SpillableOperator) sink_)
+ .computeResourceProfileIfSpill(queryOptions, maxMemPerInstance);
+ } else {
+ sink_.computeResourceProfile(queryOptions);
+ }
Preconditions.checkState(
sink_.getRuntimeFilters().isEmpty() || sink_ instanceof JoinBuildSink);
for (RuntimeFilter filter : sink_.getRuntimeFilters()) {
@@ -473,7 +486,12 @@ public class PlanFragment extends TreeNode<PlanFragment> {
producedFilters.put(filter.getFilterId(), filter);
}
for (PlanNode node : collectPlanNodes()) {
- node.computeNodeResourceProfile(analyzer.getQueryOptions());
+ if (biasToSpill && node instanceof SpillableOperator) {
+ ((SpillableOperator) node)
+ .computeResourceProfileIfSpill(queryOptions, maxMemPerInstance);
+ } else {
+ node.computeNodeResourceProfile(queryOptions);
+ }
boolean isFilterProducer = node instanceof JoinNode;
for (RuntimeFilter filter : node.getRuntimeFilters()) {
if (isFilterProducer) {
@@ -485,8 +503,7 @@ public class PlanFragment extends TreeNode<PlanFragment> {
}
for (RuntimeFilter f : producedFilters.values()) {
producedRuntimeFiltersMemReservationBytes_ += f.getFilterSize();
- if (analyzer.getQueryOptions().max_num_filters_aggregated_per_host > 1
- && !f.isBroadcast()) {
+ if (queryOptions.max_num_filters_aggregated_per_host > 1 && !f.isBroadcast()) {
// If distributed aggregation is enabled, any backend can be selected as
// an intermediate aggregator. Add the same amount of memory per backend to
// anticipate for it.
@@ -1009,7 +1026,7 @@ public class PlanFragment extends TreeNode<PlanFragment> {
/**
* Only valid to call after calling {@link #traverseEffectiveParallelism(int, int,
- * PlanFragment)}.
+ * PlanFragment, TQueryOptions)}.
*/
protected int getAdjustedInstanceCount() {
Preconditions.checkState(adjustedInstanceCount_ > -1);
@@ -1018,7 +1035,7 @@ public class PlanFragment extends TreeNode<PlanFragment> {
/**
* Only valid to call after calling {@link #traverseEffectiveParallelism(int, int,
- * PlanFragment)}.
+ * PlanFragment, TQueryOptions)}.
*/
protected int getMaxParallelism() {
Preconditions.checkState(maxParallelism_ > -1);
@@ -1363,7 +1380,8 @@ public class PlanFragment extends TreeNode<PlanFragment> {
/**
* Override parallelism of this fragment with adjusted parallelism from CPU costing
* algorithm.
- * <p>Only valid after {@link #traverseEffectiveParallelism(int, int, int)}
+ * <p>Only valid after {@link #traverseEffectiveParallelism(int, int, PlanFragment,
+ * TQueryOptions)}
* called.
*/
protected void setEffectiveNumInstance() {
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 3a8d4815b..4448eaa21 100644
--- a/fe/src/main/java/org/apache/impala/planner/PlanNode.java
+++ b/fe/src/main/java/org/apache/impala/planner/PlanNode.java
@@ -48,6 +48,7 @@ import org.apache.impala.common.ThriftSerializationCtx;
import org.apache.impala.common.TreeNode;
import org.apache.impala.planner.RuntimeFilterGenerator.RuntimeFilter;
import org.apache.impala.planner.TupleCacheInfo.IneligibilityReason;
+import org.apache.impala.service.BackendConfig;
import org.apache.impala.thrift.TExecNodePhase;
import org.apache.impala.thrift.TExecStats;
import org.apache.impala.thrift.TExplainLevel;
@@ -1421,4 +1422,19 @@ abstract public class PlanNode extends TreeNode<PlanNode> {
}
return Math.max(-1, cardinality2);
}
+
+ /**
+ * Return True if Impala Coordinator node has scratch_dirs flag configured and
+ * given 'queryOptions' allows taking spilling to disk into account [when calculating
+ * memory requirements].
+ * Otherwise, return False.
+ * Note that this method does not validate if all backend executors also have
+ * scratch_dirs flag set up.
+ */
+ public static boolean shouldComputeResourcesWithSpill(TQueryOptions queryOptions) {
+ long scratchLimit = queryOptions.getScratch_limit();
+ double memScale = queryOptions.getMem_estimate_scale_for_spilling_operator();
+ String scratchDirs = BackendConfig.INSTANCE.getScratchDirs();
+ return scratchLimit != 0 && !scratchDirs.isEmpty() && memScale > 0.0;
+ }
}
diff --git a/fe/src/main/java/org/apache/impala/planner/ResourceProfile.java b/fe/src/main/java/org/apache/impala/planner/ResourceProfile.java
index cc13e418c..8c5695f5b 100644
--- a/fe/src/main/java/org/apache/impala/planner/ResourceProfile.java
+++ b/fe/src/main/java/org/apache/impala/planner/ResourceProfile.java
@@ -33,8 +33,10 @@ public class ResourceProfile {
private final boolean isValid_;
// Estimated memory consumption in bytes. Guaranteed to be >= minReservationBytes_ if
- // both are set and guaranteed to be <= maxMemReservationBytes_ if both are set (the
- // constructor ensures both of these conditions).
+ // both are set. It is, however, not guaranteed to be <= maxMemReservationBytes_ if
+ // this ResourceProfile is not instantiated from the public constructor.
+ // See other instantiation in sum(), max(), combine(), multiply() method below where
+ // memEstimateBytes_ <= maxMemReservationBytes_ is not guaranteed.
private final long memEstimateBytes_;
// Minimum memory reservation required to execute in bytes.
@@ -60,16 +62,30 @@ public class ResourceProfile {
// -1 if the profile is invalid (i.e. isValid_ is false).
private final long threadReservation_;
- ResourceProfile(boolean isValid, long memEstimateBytes, long minMemReservationBytes,
- long maxMemReservationBytes, long spillableBufferBytes,
- long maxRowBufferBytes, long threadReservation) {
+ /**
+ * Construct a ResourceProfile with given parameters.
+ * Do several parameter validation to make sure that the numbers are OK.
+ * If 'maxMemReservationBytes' is set (greater than 0) and 'isStrictMaxMemory' is True,
+ * also validate that 'memEstimateBytes' <= 'maxMemReservationBytes'.
+ */
+ private ResourceProfile(boolean isValid, long memEstimateBytes,
+ long minMemReservationBytes, long maxMemReservationBytes, long spillableBufferBytes,
+ long maxRowBufferBytes, long threadReservation, boolean isStrictMaxMemory) {
Preconditions.checkArgument(spillableBufferBytes == -1 || maxRowBufferBytes != -1);
Preconditions.checkArgument(spillableBufferBytes == -1
|| LongMath.isPowerOfTwo(spillableBufferBytes));
Preconditions.checkArgument(maxRowBufferBytes == -1
|| LongMath.isPowerOfTwo(maxRowBufferBytes));
Preconditions.checkArgument(!isValid || threadReservation >= 0, threadReservation);
- Preconditions.checkArgument(maxMemReservationBytes >= minMemReservationBytes);
+ Preconditions.checkState(minMemReservationBytes <= maxMemReservationBytes,
+ "minMemReservationBytes (%s) is greater than maxMemReservationBytes (%s)!",
+ minMemReservationBytes, maxMemReservationBytes);
+ if (maxMemReservationBytes > 0 && isStrictMaxMemory) {
+ Preconditions.checkState(memEstimateBytes <= maxMemReservationBytes,
+ "memEstimateBytes (%s) is greater than maxMemReservationBytes (%s)!",
+ memEstimateBytes, maxMemReservationBytes);
+ }
+
isValid_ = isValid;
memEstimateBytes_ = (minMemReservationBytes != -1) ?
Math.max(memEstimateBytes, minMemReservationBytes) : memEstimateBytes;
@@ -80,14 +96,21 @@ public class ResourceProfile {
threadReservation_ = threadReservation;
}
+ public ResourceProfile(boolean isValid, long memEstimateBytes,
+ long minMemReservationBytes, long maxMemReservationBytes, long spillableBufferBytes,
+ long maxRowBufferBytes, long threadReservation) {
+ this(isValid, memEstimateBytes, minMemReservationBytes, maxMemReservationBytes,
+ spillableBufferBytes, maxRowBufferBytes, threadReservation, true);
+ }
+
// Create a resource profile with zero min or max reservation and zero required
// threads.
public static ResourceProfile noReservation(long memEstimateBytes) {
- return new ResourceProfile(true, memEstimateBytes, 0, 0, -1, -1, 0);
+ return new ResourceProfile(true, memEstimateBytes, 0, 0, -1, -1, 0, false);
}
public static ResourceProfile invalid() {
- return new ResourceProfile(false, -1, -1, -1, -1, -1, -1);
+ return new ResourceProfile(false, -1, -1, -1, -1, -1, -1, false);
}
public boolean isValid() { return isValid_; }
@@ -133,9 +156,8 @@ public class ResourceProfile {
return new ResourceProfile(true,
Math.max(getMemEstimateBytes(), other.getMemEstimateBytes()),
Math.max(getMinMemReservationBytes(), other.getMinMemReservationBytes()),
- Math.max(getMaxMemReservationBytes(), other.getMaxMemReservationBytes()),
- -1, -1,
- Math.max(getThreadReservation(), other.getThreadReservation()));
+ Math.max(getMaxMemReservationBytes(), other.getMaxMemReservationBytes()), -1, -1,
+ Math.max(getThreadReservation(), other.getThreadReservation()), false);
}
// Returns a profile with the sum of each aggregate value in 'this' and 'other'.
@@ -146,10 +168,12 @@ public class ResourceProfile {
return new ResourceProfile(true,
MathUtil.saturatingAdd(getMemEstimateBytes(), other.getMemEstimateBytes()),
MathUtil.saturatingAdd(
- getMinMemReservationBytes(),other.getMinMemReservationBytes()),
+ getMinMemReservationBytes(), other.getMinMemReservationBytes()),
MathUtil.saturatingAdd(
- getMaxMemReservationBytes(), other.getMaxMemReservationBytes()), -1, -1,
- MathUtil.saturatingAdd(getThreadReservation(), other.getThreadReservation()));
+ getMaxMemReservationBytes(), other.getMaxMemReservationBytes()),
+ -1, -1,
+ MathUtil.saturatingAdd(getThreadReservation(), other.getThreadReservation()),
+ false);
}
// Returns a profile with the sum of each aggregate value in 'this' and 'other'.
@@ -160,12 +184,13 @@ public class ResourceProfile {
return new ResourceProfile(true,
MathUtil.saturatingAdd(getMemEstimateBytes(), other.getMemEstimateBytes()),
MathUtil.saturatingAdd(
- getMinMemReservationBytes(),other.getMinMemReservationBytes()),
+ getMinMemReservationBytes(), other.getMinMemReservationBytes()),
MathUtil.saturatingAdd(
getMaxMemReservationBytes(), other.getMaxMemReservationBytes()),
Math.max(getSpillableBufferBytes(), other.getSpillableBufferBytes()),
Math.max(getMaxRowBufferBytes(), other.getMaxRowBufferBytes()),
- MathUtil.saturatingAdd(getThreadReservation(), other.getThreadReservation()));
+ MathUtil.saturatingAdd(getThreadReservation(), other.getThreadReservation()),
+ false);
}
// Returns a profile with each aggregate value multiplied by 'factor'.
@@ -177,7 +202,7 @@ public class ResourceProfile {
MathUtil.saturatingMultiply(memEstimateBytes_, factor),
MathUtil.saturatingMultiply(minMemReservationBytes_, factor),
MathUtil.saturatingMultiply(maxMemReservationBytes_, factor), -1, -1,
- MathUtil.saturatingMultiply(threadReservation_, factor));
+ MathUtil.saturatingMultiply(threadReservation_, factor), false);
}
public TBackendResourceProfile toThrift() {
diff --git a/fe/src/main/java/org/apache/impala/planner/ResourceProfileBuilder.java b/fe/src/main/java/org/apache/impala/planner/ResourceProfileBuilder.java
index 1467aad79..90126dd6d 100644
--- a/fe/src/main/java/org/apache/impala/planner/ResourceProfileBuilder.java
+++ b/fe/src/main/java/org/apache/impala/planner/ResourceProfileBuilder.java
@@ -17,9 +17,10 @@
package org.apache.impala.planner;
-import org.apache.impala.planner.ResourceProfile;
import com.google.common.base.Preconditions;
+import org.apache.impala.util.MathUtil;
+
/**
* Utility class to help set up the various parameters of a ResourceProfile.
*/
@@ -38,6 +39,13 @@ public class ResourceProfileBuilder {
// Must be set if spillableBufferBytes_ is set.
private long maxRowBufferBytes_= -1;
+ // If set, attempt to scale down memEstimateBytes_ to value between
+ // minMemReservationBytes_ and maxMemEstimateBytesAfterScaling_ at build().
+ // memEstimateBytes_ will stay unchanged if scaled memory estimate is still higher
+ // or maxMemEstimateBytesAfterScaling_ is less than minMemReservationBytes_.
+ private double scale_ = 0.0;
+ private long maxMemEstimateBytesAfterScaling_ = -1;
+
// Defaults to zero, because most ExecNodes do not create additional threads.
private long threadReservation_ = 0;
@@ -75,8 +83,30 @@ public class ResourceProfileBuilder {
return this;
}
+ public ResourceProfileBuilder setMemEstimateScale(double scale, long maxMemBound) {
+ Preconditions.checkArgument(scale > 0.0 && scale <= 1.0, "invalid scale %s", scale);
+ Preconditions.checkArgument(maxMemBound > 0, "invalid maxMemBound %s", maxMemBound);
+ scale_ = scale;
+ maxMemEstimateBytesAfterScaling_ = maxMemBound;
+ return this;
+ }
+
ResourceProfile build() {
Preconditions.checkState(memEstimateBytes_ >= 0, "Mem estimate must be set");
+
+ if (scale_ > 0.0 && maxMemEstimateBytesAfterScaling_ > 0) {
+ long scaleBasedMemEstimate = minMemReservationBytes_;
+ if (minMemReservationBytes_ <= maxMemEstimateBytesAfterScaling_) {
+ scaleBasedMemEstimate = MathUtil.saturatingAdd(minMemReservationBytes_,
+ (long) ((maxMemEstimateBytesAfterScaling_ - minMemReservationBytes_)
+ * scale_));
+ memEstimateBytes_ = Math.min(memEstimateBytes_, scaleBasedMemEstimate);
+ }
+ if (maxMemReservationBytes_ > 0) {
+ memEstimateBytes_ = Math.min(memEstimateBytes_, maxMemReservationBytes_);
+ }
+ }
+
return new ResourceProfile(true, memEstimateBytes_, minMemReservationBytes_,
maxMemReservationBytes_, spillableBufferBytes_, maxRowBufferBytes_,
threadReservation_);
diff --git a/fe/src/main/java/org/apache/impala/planner/SortNode.java b/fe/src/main/java/org/apache/impala/planner/SortNode.java
index 417128359..43295763e 100644
--- a/fe/src/main/java/org/apache/impala/planner/SortNode.java
+++ b/fe/src/main/java/org/apache/impala/planner/SortNode.java
@@ -50,7 +50,7 @@ import com.google.common.base.Preconditions;
*
* Will always materialize the new tuple info_.sortTupleDesc_.
*/
-public class SortNode extends PlanNode {
+public class SortNode extends PlanNode implements SpillableOperator {
private final static Logger LOG = LoggerFactory.getLogger(SortNode.class);
// Coefficiencts for estimating scan processing cost. Derived from benchmarking.
@@ -500,10 +500,29 @@ public class SortNode extends PlanNode {
@Override
public void computeNodeResourceProfile(TQueryOptions queryOptions) {
+ computeResourceProfileIfSpill(queryOptions, Long.MAX_VALUE);
+ }
+
+ @Override
+ public void computeResourceProfileIfSpill(
+ TQueryOptions queryOptions, long maxMemoryEstimatePerInstance) {
Preconditions.checkState(hasValidStats());
+ boolean biasToSpill = shouldComputeResourcesWithSpill(queryOptions);
+ if (!biasToSpill) {
+ Preconditions.checkArgument(maxMemoryEstimatePerInstance == Long.MAX_VALUE);
+ }
+
if (type_ == TSortType.TOPN) {
- nodeResourceProfile_ = ResourceProfile.noReservation(
- getSortInfo().estimateTopNMaterializedSize(cardinality_, offset_));
+ long memEstimate =
+ getSortInfo().estimateTopNMaterializedSize(cardinality_, offset_);
+ ResourceProfileBuilder builder =
+ new ResourceProfileBuilder().setMemEstimateBytes(memEstimate);
+ if (biasToSpill) {
+ builder.setMemEstimateScale(
+ queryOptions.getMem_estimate_scale_for_spilling_operator(),
+ maxMemoryEstimatePerInstance);
+ }
+ nodeResourceProfile_ = builder.build();
return;
}
@@ -553,10 +572,20 @@ public class SortNode extends PlanNode {
perInstanceMemEstimate = Math.min(perInstanceMemEstimate, totalHeapBytes);
}
}
- nodeResourceProfile_ = new ResourceProfileBuilder()
- .setMemEstimateBytes(perInstanceMemEstimate)
- .setMinMemReservationBytes(perInstanceMinMemReservation)
- .setSpillableBufferBytes(bufferSize).setMaxRowBufferBytes(bufferSize).build();
+ ResourceProfileBuilder builder =
+ new ResourceProfileBuilder()
+ .setMemEstimateBytes(perInstanceMemEstimate)
+ .setMinMemReservationBytes(perInstanceMinMemReservation)
+ .setSpillableBufferBytes(bufferSize)
+ .setMaxRowBufferBytes(bufferSize);
+ if (type_ != TSortType.PARTIAL && biasToSpill) {
+ // PartialSortNode use Sorter that does not support spill.
+ // Other sort node use Sorter that support spill.
+ builder.setMemEstimateScale(
+ queryOptions.getMem_estimate_scale_for_spilling_operator(),
+ maxMemoryEstimatePerInstance);
+ }
+ nodeResourceProfile_ = builder.build();
}
private static String getDisplayName(TSortType type) {
diff --git a/fe/src/main/java/org/apache/impala/planner/SpillableOperator.java b/fe/src/main/java/org/apache/impala/planner/SpillableOperator.java
new file mode 100644
index 000000000..bc81f9b99
--- /dev/null
+++ b/fe/src/main/java/org/apache/impala/planner/SpillableOperator.java
@@ -0,0 +1,39 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.impala.planner;
+
+import org.apache.impala.thrift.TQueryOptions;
+
+/**
+ * An interface for {@link PlanNode} that support spilling to disk.
+ * TODO: It is probably better to merge this functionality into PlanNode itself,
+ * but that will require major refactoring.
+ */
+public interface SpillableOperator {
+ /**
+ * Alternative for {@link PlanNode#computeNodeResourceProfile(TQueryOptions)} and {@link
+ * DataSink#computeResourceProfile(TQueryOptions)} that assumes memory estimation of
+ * fragment instance can be capped at maxMemoryEstimatePerInstance because query can
+ * spill to disk.
+ * @param queryOptions the query options to run this query.
+ * @param maxMemoryEstimatePerInstance maximum memory estimation per fragment instance
+ * in bytes. Determined by Frontend.
+ */
+ void computeResourceProfileIfSpill(
+ TQueryOptions queryOptions, long maxMemoryEstimatePerInstance);
+}
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 f1f1cfdab..9abfdb1c1 100644
--- a/fe/src/main/java/org/apache/impala/service/Frontend.java
+++ b/fe/src/main/java/org/apache/impala/service/Frontend.java
@@ -2065,7 +2065,6 @@ public class Frontend {
public static List<TExecutorGroupSet> setupThresholdsForExecutorGroupSets(
List<TExecutorGroupSet> executorGroupSets, String request_pool,
boolean default_executor_group, boolean test_replan) throws ImpalaException {
- RequestPoolService poolService = RequestPoolService.getInstance();
List<TExecutorGroupSet> result = Lists.newArrayList();
if (default_executor_group) {
@@ -2116,6 +2115,7 @@ public class Frontend {
}
// Executor groups exist in the cluster. Identify those that can be used.
+ RequestPoolService poolService = RequestPoolService.getInstance();
for (TExecutorGroupSet e : executorGroupSets) {
// If defined, request_pool can be a suffix of the group name prefix. For example
// group_set_prefix = root.queue1
@@ -2773,6 +2773,7 @@ public class Frontend {
expectedNumExecutor(planCtx.compilationState_.getGroupSet()));
analysisResult.getAnalyzer().setAvailableCoresPerNode(
Math.max(1, planCtx.compilationState_.getAvailableCoresPerNode()));
+ analysisResult.getAnalyzer().setPoolMemLimit(planCtx.compilationState_.getGroupSet());
try {
TQueryOptions queryOptions = queryCtx.client_request.query_options;
diff --git a/fe/src/main/java/org/apache/impala/util/JniRequestPoolService.java b/fe/src/main/java/org/apache/impala/util/JniRequestPoolService.java
index 844c975f0..9a1435834 100644
--- a/fe/src/main/java/org/apache/impala/util/JniRequestPoolService.java
+++ b/fe/src/main/java/org/apache/impala/util/JniRequestPoolService.java
@@ -34,7 +34,6 @@ import org.apache.impala.thrift.TPoolConfigParams;
import org.apache.impala.thrift.TPoolConfig;
import org.apache.impala.thrift.TResolveRequestPoolParams;
import org.apache.impala.thrift.TResolveRequestPoolResult;
-import org.apache.impala.thrift.TStatus;
import org.apache.thrift.TException;
import org.apache.thrift.TSerializer;
diff --git a/fe/src/main/java/org/apache/impala/util/RequestPoolService.java b/fe/src/main/java/org/apache/impala/util/RequestPoolService.java
index 9d4fc2ac2..28efaefad 100644
--- a/fe/src/main/java/org/apache/impala/util/RequestPoolService.java
+++ b/fe/src/main/java/org/apache/impala/util/RequestPoolService.java
@@ -288,6 +288,9 @@ public class RequestPoolService {
public void stop() {
Preconditions.checkState(running_.get());
stopInternal();
+
+ // For test only, unset single_instance_ if this RequestPoolService is the singleton.
+ if (single_instance_ == this) { single_instance_ = null; }
}
public boolean isRunning() { return running_.get(); }
diff --git a/fe/src/test/java/org/apache/impala/planner/PlannerTestBase.java b/fe/src/test/java/org/apache/impala/planner/PlannerTestBase.java
index e90e8e7f3..ba04c0182 100644
--- a/fe/src/test/java/org/apache/impala/planner/PlannerTestBase.java
+++ b/fe/src/test/java/org/apache/impala/planner/PlannerTestBase.java
@@ -67,10 +67,8 @@ import org.apache.impala.thrift.TPlanNode;
import org.apache.impala.thrift.TQueryCtx;
import org.apache.impala.thrift.TQueryExecRequest;
import org.apache.impala.thrift.TQueryOptions;
-import org.apache.impala.thrift.TReplicaPreference;
import org.apache.impala.thrift.TScanRangeLocationList;
import org.apache.impala.thrift.TScanRangeSpec;
-import org.apache.impala.thrift.TSlotCountStrategy;
import org.apache.impala.thrift.TTableDescriptor;
import org.apache.impala.thrift.TTableSink;
import org.apache.impala.thrift.TTupleDescriptor;
@@ -106,18 +104,21 @@ public class PlannerTestBase extends FrontendTestBase {
// Map from table ID (TTableId) to the table descriptor with that ID.
private final Map<Integer, TTableDescriptor> tableMap_ = Maps.newHashMap();
- protected static void setUpWithSize(int num_executors, int expected_num_executors)
- throws Exception {
+ protected static void setUpTestCluster(
+ int num_executors, int expected_num_executors, String exec_group_name_prefix) {
TUpdateExecutorMembershipRequest updateReq = new TUpdateExecutorMembershipRequest();
updateReq.setIp_addresses(Sets.newHashSet("127.0.0.1"));
updateReq.setHostnames(Sets.newHashSet("localhost"));
TExecutorGroupSet group_set = new TExecutorGroupSet();
group_set.curr_num_executors = num_executors;
group_set.expected_num_executors = expected_num_executors;
+ group_set.exec_group_name_prefix = exec_group_name_prefix;
updateReq.setExec_group_sets(new ArrayList<TExecutorGroupSet>());
updateReq.getExec_group_sets().add(group_set);
ExecutorMembershipSnapshot.update(updateReq);
+ }
+ protected static void setUpKuduClientAndLogDir() {
kuduClient_ = new KuduClient.KuduClientBuilder("127.0.0.1:7051").build();
String logDir = System.getenv("IMPALA_FE_TEST_LOGS_DIR");
if (logDir == null) logDir = "/tmp";
@@ -128,7 +129,8 @@ public class PlannerTestBase extends FrontendTestBase {
public static void setUp() throws Exception {
// Mimic the 3 node test mini-cluster.
// 20 is the default num_expected_executors startup flag.
- setUpWithSize(3, 20);
+ setUpTestCluster(3, 20, "");
+ setUpKuduClientAndLogDir();
}
@Before
@@ -422,18 +424,6 @@ public class PlannerTestBase extends FrontendTestBase {
return options;
}
- protected static TQueryOptions tpcdsParquetCpuCostQueryOptions() {
- return tpcdsParquetQueryOptions()
- .setCompute_processing_cost(true)
- .setMax_fragment_instances_per_node(12)
- .setReplica_preference(TReplicaPreference.REMOTE)
- .setSlot_count_strategy(TSlotCountStrategy.PLANNER_CPU_ASK)
- .setPlanner_testcase_mode(true)
- // Required so that output doesn't vary by whether scanned tables have stats &
- // numRows property or not.
- .setDisable_hdfs_num_rows_estimate(true);
- }
-
protected static Set<PlannerTestOption> tpcdsParquetTestOptions() {
return ImmutableSet.of(PlannerTestOption.EXTENDED_EXPLAIN,
PlannerTestOption.INCLUDE_RESOURCE_HEADER, PlannerTestOption.VALIDATE_RESOURCES,
diff --git a/fe/src/test/java/org/apache/impala/planner/TpcdsCpuCostPlannerTest.java b/fe/src/test/java/org/apache/impala/planner/TpcdsCpuCostPlannerTest.java
index 6ac27ee48..e498b0937 100644
--- a/fe/src/test/java/org/apache/impala/planner/TpcdsCpuCostPlannerTest.java
+++ b/fe/src/test/java/org/apache/impala/planner/TpcdsCpuCostPlannerTest.java
@@ -17,14 +17,29 @@
package org.apache.impala.planner;
+import com.google.common.collect.Sets;
+import com.google.common.io.Files;
+
import org.apache.impala.catalog.SideloadTableStats;
+import org.apache.impala.common.ByteUnits;
import org.apache.impala.common.RuntimeEnv;
+import org.apache.impala.thrift.TExecutorGroupSet;
import org.apache.impala.thrift.TQueryOptions;
+import org.apache.impala.thrift.TReplicaPreference;
+import org.apache.impala.thrift.TSlotCountStrategy;
+import org.apache.impala.thrift.TUpdateExecutorMembershipRequest;
+import org.apache.impala.util.ExecutorMembershipSnapshot;
+import org.apache.impala.util.RequestPoolService;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import java.io.File;
+import java.io.IOException;
+import java.net.URISyntaxException;
import java.nio.file.Paths;
+import java.util.ArrayList;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
@@ -39,11 +54,28 @@ import java.util.Set;
* remain unchanged.
*/
public class TpcdsCpuCostPlannerTest extends PlannerTestBase {
+ // Pool definitions and includes memory resource limits, copied to a temporary file
+ private static final String ALLOCATION_FILE = "fair-scheduler-3-groups.xml";
+
+ // Contains per-pool configurations for maximum number of running queries and queued
+ // requests.
+ private static final String LLAMA_CONFIG_FILE = "llama-site-3-groups.xml";
+
// Planner test option to run each planner test.
private static Set<PlannerTestOption> testOptions = tpcdsParquetTestOptions();
// Query option to run each planner test.
- private static TQueryOptions options = tpcdsParquetCpuCostQueryOptions();
+ private static TQueryOptions options =
+ tpcdsParquetQueryOptions()
+ .setCompute_processing_cost(true)
+ .setMax_fragment_instances_per_node(12)
+ .setReplica_preference(TReplicaPreference.REMOTE)
+ .setSlot_count_strategy(TSlotCountStrategy.PLANNER_CPU_ASK)
+ .setMem_estimate_scale_for_spilling_operator(1.0)
+ .setPlanner_testcase_mode(true)
+ // Required so that output doesn't vary by whether scanned tables have stats &
+ // numRows property or not.
+ .setDisable_hdfs_num_rows_estimate(true);
// Database name to run this test.
private static String testDb = "tpcds_partitioned_parquet_snap";
@@ -55,21 +87,54 @@ public class TpcdsCpuCostPlannerTest extends PlannerTestBase {
private static Map<String, Long> scanRangeLimit = new HashMap<String, Long>() {
{
// split a 5752989 bytes file to 10 ranges.
- put("customer", 580 * 1024L);
+ put("customer", 580 * ByteUnits.KILOBYTE);
// split a 1218792 bytes file to 10 ranges.
- put("customer_address", 125 * 1024L);
+ put("customer_address", 125 * ByteUnits.KILOBYTE);
// split a 7848768 bytes file to 10 ranges.
- put("customer_demographics", 790 * 1024L);
+ put("customer_demographics", 790 * ByteUnits.KILOBYTE);
// split a 1815300 bytes file to 4 ranges.
- put("item", 500L * 1024L);
+ put("item", 500L * ByteUnits.KILOBYTE);
}
};
+ // Temporary folder to copy admission control files into.
+ // Do not annotate with JUnit @Rule because we want to keep the tempFolder the same
+ // for entire lifetime of test class.
+ private static TemporaryFolder tempFolder;
+
+ /**
+ * Returns a {@link File} for the file on the classpath.
+ */
+ private static File getClasspathFile(String filename) throws URISyntaxException {
+ return new File(
+ TpcdsCpuCostPlannerTest.class.getClassLoader().getResource(filename).toURI());
+ }
+
+ private static void setupAdmissionControl() throws IOException, URISyntaxException {
+ // Start admission control with config file fair-scheduler-3-groups.xml
+ // and llama-site-3-groups.xml
+ tempFolder = new TemporaryFolder();
+ tempFolder.create();
+ File allocationConfFile = tempFolder.newFile(ALLOCATION_FILE);
+ Files.copy(getClasspathFile(ALLOCATION_FILE), allocationConfFile);
+
+ File llamaConfFile = tempFolder.newFile(LLAMA_CONFIG_FILE);
+ Files.copy(getClasspathFile(LLAMA_CONFIG_FILE), llamaConfFile);
+ // Intentionally mark isTest = false to cache poolService as a singleton.
+ RequestPoolService poolService =
+ RequestPoolService.getInstance(allocationConfFile.getAbsolutePath(),
+ llamaConfFile.getAbsolutePath(), /* isTest */ false);
+ poolService.start();
+ }
+
@BeforeClass
public static void setUp() throws Exception {
- // Mimic the 10 node test mini-cluster.
- // 20 is the default num_expected_executors startup flag.
- PlannerTestBase.setUpWithSize(10, 20);
+ // Mimic the 10 node test mini-cluster with admission control enabled.
+ setupAdmissionControl();
+ // Add 10 node executor group set root.large. This group set also set with
+ // impala.admission-control.max-query-mem-limit.root.large = 50GB.
+ setUpTestCluster(10, 10, "root.large");
+ setUpKuduClientAndLogDir();
Paths.get(outDir_.toString(), "tpcds_cpu_cost").toFile().mkdirs();
// Sideload stats through RuntimeEnv.
@@ -87,10 +152,13 @@ public class TpcdsCpuCostPlannerTest extends PlannerTestBase {
}
@AfterClass
- public static void unsetMetadataScale() {
+ public static void unsetMetadataScaleAndStopPoolService() {
RuntimeEnv.INSTANCE.dropSideloadStats();
RuntimeEnv.INSTANCE.dropTableScanRangeLimit();
invalidateTables();
+
+ RequestPoolService.getInstance().stop();
+ tempFolder.delete();
}
/**
@@ -653,4 +721,72 @@ public class TpcdsCpuCostPlannerTest extends PlannerTestBase {
public void testTpcdsDdlIceberg() {
runPlannerTestFile("tpcds_cpu_cost/tpcds-ddl-iceberg", testDb, options, testOptions);
}
+
+ /**
+ * Return TQueryOptions with both MEM_LIMIT_EXECUTORS and MEM_LIMIT_COORDINATORS
+ * set to 'bytes_limit'.
+ */
+ private TQueryOptions coordExecMemLimitOptions(long bytes_limit) {
+ TQueryOptions optsWithLimit = new TQueryOptions(options);
+ optsWithLimit.setMem_limit_executors(bytes_limit);
+ optsWithLimit.setMem_limit_coordinators(bytes_limit);
+ return optsWithLimit;
+ }
+
+ @Test
+ public void testSortNodeMemLimitExecutors() {
+ // Test that MEM_LIMIT_EXECUTORS & MEM_LIMIT_COORDINATORS (20GB) override
+ // impala.admission-control.max-query-mem-limit.root.large (50GB) for SortNode.
+ runPlannerTestFile("tpcds_cpu_cost/tpcds-q64-mem_limit_executors-20g", testDb,
+ coordExecMemLimitOptions(20 * ByteUnits.GIGABYTE), testOptions);
+ }
+
+ @Test
+ public void testSortNodeMemLimit() {
+ // Test that MEM_LIMIT (10GB) override MEM_LIMIT_EXECUTORS/MEM_LIMIT_COORDINATORS
+ // (20GB) and impala.admission-control.max-query-mem-limit.root.large (50GB) for
+ // SortNode.
+ TQueryOptions optsWithLimit = coordExecMemLimitOptions(20 * ByteUnits.GIGABYTE);
+ optsWithLimit.setMem_limit(10 * ByteUnits.GIGABYTE);
+ runPlannerTestFile(
+ "tpcds_cpu_cost/tpcds-q64-mem_limit-10g", testDb, optsWithLimit, testOptions);
+ }
+
+ @Test
+ public void testAggregationNodeMemLimitExecutors() {
+ // Test that MEM_LIMIT_EXECUTORS & MEM_LIMIT_COORDINATORS (20GB) override
+ // impala.admission-control.max-query-mem-limit.root.large (50GB) for AggregationNode.
+ runPlannerTestFile("tpcds_cpu_cost/tpcds-q67-mem_limit_executors-20g", testDb,
+ coordExecMemLimitOptions(20 * ByteUnits.GIGABYTE), testOptions);
+ }
+
+ @Test
+ public void testAggregationNodeMemLimit() {
+ // Test that MEM_LIMIT (10GB) override MEM_LIMIT_EXECUTORS/MEM_LIMIT_COORDINATORS
+ // (20GB) and impala.admission-control.max-query-mem-limit.root.large (50GB) for
+ // AggregationNode.
+ TQueryOptions optsWithLimit = coordExecMemLimitOptions(20 * ByteUnits.GIGABYTE);
+ optsWithLimit.setMem_limit(10 * ByteUnits.GIGABYTE);
+ runPlannerTestFile(
+ "tpcds_cpu_cost/tpcds-q67-mem_limit-10g", testDb, optsWithLimit, testOptions);
+ }
+
+ @Test
+ public void testHashJoinNodeMemLimitExecutors() {
+ // Test that MEM_LIMIT_EXECUTORS & MEM_LIMIT_COORDINATORS (20GB) override
+ // impala.admission-control.max-query-mem-limit.root.large (50GB) for HashJoinNode.
+ runPlannerTestFile("tpcds_cpu_cost/tpcds-q97-mem_limit_executors-20g", testDb,
+ coordExecMemLimitOptions(20 * ByteUnits.GIGABYTE), testOptions);
+ }
+
+ @Test
+ public void testHashJoinNodeMemLimit() {
+ // Test that MEM_LIMIT (10GB) override MEM_LIMIT_EXECUTORS/MEM_LIMIT_COORDINATORS
+ // (20GB) and impala.admission-control.max-query-mem-limit.root.large (50GB) for
+ // HashJoinNode.
+ TQueryOptions optsWithLimit = coordExecMemLimitOptions(20 * ByteUnits.GIGABYTE);
+ optsWithLimit.setMem_limit(10 * ByteUnits.GIGABYTE);
+ runPlannerTestFile(
+ "tpcds_cpu_cost/tpcds-q97-mem_limit-10g", testDb, optsWithLimit, testOptions);
+ }
}
diff --git a/fe/src/test/resources/llama-site-3-groups.xml b/fe/src/test/resources/llama-site-3-groups.xml
index 76ba67408..bd3af40eb 100644
--- a/fe/src/test/resources/llama-site-3-groups.xml
+++ b/fe/src/test/resources/llama-site-3-groups.xml
@@ -55,8 +55,8 @@
</property>
<property>
<name>impala.admission-control.max-query-mem-limit.root.large</name>
- <!-- 8 PB -->
- <value>9007199254740992</value>
+ <!-- 50 GB -->
+ <value>53687091200</value>
</property>
<property>
<name>impala.admission-control.min-query-mem-limit.root.large</name>
diff --git a/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-ddl-iceberg.test b/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-ddl-iceberg.test
index b4a6577be..d7224986b 100644
--- a/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-ddl-iceberg.test
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-ddl-iceberg.test
@@ -121,9 +121,9 @@ stored as iceberg as
select * from store_sales
---- PARALLELPLANS
Max Per-Host Resource Reservation: Memory=168.00MB Threads=24
-Per-Host Resource Estimates: Memory=78.24GB
+Per-Host Resource Estimates: Memory=50.99GB
F01:PLAN FRAGMENT [HASH(tpcds_partitioned_parquet_snap.store_sales.ss_sold_date_sk)] hosts=10 instances=120
-| Per-Instance Resources: mem-estimate=6.46GB mem-reservation=6.00MB thread-reservation=1
+| Per-Instance Resources: mem-estimate=4.19GB mem-reservation=6.00MB thread-reservation=1
| max-parallelism=120 segment-costs=[44372981821, 97047706322]
WRITE TO HDFS [tpcds_partitioned_parquet_snap.store_sales_duplicate, OVERWRITE=false, PARTITION-KEYS=(ss_sold_date_sk)]
| output exprs: ss_sold_time_sk, ss_item_sk, ss_customer_sk, ss_cdemo_sk, ss_hdemo_sk, ss_addr_sk, ss_store_sk, ss_promo_sk, ss_ticket_number, ss_quantity, ss_wholesale_cost, ss_list_price, ss_sales_price, ss_ext_discount_amt, ss_ext_sales_price, ss_ext_wholesale_cost, ss_ext_list_price, ss_ext_tax, ss_coupon_amt, ss_net_paid, ss_net_paid_inc_tax, ss_net_profit, ss_sold_date_sk
@@ -131,7 +131,7 @@ WRITE TO HDFS [tpcds_partitioned_parquet_snap.store_sales_duplicate, OVERWRITE=f
|
02:SORT
| order by: ss_sold_date_sk ASC NULLS LAST
-| mem-estimate=6.44GB mem-reservation=6.00MB spill-buffer=2.00MB thread-reservation=0
+| mem-estimate=4.17GB mem-reservation=6.00MB spill-buffer=2.00MB thread-reservation=0
| tuple-ids=2 row-size=96B cardinality=8.64G cost=39597689640
| in pipelines: 02(GETNEXT), 00(OPEN)
|
@@ -440,9 +440,9 @@ select a.*, a.ca_country as part_col
from customer_address as a, customer_address as b
---- PARALLELPLANS
Max Per-Host Resource Reservation: Memory=12.25MB Threads=3
-Per-Host Resource Estimates: Memory=4.63PB
+Per-Host Resource Estimates: Memory=50.03GB
F00:PLAN FRAGMENT [RANDOM] hosts=10 instances=10
-| Per-Instance Resources: mem-estimate=4.63PB mem-reservation=12.12MB thread-reservation=1
+| Per-Instance Resources: mem-estimate=50.02GB mem-reservation=12.12MB thread-reservation=1
| max-parallelism=10 segment-costs=[2393532728547225, 2622165786501232] cpu-comparison-result=20 [max(10 (self) vs 20 (sum children))]
WRITE TO HDFS [tpcds_partitioned_parquet_snap.customer_address_1_huge_part, OVERWRITE=false, PARTITION-KEYS=(ca_country)]
| output exprs: ca_address_sk, ca_address_id, ca_street_number, ca_street_name, ca_street_type, ca_suite_number, ca_city, ca_county, ca_state, ca_zip, ca_country, ca_gmt_offset, ca_location_type, ca_country
@@ -450,7 +450,7 @@ WRITE TO HDFS [tpcds_partitioned_parquet_snap.customer_address_1_huge_part, OVER
|
04:SORT
| order by: ca_country ASC NULLS LAST
-| mem-estimate=4.63PB mem-reservation=12.00MB spill-buffer=2.00MB thread-reservation=0
+| mem-estimate=50.00GB mem-reservation=12.00MB spill-buffer=2.00MB thread-reservation=0
| tuple-ids=4 row-size=232B cardinality=225.00T cost=2358455207032019
| in pipelines: 04(GETNEXT), 00(OPEN)
|
@@ -567,9 +567,9 @@ select a.*, (a.ss_sold_date_sk % 4) as part_col
from store_sales a, store_sales b;
---- PARALLELPLANS
Max Per-Host Resource Reservation: Memory=168.25MB Threads=27
-Per-Host Resource Estimates: Memory=8192.00PB
+Per-Host Resource Estimates: Memory=51.05GB
F02:PLAN FRAGMENT [HASH((a.ss_sold_date_sk % 4))] hosts=10 instances=120
-| Per-Instance Resources: mem-estimate=6553.60PB mem-reservation=6.00MB thread-reservation=1
+| Per-Instance Resources: mem-estimate=4.19GB mem-reservation=6.00MB thread-reservation=1
| max-parallelism=120 segment-costs=[9223372036854775807, 1079134528315963008] cpu-comparison-result=240 [max(240 (self) vs 30 (sum children))]
WRITE TO HDFS [tpcds_partitioned_parquet_snap.store_sales_4_huge_part, OVERWRITE=false, PARTITION-KEYS=((ss_sold_date_sk % 4))]
| output exprs: ss_sold_time_sk, ss_item_sk, ss_customer_sk, ss_cdemo_sk, ss_hdemo_sk, ss_addr_sk, ss_store_sk, ss_promo_sk, ss_ticket_number, ss_quantity, ss_wholesale_cost, ss_list_price, ss_sales_price, ss_ext_discount_amt, ss_ext_sales_price, ss_ext_wholesale_cost, ss_ext_list_price, ss_ext_tax, ss_coupon_amt, ss_net_paid, ss_net_paid_inc_tax, ss_net_profit, ss_sold_date_sk, (ss_sold_date_sk % CAST(4 AS INT))
@@ -577,7 +577,7 @@ WRITE TO HDFS [tpcds_partitioned_parquet_snap.store_sales_4_huge_part, OVERWRITE
|
05:SORT
| order by: (ss_sold_date_sk % 4) ASC NULLS LAST
-| mem-estimate=6553.60PB mem-reservation=6.00MB spill-buffer=2.00MB thread-reservation=0
+| mem-estimate=4.17GB mem-reservation=6.00MB spill-buffer=2.00MB thread-reservation=0
| tuple-ids=4 row-size=96B cardinality=9223372.04T cost=9223372036854775807
| in pipelines: 05(GETNEXT), 00(OPEN)
|
@@ -641,9 +641,9 @@ select a.*, a.cs_ship_mode_sk as part_col
from catalog_sales a, catalog_sales b;
---- PARALLELPLANS
Max Per-Host Resource Reservation: Memory=24.12MB Threads=16
-Per-Host Resource Estimates: Memory=8192.00PB
+Per-Host Resource Estimates: Memory=50.39GB
F02:PLAN FRAGMENT [HASH(a.cs_ship_mode_sk)] hosts=10 instances=20
-| Per-Instance Resources: mem-estimate=8192.00PB mem-reservation=6.00MB thread-reservation=1
+| Per-Instance Resources: mem-estimate=25.03GB mem-reservation=6.00MB thread-reservation=1
| max-parallelism=20 segment-costs=[9223372036854775807, 1079134528315963008] cpu-comparison-result=140 [max(140 (self) vs 20 (sum children))]
WRITE TO HDFS [tpcds_partitioned_parquet_snap.catalog_sales_20_huge_part, OVERWRITE=false, PARTITION-KEYS=(cs_ship_mode_sk)]
| output exprs: cs_sold_time_sk, cs_ship_date_sk, cs_bill_customer_sk, cs_bill_cdemo_sk, cs_bill_hdemo_sk, cs_bill_addr_sk, cs_ship_customer_sk, cs_ship_cdemo_sk, cs_ship_hdemo_sk, cs_ship_addr_sk, cs_call_center_sk, cs_catalog_page_sk, cs_ship_mode_sk, cs_warehouse_sk, cs_item_sk, cs_promo_sk, cs_order_number, cs_quantity, cs_wholesale_cost, cs_list_price, cs_sales_price, cs_ext_discount_amt, cs_ext_sales_price, cs_ext_wholesale_cost, cs_ext_list_price, cs_ext_tax, cs_coupon_amt, cs_ex [...]
@@ -651,7 +651,7 @@ WRITE TO HDFS [tpcds_partitioned_parquet_snap.catalog_sales_20_huge_part, OVERWR
|
05:SORT
| order by: cs_ship_mode_sk ASC NULLS LAST
-| mem-estimate=8192.00PB mem-reservation=6.00MB spill-buffer=2.00MB thread-reservation=0
+| mem-estimate=25.00GB mem-reservation=6.00MB spill-buffer=2.00MB thread-reservation=0
| tuple-ids=4 row-size=140B cardinality=9223372.04T cost=9223372036854775807
| in pipelines: 05(GETNEXT), 00(OPEN)
|
diff --git a/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-ddl-parquet.test b/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-ddl-parquet.test
index e92113914..95b3ddb47 100644
--- a/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-ddl-parquet.test
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-ddl-parquet.test
@@ -125,9 +125,9 @@ stored as parquet as
select * from store_sales
---- PARALLELPLANS
Max Per-Host Resource Reservation: Memory=168.00MB Threads=24
-Per-Host Resource Estimates: Memory=89.98GB
+Per-Host Resource Estimates: Memory=62.74GB
F01:PLAN FRAGMENT [HASH(tpcds_partitioned_parquet_snap.store_sales.ss_sold_date_sk)] hosts=10 instances=120
-| Per-Instance Resources: mem-estimate=7.44GB mem-reservation=6.00MB thread-reservation=1
+| Per-Instance Resources: mem-estimate=5.17GB mem-reservation=6.00MB thread-reservation=1
| max-parallelism=120 segment-costs=[44372981821, 97047706322]
WRITE TO HDFS [tpcds_partitioned_parquet_snap.store_sales_duplicate, OVERWRITE=false, PARTITION-KEYS=(ss_sold_date_sk)]
| partitions=1824
@@ -136,7 +136,7 @@ WRITE TO HDFS [tpcds_partitioned_parquet_snap.store_sales_duplicate, OVERWRITE=f
|
02:SORT
| order by: ss_sold_date_sk ASC NULLS LAST
-| mem-estimate=6.44GB mem-reservation=6.00MB spill-buffer=2.00MB thread-reservation=0
+| mem-estimate=4.17GB mem-reservation=6.00MB spill-buffer=2.00MB thread-reservation=0
| tuple-ids=2 row-size=96B cardinality=8.64G cost=39597689640
| in pipelines: 02(GETNEXT), 00(OPEN)
|
@@ -451,9 +451,9 @@ select a.*, a.ca_country as part_col
from customer_address as a, customer_address as b
---- PARALLELPLANS
Max Per-Host Resource Reservation: Memory=12.25MB Threads=3
-Per-Host Resource Estimates: Memory=4.63PB
+Per-Host Resource Estimates: Memory=51.02GB
F00:PLAN FRAGMENT [RANDOM] hosts=10 instances=10
-| Per-Instance Resources: mem-estimate=4.63PB mem-reservation=12.12MB thread-reservation=1
+| Per-Instance Resources: mem-estimate=51.00GB mem-reservation=12.12MB thread-reservation=1
| max-parallelism=10 segment-costs=[2393532728547225, 2622165786501232] cpu-comparison-result=20 [max(10 (self) vs 20 (sum children))]
WRITE TO HDFS [tpcds_partitioned_parquet_snap.customer_address_1_huge_part, OVERWRITE=false, PARTITION-KEYS=(ca_country)]
| partitions=1
@@ -462,7 +462,7 @@ WRITE TO HDFS [tpcds_partitioned_parquet_snap.customer_address_1_huge_part, OVER
|
04:SORT
| order by: ca_country ASC NULLS LAST
-| mem-estimate=4.63PB mem-reservation=12.00MB spill-buffer=2.00MB thread-reservation=0
+| mem-estimate=50.00GB mem-reservation=12.00MB spill-buffer=2.00MB thread-reservation=0
| tuple-ids=4 row-size=232B cardinality=225.00T cost=2358455207032019
| in pipelines: 04(GETNEXT), 00(OPEN)
|
@@ -580,9 +580,9 @@ select a.*, (a.ss_sold_date_sk % 4) as part_col
from store_sales a, store_sales b;
---- PARALLELPLANS
Max Per-Host Resource Reservation: Memory=168.25MB Threads=27
-Per-Host Resource Estimates: Memory=8192.00PB
+Per-Host Resource Estimates: Memory=62.79GB
F02:PLAN FRAGMENT [HASH((a.ss_sold_date_sk % 4))] hosts=10 instances=120
-| Per-Instance Resources: mem-estimate=6553.60PB mem-reservation=6.00MB thread-reservation=1
+| Per-Instance Resources: mem-estimate=5.17GB mem-reservation=6.00MB thread-reservation=1
| max-parallelism=120 segment-costs=[9223372036854775807, 1079134528315963008] cpu-comparison-result=240 [max(240 (self) vs 30 (sum children))]
WRITE TO HDFS [tpcds_partitioned_parquet_snap.store_sales_4_huge_part, OVERWRITE=false, PARTITION-KEYS=((ss_sold_date_sk % 4))]
| partitions=1824
@@ -591,7 +591,7 @@ WRITE TO HDFS [tpcds_partitioned_parquet_snap.store_sales_4_huge_part, OVERWRITE
|
05:SORT
| order by: (ss_sold_date_sk % 4) ASC NULLS LAST
-| mem-estimate=6553.60PB mem-reservation=6.00MB spill-buffer=2.00MB thread-reservation=0
+| mem-estimate=4.17GB mem-reservation=6.00MB spill-buffer=2.00MB thread-reservation=0
| tuple-ids=4 row-size=96B cardinality=9223372.04T cost=9223372036854775807
| in pipelines: 05(GETNEXT), 00(OPEN)
|
@@ -655,9 +655,9 @@ select a.*, a.cs_ship_mode_sk as part_col
from catalog_sales a, catalog_sales b;
---- PARALLELPLANS
Max Per-Host Resource Reservation: Memory=24.12MB Threads=16
-Per-Host Resource Estimates: Memory=8192.00PB
+Per-Host Resource Estimates: Memory=52.34GB
F02:PLAN FRAGMENT [HASH(a.cs_ship_mode_sk)] hosts=10 instances=20
-| Per-Instance Resources: mem-estimate=8192.00PB mem-reservation=6.00MB thread-reservation=1
+| Per-Instance Resources: mem-estimate=26.00GB mem-reservation=6.00MB thread-reservation=1
| max-parallelism=20 segment-costs=[9223372036854775807, 1079134528315963008] cpu-comparison-result=140 [max(140 (self) vs 20 (sum children))]
WRITE TO HDFS [tpcds_partitioned_parquet_snap.catalog_sales_20_huge_part, OVERWRITE=false, PARTITION-KEYS=(cs_ship_mode_sk)]
| partitions=20
@@ -666,7 +666,7 @@ WRITE TO HDFS [tpcds_partitioned_parquet_snap.catalog_sales_20_huge_part, OVERWR
|
05:SORT
| order by: cs_ship_mode_sk ASC NULLS LAST
-| mem-estimate=8192.00PB mem-reservation=6.00MB spill-buffer=2.00MB thread-reservation=0
+| mem-estimate=25.00GB mem-reservation=6.00MB spill-buffer=2.00MB thread-reservation=0
| tuple-ids=4 row-size=140B cardinality=9223372.04T cost=9223372036854775807
| in pipelines: 05(GETNEXT), 00(OPEN)
|
diff --git a/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q22.test b/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q22.test
index a55f19473..61cb1ed81 100644
--- a/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q22.test
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q22.test
@@ -19,9 +19,9 @@ order by qoh, i_product_name, i_brand, i_class, i_category
limit 100;
---- PLAN
Max Per-Host Resource Reservation: Memory=176.12MB Threads=1
-Per-Host Resource Estimates: Memory=217.41GB
+Per-Host Resource Estimates: Memory=50.10GB
F00:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1
-| Per-Instance Resources: mem-estimate=217.41GB mem-reservation=176.12MB thread-reservation=1 runtime-filters-memory=2.00MB
+| Per-Instance Resources: mem-estimate=50.10GB mem-reservation=176.12MB thread-reservation=1 runtime-filters-memory=2.00MB
| max-parallelism=1 segment-costs=[8187645470, 8499749, 16608995, 500]
PLAN-ROOT SINK
| output exprs: CASE valid_tid(4,6,8,10,12) WHEN 4 THEN i_product_name WHEN 6 THEN i_product_name WHEN 8 THEN i_product_name WHEN 10 THEN i_product_name WHEN 12 THEN NULL END, CASE valid_tid(4,6,8,10,12) WHEN 4 THEN i_brand WHEN 6 THEN i_brand WHEN 8 THEN i_brand WHEN 10 THEN NULL WHEN 12 THEN NULL END, CASE valid_tid(4,6,8,10,12) WHEN 4 THEN i_class WHEN 6 THEN i_class WHEN 8 THEN NULL WHEN 10 THEN NULL WHEN 12 THEN NULL END, CASE valid_tid(4,6,8,10,12) WHEN 4 THEN i_category WHEN 6 TH [...]
@@ -56,7 +56,7 @@ PLAN-ROOT SINK
| Class 4
| output: avg(CAST(inv_quantity_on_hand AS BIGINT))
| group by: NULL, NULL, NULL, NULL
-| mem-estimate=217.31GB mem-reservation=137.94MB thread-reservation=0
+| mem-estimate=50.00GB mem-reservation=137.94MB thread-reservation=0
| tuple-ids=3N,5N,7N,9N,11N row-size=441B cardinality=1.44M cost=7104046265
| in pipelines: 05(GETNEXT), 00(OPEN)
|
diff --git a/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q23a.test b/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q23a.test
index 59cc6ce67..7d52fa9c0 100644
--- a/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q23a.test
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q23a.test
@@ -51,9 +51,9 @@ from
limit 100;
---- PLAN
Max Per-Host Resource Reservation: Memory=301.31MB Threads=1
-Per-Host Resource Estimates: Memory=69.83GB
+Per-Host Resource Estimates: Memory=53.71GB
F00:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1
-| Per-Instance Resources: mem-estimate=69.83GB mem-reservation=301.31MB thread-reservation=1 runtime-filters-memory=70.00MB
+| Per-Instance Resources: mem-estimate=53.71GB mem-reservation=301.31MB thread-reservation=1 runtime-filters-memory=70.00MB
| max-parallelism=1 segment-costs=[48663212322, 207415777, 7394305285, 9711133854, 2100000, 48663212322, 207415777, 7394305285, 9711133854, 2100000, 722596118, 0]
PLAN-ROOT SINK
| output exprs: sum(sales)
@@ -202,7 +202,7 @@ PLAN-ROOT SINK
| | | output: count(*)
| | | group by: substr(i_item_desc, CAST(1 AS BIGINT), CAST(30 AS BIGINT)), i_item_sk, d_date
| | | having: count(*) > CAST(4 AS BIGINT)
-| | | mem-estimate=66.12GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| | | mem-estimate=50.00GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
| | | tuple-ids=27 row-size=28B cardinality=705.90M cost=41666743099
| | | in pipelines: 32(GETNEXT), 27(OPEN)
| | |
@@ -421,7 +421,7 @@ PLAN-ROOT SINK
| | output: count(*)
| | group by: substr(i_item_desc, CAST(1 AS BIGINT), CAST(30 AS BIGINT)), i_item_sk, d_date
| | having: count(*) > CAST(4 AS BIGINT)
-| | mem-estimate=66.12GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| | mem-estimate=50.00GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
| | tuple-ids=5 row-size=28B cardinality=705.90M cost=41666743099
| | in pipelines: 08(GETNEXT), 03(OPEN)
| |
diff --git a/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q23b.test b/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q23b.test
index 62589b88d..04205fb76 100644
--- a/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q23b.test
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q23b.test
@@ -58,9 +58,9 @@ with frequent_ss_items as
# end query 23 in stream 0 using template query23.tpl
---- PLAN
Max Per-Host Resource Reservation: Memory=336.31MB Threads=1
-Per-Host Resource Estimates: Memory=71.71GB
+Per-Host Resource Estimates: Memory=55.60GB
F00:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1
-| Per-Instance Resources: mem-estimate=71.71GB mem-reservation=336.31MB thread-reservation=1 runtime-filters-memory=70.00MB
+| Per-Instance Resources: mem-estimate=55.60GB mem-reservation=336.31MB thread-reservation=1 runtime-filters-memory=70.00MB
| max-parallelism=1 segment-costs=[48663212322, 207415777, 7394305285, 9711133854, 2100000, 975536243, 48663212322, 207415777, 7394305285, 9711133854, 2100000, 583723564, 721685539, 300]
PLAN-ROOT SINK
| output exprs: c_last_name, c_first_name, sales
@@ -216,7 +216,7 @@ PLAN-ROOT SINK
| | | output: count(*)
| | | group by: substr(i_item_desc, CAST(1 AS BIGINT), CAST(30 AS BIGINT)), i_item_sk, d_date
| | | having: count(*) > CAST(4 AS BIGINT)
-| | | mem-estimate=66.12GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| | | mem-estimate=50.00GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
| | | tuple-ids=30 row-size=28B cardinality=705.90M cost=41666743099
| | | in pipelines: 36(GETNEXT), 31(OPEN)
| | |
@@ -461,7 +461,7 @@ PLAN-ROOT SINK
| | output: count(*)
| | group by: substr(i_item_desc, CAST(1 AS BIGINT), CAST(30 AS BIGINT)), i_item_sk, d_date
| | having: count(*) > CAST(4 AS BIGINT)
-| | mem-estimate=66.12GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| | mem-estimate=50.00GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
| | tuple-ids=6 row-size=28B cardinality=705.90M cost=41666743099
| | in pipelines: 09(GETNEXT), 04(OPEN)
| |
diff --git a/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q38.test b/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q38.test
index 58dcec7f3..9417319e9 100644
--- a/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q38.test
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q38.test
@@ -24,9 +24,9 @@ limit 100;
# end query 38 in stream 0 using template query38.tpl
---- PLAN
Max Per-Host Resource Reservation: Memory=269.94MB Threads=1
-Per-Host Resource Estimates: Memory=536.66GB
+Per-Host Resource Estimates: Memory=224.72GB
F00:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1
-| Per-Instance Resources: mem-estimate=536.66GB mem-reservation=269.94MB thread-reservation=1 runtime-filters-memory=55.00MB
+| Per-Instance Resources: mem-estimate=224.72GB mem-reservation=269.94MB thread-reservation=1 runtime-filters-memory=55.00MB
| max-parallelism=1 segment-costs=[54790085460, 27414203232, 13758375736, 13451593165, 0]
PLAN-ROOT SINK
| output exprs: count(*)
@@ -43,7 +43,7 @@ PLAN-ROOT SINK
| hash predicates: d_date IS NOT DISTINCT FROM d_date, c_first_name IS NOT DISTINCT FROM c_first_name, c_last_name IS NOT DISTINCT FROM c_last_name
| fk/pk conjuncts: assumed fk/pk
| runtime filters: RF000[bloom] <- d_date, RF001[bloom] <- c_first_name, RF002[bloom] <- c_last_name
-| mem-estimate=121.99GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| mem-estimate=50.00GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
| tuple-ids=3,8,13 row-size=120B cardinality=7.93G cost=5459196418
| in pipelines: 05(GETNEXT), 17(OPEN)
|
@@ -107,7 +107,7 @@ PLAN-ROOT SINK
| hash predicates: d_date IS NOT DISTINCT FROM d_date, c_first_name IS NOT DISTINCT FROM c_first_name, c_last_name IS NOT DISTINCT FROM c_last_name
| fk/pk conjuncts: assumed fk/pk
| runtime filters: RF003[bloom] <- d_date, RF004[bloom] <- c_first_name, RF005[bloom] <- c_last_name
-| mem-estimate=243.61GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| mem-estimate=50.00GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
| tuple-ids=3,8 row-size=80B cardinality=7.93G cost=7437210998
| in pipelines: 05(GETNEXT), 11(OPEN)
|
@@ -169,7 +169,7 @@ PLAN-ROOT SINK
|
05:AGGREGATE [FINALIZE]
| group by: c_last_name, c_first_name, d_date
-| mem-estimate=96.34GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| mem-estimate=50.00GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
| tuple-ids=3 row-size=40B cardinality=7.93G cost=46814848608
| in pipelines: 05(GETNEXT), 00(OPEN)
|
diff --git a/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q51.test b/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q51.test
index 683b856cf..cba243dc6 100644
--- a/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q51.test
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q51.test
@@ -46,9 +46,9 @@ limit 100;
# end query 51 in stream 0 using template query51.tpl
---- PLAN
Max Per-Host Resource Reservation: Memory=97.94MB Threads=1
-Per-Host Resource Estimates: Memory=3.68PB
+Per-Host Resource Estimates: Memory=198.29GB
F00:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1
-| Per-Instance Resources: mem-estimate=3.68PB mem-reservation=97.94MB thread-reservation=1 runtime-filters-memory=2.00MB
+| Per-Instance Resources: mem-estimate=198.29GB mem-reservation=97.94MB thread-reservation=1 runtime-filters-memory=2.00MB
| max-parallelism=1 segment-costs=[56515568116, 12537410622, 14128806498, 3134329462, 210805767343546, 278820518032917, 600]
PLAN-ROOT SINK
| output exprs: item_sk, d_date, web_sales, store_sales, web_cumulative, store_cumulative
@@ -77,14 +77,14 @@ PLAN-ROOT SINK
|
13:SORT
| order by: CASE WHEN ws_item_sk IS NOT NULL THEN ws_item_sk ELSE ss_item_sk END ASC NULLS LAST, CASE WHEN d_date IS NOT NULL THEN d_date ELSE d_date END ASC
-| mem-estimate=3.68PB mem-reservation=6.00MB spill-buffer=2.00MB thread-reservation=0
+| mem-estimate=50.00GB mem-reservation=6.00MB spill-buffer=2.00MB thread-reservation=0
| tuple-ids=25 row-size=80B cardinality=51.81T cost=201402086378006
| in pipelines: 13(GETNEXT), 10(OPEN)
|
12:HASH JOIN [FULL OUTER JOIN]
| hash predicates: d_date = d_date, ss_item_sk = ws_item_sk
| fk/pk conjuncts: none
-| mem-estimate=128.47GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| mem-estimate=50.00GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
| tuple-ids=22N,21N,17N,16N row-size=80B cardinality=51.81T cost=9392881062052
| in pipelines: 10(GETNEXT), 04(OPEN)
|
@@ -156,14 +156,14 @@ PLAN-ROOT SINK
|
10:SORT
| order by: ss_item_sk ASC NULLS LAST, d_date ASC
-| mem-estimate=193.12GB mem-reservation=6.00MB spill-buffer=2.00MB thread-reservation=0
+| mem-estimate=50.00GB mem-reservation=6.00MB spill-buffer=2.00MB thread-reservation=0
| tuple-ids=22 row-size=24B cardinality=8.64G cost=12537410622
| in pipelines: 10(GETNEXT), 09(OPEN)
|
09:AGGREGATE [FINALIZE]
| output: sum(ss_sales_price)
| group by: ss_item_sk, d_date
-| mem-estimate=145.09GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| mem-estimate=50.00GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
| tuple-ids=7 row-size=24B cardinality=8.64G cost=50998081470
| in pipelines: 09(GETNEXT), 06(OPEN)
|
@@ -203,7 +203,7 @@ PLAN-ROOT SINK
in pipelines: 06(GETNEXT)
---- DISTRIBUTEDPLAN
Max Per-Host Resource Reservation: Memory=2.47GB Threads=87
-Per-Host Resource Estimates: Memory=377.07TB
+Per-Host Resource Estimates: Memory=132.30GB
F08:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1
| Per-Instance Resources: mem-estimate=4.73MB mem-reservation=4.00MB thread-reservation=1
| max-parallelism=1 segment-costs=[639] cpu-comparison-result=240 [max(1 (self) vs 240 (sum children))]
@@ -219,7 +219,7 @@ PLAN-ROOT SINK
| in pipelines: 16(GETNEXT)
|
F07:PLAN FRAGMENT [HASH(CASE WHEN ws_item_sk IS NOT NULL THEN ws_item_sk ELSE ss_item_sk END)] hosts=10 instances=100 (adjusted from 120)
-Per-Instance Resources: mem-estimate=37.70TB mem-reservation=10.00MB thread-reservation=1
+Per-Instance Resources: mem-estimate=5.02GB mem-reservation=10.00MB thread-reservation=1
max-parallelism=27882060 segment-costs=[231945815232544, 278820518032917, 220] cpu-comparison-result=240 [max(220 (self) vs 240 (sum children))]
16:TOP-N [LIMIT=100]
| order by: item_sk ASC, d_date ASC
@@ -244,7 +244,7 @@ max-parallelism=27882060 segment-costs=[231945815232544, 278820518032917, 220] c
|
13:SORT
| order by: CASE WHEN ws_item_sk IS NOT NULL THEN ws_item_sk ELSE ss_item_sk END ASC NULLS LAST, CASE WHEN d_date IS NOT NULL THEN d_date ELSE d_date END ASC
-| mem-estimate=37.70TB mem-reservation=6.00MB spill-buffer=2.00MB thread-reservation=0
+| mem-estimate=5.00GB mem-reservation=6.00MB spill-buffer=2.00MB thread-reservation=0
| tuple-ids=25 row-size=80B cardinality=51.81T cost=201402086378006
| in pipelines: 13(GETNEXT), 10(OPEN)
|
@@ -469,7 +469,7 @@ max-parallelism=1824 segment-costs=[56515544083, 17805178445] cpu-comparison-res
in pipelines: 06(GETNEXT)
---- PARALLELPLANS
Max Per-Host Resource Reservation: Memory=2.47GB Threads=87
-Per-Host Resource Estimates: Memory=377.07TB
+Per-Host Resource Estimates: Memory=132.30GB
F08:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1
| Per-Instance Resources: mem-estimate=4.73MB mem-reservation=4.00MB thread-reservation=1
| max-parallelism=1 segment-costs=[639] cpu-comparison-result=240 [max(1 (self) vs 240 (sum children))]
@@ -485,7 +485,7 @@ PLAN-ROOT SINK
| in pipelines: 16(GETNEXT)
|
F07:PLAN FRAGMENT [HASH(CASE WHEN ws_item_sk IS NOT NULL THEN ws_item_sk ELSE ss_item_sk END)] hosts=10 instances=100 (adjusted from 120)
-Per-Instance Resources: mem-estimate=37.70TB mem-reservation=10.00MB thread-reservation=1
+Per-Instance Resources: mem-estimate=5.02GB mem-reservation=10.00MB thread-reservation=1
max-parallelism=27882060 segment-costs=[231945815232544, 278820518032917, 220] cpu-comparison-result=240 [max(220 (self) vs 240 (sum children))]
16:TOP-N [LIMIT=100]
| order by: item_sk ASC, d_date ASC
@@ -510,7 +510,7 @@ max-parallelism=27882060 segment-costs=[231945815232544, 278820518032917, 220] c
|
13:SORT
| order by: CASE WHEN ws_item_sk IS NOT NULL THEN ws_item_sk ELSE ss_item_sk END ASC NULLS LAST, CASE WHEN d_date IS NOT NULL THEN d_date ELSE d_date END ASC
-| mem-estimate=37.70TB mem-reservation=6.00MB spill-buffer=2.00MB thread-reservation=0
+| mem-estimate=5.00GB mem-reservation=6.00MB spill-buffer=2.00MB thread-reservation=0
| tuple-ids=25 row-size=80B cardinality=51.81T cost=201402086378006
| in pipelines: 13(GETNEXT), 10(OPEN)
|
diff --git a/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q57.test b/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q57.test
index 37b069b29..15050bb98 100644
--- a/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q57.test
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q57.test
@@ -49,9 +49,9 @@ with v1 as(
# end query 57 in stream 0 using template query57.tpl
---- PLAN
Max Per-Host Resource Reservation: Memory=243.62MB Threads=1
-Per-Host Resource Estimates: Memory=192.76GB
+Per-Host Resource Estimates: Memory=189.05GB
F00:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1
-| Per-Instance Resources: mem-estimate=192.76GB mem-reservation=243.62MB thread-reservation=1 runtime-filters-memory=9.00MB
+| Per-Instance Resources: mem-estimate=189.05GB mem-reservation=243.62MB thread-reservation=1 runtime-filters-memory=9.00MB
| max-parallelism=1 segment-costs=[4679833259, 1710158229, 4679833259, 1710158229, 2214045214, 4679833259, 1710158229, 5732824810256569220, 900]
PLAN-ROOT SINK
| output exprs: i_category, i_brand, cc_name, d_year, d_moy, avg_monthly_sales, sum_sales, psum, nsum
@@ -66,7 +66,7 @@ PLAN-ROOT SINK
34:HASH JOIN [INNER JOIN]
| hash predicates: rank() = rank() - 1, i_category = i_category, cc_name = cc_name, i_brand = i_brand
| fk/pk conjuncts: none
-| mem-estimate=53.71GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| mem-estimate=50.00GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
| tuple-ids=44,43,36,31,52,51 row-size=327B cardinality=176488.93T cost=31979878532739128
| in pipelines: 21(GETNEXT), 31(OPEN)
|
diff --git a/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q64.test b/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q64-mem_limit-10g.test
similarity index 99%
copy from testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q64.test
copy to testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q64-mem_limit-10g.test
index 01b79303b..584897dcc 100644
--- a/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q64.test
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q64-mem_limit-10g.test
@@ -122,9 +122,9 @@ order by cs1.product_name
# end query 64 in stream 0 using template query64.tpl
---- PLAN
Max Per-Host Resource Reservation: Memory=676.89MB Threads=1
-Per-Host Resource Estimates: Memory=524.91GB
+Per-Host Resource Estimates: Memory=39.31GB
F00:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1
-| Per-Instance Resources: mem-estimate=524.91GB mem-reservation=676.89MB thread-reservation=1 runtime-filters-memory=55.00MB
+| Per-Instance Resources: mem-estimate=39.31GB mem-reservation=676.89MB thread-reservation=1 runtime-filters-memory=55.00MB
| max-parallelism=1 segment-costs=[3995470666, 7240810714, 3995470666, 7240810714, 14180818211, 18143864832]
PLAN-ROOT SINK
| output exprs: product_name, store_name, store_zip, b_street_number, b_street_name, b_city, b_zip, c_street_number, c_street_name, c_city, c_zip, syear, cnt, s1, s2, s3, s1, s2, s3, syear, cnt
@@ -132,7 +132,7 @@ PLAN-ROOT SINK
|
79:SORT
| order by: product_name ASC, store_name ASC, cnt ASC, s1 ASC, s1 ASC
-| mem-estimate=269.48GB mem-reservation=12.00MB spill-buffer=2.00MB thread-reservation=0
+| mem-estimate=10.00GB mem-reservation=12.00MB spill-buffer=2.00MB thread-reservation=0
| tuple-ids=46 row-size=335B cardinality=863.99M cost=12938654601
| in pipelines: 79(GETNEXT), 38(OPEN)
|
@@ -141,7 +141,7 @@ PLAN-ROOT SINK
| fk/pk conjuncts: i_item_sk = i_item_sk, s_store_name = s_store_name, s_zip = s_zip
| other predicates: count(*) <= count(*)
| runtime filters: RF003[min_max] <- i_item_sk, RF004[min_max] <- s_store_name, RF005[min_max] <- s_zip
-| mem-estimate=254.86GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| mem-estimate=10.00GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
| tuple-ids=21,44 row-size=574B cardinality=863.99M cost=1242163610
| in pipelines: 38(GETNEXT), 77(OPEN)
|
@@ -172,7 +172,7 @@ PLAN-ROOT SINK
| | | hash predicates: cs_item_sk = cr_item_sk, cs_order_number = cr_order_number
| | | fk/pk conjuncts: cs_item_sk = cr_item_sk, cs_order_number = cr_order_number
| | | runtime filters: RF084[min_max] <- cr_item_sk, RF085[min_max] <- cr_order_number
-| | | mem-estimate=21.66GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| | | mem-estimate=10.00GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
| | | tuple-ids=25,26 row-size=40B cardinality=2.56G cost=1922201419
| | | in pipelines: 41(GETNEXT), 42(OPEN)
| | |
@@ -539,7 +539,7 @@ PLAN-ROOT SINK
| | hash predicates: cs_item_sk = cr_item_sk, cs_order_number = cr_order_number
| | fk/pk conjuncts: cs_item_sk = cr_item_sk, cs_order_number = cr_order_number
| | runtime filters: RF042[bloom] <- cr_item_sk, RF043[bloom] <- cr_order_number, RF044[min_max] <- cr_item_sk, RF045[min_max] <- cr_order_number
-| | mem-estimate=21.66GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| | mem-estimate=10.00GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
| | tuple-ids=2,3 row-size=40B cardinality=2.56G cost=1922201419
| | in pipelines: 02(GETNEXT), 03(OPEN)
| |
@@ -882,7 +882,7 @@ PLAN-ROOT SINK
in pipelines: 01(GETNEXT)
---- DISTRIBUTEDPLAN
Max Per-Host Resource Reservation: Memory=6.10GB Threads=325
-Per-Host Resource Estimates: Memory=101.04GB
+Per-Host Resource Estimates: Memory=68.61GB
F53:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1
| Per-Instance Resources: mem-estimate=1.21GB mem-reservation=4.00MB thread-reservation=1
| max-parallelism=1 segment-costs=[18941077893] cpu-comparison-result=1516 [max(1 (self) vs 1516 (sum children))]
@@ -897,11 +897,11 @@ PLAN-ROOT SINK
| in pipelines: 79(GETNEXT)
|
F52:PLAN FRAGMENT [HASH(i_item_sk,s_store_name,s_zip)] hosts=10 instances=110 (adjusted from 120)
-Per-Instance Resources: mem-estimate=2.49GB mem-reservation=12.00MB thread-reservation=1
+Per-Instance Resources: mem-estimate=975.00MB mem-reservation=12.00MB thread-reservation=1
max-parallelism=1460 segment-costs=[14553075383, 7852642112] cpu-comparison-result=1516 [max(110 (self) vs 1516 (sum children))]
79:SORT
| order by: product_name ASC, store_name ASC, cnt ASC, s1 ASC, s1 ASC
-| mem-estimate=2.45GB mem-reservation=12.00MB spill-buffer=2.00MB thread-reservation=0
+| mem-estimate=930.91MB mem-reservation=12.00MB spill-buffer=2.00MB thread-reservation=0
| tuple-ids=46 row-size=335B cardinality=863.99M cost=12938654601
| in pipelines: 79(GETNEXT), 106(OPEN)
|
@@ -915,13 +915,13 @@ max-parallelism=1460 segment-costs=[14553075383, 7852642112] cpu-comparison-resu
| in pipelines: 106(GETNEXT), 133(OPEN)
|
|--F54:PLAN FRAGMENT [HASH(i_item_sk,s_store_name,s_zip)] hosts=10 instances=110 (adjusted from 120)
-| | Per-Instance Resources: mem-estimate=2.36GB mem-reservation=34.00MB thread-reservation=1
+| | Per-Instance Resources: mem-estimate=975.00MB mem-reservation=34.00MB thread-reservation=1
| | max-parallelism=1460 segment-costs=[2100244388] cpu-comparison-result=758 [max(110 (self) vs 758 (sum children))]
| JOIN BUILD
| | join-table-id=00 plan-id=01 cohort-id=01
| | build expressions: i_item_sk, s_store_name, s_zip
| | runtime filters: RF003[min_max] <- i_item_sk, RF004[min_max] <- s_store_name, RF005[min_max] <- s_zip
-| | mem-estimate=2.32GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0 cost=863993608
+| | mem-estimate=930.91MB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0 cost=863993608
| |
| 135:EXCHANGE [HASH(i_item_sk,s_store_name,s_zip)]
| | mem-estimate=44.09MB mem-reservation=0B thread-reservation=0
@@ -2433,7 +2433,7 @@ max-parallelism=30 segment-costs=[232690522]
in pipelines: 01(GETNEXT)
---- PARALLELPLANS
Max Per-Host Resource Reservation: Memory=6.10GB Threads=325
-Per-Host Resource Estimates: Memory=101.04GB
+Per-Host Resource Estimates: Memory=68.61GB
F53:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1
| Per-Instance Resources: mem-estimate=1.21GB mem-reservation=4.00MB thread-reservation=1
| max-parallelism=1 segment-costs=[18941077893] cpu-comparison-result=1516 [max(1 (self) vs 1516 (sum children))]
@@ -2448,11 +2448,11 @@ PLAN-ROOT SINK
| in pipelines: 79(GETNEXT)
|
F52:PLAN FRAGMENT [HASH(i_item_sk,s_store_name,s_zip)] hosts=10 instances=110 (adjusted from 120)
-Per-Instance Resources: mem-estimate=2.49GB mem-reservation=12.00MB thread-reservation=1
+Per-Instance Resources: mem-estimate=975.00MB mem-reservation=12.00MB thread-reservation=1
max-parallelism=1460 segment-costs=[14553075383, 7852642112] cpu-comparison-result=1516 [max(110 (self) vs 1516 (sum children))]
79:SORT
| order by: product_name ASC, store_name ASC, cnt ASC, s1 ASC, s1 ASC
-| mem-estimate=2.45GB mem-reservation=12.00MB spill-buffer=2.00MB thread-reservation=0
+| mem-estimate=930.91MB mem-reservation=12.00MB spill-buffer=2.00MB thread-reservation=0
| tuple-ids=46 row-size=335B cardinality=863.99M cost=12938654601
| in pipelines: 79(GETNEXT), 106(OPEN)
|
@@ -2466,13 +2466,13 @@ max-parallelism=1460 segment-costs=[14553075383, 7852642112] cpu-comparison-resu
| in pipelines: 106(GETNEXT), 133(OPEN)
|
|--F54:PLAN FRAGMENT [HASH(i_item_sk,s_store_name,s_zip)] hosts=10 instances=110 (adjusted from 120)
-| | Per-Instance Resources: mem-estimate=2.36GB mem-reservation=34.00MB thread-reservation=1
+| | Per-Instance Resources: mem-estimate=975.00MB mem-reservation=34.00MB thread-reservation=1
| | max-parallelism=1460 segment-costs=[2100244388] cpu-comparison-result=758 [max(110 (self) vs 758 (sum children))]
| JOIN BUILD
| | join-table-id=00 plan-id=01 cohort-id=01
| | build expressions: i_item_sk, s_store_name, s_zip
| | runtime filters: RF003[min_max] <- i_item_sk, RF004[min_max] <- s_store_name, RF005[min_max] <- s_zip
-| | mem-estimate=2.32GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0 cost=863993608
+| | mem-estimate=930.91MB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0 cost=863993608
| |
| 135:EXCHANGE [HASH(i_item_sk,s_store_name,s_zip)]
| | mem-estimate=44.09MB mem-reservation=0B thread-reservation=0
diff --git a/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q64.test b/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q64-mem_limit_executors-20g.test
similarity index 99%
copy from testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q64.test
copy to testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q64-mem_limit_executors-20g.test
index 01b79303b..a297dac6e 100644
--- a/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q64.test
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q64-mem_limit_executors-20g.test
@@ -122,9 +122,9 @@ order by cs1.product_name
# end query 64 in stream 0 using template query64.tpl
---- PLAN
Max Per-Host Resource Reservation: Memory=676.89MB Threads=1
-Per-Host Resource Estimates: Memory=524.91GB
+Per-Host Resource Estimates: Memory=59.31GB
F00:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1
-| Per-Instance Resources: mem-estimate=524.91GB mem-reservation=676.89MB thread-reservation=1 runtime-filters-memory=55.00MB
+| Per-Instance Resources: mem-estimate=59.31GB mem-reservation=676.89MB thread-reservation=1 runtime-filters-memory=55.00MB
| max-parallelism=1 segment-costs=[3995470666, 7240810714, 3995470666, 7240810714, 14180818211, 18143864832]
PLAN-ROOT SINK
| output exprs: product_name, store_name, store_zip, b_street_number, b_street_name, b_city, b_zip, c_street_number, c_street_name, c_city, c_zip, syear, cnt, s1, s2, s3, s1, s2, s3, syear, cnt
@@ -132,7 +132,7 @@ PLAN-ROOT SINK
|
79:SORT
| order by: product_name ASC, store_name ASC, cnt ASC, s1 ASC, s1 ASC
-| mem-estimate=269.48GB mem-reservation=12.00MB spill-buffer=2.00MB thread-reservation=0
+| mem-estimate=20.00GB mem-reservation=12.00MB spill-buffer=2.00MB thread-reservation=0
| tuple-ids=46 row-size=335B cardinality=863.99M cost=12938654601
| in pipelines: 79(GETNEXT), 38(OPEN)
|
@@ -141,7 +141,7 @@ PLAN-ROOT SINK
| fk/pk conjuncts: i_item_sk = i_item_sk, s_store_name = s_store_name, s_zip = s_zip
| other predicates: count(*) <= count(*)
| runtime filters: RF003[min_max] <- i_item_sk, RF004[min_max] <- s_store_name, RF005[min_max] <- s_zip
-| mem-estimate=254.86GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| mem-estimate=20.00GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
| tuple-ids=21,44 row-size=574B cardinality=863.99M cost=1242163610
| in pipelines: 38(GETNEXT), 77(OPEN)
|
@@ -172,7 +172,7 @@ PLAN-ROOT SINK
| | | hash predicates: cs_item_sk = cr_item_sk, cs_order_number = cr_order_number
| | | fk/pk conjuncts: cs_item_sk = cr_item_sk, cs_order_number = cr_order_number
| | | runtime filters: RF084[min_max] <- cr_item_sk, RF085[min_max] <- cr_order_number
-| | | mem-estimate=21.66GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| | | mem-estimate=20.00GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
| | | tuple-ids=25,26 row-size=40B cardinality=2.56G cost=1922201419
| | | in pipelines: 41(GETNEXT), 42(OPEN)
| | |
@@ -539,7 +539,7 @@ PLAN-ROOT SINK
| | hash predicates: cs_item_sk = cr_item_sk, cs_order_number = cr_order_number
| | fk/pk conjuncts: cs_item_sk = cr_item_sk, cs_order_number = cr_order_number
| | runtime filters: RF042[bloom] <- cr_item_sk, RF043[bloom] <- cr_order_number, RF044[min_max] <- cr_item_sk, RF045[min_max] <- cr_order_number
-| | mem-estimate=21.66GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| | mem-estimate=20.00GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
| | tuple-ids=2,3 row-size=40B cardinality=2.56G cost=1922201419
| | in pipelines: 02(GETNEXT), 03(OPEN)
| |
@@ -882,7 +882,7 @@ PLAN-ROOT SINK
in pipelines: 01(GETNEXT)
---- DISTRIBUTEDPLAN
Max Per-Host Resource Reservation: Memory=6.10GB Threads=325
-Per-Host Resource Estimates: Memory=101.04GB
+Per-Host Resource Estimates: Memory=88.61GB
F53:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1
| Per-Instance Resources: mem-estimate=1.21GB mem-reservation=4.00MB thread-reservation=1
| max-parallelism=1 segment-costs=[18941077893] cpu-comparison-result=1516 [max(1 (self) vs 1516 (sum children))]
@@ -897,11 +897,11 @@ PLAN-ROOT SINK
| in pipelines: 79(GETNEXT)
|
F52:PLAN FRAGMENT [HASH(i_item_sk,s_store_name,s_zip)] hosts=10 instances=110 (adjusted from 120)
-Per-Instance Resources: mem-estimate=2.49GB mem-reservation=12.00MB thread-reservation=1
+Per-Instance Resources: mem-estimate=1.86GB mem-reservation=12.00MB thread-reservation=1
max-parallelism=1460 segment-costs=[14553075383, 7852642112] cpu-comparison-result=1516 [max(110 (self) vs 1516 (sum children))]
79:SORT
| order by: product_name ASC, store_name ASC, cnt ASC, s1 ASC, s1 ASC
-| mem-estimate=2.45GB mem-reservation=12.00MB spill-buffer=2.00MB thread-reservation=0
+| mem-estimate=1.82GB mem-reservation=12.00MB spill-buffer=2.00MB thread-reservation=0
| tuple-ids=46 row-size=335B cardinality=863.99M cost=12938654601
| in pipelines: 79(GETNEXT), 106(OPEN)
|
@@ -915,13 +915,13 @@ max-parallelism=1460 segment-costs=[14553075383, 7852642112] cpu-comparison-resu
| in pipelines: 106(GETNEXT), 133(OPEN)
|
|--F54:PLAN FRAGMENT [HASH(i_item_sk,s_store_name,s_zip)] hosts=10 instances=110 (adjusted from 120)
-| | Per-Instance Resources: mem-estimate=2.36GB mem-reservation=34.00MB thread-reservation=1
+| | Per-Instance Resources: mem-estimate=1.86GB mem-reservation=34.00MB thread-reservation=1
| | max-parallelism=1460 segment-costs=[2100244388] cpu-comparison-result=758 [max(110 (self) vs 758 (sum children))]
| JOIN BUILD
| | join-table-id=00 plan-id=01 cohort-id=01
| | build expressions: i_item_sk, s_store_name, s_zip
| | runtime filters: RF003[min_max] <- i_item_sk, RF004[min_max] <- s_store_name, RF005[min_max] <- s_zip
-| | mem-estimate=2.32GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0 cost=863993608
+| | mem-estimate=1.82GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0 cost=863993608
| |
| 135:EXCHANGE [HASH(i_item_sk,s_store_name,s_zip)]
| | mem-estimate=44.09MB mem-reservation=0B thread-reservation=0
@@ -2433,7 +2433,7 @@ max-parallelism=30 segment-costs=[232690522]
in pipelines: 01(GETNEXT)
---- PARALLELPLANS
Max Per-Host Resource Reservation: Memory=6.10GB Threads=325
-Per-Host Resource Estimates: Memory=101.04GB
+Per-Host Resource Estimates: Memory=88.61GB
F53:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1
| Per-Instance Resources: mem-estimate=1.21GB mem-reservation=4.00MB thread-reservation=1
| max-parallelism=1 segment-costs=[18941077893] cpu-comparison-result=1516 [max(1 (self) vs 1516 (sum children))]
@@ -2448,11 +2448,11 @@ PLAN-ROOT SINK
| in pipelines: 79(GETNEXT)
|
F52:PLAN FRAGMENT [HASH(i_item_sk,s_store_name,s_zip)] hosts=10 instances=110 (adjusted from 120)
-Per-Instance Resources: mem-estimate=2.49GB mem-reservation=12.00MB thread-reservation=1
+Per-Instance Resources: mem-estimate=1.86GB mem-reservation=12.00MB thread-reservation=1
max-parallelism=1460 segment-costs=[14553075383, 7852642112] cpu-comparison-result=1516 [max(110 (self) vs 1516 (sum children))]
79:SORT
| order by: product_name ASC, store_name ASC, cnt ASC, s1 ASC, s1 ASC
-| mem-estimate=2.45GB mem-reservation=12.00MB spill-buffer=2.00MB thread-reservation=0
+| mem-estimate=1.82GB mem-reservation=12.00MB spill-buffer=2.00MB thread-reservation=0
| tuple-ids=46 row-size=335B cardinality=863.99M cost=12938654601
| in pipelines: 79(GETNEXT), 106(OPEN)
|
@@ -2466,13 +2466,13 @@ max-parallelism=1460 segment-costs=[14553075383, 7852642112] cpu-comparison-resu
| in pipelines: 106(GETNEXT), 133(OPEN)
|
|--F54:PLAN FRAGMENT [HASH(i_item_sk,s_store_name,s_zip)] hosts=10 instances=110 (adjusted from 120)
-| | Per-Instance Resources: mem-estimate=2.36GB mem-reservation=34.00MB thread-reservation=1
+| | Per-Instance Resources: mem-estimate=1.86GB mem-reservation=34.00MB thread-reservation=1
| | max-parallelism=1460 segment-costs=[2100244388] cpu-comparison-result=758 [max(110 (self) vs 758 (sum children))]
| JOIN BUILD
| | join-table-id=00 plan-id=01 cohort-id=01
| | build expressions: i_item_sk, s_store_name, s_zip
| | runtime filters: RF003[min_max] <- i_item_sk, RF004[min_max] <- s_store_name, RF005[min_max] <- s_zip
-| | mem-estimate=2.32GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0 cost=863993608
+| | mem-estimate=1.82GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0 cost=863993608
| |
| 135:EXCHANGE [HASH(i_item_sk,s_store_name,s_zip)]
| | mem-estimate=44.09MB mem-reservation=0B thread-reservation=0
diff --git a/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q64.test b/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q64.test
index 01b79303b..77bec8cd4 100644
--- a/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q64.test
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q64.test
@@ -122,9 +122,9 @@ order by cs1.product_name
# end query 64 in stream 0 using template query64.tpl
---- PLAN
Max Per-Host Resource Reservation: Memory=676.89MB Threads=1
-Per-Host Resource Estimates: Memory=524.91GB
+Per-Host Resource Estimates: Memory=100.57GB
F00:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1
-| Per-Instance Resources: mem-estimate=524.91GB mem-reservation=676.89MB thread-reservation=1 runtime-filters-memory=55.00MB
+| Per-Instance Resources: mem-estimate=100.57GB mem-reservation=676.89MB thread-reservation=1 runtime-filters-memory=55.00MB
| max-parallelism=1 segment-costs=[3995470666, 7240810714, 3995470666, 7240810714, 14180818211, 18143864832]
PLAN-ROOT SINK
| output exprs: product_name, store_name, store_zip, b_street_number, b_street_name, b_city, b_zip, c_street_number, c_street_name, c_city, c_zip, syear, cnt, s1, s2, s3, s1, s2, s3, syear, cnt
@@ -132,7 +132,7 @@ PLAN-ROOT SINK
|
79:SORT
| order by: product_name ASC, store_name ASC, cnt ASC, s1 ASC, s1 ASC
-| mem-estimate=269.48GB mem-reservation=12.00MB spill-buffer=2.00MB thread-reservation=0
+| mem-estimate=50.00GB mem-reservation=12.00MB spill-buffer=2.00MB thread-reservation=0
| tuple-ids=46 row-size=335B cardinality=863.99M cost=12938654601
| in pipelines: 79(GETNEXT), 38(OPEN)
|
@@ -141,7 +141,7 @@ PLAN-ROOT SINK
| fk/pk conjuncts: i_item_sk = i_item_sk, s_store_name = s_store_name, s_zip = s_zip
| other predicates: count(*) <= count(*)
| runtime filters: RF003[min_max] <- i_item_sk, RF004[min_max] <- s_store_name, RF005[min_max] <- s_zip
-| mem-estimate=254.86GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| mem-estimate=50.00GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
| tuple-ids=21,44 row-size=574B cardinality=863.99M cost=1242163610
| in pipelines: 38(GETNEXT), 77(OPEN)
|
diff --git a/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q67.test b/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q67-mem_limit-10g.test
similarity index 98%
copy from testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q67.test
copy to testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q67-mem_limit-10g.test
index 1ff839aa6..85060ab6d 100644
--- a/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q67.test
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q67-mem_limit-10g.test
@@ -45,9 +45,9 @@ limit 100;
# end query 67 in stream 0 using template query67.tpl
---- PLAN
Max Per-Host Resource Reservation: Memory=263.62MB Threads=1
-Per-Host Resource Estimates: Memory=2.16TB
+Per-Host Resource Estimates: Memory=15.19GB
F00:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1
-| Per-Instance Resources: mem-estimate=2.16TB mem-reservation=263.62MB thread-reservation=1 runtime-filters-memory=3.00MB
+| Per-Instance Resources: mem-estimate=15.19GB mem-reservation=263.62MB thread-reservation=1 runtime-filters-memory=3.00MB
| max-parallelism=1 segment-costs=[176669034554, 73050686125, 233936768446, 1261, 1000]
PLAN-ROOT SINK
| output exprs: i_category, i_class, i_brand, i_product_name, d_year, d_qoy, d_moy, s_store_id, sumsales, rk
@@ -117,7 +117,7 @@ PLAN-ROOT SINK
| Class 8
| output: sum(coalesce(ss_sales_price * CAST(ss_quantity AS DECIMAL(10,0)), CAST(0 AS DECIMAL(18,2))))
| group by: NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL
-| mem-estimate=2.16TB mem-reservation=210.75MB thread-reservation=0
+| mem-estimate=10.00GB mem-reservation=210.75MB thread-reservation=0
| tuple-ids=4N,5N,6N,7N,8N,9N,10N,11N,12N row-size=1.09KB cardinality=12.38G cost=162840615894
| in pipelines: 07(GETNEXT), 00(OPEN)
|
@@ -191,7 +191,7 @@ PLAN-ROOT SINK
in pipelines: 00(GETNEXT)
---- DISTRIBUTEDPLAN
Max Per-Host Resource Reservation: Memory=4.70GB Threads=27
-Per-Host Resource Estimates: Memory=255.46GB
+Per-Host Resource Estimates: Memory=28.37GB
F06:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1
| Per-Instance Resources: mem-estimate=4.10MB mem-reservation=4.00MB thread-reservation=1
| max-parallelism=1 segment-costs=[1044] cpu-comparison-result=120 [max(1 (self) vs 120 (sum children))]
@@ -243,7 +243,7 @@ max-parallelism=10 segment-costs=[967, 1261, 282] cpu-comparison-result=120 [max
| in pipelines: 09(GETNEXT)
|
F04:PLAN FRAGMENT [HASH(CASE valid_tid(4,5,6,7,8,9,10,11,12) WHEN 4 THEN murmur_hash(i_category) WHEN 5 THEN murmur_hash(i_category) WHEN 6 THEN murmur_hash(i_category) WHEN 7 THEN murmur_hash(i_category) WHEN 8 THEN murmur_hash(i_category) WHEN 9 THEN murmur_hash(i_category) WHEN 10 THEN murmur_hash(i_category) WHEN 11 THEN murmur_hash(i_category) WHEN 12 THEN murmur_hash(NULL) END,CASE valid_tid(4,5,6,7,8,9,10,11,12) WHEN 4 THEN murmur_hash(i_class) WHEN 5 THEN murmur_hash(i_class) WHE [...]
-Per-Instance Resources: mem-estimate=7.14GB mem-reservation=244.75MB thread-reservation=1
+Per-Instance Resources: mem-estimate=1.99GB mem-reservation=244.75MB thread-reservation=1
max-parallelism=26630 segment-costs=[266227423359, 73050686125, 233936768446, 1288] cpu-comparison-result=120 [max(70 (self) vs 120 (sum children))]
09:TOP-N
| order by: CASE valid_tid(4,5,6,7,8,9,10,11,12) WHEN 4 THEN i_category WHEN 5 THEN i_category WHEN 6 THEN i_category WHEN 7 THEN i_category WHEN 8 THEN i_category WHEN 9 THEN i_category WHEN 10 THEN i_category WHEN 11 THEN i_category WHEN 12 THEN NULL END ASC NULLS LAST, aggif(valid_tid(4,5,6,7,8,9,10,11,12) IN (4, 5, 6, 7, 8, 9, 10, 11, 12), CASE valid_tid(4,5,6,7,8,9,10,11,12) WHEN 4 THEN sum(coalesce(ss_sales_price * ss_quantity, 0)) WHEN 5 THEN sum(coalesce(ss_sales_price * ss_quan [...]
@@ -288,7 +288,7 @@ max-parallelism=26630 segment-costs=[266227423359, 73050686125, 233936768446, 12
| Class 8
| output: sum:merge(coalesce(ss_sales_price * ss_quantity, 0))
| group by: NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL
-| mem-estimate=6.57GB mem-reservation=210.75MB thread-reservation=0
+| mem-estimate=1.43GB mem-reservation=210.75MB thread-reservation=0
| tuple-ids=4N,5N,6N,7N,8N,9N,10N,11N,12N row-size=1.09KB cardinality=12.38G cost=97683994945
| in pipelines: 17(GETNEXT), 00(OPEN)
|
@@ -299,7 +299,7 @@ max-parallelism=26630 segment-costs=[266227423359, 73050686125, 233936768446, 12
|
F00:PLAN FRAGMENT [RANDOM] hosts=10 instances=120
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=17.08GB mem-reservation=219.00MB thread-reservation=1
+Per-Instance Resources: mem-estimate=1.16GB mem-reservation=219.00MB thread-reservation=1
max-parallelism=1824 segment-costs=[257850166519, 2328086935759] cpu-comparison-result=120 [max(120 (self) vs 36 (sum children))]
07:AGGREGATE [STREAMING]
| Class 0
@@ -329,7 +329,7 @@ max-parallelism=1824 segment-costs=[257850166519, 2328086935759] cpu-comparison-
| Class 8
| output: sum(coalesce(ss_sales_price * CAST(ss_quantity AS DECIMAL(10,0)), CAST(0 AS DECIMAL(18,2))))
| group by: NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL
-| mem-estimate=16.76GB mem-reservation=211.00MB thread-reservation=0
+| mem-estimate=853.33MB mem-reservation=211.00MB thread-reservation=0
| tuple-ids=4N,5N,6N,7N,8N,9N,10N,11N,12N row-size=1.09KB cardinality=30.30G cost=244022435997
| in pipelines: 00(GETNEXT)
|
@@ -454,7 +454,7 @@ max-parallelism=1824 segment-costs=[257850166519, 2328086935759] cpu-comparison-
in pipelines: 00(GETNEXT)
---- PARALLELPLANS
Max Per-Host Resource Reservation: Memory=4.70GB Threads=27
-Per-Host Resource Estimates: Memory=255.46GB
+Per-Host Resource Estimates: Memory=28.37GB
F06:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1
| Per-Instance Resources: mem-estimate=4.10MB mem-reservation=4.00MB thread-reservation=1
| max-parallelism=1 segment-costs=[1044] cpu-comparison-result=120 [max(1 (self) vs 120 (sum children))]
@@ -506,7 +506,7 @@ max-parallelism=10 segment-costs=[967, 1261, 282] cpu-comparison-result=120 [max
| in pipelines: 09(GETNEXT)
|
F04:PLAN FRAGMENT [HASH(CASE valid_tid(4,5,6,7,8,9,10,11,12) WHEN 4 THEN murmur_hash(i_category) WHEN 5 THEN murmur_hash(i_category) WHEN 6 THEN murmur_hash(i_category) WHEN 7 THEN murmur_hash(i_category) WHEN 8 THEN murmur_hash(i_category) WHEN 9 THEN murmur_hash(i_category) WHEN 10 THEN murmur_hash(i_category) WHEN 11 THEN murmur_hash(i_category) WHEN 12 THEN murmur_hash(NULL) END,CASE valid_tid(4,5,6,7,8,9,10,11,12) WHEN 4 THEN murmur_hash(i_class) WHEN 5 THEN murmur_hash(i_class) WHE [...]
-Per-Instance Resources: mem-estimate=7.14GB mem-reservation=244.75MB thread-reservation=1
+Per-Instance Resources: mem-estimate=1.99GB mem-reservation=244.75MB thread-reservation=1
max-parallelism=26630 segment-costs=[266227423359, 73050686125, 233936768446, 1288] cpu-comparison-result=120 [max(70 (self) vs 120 (sum children))]
09:TOP-N
| order by: CASE valid_tid(4,5,6,7,8,9,10,11,12) WHEN 4 THEN i_category WHEN 5 THEN i_category WHEN 6 THEN i_category WHEN 7 THEN i_category WHEN 8 THEN i_category WHEN 9 THEN i_category WHEN 10 THEN i_category WHEN 11 THEN i_category WHEN 12 THEN NULL END ASC NULLS LAST, aggif(valid_tid(4,5,6,7,8,9,10,11,12) IN (4, 5, 6, 7, 8, 9, 10, 11, 12), CASE valid_tid(4,5,6,7,8,9,10,11,12) WHEN 4 THEN sum(coalesce(ss_sales_price * ss_quantity, 0)) WHEN 5 THEN sum(coalesce(ss_sales_price * ss_quan [...]
@@ -551,7 +551,7 @@ max-parallelism=26630 segment-costs=[266227423359, 73050686125, 233936768446, 12
| Class 8
| output: sum:merge(coalesce(ss_sales_price * ss_quantity, 0))
| group by: NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL
-| mem-estimate=6.57GB mem-reservation=210.75MB thread-reservation=0
+| mem-estimate=1.43GB mem-reservation=210.75MB thread-reservation=0
| tuple-ids=4N,5N,6N,7N,8N,9N,10N,11N,12N row-size=1.09KB cardinality=12.38G cost=97683994945
| in pipelines: 17(GETNEXT), 00(OPEN)
|
@@ -562,7 +562,7 @@ max-parallelism=26630 segment-costs=[266227423359, 73050686125, 233936768446, 12
|
F00:PLAN FRAGMENT [RANDOM] hosts=10 instances=120
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=17.08GB mem-reservation=219.00MB thread-reservation=1
+Per-Instance Resources: mem-estimate=1.16GB mem-reservation=219.00MB thread-reservation=1
max-parallelism=1824 segment-costs=[257850166519, 2328086935759] cpu-comparison-result=120 [max(120 (self) vs 36 (sum children))]
07:AGGREGATE [STREAMING]
| Class 0
@@ -592,7 +592,7 @@ max-parallelism=1824 segment-costs=[257850166519, 2328086935759] cpu-comparison-
| Class 8
| output: sum(coalesce(ss_sales_price * CAST(ss_quantity AS DECIMAL(10,0)), CAST(0 AS DECIMAL(18,2))))
| group by: NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL
-| mem-estimate=16.76GB mem-reservation=211.00MB thread-reservation=0
+| mem-estimate=853.33MB mem-reservation=211.00MB thread-reservation=0
| tuple-ids=4N,5N,6N,7N,8N,9N,10N,11N,12N row-size=1.09KB cardinality=30.30G cost=244022435997
| in pipelines: 00(GETNEXT)
|
diff --git a/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q67.test b/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q67-mem_limit_executors-20g.test
similarity index 98%
copy from testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q67.test
copy to testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q67-mem_limit_executors-20g.test
index 1ff839aa6..e628bfa34 100644
--- a/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q67.test
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q67-mem_limit_executors-20g.test
@@ -45,9 +45,9 @@ limit 100;
# end query 67 in stream 0 using template query67.tpl
---- PLAN
Max Per-Host Resource Reservation: Memory=263.62MB Threads=1
-Per-Host Resource Estimates: Memory=2.16TB
+Per-Host Resource Estimates: Memory=25.19GB
F00:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1
-| Per-Instance Resources: mem-estimate=2.16TB mem-reservation=263.62MB thread-reservation=1 runtime-filters-memory=3.00MB
+| Per-Instance Resources: mem-estimate=25.19GB mem-reservation=263.62MB thread-reservation=1 runtime-filters-memory=3.00MB
| max-parallelism=1 segment-costs=[176669034554, 73050686125, 233936768446, 1261, 1000]
PLAN-ROOT SINK
| output exprs: i_category, i_class, i_brand, i_product_name, d_year, d_qoy, d_moy, s_store_id, sumsales, rk
@@ -117,7 +117,7 @@ PLAN-ROOT SINK
| Class 8
| output: sum(coalesce(ss_sales_price * CAST(ss_quantity AS DECIMAL(10,0)), CAST(0 AS DECIMAL(18,2))))
| group by: NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL
-| mem-estimate=2.16TB mem-reservation=210.75MB thread-reservation=0
+| mem-estimate=20.00GB mem-reservation=210.75MB thread-reservation=0
| tuple-ids=4N,5N,6N,7N,8N,9N,10N,11N,12N row-size=1.09KB cardinality=12.38G cost=162840615894
| in pipelines: 07(GETNEXT), 00(OPEN)
|
@@ -191,7 +191,7 @@ PLAN-ROOT SINK
in pipelines: 00(GETNEXT)
---- DISTRIBUTEDPLAN
Max Per-Host Resource Reservation: Memory=4.70GB Threads=27
-Per-Host Resource Estimates: Memory=255.46GB
+Per-Host Resource Estimates: Memory=48.37GB
F06:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1
| Per-Instance Resources: mem-estimate=4.10MB mem-reservation=4.00MB thread-reservation=1
| max-parallelism=1 segment-costs=[1044] cpu-comparison-result=120 [max(1 (self) vs 120 (sum children))]
@@ -243,7 +243,7 @@ max-parallelism=10 segment-costs=[967, 1261, 282] cpu-comparison-result=120 [max
| in pipelines: 09(GETNEXT)
|
F04:PLAN FRAGMENT [HASH(CASE valid_tid(4,5,6,7,8,9,10,11,12) WHEN 4 THEN murmur_hash(i_category) WHEN 5 THEN murmur_hash(i_category) WHEN 6 THEN murmur_hash(i_category) WHEN 7 THEN murmur_hash(i_category) WHEN 8 THEN murmur_hash(i_category) WHEN 9 THEN murmur_hash(i_category) WHEN 10 THEN murmur_hash(i_category) WHEN 11 THEN murmur_hash(i_category) WHEN 12 THEN murmur_hash(NULL) END,CASE valid_tid(4,5,6,7,8,9,10,11,12) WHEN 4 THEN murmur_hash(i_class) WHEN 5 THEN murmur_hash(i_class) WHE [...]
-Per-Instance Resources: mem-estimate=7.14GB mem-reservation=244.75MB thread-reservation=1
+Per-Instance Resources: mem-estimate=3.42GB mem-reservation=244.75MB thread-reservation=1
max-parallelism=26630 segment-costs=[266227423359, 73050686125, 233936768446, 1288] cpu-comparison-result=120 [max(70 (self) vs 120 (sum children))]
09:TOP-N
| order by: CASE valid_tid(4,5,6,7,8,9,10,11,12) WHEN 4 THEN i_category WHEN 5 THEN i_category WHEN 6 THEN i_category WHEN 7 THEN i_category WHEN 8 THEN i_category WHEN 9 THEN i_category WHEN 10 THEN i_category WHEN 11 THEN i_category WHEN 12 THEN NULL END ASC NULLS LAST, aggif(valid_tid(4,5,6,7,8,9,10,11,12) IN (4, 5, 6, 7, 8, 9, 10, 11, 12), CASE valid_tid(4,5,6,7,8,9,10,11,12) WHEN 4 THEN sum(coalesce(ss_sales_price * ss_quantity, 0)) WHEN 5 THEN sum(coalesce(ss_sales_price * ss_quan [...]
@@ -288,7 +288,7 @@ max-parallelism=26630 segment-costs=[266227423359, 73050686125, 233936768446, 12
| Class 8
| output: sum:merge(coalesce(ss_sales_price * ss_quantity, 0))
| group by: NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL
-| mem-estimate=6.57GB mem-reservation=210.75MB thread-reservation=0
+| mem-estimate=2.86GB mem-reservation=210.75MB thread-reservation=0
| tuple-ids=4N,5N,6N,7N,8N,9N,10N,11N,12N row-size=1.09KB cardinality=12.38G cost=97683994945
| in pipelines: 17(GETNEXT), 00(OPEN)
|
@@ -299,7 +299,7 @@ max-parallelism=26630 segment-costs=[266227423359, 73050686125, 233936768446, 12
|
F00:PLAN FRAGMENT [RANDOM] hosts=10 instances=120
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=17.08GB mem-reservation=219.00MB thread-reservation=1
+Per-Instance Resources: mem-estimate=1.99GB mem-reservation=219.00MB thread-reservation=1
max-parallelism=1824 segment-costs=[257850166519, 2328086935759] cpu-comparison-result=120 [max(120 (self) vs 36 (sum children))]
07:AGGREGATE [STREAMING]
| Class 0
@@ -329,7 +329,7 @@ max-parallelism=1824 segment-costs=[257850166519, 2328086935759] cpu-comparison-
| Class 8
| output: sum(coalesce(ss_sales_price * CAST(ss_quantity AS DECIMAL(10,0)), CAST(0 AS DECIMAL(18,2))))
| group by: NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL
-| mem-estimate=16.76GB mem-reservation=211.00MB thread-reservation=0
+| mem-estimate=1.67GB mem-reservation=211.00MB thread-reservation=0
| tuple-ids=4N,5N,6N,7N,8N,9N,10N,11N,12N row-size=1.09KB cardinality=30.30G cost=244022435997
| in pipelines: 00(GETNEXT)
|
@@ -454,7 +454,7 @@ max-parallelism=1824 segment-costs=[257850166519, 2328086935759] cpu-comparison-
in pipelines: 00(GETNEXT)
---- PARALLELPLANS
Max Per-Host Resource Reservation: Memory=4.70GB Threads=27
-Per-Host Resource Estimates: Memory=255.46GB
+Per-Host Resource Estimates: Memory=48.37GB
F06:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1
| Per-Instance Resources: mem-estimate=4.10MB mem-reservation=4.00MB thread-reservation=1
| max-parallelism=1 segment-costs=[1044] cpu-comparison-result=120 [max(1 (self) vs 120 (sum children))]
@@ -506,7 +506,7 @@ max-parallelism=10 segment-costs=[967, 1261, 282] cpu-comparison-result=120 [max
| in pipelines: 09(GETNEXT)
|
F04:PLAN FRAGMENT [HASH(CASE valid_tid(4,5,6,7,8,9,10,11,12) WHEN 4 THEN murmur_hash(i_category) WHEN 5 THEN murmur_hash(i_category) WHEN 6 THEN murmur_hash(i_category) WHEN 7 THEN murmur_hash(i_category) WHEN 8 THEN murmur_hash(i_category) WHEN 9 THEN murmur_hash(i_category) WHEN 10 THEN murmur_hash(i_category) WHEN 11 THEN murmur_hash(i_category) WHEN 12 THEN murmur_hash(NULL) END,CASE valid_tid(4,5,6,7,8,9,10,11,12) WHEN 4 THEN murmur_hash(i_class) WHEN 5 THEN murmur_hash(i_class) WHE [...]
-Per-Instance Resources: mem-estimate=7.14GB mem-reservation=244.75MB thread-reservation=1
+Per-Instance Resources: mem-estimate=3.42GB mem-reservation=244.75MB thread-reservation=1
max-parallelism=26630 segment-costs=[266227423359, 73050686125, 233936768446, 1288] cpu-comparison-result=120 [max(70 (self) vs 120 (sum children))]
09:TOP-N
| order by: CASE valid_tid(4,5,6,7,8,9,10,11,12) WHEN 4 THEN i_category WHEN 5 THEN i_category WHEN 6 THEN i_category WHEN 7 THEN i_category WHEN 8 THEN i_category WHEN 9 THEN i_category WHEN 10 THEN i_category WHEN 11 THEN i_category WHEN 12 THEN NULL END ASC NULLS LAST, aggif(valid_tid(4,5,6,7,8,9,10,11,12) IN (4, 5, 6, 7, 8, 9, 10, 11, 12), CASE valid_tid(4,5,6,7,8,9,10,11,12) WHEN 4 THEN sum(coalesce(ss_sales_price * ss_quantity, 0)) WHEN 5 THEN sum(coalesce(ss_sales_price * ss_quan [...]
@@ -551,7 +551,7 @@ max-parallelism=26630 segment-costs=[266227423359, 73050686125, 233936768446, 12
| Class 8
| output: sum:merge(coalesce(ss_sales_price * ss_quantity, 0))
| group by: NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL
-| mem-estimate=6.57GB mem-reservation=210.75MB thread-reservation=0
+| mem-estimate=2.86GB mem-reservation=210.75MB thread-reservation=0
| tuple-ids=4N,5N,6N,7N,8N,9N,10N,11N,12N row-size=1.09KB cardinality=12.38G cost=97683994945
| in pipelines: 17(GETNEXT), 00(OPEN)
|
@@ -562,7 +562,7 @@ max-parallelism=26630 segment-costs=[266227423359, 73050686125, 233936768446, 12
|
F00:PLAN FRAGMENT [RANDOM] hosts=10 instances=120
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=17.08GB mem-reservation=219.00MB thread-reservation=1
+Per-Instance Resources: mem-estimate=1.99GB mem-reservation=219.00MB thread-reservation=1
max-parallelism=1824 segment-costs=[257850166519, 2328086935759] cpu-comparison-result=120 [max(120 (self) vs 36 (sum children))]
07:AGGREGATE [STREAMING]
| Class 0
@@ -592,7 +592,7 @@ max-parallelism=1824 segment-costs=[257850166519, 2328086935759] cpu-comparison-
| Class 8
| output: sum(coalesce(ss_sales_price * CAST(ss_quantity AS DECIMAL(10,0)), CAST(0 AS DECIMAL(18,2))))
| group by: NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL
-| mem-estimate=16.76GB mem-reservation=211.00MB thread-reservation=0
+| mem-estimate=1.67GB mem-reservation=211.00MB thread-reservation=0
| tuple-ids=4N,5N,6N,7N,8N,9N,10N,11N,12N row-size=1.09KB cardinality=30.30G cost=244022435997
| in pipelines: 00(GETNEXT)
|
diff --git a/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q67.test b/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q67.test
index 1ff839aa6..1034caaca 100644
--- a/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q67.test
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q67.test
@@ -45,9 +45,9 @@ limit 100;
# end query 67 in stream 0 using template query67.tpl
---- PLAN
Max Per-Host Resource Reservation: Memory=263.62MB Threads=1
-Per-Host Resource Estimates: Memory=2.16TB
+Per-Host Resource Estimates: Memory=55.19GB
F00:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1
-| Per-Instance Resources: mem-estimate=2.16TB mem-reservation=263.62MB thread-reservation=1 runtime-filters-memory=3.00MB
+| Per-Instance Resources: mem-estimate=55.19GB mem-reservation=263.62MB thread-reservation=1 runtime-filters-memory=3.00MB
| max-parallelism=1 segment-costs=[176669034554, 73050686125, 233936768446, 1261, 1000]
PLAN-ROOT SINK
| output exprs: i_category, i_class, i_brand, i_product_name, d_year, d_qoy, d_moy, s_store_id, sumsales, rk
@@ -117,7 +117,7 @@ PLAN-ROOT SINK
| Class 8
| output: sum(coalesce(ss_sales_price * CAST(ss_quantity AS DECIMAL(10,0)), CAST(0 AS DECIMAL(18,2))))
| group by: NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL
-| mem-estimate=2.16TB mem-reservation=210.75MB thread-reservation=0
+| mem-estimate=50.00GB mem-reservation=210.75MB thread-reservation=0
| tuple-ids=4N,5N,6N,7N,8N,9N,10N,11N,12N row-size=1.09KB cardinality=12.38G cost=162840615894
| in pipelines: 07(GETNEXT), 00(OPEN)
|
@@ -191,7 +191,7 @@ PLAN-ROOT SINK
in pipelines: 00(GETNEXT)
---- DISTRIBUTEDPLAN
Max Per-Host Resource Reservation: Memory=4.70GB Threads=27
-Per-Host Resource Estimates: Memory=255.46GB
+Per-Host Resource Estimates: Memory=104.39GB
F06:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1
| Per-Instance Resources: mem-estimate=4.10MB mem-reservation=4.00MB thread-reservation=1
| max-parallelism=1 segment-costs=[1044] cpu-comparison-result=120 [max(1 (self) vs 120 (sum children))]
@@ -299,7 +299,7 @@ max-parallelism=26630 segment-costs=[266227423359, 73050686125, 233936768446, 12
|
F00:PLAN FRAGMENT [RANDOM] hosts=10 instances=120
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=17.08GB mem-reservation=219.00MB thread-reservation=1
+Per-Instance Resources: mem-estimate=4.49GB mem-reservation=219.00MB thread-reservation=1
max-parallelism=1824 segment-costs=[257850166519, 2328086935759] cpu-comparison-result=120 [max(120 (self) vs 36 (sum children))]
07:AGGREGATE [STREAMING]
| Class 0
@@ -329,7 +329,7 @@ max-parallelism=1824 segment-costs=[257850166519, 2328086935759] cpu-comparison-
| Class 8
| output: sum(coalesce(ss_sales_price * CAST(ss_quantity AS DECIMAL(10,0)), CAST(0 AS DECIMAL(18,2))))
| group by: NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL
-| mem-estimate=16.76GB mem-reservation=211.00MB thread-reservation=0
+| mem-estimate=4.17GB mem-reservation=211.00MB thread-reservation=0
| tuple-ids=4N,5N,6N,7N,8N,9N,10N,11N,12N row-size=1.09KB cardinality=30.30G cost=244022435997
| in pipelines: 00(GETNEXT)
|
@@ -454,7 +454,7 @@ max-parallelism=1824 segment-costs=[257850166519, 2328086935759] cpu-comparison-
in pipelines: 00(GETNEXT)
---- PARALLELPLANS
Max Per-Host Resource Reservation: Memory=4.70GB Threads=27
-Per-Host Resource Estimates: Memory=255.46GB
+Per-Host Resource Estimates: Memory=104.39GB
F06:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1
| Per-Instance Resources: mem-estimate=4.10MB mem-reservation=4.00MB thread-reservation=1
| max-parallelism=1 segment-costs=[1044] cpu-comparison-result=120 [max(1 (self) vs 120 (sum children))]
@@ -562,7 +562,7 @@ max-parallelism=26630 segment-costs=[266227423359, 73050686125, 233936768446, 12
|
F00:PLAN FRAGMENT [RANDOM] hosts=10 instances=120
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=17.08GB mem-reservation=219.00MB thread-reservation=1
+Per-Instance Resources: mem-estimate=4.49GB mem-reservation=219.00MB thread-reservation=1
max-parallelism=1824 segment-costs=[257850166519, 2328086935759] cpu-comparison-result=120 [max(120 (self) vs 36 (sum children))]
07:AGGREGATE [STREAMING]
| Class 0
@@ -592,7 +592,7 @@ max-parallelism=1824 segment-costs=[257850166519, 2328086935759] cpu-comparison-
| Class 8
| output: sum(coalesce(ss_sales_price * CAST(ss_quantity AS DECIMAL(10,0)), CAST(0 AS DECIMAL(18,2))))
| group by: NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL
-| mem-estimate=16.76GB mem-reservation=211.00MB thread-reservation=0
+| mem-estimate=4.17GB mem-reservation=211.00MB thread-reservation=0
| tuple-ids=4N,5N,6N,7N,8N,9N,10N,11N,12N row-size=1.09KB cardinality=30.30G cost=244022435997
| in pipelines: 00(GETNEXT)
|
diff --git a/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q72.test b/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q72.test
index 908441a82..e70098b3a 100644
--- a/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q72.test
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q72.test
@@ -30,9 +30,9 @@ limit 100;
# end query 72 in stream 0 using template query72.tpl
---- PLAN
Max Per-Host Resource Reservation: Memory=197.25MB Threads=1
-Per-Host Resource Estimates: Memory=2.50TB
+Per-Host Resource Estimates: Memory=116.95GB
F00:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1
-| Per-Instance Resources: mem-estimate=2.50TB mem-reservation=197.25MB thread-reservation=1 runtime-filters-memory=9.00MB
+| Per-Instance Resources: mem-estimate=116.95GB mem-reservation=197.25MB thread-reservation=1 runtime-filters-memory=9.00MB
| max-parallelism=1 segment-costs=[1946370237431, 1196340444652, 600]
PLAN-ROOT SINK
| output exprs: i_item_desc, w_warehouse_name, d1.d_week_seq, sum(CASE WHEN p_promo_sk IS NULL THEN 1 ELSE 0 END), sum(CASE WHEN p_promo_sk IS NOT NULL THEN 1 ELSE 0 END), count(*)
@@ -47,7 +47,7 @@ PLAN-ROOT SINK
21:AGGREGATE [FINALIZE]
| output: sum(CAST(CASE WHEN p_promo_sk IS NULL THEN CAST(1 AS TINYINT) ELSE CAST(0 AS TINYINT) END AS BIGINT)), sum(CAST(CASE WHEN p_promo_sk IS NOT NULL THEN CAST(1 AS TINYINT) ELSE CAST(0 AS TINYINT) END AS BIGINT)), count(*)
| group by: i_item_desc, w_warehouse_name, d1.d_week_seq
-| mem-estimate=2.44TB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| mem-estimate=50.00GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
| tuple-ids=11 row-size=169B cardinality=59.29G cost=1111641993133
| in pipelines: 21(GETNEXT), 00(OPEN)
|
@@ -109,7 +109,7 @@ PLAN-ROOT SINK
| fk/pk conjuncts: none
| other predicates: inv_quantity_on_hand < cs_quantity
| runtime filters: RF002[bloom] <- inv_item_sk, RF003[bloom] <- inv_date_sk, RF004[min_max] <- inv_item_sk, RF005[min_max] <- inv_date_sk
-| mem-estimate=53.40GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| mem-estimate=50.00GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
| tuple-ids=0,5,4,6,3,8,7,1 row-size=241B cardinality=556.85G cost=101985452242
| in pipelines: 00(GETNEXT), 01(OPEN)
|
@@ -259,7 +259,7 @@ PLAN-ROOT SINK
in pipelines: 00(GETNEXT)
---- DISTRIBUTEDPLAN
Max Per-Host Resource Reservation: Memory=2.66GB Threads=79
-Per-Host Resource Estimates: Memory=1.99TB
+Per-Host Resource Estimates: Memory=127.17GB
F13:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1
| Per-Instance Resources: mem-estimate=5.98MB mem-reservation=4.00MB thread-reservation=1
| max-parallelism=1 segment-costs=[659] cpu-comparison-result=360 [max(1 (self) vs 360 (sum children))]
@@ -275,7 +275,7 @@ PLAN-ROOT SINK
| in pipelines: 22(GETNEXT)
|
F12:PLAN FRAGMENT [HASH(i_item_desc,w_warehouse_name,d1.d_week_seq)] hosts=10 instances=120
-Per-Instance Resources: mem-estimate=21.19GB mem-reservation=34.00MB thread-reservation=1
+Per-Instance Resources: mem-estimate=4.20GB mem-reservation=34.00MB thread-reservation=1
max-parallelism=155460 segment-costs=[1554521531826, 1196340444652, 473] cpu-comparison-result=360 [max(120 (self) vs 360 (sum children))]
22:TOP-N [LIMIT=100]
| order by: count(*) DESC, i_item_desc ASC, w_warehouse_name ASC, d1.d_week_seq ASC
@@ -286,7 +286,7 @@ max-parallelism=155460 segment-costs=[1554521531826, 1196340444652, 473] cpu-com
35:AGGREGATE [FINALIZE]
| output: sum:merge(CASE WHEN p_promo_sk IS NULL THEN 1 ELSE 0 END), sum:merge(CASE WHEN p_promo_sk IS NOT NULL THEN 1 ELSE 0 END), count:merge(*)
| group by: i_item_desc, w_warehouse_name, d1.d_week_seq
-| mem-estimate=21.16GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| mem-estimate=4.17GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
| tuple-ids=11 row-size=169B cardinality=59.29G cost=1079491668669
| in pipelines: 35(GETNEXT), 00(OPEN)
|
@@ -296,12 +296,12 @@ max-parallelism=155460 segment-costs=[1554521531826, 1196340444652, 473] cpu-com
| in pipelines: 00(GETNEXT)
|
F08:PLAN FRAGMENT [HASH(cs_item_sk,d2.d_date_sk)] hosts=10 instances=120
-Per-Instance Resources: mem-estimate=146.88GB mem-reservation=34.00MB thread-reservation=1
+Per-Instance Resources: mem-estimate=4.28GB mem-reservation=34.00MB thread-reservation=1
max-parallelism=609330 segment-costs=[4101117248955, 6093265120169] cpu-comparison-result=360 [max(240 (self) vs 360 (sum children))]
21:AGGREGATE [STREAMING]
| output: sum(CAST(CASE WHEN p_promo_sk IS NULL THEN CAST(1 AS TINYINT) ELSE CAST(0 AS TINYINT) END AS BIGINT)), sum(CAST(CASE WHEN p_promo_sk IS NOT NULL THEN CAST(1 AS TINYINT) ELSE CAST(0 AS TINYINT) END AS BIGINT)), count(*)
| group by: i_item_desc, w_warehouse_name, d1.d_week_seq
-| mem-estimate=146.76GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| mem-estimate=4.17GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
| tuple-ids=11 row-size=169B cardinality=535.62G cost=3268693778097
| in pipelines: 00(GETNEXT)
|
@@ -696,7 +696,7 @@ max-parallelism=374 segment-costs=[4303039497]
in pipelines: 00(GETNEXT)
---- PARALLELPLANS
Max Per-Host Resource Reservation: Memory=2.66GB Threads=79
-Per-Host Resource Estimates: Memory=1.99TB
+Per-Host Resource Estimates: Memory=127.17GB
F13:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1
| Per-Instance Resources: mem-estimate=5.98MB mem-reservation=4.00MB thread-reservation=1
| max-parallelism=1 segment-costs=[659] cpu-comparison-result=360 [max(1 (self) vs 360 (sum children))]
@@ -712,7 +712,7 @@ PLAN-ROOT SINK
| in pipelines: 22(GETNEXT)
|
F12:PLAN FRAGMENT [HASH(i_item_desc,w_warehouse_name,d1.d_week_seq)] hosts=10 instances=120
-Per-Instance Resources: mem-estimate=21.19GB mem-reservation=34.00MB thread-reservation=1
+Per-Instance Resources: mem-estimate=4.20GB mem-reservation=34.00MB thread-reservation=1
max-parallelism=155460 segment-costs=[1554521531826, 1196340444652, 473] cpu-comparison-result=360 [max(120 (self) vs 360 (sum children))]
22:TOP-N [LIMIT=100]
| order by: count(*) DESC, i_item_desc ASC, w_warehouse_name ASC, d1.d_week_seq ASC
@@ -723,7 +723,7 @@ max-parallelism=155460 segment-costs=[1554521531826, 1196340444652, 473] cpu-com
35:AGGREGATE [FINALIZE]
| output: sum:merge(CASE WHEN p_promo_sk IS NULL THEN 1 ELSE 0 END), sum:merge(CASE WHEN p_promo_sk IS NOT NULL THEN 1 ELSE 0 END), count:merge(*)
| group by: i_item_desc, w_warehouse_name, d1.d_week_seq
-| mem-estimate=21.16GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| mem-estimate=4.17GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
| tuple-ids=11 row-size=169B cardinality=59.29G cost=1079491668669
| in pipelines: 35(GETNEXT), 00(OPEN)
|
@@ -733,12 +733,12 @@ max-parallelism=155460 segment-costs=[1554521531826, 1196340444652, 473] cpu-com
| in pipelines: 00(GETNEXT)
|
F08:PLAN FRAGMENT [HASH(cs_item_sk,d2.d_date_sk)] hosts=10 instances=120
-Per-Instance Resources: mem-estimate=146.88GB mem-reservation=34.00MB thread-reservation=1
+Per-Instance Resources: mem-estimate=4.28GB mem-reservation=34.00MB thread-reservation=1
max-parallelism=609330 segment-costs=[4101117248955, 6093265120169] cpu-comparison-result=360 [max(240 (self) vs 360 (sum children))]
21:AGGREGATE [STREAMING]
| output: sum(CAST(CASE WHEN p_promo_sk IS NULL THEN CAST(1 AS TINYINT) ELSE CAST(0 AS TINYINT) END AS BIGINT)), sum(CAST(CASE WHEN p_promo_sk IS NOT NULL THEN CAST(1 AS TINYINT) ELSE CAST(0 AS TINYINT) END AS BIGINT)), count(*)
| group by: i_item_desc, w_warehouse_name, d1.d_week_seq
-| mem-estimate=146.76GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| mem-estimate=4.17GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
| tuple-ids=11 row-size=169B cardinality=535.62G cost=3268693778097
| in pipelines: 00(GETNEXT)
|
diff --git a/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q87.test b/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q87.test
index 234d0342e..8343d5574 100644
--- a/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q87.test
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q87.test
@@ -24,9 +24,9 @@ from ((select distinct c_last_name, c_first_name, d_date
# end query 87 in stream 0 using template query87.tpl
---- PLAN
Max Per-Host Resource Reservation: Memory=265.94MB Threads=1
-Per-Host Resource Estimates: Memory=536.66GB
+Per-Host Resource Estimates: Memory=224.72GB
F00:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1
-| Per-Instance Resources: mem-estimate=536.66GB mem-reservation=265.94MB thread-reservation=1 runtime-filters-memory=51.00MB
+| Per-Instance Resources: mem-estimate=224.72GB mem-reservation=265.94MB thread-reservation=1 runtime-filters-memory=51.00MB
| max-parallelism=1 segment-costs=[54790085460, 27414203232, 13758375736, 13451593165, 0]
PLAN-ROOT SINK
| output exprs: count(*)
@@ -40,7 +40,7 @@ PLAN-ROOT SINK
|
19:HASH JOIN [LEFT ANTI JOIN]
| hash predicates: d_date IS NOT DISTINCT FROM d_date, c_first_name IS NOT DISTINCT FROM c_first_name, c_last_name IS NOT DISTINCT FROM c_last_name
-| mem-estimate=121.99GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| mem-estimate=50.00GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
| tuple-ids=3 row-size=40B cardinality=7.93G cost=5459196418
| in pipelines: 05(GETNEXT), 17(OPEN)
|
@@ -103,7 +103,7 @@ PLAN-ROOT SINK
|
18:HASH JOIN [LEFT ANTI JOIN]
| hash predicates: d_date IS NOT DISTINCT FROM d_date, c_first_name IS NOT DISTINCT FROM c_first_name, c_last_name IS NOT DISTINCT FROM c_last_name
-| mem-estimate=243.61GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| mem-estimate=50.00GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
| tuple-ids=3 row-size=40B cardinality=7.93G cost=7437210998
| in pipelines: 05(GETNEXT), 11(OPEN)
|
@@ -166,7 +166,7 @@ PLAN-ROOT SINK
|
05:AGGREGATE [FINALIZE]
| group by: c_last_name, c_first_name, d_date
-| mem-estimate=96.34GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| mem-estimate=50.00GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
| tuple-ids=3 row-size=40B cardinality=7.93G cost=46814848608
| in pipelines: 05(GETNEXT), 00(OPEN)
|
diff --git a/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q95.test b/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q95.test
index e14811e46..e1a7a652f 100644
--- a/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q95.test
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q95.test
@@ -33,9 +33,9 @@ limit 100;
# end query 95 in stream 0 using template query95.tpl
---- PLAN
Max Per-Host Resource Reservation: Memory=278.25MB Threads=1
-Per-Host Resource Estimates: Memory=218.14GB
+Per-Host Resource Estimates: Memory=114.74GB
F00:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1
-| Per-Instance Resources: mem-estimate=218.14GB mem-reservation=278.25MB thread-reservation=1 runtime-filters-memory=53.00MB
+| Per-Instance Resources: mem-estimate=114.74GB mem-reservation=278.25MB thread-reservation=1 runtime-filters-memory=53.00MB
| max-parallelism=1 segment-costs=[16453556987, 30961638418, 979524347, 43005, 0, 0]
PLAN-ROOT SINK
| output exprs: count(ws_order_number), sum(ws_ext_ship_cost), sum(ws_net_profit)
@@ -160,7 +160,7 @@ PLAN-ROOT SINK
| | fk/pk conjuncts: none
| | other predicates: ws1.ws_warehouse_sk != ws2.ws_warehouse_sk
| | runtime filters: RF008[bloom] <- ws2.ws_order_number, RF009[min_max] <- ws2.ws_order_number
-| | mem-estimate=101.70GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| | mem-estimate=50.00GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
| | tuple-ids=4,5 row-size=24B cardinality=26.46G cost=7508259980
| | in pipelines: 04(GETNEXT), 05(OPEN)
| |
@@ -219,7 +219,7 @@ PLAN-ROOT SINK
| fk/pk conjuncts: none
| other predicates: ws1.ws_warehouse_sk != ws2.ws_warehouse_sk
| runtime filters: RF004[bloom] <- ws2.ws_order_number, RF005[min_max] <- ws2.ws_order_number
-| mem-estimate=101.70GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| mem-estimate=50.00GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
| tuple-ids=9,10 row-size=24B cardinality=26.46G cost=7508259980
| in pipelines: 08(GETNEXT), 09(OPEN)
|
diff --git a/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q97.test b/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q97-mem_limit-10g.test
similarity index 95%
copy from testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q97.test
copy to testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q97-mem_limit-10g.test
index 03619a839..1b84864e6 100644
--- a/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q97.test
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q97-mem_limit-10g.test
@@ -26,9 +26,9 @@ limit 100;
# end query 97 in stream 0 using template query97.tpl
---- PLAN
Max Per-Host Resource Reservation: Memory=113.94MB Threads=1
-Per-Host Resource Estimates: Memory=249.51GB
+Per-Host Resource Estimates: Memory=30.02GB
F00:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1
-| Per-Instance Resources: mem-estimate=249.51GB mem-reservation=113.94MB thread-reservation=1 runtime-filters-memory=2.00MB
+| Per-Instance Resources: mem-estimate=30.02GB mem-reservation=113.94MB thread-reservation=1 runtime-filters-memory=2.00MB
| max-parallelism=1 segment-costs=[56272781729, 28137119708, 11606178508, 0]
PLAN-ROOT SINK
| output exprs: sum(CASE WHEN ssci.customer_sk IS NOT NULL AND csci.customer_sk IS NULL THEN 1 ELSE 0 END), sum(CASE WHEN ssci.customer_sk IS NULL AND csci.customer_sk IS NOT NULL THEN 1 ELSE 0 END), sum(CASE WHEN ssci.customer_sk IS NOT NULL AND csci.customer_sk IS NOT NULL THEN 1 ELSE 0 END)
@@ -44,13 +44,13 @@ PLAN-ROOT SINK
08:HASH JOIN [FULL OUTER JOIN]
| hash predicates: ss_customer_sk = cs_bill_customer_sk, ss_item_sk = cs_item_sk
| fk/pk conjuncts: ss_customer_sk = cs_bill_customer_sk, ss_item_sk = cs_item_sk
-| mem-estimate=128.19GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| mem-estimate=10.00GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
| tuple-ids=2N,6N row-size=16B cardinality=12.96G cost=8884575742
| in pipelines: 03(GETNEXT), 07(OPEN)
|
|--07:AGGREGATE [FINALIZE]
| | group by: cs_bill_customer_sk, cs_item_sk
-| | mem-estimate=40.54GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| | mem-estimate=10.00GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
| | tuple-ids=6 row-size=8B cardinality=4.32G cost=25499692278
| | in pipelines: 07(GETNEXT), 04(OPEN)
| |
@@ -89,7 +89,7 @@ PLAN-ROOT SINK
|
03:AGGREGATE [FINALIZE]
| group by: ss_customer_sk, ss_item_sk
-| mem-estimate=80.77GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| mem-estimate=10.00GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
| tuple-ids=2 row-size=8B cardinality=8.64G cost=50998081470
| in pipelines: 03(GETNEXT), 00(OPEN)
|
@@ -127,7 +127,7 @@ PLAN-ROOT SINK
in pipelines: 00(GETNEXT)
---- DISTRIBUTEDPLAN
Max Per-Host Resource Reservation: Memory=2.15GB Threads=53
-Per-Host Resource Estimates: Memory=48.64GB
+Per-Host Resource Estimates: Memory=44.12GB
F06:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1
| Per-Instance Resources: mem-estimate=410.03KB mem-reservation=0B thread-reservation=1
| max-parallelism=1 segment-costs=[51, 0] cpu-comparison-result=240 [max(1 (self) vs 240 (sum children))]
@@ -148,7 +148,7 @@ PLAN-ROOT SINK
| in pipelines: 09(GETNEXT)
|
F02:PLAN FRAGMENT [HASH(ss_customer_sk,ss_item_sk)] hosts=10 instances=120
-Per-Instance Resources: mem-estimate=1009.71MB mem-reservation=34.00MB thread-reservation=1
+Per-Instance Resources: mem-estimate=864.74MB mem-reservation=34.00MB thread-reservation=1
max-parallelism=5230 segment-costs=[52275927885, 7286100529, 247] cpu-comparison-result=240 [max(120 (self) vs 240 (sum children))]
09:AGGREGATE
| output: sum(CAST(CASE WHEN ss_customer_sk IS NOT NULL AND cs_bill_customer_sk IS NULL THEN CAST(1 AS TINYINT) ELSE CAST(0 AS TINYINT) END AS BIGINT)), sum(CAST(CASE WHEN ss_customer_sk IS NULL AND cs_bill_customer_sk IS NOT NULL THEN CAST(1 AS TINYINT) ELSE CAST(0 AS TINYINT) END AS BIGINT)), sum(CAST(CASE WHEN ss_customer_sk IS NOT NULL AND cs_bill_customer_sk IS NOT NULL THEN CAST(1 AS TINYINT) ELSE CAST(0 AS TINYINT) END AS BIGINT))
@@ -165,12 +165,12 @@ max-parallelism=5230 segment-costs=[52275927885, 7286100529, 247] cpu-comparison
| in pipelines: 12(GETNEXT), 15(OPEN)
|
|--F07:PLAN FRAGMENT [HASH(ss_customer_sk,ss_item_sk)] hosts=10 instances=120
-| | Per-Instance Resources: mem-estimate=1.71GB mem-reservation=68.00MB thread-reservation=1
+| | Per-Instance Resources: mem-estimate=1.47GB mem-reservation=68.00MB thread-reservation=1
| | max-parallelism=5230 segment-costs=[26138631811, 4320077979] cpu-comparison-result=120 [max(120 (self) vs 120 (sum children))]
| JOIN BUILD
| | join-table-id=00 plan-id=01 cohort-id=01
| | build expressions: cs_bill_customer_sk, cs_item_sk
-| | mem-estimate=1.07GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0 cost=4320077979
+| | mem-estimate=853.33MB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0 cost=4320077979
| |
| 15:AGGREGATE [FINALIZE]
| | group by: cs_bill_customer_sk, cs_item_sk
@@ -245,7 +245,7 @@ max-parallelism=5230 segment-costs=[52275927885, 7286100529, 247] cpu-comparison
|
12:AGGREGATE [FINALIZE]
| group by: ss_customer_sk, ss_item_sk
-| mem-estimate=998.31MB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| mem-estimate=853.33MB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
| tuple-ids=2 row-size=8B cardinality=8.64G cost=50998081470
| in pipelines: 12(GETNEXT), 00(OPEN)
|
@@ -315,7 +315,7 @@ max-parallelism=1824 segment-costs=[56272761904, 8902589222] cpu-comparison-resu
in pipelines: 00(GETNEXT)
---- PARALLELPLANS
Max Per-Host Resource Reservation: Memory=2.15GB Threads=53
-Per-Host Resource Estimates: Memory=48.64GB
+Per-Host Resource Estimates: Memory=44.12GB
F06:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1
| Per-Instance Resources: mem-estimate=410.03KB mem-reservation=0B thread-reservation=1
| max-parallelism=1 segment-costs=[51, 0] cpu-comparison-result=240 [max(1 (self) vs 240 (sum children))]
@@ -336,7 +336,7 @@ PLAN-ROOT SINK
| in pipelines: 09(GETNEXT)
|
F02:PLAN FRAGMENT [HASH(ss_customer_sk,ss_item_sk)] hosts=10 instances=120
-Per-Instance Resources: mem-estimate=1009.71MB mem-reservation=34.00MB thread-reservation=1
+Per-Instance Resources: mem-estimate=864.74MB mem-reservation=34.00MB thread-reservation=1
max-parallelism=5230 segment-costs=[52275927885, 7286100529, 247] cpu-comparison-result=240 [max(120 (self) vs 240 (sum children))]
09:AGGREGATE
| output: sum(CAST(CASE WHEN ss_customer_sk IS NOT NULL AND cs_bill_customer_sk IS NULL THEN CAST(1 AS TINYINT) ELSE CAST(0 AS TINYINT) END AS BIGINT)), sum(CAST(CASE WHEN ss_customer_sk IS NULL AND cs_bill_customer_sk IS NOT NULL THEN CAST(1 AS TINYINT) ELSE CAST(0 AS TINYINT) END AS BIGINT)), sum(CAST(CASE WHEN ss_customer_sk IS NOT NULL AND cs_bill_customer_sk IS NOT NULL THEN CAST(1 AS TINYINT) ELSE CAST(0 AS TINYINT) END AS BIGINT))
@@ -353,12 +353,12 @@ max-parallelism=5230 segment-costs=[52275927885, 7286100529, 247] cpu-comparison
| in pipelines: 12(GETNEXT), 15(OPEN)
|
|--F07:PLAN FRAGMENT [HASH(ss_customer_sk,ss_item_sk)] hosts=10 instances=120
-| | Per-Instance Resources: mem-estimate=1.71GB mem-reservation=68.00MB thread-reservation=1
+| | Per-Instance Resources: mem-estimate=1.47GB mem-reservation=68.00MB thread-reservation=1
| | max-parallelism=5230 segment-costs=[26138631811, 4320077979] cpu-comparison-result=120 [max(120 (self) vs 120 (sum children))]
| JOIN BUILD
| | join-table-id=00 plan-id=01 cohort-id=01
| | build expressions: cs_bill_customer_sk, cs_item_sk
-| | mem-estimate=1.07GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0 cost=4320077979
+| | mem-estimate=853.33MB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0 cost=4320077979
| |
| 15:AGGREGATE [FINALIZE]
| | group by: cs_bill_customer_sk, cs_item_sk
@@ -433,7 +433,7 @@ max-parallelism=5230 segment-costs=[52275927885, 7286100529, 247] cpu-comparison
|
12:AGGREGATE [FINALIZE]
| group by: ss_customer_sk, ss_item_sk
-| mem-estimate=998.31MB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| mem-estimate=853.33MB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
| tuple-ids=2 row-size=8B cardinality=8.64G cost=50998081470
| in pipelines: 12(GETNEXT), 00(OPEN)
|
diff --git a/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q97.test b/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q97-mem_limit_executors-20g.test
similarity index 98%
copy from testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q97.test
copy to testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q97-mem_limit_executors-20g.test
index 03619a839..403071e1d 100644
--- a/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q97.test
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q97-mem_limit_executors-20g.test
@@ -26,9 +26,9 @@ limit 100;
# end query 97 in stream 0 using template query97.tpl
---- PLAN
Max Per-Host Resource Reservation: Memory=113.94MB Threads=1
-Per-Host Resource Estimates: Memory=249.51GB
+Per-Host Resource Estimates: Memory=60.02GB
F00:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1
-| Per-Instance Resources: mem-estimate=249.51GB mem-reservation=113.94MB thread-reservation=1 runtime-filters-memory=2.00MB
+| Per-Instance Resources: mem-estimate=60.02GB mem-reservation=113.94MB thread-reservation=1 runtime-filters-memory=2.00MB
| max-parallelism=1 segment-costs=[56272781729, 28137119708, 11606178508, 0]
PLAN-ROOT SINK
| output exprs: sum(CASE WHEN ssci.customer_sk IS NOT NULL AND csci.customer_sk IS NULL THEN 1 ELSE 0 END), sum(CASE WHEN ssci.customer_sk IS NULL AND csci.customer_sk IS NOT NULL THEN 1 ELSE 0 END), sum(CASE WHEN ssci.customer_sk IS NOT NULL AND csci.customer_sk IS NOT NULL THEN 1 ELSE 0 END)
@@ -44,13 +44,13 @@ PLAN-ROOT SINK
08:HASH JOIN [FULL OUTER JOIN]
| hash predicates: ss_customer_sk = cs_bill_customer_sk, ss_item_sk = cs_item_sk
| fk/pk conjuncts: ss_customer_sk = cs_bill_customer_sk, ss_item_sk = cs_item_sk
-| mem-estimate=128.19GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| mem-estimate=20.00GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
| tuple-ids=2N,6N row-size=16B cardinality=12.96G cost=8884575742
| in pipelines: 03(GETNEXT), 07(OPEN)
|
|--07:AGGREGATE [FINALIZE]
| | group by: cs_bill_customer_sk, cs_item_sk
-| | mem-estimate=40.54GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| | mem-estimate=20.00GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
| | tuple-ids=6 row-size=8B cardinality=4.32G cost=25499692278
| | in pipelines: 07(GETNEXT), 04(OPEN)
| |
@@ -89,7 +89,7 @@ PLAN-ROOT SINK
|
03:AGGREGATE [FINALIZE]
| group by: ss_customer_sk, ss_item_sk
-| mem-estimate=80.77GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| mem-estimate=20.00GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
| tuple-ids=2 row-size=8B cardinality=8.64G cost=50998081470
| in pipelines: 03(GETNEXT), 00(OPEN)
|
diff --git a/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q97.test b/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q97.test
index 03619a839..d93fc7724 100644
--- a/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q97.test
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/tpcds_cpu_cost/tpcds-q97.test
@@ -26,9 +26,9 @@ limit 100;
# end query 97 in stream 0 using template query97.tpl
---- PLAN
Max Per-Host Resource Reservation: Memory=113.94MB Threads=1
-Per-Host Resource Estimates: Memory=249.51GB
+Per-Host Resource Estimates: Memory=140.56GB
F00:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1
-| Per-Instance Resources: mem-estimate=249.51GB mem-reservation=113.94MB thread-reservation=1 runtime-filters-memory=2.00MB
+| Per-Instance Resources: mem-estimate=140.56GB mem-reservation=113.94MB thread-reservation=1 runtime-filters-memory=2.00MB
| max-parallelism=1 segment-costs=[56272781729, 28137119708, 11606178508, 0]
PLAN-ROOT SINK
| output exprs: sum(CASE WHEN ssci.customer_sk IS NOT NULL AND csci.customer_sk IS NULL THEN 1 ELSE 0 END), sum(CASE WHEN ssci.customer_sk IS NULL AND csci.customer_sk IS NOT NULL THEN 1 ELSE 0 END), sum(CASE WHEN ssci.customer_sk IS NOT NULL AND csci.customer_sk IS NOT NULL THEN 1 ELSE 0 END)
@@ -44,7 +44,7 @@ PLAN-ROOT SINK
08:HASH JOIN [FULL OUTER JOIN]
| hash predicates: ss_customer_sk = cs_bill_customer_sk, ss_item_sk = cs_item_sk
| fk/pk conjuncts: ss_customer_sk = cs_bill_customer_sk, ss_item_sk = cs_item_sk
-| mem-estimate=128.19GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| mem-estimate=50.00GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
| tuple-ids=2N,6N row-size=16B cardinality=12.96G cost=8884575742
| in pipelines: 03(GETNEXT), 07(OPEN)
|
@@ -89,7 +89,7 @@ PLAN-ROOT SINK
|
03:AGGREGATE [FINALIZE]
| group by: ss_customer_sk, ss_item_sk
-| mem-estimate=80.77GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| mem-estimate=50.00GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
| tuple-ids=2 row-size=8B cardinality=8.64G cost=50998081470
| in pipelines: 03(GETNEXT), 00(OPEN)
|
diff --git a/tests/custom_cluster/test_executor_groups.py b/tests/custom_cluster/test_executor_groups.py
index 965aab1ae..70a11a3c2 100644
--- a/tests/custom_cluster/test_executor_groups.py
+++ b/tests/custom_cluster/test_executor_groups.py
@@ -74,6 +74,61 @@ order by c_customer_id
limit 100
"""
+# TPC-DS Q67 to test memory bounding against big memory estimation.
+TPCDS_Q67 = """
+select *
+from (select i_category
+ ,i_class
+ ,i_brand
+ ,i_product_name
+ ,d_year
+ ,d_qoy
+ ,d_moy
+ ,s_store_id
+ ,sumsales
+ ,rank() over (partition by i_category order by sumsales desc) rk
+ from (select i_category
+ ,i_class
+ ,i_brand
+ ,i_product_name
+ ,d_year
+ ,d_qoy
+ ,d_moy
+ ,s_store_id
+ ,sum(coalesce(ss_sales_price*ss_quantity,0)) sumsales
+ from tpcds_partitioned_parquet_snap.store_sales
+ ,tpcds_partitioned_parquet_snap.date_dim
+ ,tpcds_partitioned_parquet_snap.store
+ ,tpcds_partitioned_parquet_snap.item
+ where ss_sold_date_sk=d_date_sk
+ and ss_item_sk=i_item_sk
+ and ss_store_sk = s_store_sk
+ and d_month_seq between 1196 and 1196+11
+ group by rollup(
+ i_category,
+ i_class,
+ i_brand,
+ i_product_name,
+ d_year,
+ d_qoy,
+ d_moy,
+ s_store_id)
+ ) dw1
+) dw2
+where rk <= 100
+order by i_category
+ ,i_class
+ ,i_brand
+ ,i_product_name
+ ,d_year
+ ,d_qoy
+ ,d_moy
+ ,s_store_id
+ ,sumsales
+ ,rk
+limit 100;
+"""
+
DEFAULT_RESOURCE_POOL = "default-pool"
DEBUG_ACTION_DELAY_SCAN = "HDFS_SCANNER_THREAD_OBTAINED_RANGE:SLEEP@1000"
@@ -920,9 +975,9 @@ class TestExecutorGroups(CustomClusterTestSuite):
# Add an exec group with 8 admission slots and 2 executors.
self._add_executor_group("group", 2, admission_control_slots=8,
resource_pool="root.small", extra_args="-mem_limit=2g")
- # Add another exec group with 64 admission slots and 3 executors.
+ # Add another exec group with 64 admission slots, 3 executors, and 4g mem_limit.
self._add_executor_group("group", 3, admission_control_slots=64,
- resource_pool="root.large", extra_args="-mem_limit=2g")
+ resource_pool="root.large", extra_args="-mem_limit=4g")
assert self._get_num_executor_groups(only_healthy=True) == 3
assert self._get_num_executor_groups(only_healthy=True,
exec_group_set_prefix="root.tiny") == 1
@@ -1034,14 +1089,22 @@ class TestExecutorGroups(CustomClusterTestSuite):
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.
+ # for COMPUTE STATS test. Forcing large parallelism to speed up CTAS.
+ # Otherwise, query will go to tiny pool.
+ self.client.set_configuration({
+ 'REQUEST_POOL': 'root.large',
+ 'PROCESSING_COST_MIN_THREADS': '4'})
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.tiny", "ExecutorGroupsConsidered: 1",
- "Verdict: Match", "CpuAsk: 1", "AvgAdmissionSlotsPerExecutor: 1"])
+ ["Executor Group: root.large", "ExecutorGroupsConsidered: 1",
+ "Verdict: query option REQUEST_POOL=root.large is set. "
+ "Memory and cpu limit checking is skipped."])
+ self.client.set_configuration({
+ 'REQUEST_POOL': '',
+ 'PROCESSING_COST_MIN_THREADS': ''})
compute_stats_query = ("compute stats {0}.{1}").format(
unique_database, "store_sales_subset")
@@ -1050,18 +1113,18 @@ class TestExecutorGroups(CustomClusterTestSuite):
# parent query. Currently both child query assign to root.tiny.
# TODO: Revise the CTAS query above such that the two child queries assigned to
# distinct executor group set.
- self._verify_total_admitted_queries("root.tiny", 4)
+ self._verify_total_admitted_queries("root.tiny", 3)
self._verify_total_admitted_queries("root.small", 0)
- self._verify_total_admitted_queries("root.large", 1)
+ self._verify_total_admitted_queries("root.large", 2)
self._run_query_and_verify_profile(compute_stats_query,
["ExecutorGroupsConsidered: 1",
"Verdict: Assign to first group because query is not auto-scalable"],
["Query Options (set by configuration): REQUEST_POOL=",
"EffectiveParallelism:", "CpuAsk:", "AvgAdmissionSlotsPerExecutor:",
"Executor Group:"])
- self._verify_total_admitted_queries("root.tiny", 6)
+ self._verify_total_admitted_queries("root.tiny", 5)
self._verify_total_admitted_queries("root.small", 0)
- self._verify_total_admitted_queries("root.large", 1)
+ self._verify_total_admitted_queries("root.large", 2)
# Test that child queries follow REQUEST_POOL that is set through client
# configuration. Two child queries should all run in root.small.
@@ -1082,7 +1145,7 @@ class TestExecutorGroups(CustomClusterTestSuite):
"ExecutorGroupsConsidered: 1",
"Verdict: Assign to first group because query is not auto-scalable"],
["EffectiveParallelism:", "CpuAsk:", "AvgAdmissionSlotsPerExecutor:"])
- self._verify_total_admitted_queries("root.large", 3)
+ 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,
@@ -1249,13 +1312,80 @@ class TestExecutorGroups(CustomClusterTestSuite):
])
# END test slot count strategy
+ # BEGIN test memory capping.
+ # Set MAX_FRAGMENT_INSTANCES_PER_NODE=4 and
+ # disable MEM_ESTIMATE_SCALE_FOR_SPILLING_OPERATOR.
+ self._set_query_options({
+ 'MEM_ESTIMATE_SCALE_FOR_SPILLING_OPERATOR': 0.0,
+ 'MAX_FRAGMENT_INSTANCES_PER_NODE': 4})
+ self._run_query_and_verify_profile(
+ TPCDS_Q67,
+ ["Executor Group: root.large-group", "ExecutorGroupsConsidered: 3",
+ "Per-Host Resource Estimates: Memory=9.25GB",
+ "mem-estimate=1.38GB", # memory estimate of 17:AGGREGATE.
+ "mem-estimate=1.11GB", # memory estimate of 07:AGGREGATE.
+ "Cluster Memory Admitted: 10.90 GB",
+ "MemoryAsk: 27.76 GB"])
+
+ # Set very low MEM_ESTIMATE_SCALE_FOR_SPILLING_OPERATOR.
+ self._set_query_options({'MEM_ESTIMATE_SCALE_FOR_SPILLING_OPERATOR': 0.00001})
+ self._run_query_and_verify_profile(
+ TPCDS_Q67,
+ ["Executor Group: root.large-group", "ExecutorGroupsConsidered: 3",
+ "Per-Host Resource Estimates: Memory=1.86GB",
+ "mem-estimate=210.92MB", # memory estimate of 17:AGGREGATE.
+ "mem-estimate=211.13MB", # memory estimate of 07:AGGREGATE.
+ "Cluster Memory Admitted: 6.07 GB",
+ "MemoryAsk: 5.57 GB"])
+
+ # Set MEM_ESTIMATE_SCALE_FOR_SPILLING_OPERATOR to maximum.
+ self._set_query_options({'MEM_ESTIMATE_SCALE_FOR_SPILLING_OPERATOR': 1.0})
+ self._run_query_and_verify_profile(
+ TPCDS_Q67,
+ ["Executor Group: root.large-group", "ExecutorGroupsConsidered: 3",
+ "Per-Host Resource Estimates: Memory=9.25GB",
+ "mem-estimate=1.38GB", # memory estimate of 17:AGGREGATE.
+ "mem-estimate=1.11GB", # memory estimate of 07:AGGREGATE.
+ "Cluster Memory Admitted: 10.90 GB",
+ "MemoryAsk: 27.76 GB"])
+
+ # Set MEM_LIMIT_EXECUTORS to 3GB, lower than --mem_limit=4g of root.large's
+ # backend executors.
+ self._set_query_options({'MEM_LIMIT_EXECUTORS': '3GB'})
+ self._run_query_and_verify_profile(
+ TPCDS_Q67,
+ ["Executor Group: root.large-group", "ExecutorGroupsConsidered: 3",
+ "Per-Host Resource Estimates: Memory=6.68GB",
+ "mem-estimate=1.00GB", # memory estimate of 17:AGGREGATE.
+ "mem-estimate=768.00MB", # memory estimate of 07:AGGREGATE.
+ "Cluster Memory Admitted: 9.10 GB",
+ "MemoryAsk: 20.03 GB"])
+
+ # Set MEM_LIMIT to 2GB, lower than MEM_LIMIT_EXECUTORS.
+ self._set_query_options({'MEM_LIMIT': '2GB'})
+ self._run_query_and_verify_profile(
+ TPCDS_Q67,
+ ["Executor Group: root.large-group", "ExecutorGroupsConsidered: 3",
+ "Per-Host Resource Estimates: Memory=4.68GB",
+ "mem-estimate=682.67MB", # memory estimate of 17:AGGREGATE.
+ "mem-estimate=512.00MB", # memory estimate of 07:AGGREGATE.
+ "Cluster Memory Admitted: 8.00 GB",
+ "MemoryAsk: 14.03 GB"])
+
+ self._set_query_options({
+ 'MEM_ESTIMATE_SCALE_FOR_SPILLING_OPERATOR': 0.0,
+ 'MAX_FRAGMENT_INSTANCES_PER_NODE': '',
+ 'MEM_LIMIT_EXECUTORS': '',
+ 'MEM_LIMIT': ''})
+ # END test memory capping.
+
# Check resource pools on the Web queries site and admission site
- self._verify_query_num_for_resource_pool("root.tiny", 18)
+ self._verify_query_num_for_resource_pool("root.tiny", 17)
self._verify_query_num_for_resource_pool("root.small", 3)
- self._verify_query_num_for_resource_pool("root.large", 7)
- self._verify_total_admitted_queries("root.tiny", 18)
+ self._verify_query_num_for_resource_pool("root.large", 13)
+ self._verify_total_admitted_queries("root.tiny", 17)
self._verify_total_admitted_queries("root.small", 3)
- self._verify_total_admitted_queries("root.large", 7)
+ self._verify_total_admitted_queries("root.large", 13)
@UniqueDatabase.parametrize(sync_ddl=True)
@pytest.mark.execute_serially