This is an automated email from the ASF dual-hosted git repository.
HappenLee pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 97705a7f457 [improvement](be) Adapt bucket counts for bucketed
aggregation (#69425)
97705a7f457 is described below
commit 97705a7f4578dd15ff277b6a5e8dd70199edb996
Author: HappenLee <[email protected]>
AuthorDate: Mon Oct 26 19:07:01 2026 +0800
[improvement](be) Adapt bucket counts for bucketed aggregation (#69425)
### What problem does this PR solve?
Issue Number: N/A
Related PR: N/A
Problem Summary: Bucketed aggregation allocates 256 hash tables per sink
even for serial execution, and routing with mixed-hash bits 24-31 overlaps
phmap probe-index bits as tables grow. Add bucketed_agg_num_buckets with
automatic or explicit power-of-two bucket selection, capture it in the plan and
forward it in an optional Thrift field. Automatic mode uses one bucket when
both sink and source have one instance on a BE, and 256 otherwise; absent
fields retain 256 buckets. Route with [...]
### Release note
Add the bucketed_agg_num_buckets session variable. The default 0 selects
one bucket for a single sink and source on a BE, and 256 otherwise. Explicit
values must be powers of two from 1 through 256. SQL aggregation results are
unchanged.
### Check List (For Author)
- Test: TeamCity COMPILE, FE UT, BE UT, Cloud UT, P0, Cloud P0, Vault P0,
External and NonConcurrent regression, performance, and BE/FE coverage checks
passed on 576b4676128a51a6f773adb903f3af730e1973fd. GitHub formatting,
Checkstyle, license and code-review checks passed. Independent GCC 15.1.0 -O3
-Werror compile reproducer passed after the completion-flag allocation fix.
- Behavior changed: Yes; automatic bucket selection and a per-session
bucket-count override, with unchanged aggregation semantics.
- Does this need documentation: Yes; session-variable reference. The inline
setting description is included; the website documentation PR remains pending.
---
.../bucketed_aggregation_sink_operator.cpp | 143 ++-
.../operator/bucketed_aggregation_sink_operator.h | 10 +-
.../bucketed_aggregation_source_operator.cpp | 19 +-
be/src/exec/pipeline/dependency.cpp | 35 +-
be/src/exec/pipeline/dependency.h | 23 +-
be/test/exec/operator/agg_operator_test.cpp | 4 +-
.../pipeline/bucketed_agg_shared_state_test.cpp | 91 ++
.../glue/translator/PhysicalPlanTranslator.java | 3 +-
.../doris/planner/BucketedAggregationNode.java | 7 +-
.../java/org/apache/doris/qe/SessionVariable.java | 23 +-
.../BucketedAggregateTranslatorTest.java | 48 +
.../rules/analysis/AnalyzeWhereSubqueryTest.java | 5 +-
.../doris/qe/SessionVariableBucketedAggTest.java | 65 +
gensrc/thrift/PlanNodes.thrift | 3 +
.../agg_strategy/bucketed_hash_agg_num_buckets.out | 1236 ++++++++++++++++++++
.../bucketed_hash_agg_num_buckets.groovy | 177 +++
16 files changed, 1801 insertions(+), 91 deletions(-)
diff --git a/be/src/exec/operator/bucketed_aggregation_sink_operator.cpp
b/be/src/exec/operator/bucketed_aggregation_sink_operator.cpp
index 9c76ec7e8a7..2471735a80f 100644
--- a/be/src/exec/operator/bucketed_aggregation_sink_operator.cpp
+++ b/be/src/exec/operator/bucketed_aggregation_sink_operator.cpp
@@ -54,6 +54,7 @@ Status BucketedAggSinkLocalState::init(RuntimeState* state,
LocalSinkStateInfo&
_expr_timer = ADD_TIMER(Base::custom_profile(), "ExprTime");
_hash_table_compute_timer = ADD_TIMER(Base::custom_profile(),
"HashTableComputeTime");
_hash_table_emplace_timer = ADD_TIMER(Base::custom_profile(),
"HashTableEmplaceTime");
+ _partition_timer = ADD_TIMER(Base::custom_profile(), "PartitionTime");
_hash_table_input_counter =
ADD_COUNTER(Base::custom_profile(), "HashTableInputCount",
TUnit::UNIT);
_memory_usage_arena = ADD_COUNTER(custom_profile(), "MemoryUsageArena",
TUnit::BYTES);
@@ -72,29 +73,40 @@ Status BucketedAggSinkLocalState::open(RuntimeState* state)
{
// Initialize per-instance data and shared metadata. Multiple sink
instances call open()
// concurrently, so all shared-state writes must be inside call_once to
avoid data races.
// init_instances stores the init status in shared state so all threads
see failures.
- RETURN_IF_ERROR(shared_state.init_instances(state->task_num(), [&]() ->
Status {
- // Copy metadata to shared state (once, from the first instance to
reach here).
- shared_state.align_aggregate_states = p._align_aggregate_states;
- shared_state.total_size_of_aggregate_states =
p._total_size_of_aggregate_states;
- shared_state.offsets_of_aggregate_states =
p._offsets_of_aggregate_states;
- shared_state.make_nullable_keys = p._make_nullable_keys;
-
- shared_state.probe_expr_ctxs.resize(p._probe_expr_ctxs.size());
- for (size_t i = 0; i < shared_state.probe_expr_ctxs.size(); i++) {
- RETURN_IF_ERROR(p._probe_expr_ctxs[i]->clone(state,
shared_state.probe_expr_ctxs[i]));
- }
-
- for (auto& evaluator : p._aggregate_evaluators) {
-
shared_state.aggregate_evaluators.push_back(evaluator->clone(state, p._pool));
- }
-
- // Detect simple_count: exactly one COUNT(*) / COUNT(slot), with GROUP
BY present.
- // Bucketed agg always has GROUP BY (without-key not supported).
- if (p._aggregate_evaluators.size() == 1 &&
p._aggregate_evaluators[0]->is_simple_count()) {
- shared_state.use_simple_count = true;
- }
- return Status::OK();
- }));
+ RETURN_IF_ERROR(shared_state.init_instances(
+ state->task_num(),
+ [&]() -> Status {
+ // Copy metadata to shared state (once, from the first
instance to reach here).
+ shared_state.align_aggregate_states =
p._align_aggregate_states;
+ shared_state.total_size_of_aggregate_states =
p._total_size_of_aggregate_states;
+ shared_state.offsets_of_aggregate_states =
p._offsets_of_aggregate_states;
+ shared_state.make_nullable_keys = p._make_nullable_keys;
+
+ shared_state.probe_expr_ctxs.resize(p._probe_expr_ctxs.size());
+ for (size_t i = 0; i < shared_state.probe_expr_ctxs.size();
i++) {
+ RETURN_IF_ERROR(
+ p._probe_expr_ctxs[i]->clone(state,
shared_state.probe_expr_ctxs[i]));
+ }
+
+ for (auto& evaluator : p._aggregate_evaluators) {
+
shared_state.aggregate_evaluators.push_back(evaluator->clone(state, p._pool));
+ }
+
+ // Detect simple_count: exactly one COUNT(*) / COUNT(slot),
with GROUP BY present.
+ // Bucketed agg always has GROUP BY (without-key not
supported).
+ if (p._aggregate_evaluators.size() == 1 &&
+ p._aggregate_evaluators[0]->is_simple_count()) {
+ shared_state.use_simple_count = true;
+ }
+ return Status::OK();
+ },
+ p._num_buckets));
+
+ _num_buckets = shared_state.num_buckets;
+ COUNTER_SET(ADD_COUNTER(custom_profile(), "NumBuckets", TUnit::UNIT),
+ static_cast<int64_t>(_num_buckets));
+ COUNTER_SET(ADD_COUNTER(custom_profile(), "NumSinkInstances", TUnit::UNIT),
+ static_cast<int64_t>(shared_state.num_sink_instances));
// Now safe to access per_instance_data since init_instances has been
called.
auto& inst = shared_state.per_instance_data[_instance_idx];
@@ -117,13 +129,13 @@ Status BucketedAggSinkLocalState::open(RuntimeState*
state) {
_aggregate_evaluators.push_back(evaluator->clone(state, p._pool));
}
- // Set up _bucket_agg_data as 256 pointers into this instance's bucket
hash tables.
- _bucket_agg_data.resize(BUCKETED_AGG_NUM_BUCKETS);
- for (int b = 0; b < BUCKETED_AGG_NUM_BUCKETS; ++b) {
+ // Set up _bucket_agg_data as pointers into this instance's bucket hash
tables.
+ _bucket_agg_data.resize(_num_buckets);
+ for (int b = 0; b < _num_buckets; ++b) {
_bucket_agg_data[b] = inst.bucket_agg_data[b].get();
}
- // Initialize hash method for all 256 bucket hash tables. Each bucket uses
+ // Initialize hash method for all bucket hash tables. Each bucket uses
// the same hash key type. We use per-instance _probe_expr_ctxs here
(read-only
// for data types), which is safe since they were cloned above.
RETURN_IF_ERROR(_init_hash_method(_probe_expr_ctxs));
@@ -183,7 +195,8 @@ Status
BucketedAggSinkLocalState::_execute_with_serialized_key(Block* block) {
}
auto rows = (uint32_t)block->rows();
- if (_places.size() < rows) {
+ // Inline COUNT updates mapped counters and never reads the places array.
+ if (!shared_state.use_simple_count && _places.size() < rows) {
_places.resize(rows);
}
@@ -205,7 +218,7 @@ void
BucketedAggSinkLocalState::_emplace_into_hash_table(AggregateDataPtr* place
auto& p = Base::_parent->template cast<BucketedAggSinkOperatorX>();
// Use bucket 0's method to compute keys and hash values for all rows.
- // All 256 buckets use the same hash key type, so the hash function is
identical.
+ // All buckets use the same hash key type, so the hash function is
identical.
std::visit(
Overload {[&](std::monostate& arg) -> void {
throw doris::Exception(ErrorCode::INTERNAL_ERROR,
"uninited hash table");
@@ -256,23 +269,37 @@ void
BucketedAggSinkLocalState::_emplace_into_hash_table(AggregateDataPtr* place
{
SCOPED_TIMER(_hash_table_emplace_timer);
- // Phase 1: Pre-group rows by bucket index.
- // Single pass over hash_values to partition row
indices
- // into 256 bucket groups. This converts random
256-way scatter
- // into sequential per-bucket batches with good
cache locality.
- for (int b = 0; b < BUCKETED_AGG_NUM_BUCKETS;
++b) {
- _bucket_row_indices[b].clear();
+ // A single bucket can consume the whole batch
without scattering.
+ if (_num_buckets == 1) {
+ lazy_emplace_batch(agg_method, state,
num_rows, creator,
+ creator_for_null_key,
+ [&](uint32_t row, auto&
mapped) {
+ if (use_simple_count)
{
+
++reinterpret_cast<UInt64&>(mapped);
+ } else {
+ places[row] =
mapped;
+ }
+ });
+ COUNTER_UPDATE(_hash_table_input_counter,
num_rows);
+ return;
}
- for (uint32_t i = 0; i < num_rows; ++i) {
- int bucket =
static_cast<int>((agg_method.hash_values[i] >> 24) &
- 0xFF);
- _bucket_row_indices[bucket].push_back(i);
+
+ // Phase 1: Group rows by bucket to turn random
scatter into sequential batches.
+ {
+ SCOPED_TIMER(_partition_timer);
+ for (int b = 0; b < _num_buckets; ++b) {
+ _bucket_row_indices[b].clear();
+ }
+ for (uint32_t i = 0; i < num_rows; ++i) {
+ int bucket =
+
static_cast<int>(BucketedAggSharedState::bucket_index(
+
agg_method.hash_values[i], _num_buckets));
+ _bucket_row_indices[bucket].push_back(i);
+ }
}
- // Phase 2: Process each bucket's rows as a
batch.
- // All rows in a bucket hit the same hash table,
enabling
- // effective prefetch and cache-friendly access
patterns.
- for (int b = 0; b < BUCKETED_AGG_NUM_BUCKETS;
++b) {
+ // Phase 2: Process each bucket's rows as a
batch for cache locality and prefetch.
+ for (int b = 0; b < _num_buckets; ++b) {
auto& indices = _bucket_row_indices[b];
if (indices.empty()) {
continue;
@@ -282,9 +309,7 @@ void
BucketedAggSinkLocalState::_emplace_into_hash_table(AggregateDataPtr* place
_bucket_agg_data[b]->method_variant);
auto& bucket_ht = *bucket_method.hash_table;
- // Batch emplace with prefetch into this
bucket's hash table.
for (size_t j = 0; j < indices.size(); ++j) {
- // Prefetch ahead within this bucket's
hash table.
if (j + HASH_MAP_PREFETCH_DIST <
indices.size()) {
uint32_t prefetch_row =
indices[j +
HASH_MAP_PREFETCH_DIST];
@@ -313,7 +338,7 @@ void
BucketedAggSinkLocalState::_emplace_into_hash_table(AggregateDataPtr* place
size_t BucketedAggSinkLocalState::_get_hash_table_size() const {
size_t total = 0;
- for (int b = 0; b < BUCKETED_AGG_NUM_BUCKETS; ++b) {
+ for (int b = 0; b < _num_buckets; ++b) {
total += std::visit(Overload {[&](std::monostate& arg) -> size_t {
return 0; },
[&](auto& agg_method) -> size_t {
return agg_method.hash_table->size();
@@ -324,9 +349,9 @@ size_t BucketedAggSinkLocalState::_get_hash_table_size()
const {
}
Status BucketedAggSinkLocalState::_init_hash_method(const VExprContextSPtrs&
probe_exprs) {
- // Initialize all 256 bucket hash tables with the same hash key type.
+ // Initialize all bucket hash tables with the same hash key type.
auto data_types = get_data_types(probe_exprs);
- for (int b = 0; b < BUCKETED_AGG_NUM_BUCKETS; ++b) {
+ for (int b = 0; b < _num_buckets; ++b) {
RETURN_IF_ERROR(init_hash_method<BucketedAggDataVariants>(_bucket_agg_data[b],
data_types,
true /*
is_first_phase */));
}
@@ -335,7 +360,7 @@ Status BucketedAggSinkLocalState::_init_hash_method(const
VExprContextSPtrs& pro
void BucketedAggSinkLocalState::_update_memusage() {
int64_t hash_table_memory_usage = 0;
- for (int b = 0; b < BUCKETED_AGG_NUM_BUCKETS; ++b) {
+ for (int b = 0; b < _num_buckets; ++b) {
std::visit(Overload {[&](std::monostate& arg) -> void {},
[&](auto& agg_method) -> void {
hash_table_memory_usage +=
@@ -351,23 +376,15 @@ void BucketedAggSinkLocalState::_update_memusage() {
}
size_t BucketedAggSinkLocalState::get_reserve_mem_size(RuntimeState* state,
bool eos) const {
- // Estimate the memory needed for the next batch across all 256 bucket
hash tables.
- // Each bucket's hash table may need to resize, so we take the max
estimate across
- // all buckets and multiply conservatively. In practice, data is
distributed across
- // buckets, so the per-bucket growth is batch_size/256 on average, but we
use the
- // full batch_size estimate from the largest bucket for safety.
+ // Sum the growth estimates for all active buckets. Allow twice the average
+ // batch contribution per bucket, capped at the full batch for a single
bucket.
size_t size_to_reserve = 0;
- for (int b = 0; b < BUCKETED_AGG_NUM_BUCKETS; ++b) {
+ for (int b = 0; b < _num_buckets; ++b) {
size_to_reserve += std::visit(
Overload {[&](std::monostate& arg) -> size_t { return 0; },
[&](auto& agg_method) -> size_t {
- // estimate_memory returns the delta needed if
we insert
- // num_elem more entries. Since rows are
distributed across
- // 256 buckets, each bucket sees ~batch_size/256
rows on average.
- // However, skew is possible, so we use a
conservative per-bucket
- // estimate of batch_size/128 (2x average).
auto per_bucket_rows = std::max<size_t>(
- 1, state->batch_size() /
(BUCKETED_AGG_NUM_BUCKETS / 2));
+ 1, state->batch_size() / std::max(1,
_num_buckets / 2));
return
agg_method.hash_table->estimate_memory(per_bucket_rows);
}},
_bucket_agg_data[b]->method_variant);
@@ -408,6 +425,10 @@ bool BucketedAggSinkOperatorX::has_blockable_aggregate()
const {
Status BucketedAggSinkOperatorX::init(const TPlanNode& tnode, RuntimeState*
state) {
RETURN_IF_ERROR(DataSinkOperatorX<BucketedAggSinkLocalState>::init(tnode,
state));
+ if (tnode.bucketed_agg_node.__isset.num_buckets) {
+ _num_buckets = tnode.bucketed_agg_node.num_buckets;
+ }
+
RETURN_IF_ERROR(
VExpr::create_expr_trees(tnode.bucketed_agg_node.grouping_exprs,
_probe_expr_ctxs));
diff --git a/be/src/exec/operator/bucketed_aggregation_sink_operator.h
b/be/src/exec/operator/bucketed_aggregation_sink_operator.h
index 1cd6a16fbe6..6096f2e1963 100644
--- a/be/src/exec/operator/bucketed_aggregation_sink_operator.h
+++ b/be/src/exec/operator/bucketed_aggregation_sink_operator.h
@@ -29,7 +29,7 @@ namespace doris {
class BucketedAggSinkOperatorX;
/// Sink-side local state for bucketed hash aggregation.
-/// Each pipeline instance builds 256 per-bucket hash tables (two-level hash
table).
+/// Each pipeline instance builds a configurable number of per-bucket hash
tables.
/// No locking: each instance writes to per_instance_data[_instance_idx].
class BucketedAggSinkLocalState : public
PipelineXSinkLocalState<BucketedAggSharedState> {
public:
@@ -59,8 +59,9 @@ private:
void _update_memusage();
int _instance_idx = 0;
+ int _num_buckets = 0;
/// Pointers into shared_state->per_instance_data[_instance_idx].
- std::vector<BucketedAggDataVariants*> _bucket_agg_data; // [256]
+ std::vector<BucketedAggDataVariants*> _bucket_agg_data;
Arena* _arena = nullptr;
/// Per-instance clones of probe expression contexts. Required because
@@ -83,6 +84,7 @@ private:
RuntimeProfile::Counter* _hash_table_compute_timer = nullptr;
RuntimeProfile::Counter* _hash_table_emplace_timer = nullptr;
+ RuntimeProfile::Counter* _partition_timer = nullptr;
RuntimeProfile::Counter* _hash_table_input_counter = nullptr;
RuntimeProfile::Counter* _build_timer = nullptr;
RuntimeProfile::Counter* _expr_timer = nullptr;
@@ -98,7 +100,7 @@ private:
/// Bucketed hash aggregation sink operator.
/// Fuses local + global aggregation for single-BE deployments.
-/// Each pipeline instance builds 256 per-bucket hash tables from raw input.
+/// Each pipeline instance builds per-bucket hash tables from raw input.
/// The source operator then merges across instances per-bucket (second-phase
agg).
class BucketedAggSinkOperatorX final : public
DataSinkOperatorX<BucketedAggSinkLocalState> {
public:
@@ -136,6 +138,8 @@ public:
private:
friend class BucketedAggSinkLocalState;
+ // Missing Thrift field preserves the behavior of an older FE. Zero
selects auto.
+ int _num_buckets = BUCKETED_AGG_NUM_BUCKETS;
std::vector<AggFnEvaluator*> _aggregate_evaluators;
// Bucketed agg is one-phase: intermediate and output tuples are identical.
diff --git a/be/src/exec/operator/bucketed_aggregation_source_operator.cpp
b/be/src/exec/operator/bucketed_aggregation_source_operator.cpp
index 44221f14a4c..b1d20022bd8 100644
--- a/be/src/exec/operator/bucketed_aggregation_source_operator.cpp
+++ b/be/src/exec/operator/bucketed_aggregation_source_operator.cpp
@@ -217,6 +217,11 @@ void BucketedAggLocalState::_wake_up_other_sources() {
int BucketedAggLocalState::_merge_bucket(int bucket, int merge_target) {
SCOPED_TIMER(_merge_timer);
auto& shared_state = *_shared_state;
+ // The sole sink already owns the final states. Source-side bucket
ownership
+ // still coordinates output when the number of source tasks differs.
+ if (shared_state.num_sink_instances == 1) {
+ return 0;
+ }
auto& bs = shared_state.bucket_states[bucket];
// Other source instances may merge other buckets at the same time, so
aggregate
// function merges must allocate from this source instance's own arena.
@@ -539,7 +544,7 @@ Status
BucketedAggLocalState::_merge_and_output_null_keys(RuntimeState* state, B
size_t key_size = shared_state.probe_expr_ctxs.size();
int merge_target =
shared_state.merge_target_instance.load(std::memory_order_relaxed);
- // Merge null keys from all 256 buckets (in merge target) into bucket 0.
+ // Merge null keys from all buckets (in merge target) into bucket 0.
// After per-bucket merge, each bucket in merge target may have its own
null key data.
// We merge them all into bucket 0's null key.
auto& target_inst = shared_state.per_instance_data[merge_target];
@@ -555,8 +560,8 @@ Status
BucketedAggLocalState::_merge_and_output_null_keys(RuntimeState* state, B
if constexpr (has_nullable_key_v<
std::remove_reference_t<decltype(dst_data)>>) {
const bool use_simple_count =
shared_state.use_simple_count;
- // Merge null keys from buckets [1..255] into
bucket 0.
- for (int b = 1; b < BUCKETED_AGG_NUM_BUCKETS;
++b) {
+ // Merge null keys from the other active
buckets into bucket 0.
+ for (int b = 1; b < shared_state.num_buckets;
++b) {
auto& src_method =
std::get<AggMethodType>(
target_inst.bucket_agg_data[b]->method_variant);
auto& src_data = *src_method.hash_table;
@@ -654,7 +659,7 @@ Status BucketedAggLocalState::_get_results(RuntimeState*
state, Block* block, bo
shared_state.num_sink_instances;
bool did_work = false;
- for (int b = 0; b < BUCKETED_AGG_NUM_BUCKETS; ++b) {
+ for (int b = 0; b < shared_state.num_buckets; ++b) {
auto& bs = shared_state.bucket_states[b];
// Skip buckets already fully output.
@@ -713,15 +718,15 @@ Status BucketedAggLocalState::_get_results(RuntimeState*
state, Block* block, bo
}
}
- // Check if all 256 buckets have been output.
+ // Check if all buckets have been output.
{
int actually_done = 0;
- for (int b = 0; b < BUCKETED_AGG_NUM_BUCKETS; ++b) {
+ for (int b = 0; b < shared_state.num_buckets; ++b) {
if
(shared_state.bucket_states[b].output_done.load(std::memory_order_acquire)) {
++actually_done;
}
}
- if (actually_done == BUCKETED_AGG_NUM_BUCKETS) {
+ if (actually_done == shared_state.num_buckets) {
// All buckets output. Handle null key.
bool expected = false;
if
(shared_state.null_key_output_claimed.compare_exchange_strong(expected, true)) {
diff --git a/be/src/exec/pipeline/dependency.cpp
b/be/src/exec/pipeline/dependency.cpp
index 85ca4f0c21b..2254fecced3 100644
--- a/be/src/exec/pipeline/dependency.cpp
+++ b/be/src/exec/pipeline/dependency.cpp
@@ -249,8 +249,9 @@ void AggSharedState::_close_without_key() {
}
}
-BucketedAggSharedState::PerInstanceData::PerInstanceData() :
arena(std::make_unique<Arena>()) {
- bucket_agg_data.resize(BUCKETED_AGG_NUM_BUCKETS);
+BucketedAggSharedState::PerInstanceData::PerInstanceData(int num_buckets)
+ : arena(std::make_unique<Arena>()) {
+ bucket_agg_data.resize(num_buckets);
for (auto& p : bucket_agg_data) {
p = std::make_unique<BucketedAggDataVariants>();
}
@@ -261,16 +262,36 @@ BucketedAggSharedState::~BucketedAggSharedState() {
}
Status BucketedAggSharedState::init_instances(int num_instances,
- const std::function<Status()>&
metadata_init) {
+ const std::function<Status()>&
metadata_init,
+ int requested_num_buckets) {
std::call_once(_init_once, [&]() {
+ // Mask-based routing requires a power of two, and the merge state has
256 slots.
+ if (requested_num_buckets < 0 || requested_num_buckets >
BUCKETED_AGG_NUM_BUCKETS ||
+ (requested_num_buckets & (requested_num_buckets - 1)) != 0) {
+ _init_status = Status::InvalidArgument(
+ "Bucketed aggregation bucket count must be 0 or a power of
two in [1, {}], got "
+ "{}",
+ BUCKETED_AGG_NUM_BUCKETS, requested_num_buckets);
+ return;
+ }
+ // One sink can feed several sources. Keep multiple buckets in that
case,
+ // otherwise one source holds the only bucket lock across every output
batch.
+ num_buckets = requested_num_buckets;
+ if (num_buckets == 0) {
+ num_buckets =
+ num_instances == 1 && source_deps.size() == 1 ? 1 :
BUCKETED_AGG_NUM_BUCKETS;
+ }
num_sink_instances = num_instances;
- per_instance_data.resize(num_instances);
- sink_finished = std::make_unique<std::atomic<bool>[]>(num_instances);
+ per_instance_data.reserve(num_instances);
+ for (int i = 0; i < num_instances; ++i) {
+ per_instance_data.emplace_back(num_buckets);
+ }
+ sink_finished =
std::make_unique<std::atomic<bool>[]>(per_instance_data.size());
for (int i = 0; i < num_instances; ++i) {
sink_finished[i].store(false, std::memory_order_relaxed);
}
- for (auto& bs : bucket_states) {
- bs.merged_instances.resize(num_instances, false);
+ for (int b = 0; b < num_buckets; ++b) {
+ bucket_states[b].merged_instances.resize(num_instances, false);
}
source_merge_arenas.resize(source_deps.size());
for (auto& arena : source_merge_arenas) {
diff --git a/be/src/exec/pipeline/dependency.h
b/be/src/exec/pipeline/dependency.h
index 714b9c10a52..63f0bf50a1a 100644
--- a/be/src/exec/pipeline/dependency.h
+++ b/be/src/exec/pipeline/dependency.h
@@ -416,9 +416,8 @@ static constexpr int BUCKETED_AGG_NUM_BUCKETS = 256;
/// Shared state for BucketedAggSinkOperatorX / BucketedAggSourceOperatorX.
///
-/// Each sink pipeline instance owns 256 per-bucket hash tables (two-level
hash table
-/// approach, inspired by ClickHouse). During sink, each row is routed to
bucket
-/// (hash >> 24) & 0xFF.
+/// Each sink pipeline instance owns num_buckets hash tables. Bucket selection
uses
+/// the high bits of the mixed hash, independently of the low bits used by
phmap.
///
/// Source-side merge is pipelined with sink completion: as each sink instance
finishes,
/// it unblocks all source dependencies. Source instances scan buckets and
merge data
@@ -440,14 +439,14 @@ public:
~BucketedAggSharedState() override; // defined in dependency.cpp (variant
machinery)
/// Per-instance data. One per sink pipeline instance.
- /// Each instance has 256 bucket hash tables + 1 shared arena.
+ /// Each instance has num_buckets hash tables + 1 shared arena.
struct PerInstanceData {
- /// 256 per-bucket hash tables. Each bucket has its own
BucketedAggDataVariants.
+ /// Each bucket has its own BucketedAggDataVariants.
/// Uses PHHashMap<StringRef> for string keys instead of StringHashMap.
std::vector<std::unique_ptr<BucketedAggDataVariants>> bucket_agg_data;
std::unique_ptr<Arena> arena;
- PerInstanceData(); // defined in dependency.cpp (variant machinery)
+ explicit PerInstanceData(int num_buckets); // defined in dependency.cpp
};
/// Per-bucket merge state for pipelined source-side processing.
@@ -465,6 +464,15 @@ public:
std::vector<PerInstanceData> per_instance_data;
int num_sink_instances = 0;
+ int num_buckets = BUCKETED_AGG_NUM_BUCKETS;
+
+ /// phmap uses hash bits [7, ...] for the initial probe and [0, 6] for its
+ /// fingerprint. Using bits [24, 31] for routing fixes an index bit once a
+ /// bucket reaches capacity 2^18, clustering probes in half the table.
+ /// The hash has already been mixed by phmap, including for CRC32 hashers.
+ static size_t bucket_index(size_t hash, int num_buckets) {
+ return (hash >> (sizeof(size_t) * 8 - 8)) & (num_buckets - 1);
+ }
/// Tracks how many sinks have finished. Incremented by each sink on EOS.
std::atomic<int> num_sinks_finished = 0;
@@ -526,7 +534,8 @@ public:
/// Once-per-query cold path; defined in dependency.cpp so the body (which
/// materializes the per-bucket BucketedAggDataVariants storage) stays out
/// of every includer's parse.
- Status init_instances(int num_instances, const std::function<Status()>&
metadata_init);
+ Status init_instances(int num_instances, const std::function<Status()>&
metadata_init,
+ int requested_num_buckets =
BUCKETED_AGG_NUM_BUCKETS);
private:
std::once_flag _init_once;
diff --git a/be/test/exec/operator/agg_operator_test.cpp
b/be/test/exec/operator/agg_operator_test.cpp
index f8e866eb4ce..35fdd05cc34 100644
--- a/be/test/exec/operator/agg_operator_test.cpp
+++ b/be/test/exec/operator/agg_operator_test.cpp
@@ -389,7 +389,9 @@ TEST_F(BucketedAggOperatorTest, merges_two_sinks) {
// still destroys it when the failed query is torn down.
TEST_F(BucketedAggOperatorTest,
failed_source_merge_keeps_every_state_for_cleanup) {
ASSERT_TRUE(sink(0, {1, 2, 3}, true).ok());
- ASSERT_TRUE(sink(1, {1, 2, 4}, true).ok());
+ // Every non-empty bucket must merge a duplicate before get_block can
return output,
+ // independently of the bucket routing and traversal order.
+ ASSERT_TRUE(sink(1, {1, 2, 3}, true).ok());
ASSERT_EQ(count_live_states(), 6);
DebugPoints::instance()->add(FAIL_MERGE_DEBUG_POINT);
diff --git a/be/test/exec/pipeline/bucketed_agg_shared_state_test.cpp
b/be/test/exec/pipeline/bucketed_agg_shared_state_test.cpp
index 39ecaba7a67..35370ab06d8 100644
--- a/be/test/exec/pipeline/bucketed_agg_shared_state_test.cpp
+++ b/be/test/exec/pipeline/bucketed_agg_shared_state_test.cpp
@@ -17,6 +17,8 @@
#include <gtest/gtest.h>
+#include <thread>
+
#include "exec/common/agg_utils.h"
#include "exec/pipeline/dependency.h"
@@ -41,4 +43,93 @@ TEST(BucketedAggSharedStateTest,
InitInstancesCreatesOneMergeArenaPerSource) {
}
}
+TEST(BucketedAggSharedStateTest,
BucketRoutingDoesNotConstrainPhmapProbeOrFingerprint) {
+ // At capacity 2^18 - 1, phmap uses bits [7, 24] for its initial probe.
+ // Fixing a bucket must leave both halves of that index space available.
+ // The old (hash >> 24) & 255 routing fails this invariant.
+ for (int num_buckets : {1, 2, 4, 8, 16, 32, 64, 128, 256}) {
+ for (int bucket = 0; bucket < num_buckets; ++bucket) {
+ for (size_t slot : {0UL, (1UL << 17) - 1, 1UL << 17, (1UL << 18) -
1}) {
+ for (size_t fingerprint : {0UL, 127UL}) {
+ size_t hash = (size_t(bucket) << 56) | (slot << 7) |
fingerprint;
+ EXPECT_EQ(BucketedAggSharedState::bucket_index(hash,
num_buckets), bucket);
+ EXPECT_EQ((hash >> 7) & ((1UL << 18) - 1), slot);
+ }
+ }
+ }
+ }
+}
+
+// GTest assertion macros inflate the complexity of this configuration matrix.
+// NOLINTNEXTLINE(readability-function-cognitive-complexity)
+TEST(BucketedAggSharedStateTest, ConfiguredBucketsAndAutomaticSelection) {
+ for (int sinks : {1, 4}) {
+ for (int requested : {0, 1, 2, 4, 8, 16, 32, 64, 128, 256}) {
+ BucketedAggSharedState state;
+ state.create_source_dependencies(3, 0, 0, "BUCKETED_AGG_SOURCE");
+ ASSERT_TRUE(state.init_instances(
+ sinks, [] { return Status::OK(); },
requested)
+ .ok());
+ int expected = requested == 0 ? 256 : requested;
+ ASSERT_EQ(state.num_buckets, expected);
+ ASSERT_EQ(state.per_instance_data.size(), sinks);
+ for (const auto& instance : state.per_instance_data) {
+ ASSERT_EQ(instance.bucket_agg_data.size(), expected);
+ }
+ for (int b = 0; b < BUCKETED_AGG_NUM_BUCKETS; ++b) {
+ EXPECT_EQ(state.bucket_states[b].merged_instances.size(), b <
expected ? sinks : 0);
+ }
+ EXPECT_EQ(state.source_merge_arenas.size(), 3);
+ }
+ }
+ BucketedAggSharedState serial_state;
+ serial_state.create_source_dependencies(1, 0, 0, "BUCKETED_AGG_SOURCE");
+ ASSERT_TRUE(serial_state
+ .init_instances(
+ 1, [] { return Status::OK(); }, 0)
+ .ok());
+ EXPECT_EQ(serial_state.num_buckets, 1);
+ EXPECT_EQ(serial_state.per_instance_data[0].bucket_agg_data.size(), 1);
+}
+
+TEST(BucketedAggSharedStateTest,
ConcurrentInitializationPublishesOneConfiguration) {
+ BucketedAggSharedState state;
+ state.create_source_dependencies(3, 0, 0, "BUCKETED_AGG_SOURCE");
+ std::atomic<int> metadata_calls {0};
+ auto initialize = [&] {
+ EXPECT_TRUE(state.init_instances(
+ 4,
+ [&] {
+ ++metadata_calls;
+ return Status::OK();
+ },
+ 16)
+ .ok());
+ EXPECT_EQ(state.num_buckets, 16);
+ EXPECT_EQ(state.per_instance_data.size(), 4);
+ EXPECT_EQ(state.per_instance_data[0].bucket_agg_data.size(), 16);
+ };
+ std::thread first(initialize);
+ std::thread second(initialize);
+ initialize();
+ first.join();
+ second.join();
+ EXPECT_EQ(metadata_calls, 1);
+}
+
+TEST(BucketedAggSharedStateTest,
InvalidConfigurationIsRejectedBeforeAllocation) {
+ for (int requested : {-1, 3, 255, 257, 512}) {
+ BucketedAggSharedState state;
+ bool metadata_called = false;
+ auto initialize = [&] {
+ metadata_called = true;
+ return Status::OK();
+ };
+ EXPECT_FALSE(state.init_instances(2, initialize, requested).ok());
+ EXPECT_FALSE(state.init_instances(2, initialize, requested).ok());
+ EXPECT_FALSE(metadata_called);
+ EXPECT_TRUE(state.per_instance_data.empty());
+ }
+}
+
} // namespace doris
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java
index 35e271b703a..e648f333560 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java
@@ -3491,7 +3491,8 @@ public class PhysicalPlanTranslator extends
DefaultPlanVisitor<PlanFragment, Pla
AggregateInfo.AggPhase.FIRST);
BucketedAggregationNode bucketedAggNode = new BucketedAggregationNode(
- context.nextPlanNodeId(), inputPlanFragment.getPlanRoot(),
aggInfo, true);
+ context.nextPlanNodeId(), inputPlanFragment.getPlanRoot(),
aggInfo, true,
+ context.getSessionVariable().bucketedAggNumBuckets);
bucketedAggNode.setNereidsId(aggregate.getId());
context.getNereidsIdToPlanNodeIdMap().put(aggregate.getId(),
bucketedAggNode.getId());
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/planner/BucketedAggregationNode.java
b/fe/fe-core/src/main/java/org/apache/doris/planner/BucketedAggregationNode.java
index 5c36ead71fe..354e354e97d 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/planner/BucketedAggregationNode.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/planner/BucketedAggregationNode.java
@@ -40,12 +40,14 @@ import java.util.List;
public class BucketedAggregationNode extends PlanNode {
private final AggregateInfo aggInfo;
private final boolean needsFinalize;
+ private final int numBuckets;
public BucketedAggregationNode(PlanNodeId id, PlanNode input,
AggregateInfo aggInfo,
- boolean needsFinalize) {
+ boolean needsFinalize, int numBuckets) {
super(id, aggInfo.getOutputTupleId().asList(), "BUCKETED AGGREGATE");
this.aggInfo = aggInfo;
this.needsFinalize = needsFinalize;
+ this.numBuckets = numBuckets;
this.children.add(input);
}
@@ -68,6 +70,7 @@ public class BucketedAggregationNode extends PlanNode {
bucketedAggNode.setAggregateFunctions(aggregateFunctions);
bucketedAggNode.setTupleId(aggInfo.getOutputTupleId().asInt());
bucketedAggNode.setNeedFinalize(needsFinalize);
+ bucketedAggNode.setNumBuckets(numBuckets);
msg.bucketed_agg_node = bucketedAggNode;
}
@@ -76,6 +79,8 @@ public class BucketedAggregationNode extends PlanNode {
public String getNodeExplainString(String detailPrefix, TExplainLevel
detailLevel) {
StringBuilder output = new StringBuilder();
output.append(detailPrefix).append("BUCKETED").append("\n");
+ output.append(detailPrefix).append("buckets: ")
+ .append(numBuckets == 0 ? "auto" :
Integer.toString(numBuckets)).append("\n");
if (detailLevel == TExplainLevel.BRIEF) {
output.append(detailPrefix).append(String.format("cardinality=%,d",
cardinality)).append("\n");
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
index f16f9639e13..637fdf5793c 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
@@ -1044,6 +1044,7 @@ public class SessionVariable implements Serializable,
Writable {
public static final String MULTI_DISTINCT_STRATEGY =
"multi_distinct_strategy";
public static final String AGG_PHASE = "agg_phase";
public static final String ENABLE_BUCKETED_HASH_AGG =
"enable_bucketed_hash_agg";
+ public static final String BUCKETED_AGG_NUM_BUCKETS =
"bucketed_agg_num_buckets";
public static final String BUCKETED_AGG_MIN_INPUT_ROWS =
"bucketed_agg_min_input_rows";
public static final String BUCKETED_AGG_MAX_GROUP_KEYS =
"bucketed_agg_max_group_keys";
public static final String BUCKETED_AGG_HIGH_CARD_THRESHOLD =
"bucketed_agg_high_card_threshold";
@@ -3113,9 +3114,16 @@ public class SessionVariable implements Serializable,
Writable {
+ "and serialization/deserialization costs. Enabled by default.")
public boolean enableBucketedHashAgg = true;
+ @VarAttrDef.VarAttr(name = BUCKETED_AGG_NUM_BUCKETS, needForward = true,
+ checker = "checkBucketedAggNumBuckets",
+ description = "Number of hash buckets used by bucketed
aggregation. "
+ + "Must be 0 (automatic) or a power of two between 1 and 256.
Automatic uses one bucket when both "
+ + "sink and source have one instance on a BE, and 256 buckets
otherwise. Default 0.")
+ public int bucketedAggNumBuckets = 0;
+
@VarAttrDef.VarAttr(name = BUCKETED_AGG_MIN_INPUT_ROWS, fuzzy = true,
needForward = true, description = "Minimum "
+ "estimated input rows required for bucketed hash aggregation.
When estimated input "
- + "rows are below this threshold, the data volume is too small for
the 256-bucket two-level "
+ + "rows are below this threshold, the data volume is too small for
the bucketed "
+ "hash table overhead to be worthwhile. Set to 0 to disable this
check. Default 100000.")
public long bucketedAggMinInputRows = 100000;
@@ -6464,6 +6472,19 @@ public class SessionVariable implements Serializable,
Writable {
}
}
+ public void checkBucketedAggNumBuckets(String numBucketsStr) {
+ try {
+ int numBuckets = Integer.parseInt(numBucketsStr);
+ if (numBuckets < 0 || numBuckets > 256 || (numBuckets &
(numBuckets - 1)) != 0) {
+ throw new InvalidParameterException(
+ BUCKETED_AGG_NUM_BUCKETS + " must be 0 or a power of
two between 1 and 256");
+ }
+ } catch (NumberFormatException e) {
+ throw new InvalidParameterException(
+ BUCKETED_AGG_NUM_BUCKETS + " must be 0 or a power of two
between 1 and 256");
+ }
+ }
+
public void checkAggPhase(String aggPhaseStr) {
try {
long aggPhase = Long.parseLong(aggPhaseStr);
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/glue/translator/BucketedAggregateTranslatorTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/glue/translator/BucketedAggregateTranslatorTest.java
index 40b241a0b1a..760ef0f83ca 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/nereids/glue/translator/BucketedAggregateTranslatorTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/glue/translator/BucketedAggregateTranslatorTest.java
@@ -30,6 +30,7 @@ import org.apache.doris.planner.PlanFragment;
import org.apache.doris.planner.PlanNode;
import org.apache.doris.planner.Planner;
import org.apache.doris.qe.SessionVariable;
+import org.apache.doris.thrift.TBucketedAggregationNode;
import org.apache.doris.thrift.TExplainLevel;
import org.apache.doris.utframe.TestWithFeService;
@@ -684,6 +685,53 @@ public class BucketedAggregateTranslatorTest extends
TestWithFeService {
.collect(Collectors.joining("\n"));
}
+ @Test
+ public void testConfiguredBucketCountIsCapturedInPlan() throws Exception {
+ SessionVariable sessionVariable = connectContext.getSessionVariable();
+ int oldAggPhase = sessionVariable.aggPhase;
+ int oldBeNumberForTest = sessionVariable.getBeNumberForTest();
+ int oldParallelPipelineTaskNum =
sessionVariable.parallelPipelineTaskNum;
+ int oldBucketedAggNumBuckets = sessionVariable.bucketedAggNumBuckets;
+ long oldBucketedAggMinInputRows =
sessionVariable.bucketedAggMinInputRows;
+ long oldBucketedAggMaxGroupKeys =
sessionVariable.bucketedAggMaxGroupKeys;
+ double oldBucketedAggHighCardThreshold =
sessionVariable.bucketedAggHighCardThreshold;
+ boolean oldEnableBucketedHashAgg =
sessionVariable.enableBucketedHashAgg;
+ try {
+ sessionVariable.aggPhase = 1;
+ sessionVariable.setBeNumberForTest(1);
+ sessionVariable.bucketedAggMinInputRows = 0;
+ sessionVariable.bucketedAggMaxGroupKeys = 0;
+ sessionVariable.bucketedAggHighCardThreshold = 1.0;
+ sessionVariable.enableBucketedHashAgg = true;
+ for (int parallelism : new int[] {1, 4}) {
+ sessionVariable.parallelPipelineTaskNum = parallelism;
+ for (int numBuckets : new int[] {0, 1, 16, 256}) {
+ sessionVariable.bucketedAggNumBuckets = numBuckets;
+ List<BucketedAggregationNode> nodes =
collectBucketedAggregationNodes("sum(kint)");
+ Assertions.assertEquals(1, nodes.size());
+ BucketedAggregationNode node = nodes.get(0);
+ // Serialization must use the plan's captured value, even
if the session changes later.
+ sessionVariable.bucketedAggNumBuckets = numBuckets == 256
? 0 : 256;
+ TBucketedAggregationNode thriftNode =
node.treeToThrift().getNodes().get(0).getBucketedAggNode();
+ Assertions.assertTrue(thriftNode.isSetNumBuckets());
+ Assertions.assertEquals(numBuckets,
thriftNode.getNumBuckets());
+ String expected = "buckets: " + (numBuckets == 0 ? "auto"
: Integer.toString(numBuckets)) + "\n";
+ Assertions.assertTrue(node.getNodeExplainString("",
TExplainLevel.NORMAL).contains(expected));
+ Assertions.assertTrue(node.getNodeExplainString("",
TExplainLevel.BRIEF).contains(expected));
+ }
+ }
+ } finally {
+ sessionVariable.aggPhase = oldAggPhase;
+ sessionVariable.setBeNumberForTest(oldBeNumberForTest);
+ sessionVariable.parallelPipelineTaskNum =
oldParallelPipelineTaskNum;
+ sessionVariable.bucketedAggNumBuckets = oldBucketedAggNumBuckets;
+ sessionVariable.bucketedAggMinInputRows =
oldBucketedAggMinInputRows;
+ sessionVariable.bucketedAggMaxGroupKeys =
oldBucketedAggMaxGroupKeys;
+ sessionVariable.bucketedAggHighCardThreshold =
oldBucketedAggHighCardThreshold;
+ sessionVariable.enableBucketedHashAgg = oldEnableBucketedHashAgg;
+ }
+ }
+
private void assertUsesRegularAggregation(String aggregateFunction) throws
Exception {
Planner planner = planAggregate(aggregateFunction);
Assertions.assertTrue(collectNodes(planner,
BucketedAggregationNode.class).isEmpty());
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/analysis/AnalyzeWhereSubqueryTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/analysis/AnalyzeWhereSubqueryTest.java
index 0695b05ef70..bc8748ea7c8 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/analysis/AnalyzeWhereSubqueryTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/analysis/AnalyzeWhereSubqueryTest.java
@@ -124,12 +124,13 @@ public class AnalyzeWhereSubqueryTest extends
TestWithFeService implements MemoP
try {
StatementScopeIdGenerator.clear();
StatementContext statementContext =
MemoTestUtils.createStatementContext(connectContext, sql);
- PhysicalPlan plan = new
NereidsPlanner(statementContext).planWithLock(
+ NereidsPlanner planner = new NereidsPlanner(statementContext);
+ PhysicalPlan plan = planner.planWithLock(
parser.parseSingle(sql),
PhysicalProperties.ANY
);
// Just to check whether translate will throw exception
- new PhysicalPlanTranslator(new
PlanTranslatorContext()).translatePlan(plan);
+ new PhysicalPlanTranslator(new
PlanTranslatorContext(planner.getCascadesContext())).translatePlan(plan);
} catch (Throwable t) {
throw new IllegalStateException("Test sql failed: " +
t.getMessage() + ", sql:\n" + sql, t);
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/qe/SessionVariableBucketedAggTest.java
b/fe/fe-core/src/test/java/org/apache/doris/qe/SessionVariableBucketedAggTest.java
new file mode 100644
index 00000000000..864fe250fac
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/qe/SessionVariableBucketedAggTest.java
@@ -0,0 +1,65 @@
+// 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.doris.qe;
+
+import org.apache.doris.analysis.SetType;
+import org.apache.doris.analysis.SetVar;
+import org.apache.doris.analysis.StringLiteral;
+import org.apache.doris.common.DdlException;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+class SessionVariableBucketedAggTest {
+
+ @Test
+ void testDefaultAndForwarding() throws Exception {
+ SessionVariable sessionVariable = new SessionVariable();
+ Assertions.assertEquals(0, sessionVariable.bucketedAggNumBuckets);
+ VariableMgr.setVar(sessionVariable, new SetVar(SetType.SESSION,
+ SessionVariable.BUCKETED_AGG_NUM_BUCKETS, new
StringLiteral("16")));
+
+ SessionVariable forwarded = new SessionVariable();
+
forwarded.setForwardedSessionVariables(sessionVariable.getForwardVariables());
+ Assertions.assertEquals(16, forwarded.bucketedAggNumBuckets);
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"0", "1", "2", "4", "8", "16", "32", "64", "128",
"256"})
+ void testAcceptsAutomaticAndPowersOfTwo(String value) throws Exception {
+ SessionVariable sessionVariable = new SessionVariable();
+ VariableMgr.setVar(sessionVariable, new SetVar(SetType.SESSION,
+ SessionVariable.BUCKETED_AGG_NUM_BUCKETS, new
StringLiteral(value)));
+ Assertions.assertEquals(Integer.parseInt(value),
sessionVariable.bucketedAggNumBuckets);
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"-1", "-2147483648", "3", "7", "255", "257",
"512", "2147483648", "1.5", "auto", ""})
+ void testRejectsInvalidBucketCounts(String value) {
+ SessionVariable sessionVariable = new SessionVariable();
+ sessionVariable.bucketedAggNumBuckets = 16;
+ DdlException exception = Assertions.assertThrows(DdlException.class,
() ->
+ VariableMgr.setVar(sessionVariable, new SetVar(SetType.SESSION,
+ SessionVariable.BUCKETED_AGG_NUM_BUCKETS, new
StringLiteral(value))));
+ Assertions.assertTrue(exception.getMessage().contains(
+ "bucketed_agg_num_buckets must be 0 or a power of two between
1 and 256"));
+ Assertions.assertEquals(16, sessionVariable.bucketedAggNumBuckets);
+ }
+}
diff --git a/gensrc/thrift/PlanNodes.thrift b/gensrc/thrift/PlanNodes.thrift
index 1cb49705c5b..4a84b7ed955 100644
--- a/gensrc/thrift/PlanNodes.thrift
+++ b/gensrc/thrift/PlanNodes.thrift
@@ -1322,6 +1322,9 @@ struct TBucketedAggregationNode {
// so intermediate and output tuples are always identical.
3: optional Types.TTupleId tuple_id
5: optional bool need_finalize
+ // 0 selects one bucket when both sink and source have one instance,
otherwise 256 buckets.
+ // Positive values must be powers of two in [1, 256]. Missing retains 256
buckets.
+ 6: optional i32 num_buckets
}
struct TRepeatNode {
diff --git
a/regression-test/data/nereids_rules_p0/agg_strategy/bucketed_hash_agg_num_buckets.out
b/regression-test/data/nereids_rules_p0/agg_strategy/bucketed_hash_agg_num_buckets.out
new file mode 100644
index 00000000000..0ec53f56edb
--- /dev/null
+++
b/regression-test/data/nereids_rules_p0/agg_strategy/bucketed_hash_agg_num_buckets.out
@@ -0,0 +1,1236 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !string_1_0 --
+\N 4096 3276 53660880 8 32752
+0 4096 3277 53693647 7 32767
+1 4096 3277 53693641 1 32761
+2 4096 3277 53693642 2 32762
+3 4096 3277 53693643 3 32763
+4 4096 3277 53693644 4 32764
+5 4096 3276 53660880 12 32758
+6 4096 3277 53693646 6 32766
+
+-- !numeric_1_0 --
+0 2979 39040991
+1 2979 39049920
+10 2978 39032053
+2 2979 39026089
+3 2979 39035028
+4 2979 39043967
+5 2979 39052906
+6 2979 39061840
+7 2979 39038004
+8 2979 39046943
+9 2979 39055882
+
+-- !composite_1_0 --
+0 \N 1366 17886960
+0 0 1365 17886960
+0 1 1365 17886960
+0 2 1365 17886960
+0 3 1366 17919723
+0 4 1365 17886960
+0 5 1365 17886960
+0 6 1366 17919726
+1 \N 1365 17897880
+1 0 1366 17897887
+1 1 1366 17897881
+1 2 1365 17897880
+1 3 1365 17897880
+1 4 1366 17930644
+1 5 1365 17897880
+1 6 1365 17865120
+2 \N 1365 17876040
+2 0 1365 17908800
+2 1 1365 17908800
+2 2 1366 17908802
+2 3 1365 17876040
+2 4 1365 17876040
+2 5 1366 17876040
+2 6 1365 17908800
+
+-- !count_numeric_1_0 --
+0 2979
+1 2979
+10 2978
+2 2979
+3 2979
+4 2979
+5 2979
+6 2979
+7 2979
+8 2979
+9 2979
+
+-- !count_numeric_nullable_1_0 --
+\N 4096
+0 2606
+1 2607
+10 2606
+2 2606
+3 2607
+4 2607
+5 2606
+6 2607
+7 2607
+8 2606
+9 2607
+
+-- !count_string_1_0 --
+0 4682
+1 4681
+2 4681
+3 4681
+4 4681
+5 4681
+6 4681
+
+-- !count_string_nullable_1_0 --
+\N 4096
+0 4096
+1 4096
+2 4096
+3 4096
+4 4096
+5 4096
+6 4096
+
+-- !distinct_numeric_1_0 --
+32768 536854528 32768 1 1
+
+-- !distinct_composite_1_0 --
+32768 536854528 32768 429483623 4096
+
+-- !empty_1_0 --
+
+-- !string_1_1 --
+\N 4096 3276 53660880 8 32752
+0 4096 3277 53693647 7 32767
+1 4096 3277 53693641 1 32761
+2 4096 3277 53693642 2 32762
+3 4096 3277 53693643 3 32763
+4 4096 3277 53693644 4 32764
+5 4096 3276 53660880 12 32758
+6 4096 3277 53693646 6 32766
+
+-- !numeric_1_1 --
+0 2979 39040991
+1 2979 39049920
+10 2978 39032053
+2 2979 39026089
+3 2979 39035028
+4 2979 39043967
+5 2979 39052906
+6 2979 39061840
+7 2979 39038004
+8 2979 39046943
+9 2979 39055882
+
+-- !composite_1_1 --
+0 \N 1366 17886960
+0 0 1365 17886960
+0 1 1365 17886960
+0 2 1365 17886960
+0 3 1366 17919723
+0 4 1365 17886960
+0 5 1365 17886960
+0 6 1366 17919726
+1 \N 1365 17897880
+1 0 1366 17897887
+1 1 1366 17897881
+1 2 1365 17897880
+1 3 1365 17897880
+1 4 1366 17930644
+1 5 1365 17897880
+1 6 1365 17865120
+2 \N 1365 17876040
+2 0 1365 17908800
+2 1 1365 17908800
+2 2 1366 17908802
+2 3 1365 17876040
+2 4 1365 17876040
+2 5 1366 17876040
+2 6 1365 17908800
+
+-- !count_numeric_1_1 --
+0 2979
+1 2979
+10 2978
+2 2979
+3 2979
+4 2979
+5 2979
+6 2979
+7 2979
+8 2979
+9 2979
+
+-- !count_numeric_nullable_1_1 --
+\N 4096
+0 2606
+1 2607
+10 2606
+2 2606
+3 2607
+4 2607
+5 2606
+6 2607
+7 2607
+8 2606
+9 2607
+
+-- !count_string_1_1 --
+0 4682
+1 4681
+2 4681
+3 4681
+4 4681
+5 4681
+6 4681
+
+-- !count_string_nullable_1_1 --
+\N 4096
+0 4096
+1 4096
+2 4096
+3 4096
+4 4096
+5 4096
+6 4096
+
+-- !distinct_numeric_1_1 --
+32768 536854528 32768 1 1
+
+-- !distinct_composite_1_1 --
+32768 536854528 32768 429483623 4096
+
+-- !empty_1_1 --
+
+-- !string_1_4 --
+\N 4096 3276 53660880 8 32752
+0 4096 3277 53693647 7 32767
+1 4096 3277 53693641 1 32761
+2 4096 3277 53693642 2 32762
+3 4096 3277 53693643 3 32763
+4 4096 3277 53693644 4 32764
+5 4096 3276 53660880 12 32758
+6 4096 3277 53693646 6 32766
+
+-- !numeric_1_4 --
+0 2979 39040991
+1 2979 39049920
+10 2978 39032053
+2 2979 39026089
+3 2979 39035028
+4 2979 39043967
+5 2979 39052906
+6 2979 39061840
+7 2979 39038004
+8 2979 39046943
+9 2979 39055882
+
+-- !composite_1_4 --
+0 \N 1366 17886960
+0 0 1365 17886960
+0 1 1365 17886960
+0 2 1365 17886960
+0 3 1366 17919723
+0 4 1365 17886960
+0 5 1365 17886960
+0 6 1366 17919726
+1 \N 1365 17897880
+1 0 1366 17897887
+1 1 1366 17897881
+1 2 1365 17897880
+1 3 1365 17897880
+1 4 1366 17930644
+1 5 1365 17897880
+1 6 1365 17865120
+2 \N 1365 17876040
+2 0 1365 17908800
+2 1 1365 17908800
+2 2 1366 17908802
+2 3 1365 17876040
+2 4 1365 17876040
+2 5 1366 17876040
+2 6 1365 17908800
+
+-- !count_numeric_1_4 --
+0 2979
+1 2979
+10 2978
+2 2979
+3 2979
+4 2979
+5 2979
+6 2979
+7 2979
+8 2979
+9 2979
+
+-- !count_numeric_nullable_1_4 --
+\N 4096
+0 2606
+1 2607
+10 2606
+2 2606
+3 2607
+4 2607
+5 2606
+6 2607
+7 2607
+8 2606
+9 2607
+
+-- !count_string_1_4 --
+0 4682
+1 4681
+2 4681
+3 4681
+4 4681
+5 4681
+6 4681
+
+-- !count_string_nullable_1_4 --
+\N 4096
+0 4096
+1 4096
+2 4096
+3 4096
+4 4096
+5 4096
+6 4096
+
+-- !distinct_numeric_1_4 --
+32768 536854528 32768 1 1
+
+-- !distinct_composite_1_4 --
+32768 536854528 32768 429483623 4096
+
+-- !empty_1_4 --
+
+-- !string_1_16 --
+\N 4096 3276 53660880 8 32752
+0 4096 3277 53693647 7 32767
+1 4096 3277 53693641 1 32761
+2 4096 3277 53693642 2 32762
+3 4096 3277 53693643 3 32763
+4 4096 3277 53693644 4 32764
+5 4096 3276 53660880 12 32758
+6 4096 3277 53693646 6 32766
+
+-- !numeric_1_16 --
+0 2979 39040991
+1 2979 39049920
+10 2978 39032053
+2 2979 39026089
+3 2979 39035028
+4 2979 39043967
+5 2979 39052906
+6 2979 39061840
+7 2979 39038004
+8 2979 39046943
+9 2979 39055882
+
+-- !composite_1_16 --
+0 \N 1366 17886960
+0 0 1365 17886960
+0 1 1365 17886960
+0 2 1365 17886960
+0 3 1366 17919723
+0 4 1365 17886960
+0 5 1365 17886960
+0 6 1366 17919726
+1 \N 1365 17897880
+1 0 1366 17897887
+1 1 1366 17897881
+1 2 1365 17897880
+1 3 1365 17897880
+1 4 1366 17930644
+1 5 1365 17897880
+1 6 1365 17865120
+2 \N 1365 17876040
+2 0 1365 17908800
+2 1 1365 17908800
+2 2 1366 17908802
+2 3 1365 17876040
+2 4 1365 17876040
+2 5 1366 17876040
+2 6 1365 17908800
+
+-- !count_numeric_1_16 --
+0 2979
+1 2979
+10 2978
+2 2979
+3 2979
+4 2979
+5 2979
+6 2979
+7 2979
+8 2979
+9 2979
+
+-- !count_numeric_nullable_1_16 --
+\N 4096
+0 2606
+1 2607
+10 2606
+2 2606
+3 2607
+4 2607
+5 2606
+6 2607
+7 2607
+8 2606
+9 2607
+
+-- !count_string_1_16 --
+0 4682
+1 4681
+2 4681
+3 4681
+4 4681
+5 4681
+6 4681
+
+-- !count_string_nullable_1_16 --
+\N 4096
+0 4096
+1 4096
+2 4096
+3 4096
+4 4096
+5 4096
+6 4096
+
+-- !distinct_numeric_1_16 --
+32768 536854528 32768 1 1
+
+-- !distinct_composite_1_16 --
+32768 536854528 32768 429483623 4096
+
+-- !empty_1_16 --
+
+-- !string_1_64 --
+\N 4096 3276 53660880 8 32752
+0 4096 3277 53693647 7 32767
+1 4096 3277 53693641 1 32761
+2 4096 3277 53693642 2 32762
+3 4096 3277 53693643 3 32763
+4 4096 3277 53693644 4 32764
+5 4096 3276 53660880 12 32758
+6 4096 3277 53693646 6 32766
+
+-- !numeric_1_64 --
+0 2979 39040991
+1 2979 39049920
+10 2978 39032053
+2 2979 39026089
+3 2979 39035028
+4 2979 39043967
+5 2979 39052906
+6 2979 39061840
+7 2979 39038004
+8 2979 39046943
+9 2979 39055882
+
+-- !composite_1_64 --
+0 \N 1366 17886960
+0 0 1365 17886960
+0 1 1365 17886960
+0 2 1365 17886960
+0 3 1366 17919723
+0 4 1365 17886960
+0 5 1365 17886960
+0 6 1366 17919726
+1 \N 1365 17897880
+1 0 1366 17897887
+1 1 1366 17897881
+1 2 1365 17897880
+1 3 1365 17897880
+1 4 1366 17930644
+1 5 1365 17897880
+1 6 1365 17865120
+2 \N 1365 17876040
+2 0 1365 17908800
+2 1 1365 17908800
+2 2 1366 17908802
+2 3 1365 17876040
+2 4 1365 17876040
+2 5 1366 17876040
+2 6 1365 17908800
+
+-- !count_numeric_1_64 --
+0 2979
+1 2979
+10 2978
+2 2979
+3 2979
+4 2979
+5 2979
+6 2979
+7 2979
+8 2979
+9 2979
+
+-- !count_numeric_nullable_1_64 --
+\N 4096
+0 2606
+1 2607
+10 2606
+2 2606
+3 2607
+4 2607
+5 2606
+6 2607
+7 2607
+8 2606
+9 2607
+
+-- !count_string_1_64 --
+0 4682
+1 4681
+2 4681
+3 4681
+4 4681
+5 4681
+6 4681
+
+-- !count_string_nullable_1_64 --
+\N 4096
+0 4096
+1 4096
+2 4096
+3 4096
+4 4096
+5 4096
+6 4096
+
+-- !distinct_numeric_1_64 --
+32768 536854528 32768 1 1
+
+-- !distinct_composite_1_64 --
+32768 536854528 32768 429483623 4096
+
+-- !empty_1_64 --
+
+-- !string_1_256 --
+\N 4096 3276 53660880 8 32752
+0 4096 3277 53693647 7 32767
+1 4096 3277 53693641 1 32761
+2 4096 3277 53693642 2 32762
+3 4096 3277 53693643 3 32763
+4 4096 3277 53693644 4 32764
+5 4096 3276 53660880 12 32758
+6 4096 3277 53693646 6 32766
+
+-- !numeric_1_256 --
+0 2979 39040991
+1 2979 39049920
+10 2978 39032053
+2 2979 39026089
+3 2979 39035028
+4 2979 39043967
+5 2979 39052906
+6 2979 39061840
+7 2979 39038004
+8 2979 39046943
+9 2979 39055882
+
+-- !composite_1_256 --
+0 \N 1366 17886960
+0 0 1365 17886960
+0 1 1365 17886960
+0 2 1365 17886960
+0 3 1366 17919723
+0 4 1365 17886960
+0 5 1365 17886960
+0 6 1366 17919726
+1 \N 1365 17897880
+1 0 1366 17897887
+1 1 1366 17897881
+1 2 1365 17897880
+1 3 1365 17897880
+1 4 1366 17930644
+1 5 1365 17897880
+1 6 1365 17865120
+2 \N 1365 17876040
+2 0 1365 17908800
+2 1 1365 17908800
+2 2 1366 17908802
+2 3 1365 17876040
+2 4 1365 17876040
+2 5 1366 17876040
+2 6 1365 17908800
+
+-- !count_numeric_1_256 --
+0 2979
+1 2979
+10 2978
+2 2979
+3 2979
+4 2979
+5 2979
+6 2979
+7 2979
+8 2979
+9 2979
+
+-- !count_numeric_nullable_1_256 --
+\N 4096
+0 2606
+1 2607
+10 2606
+2 2606
+3 2607
+4 2607
+5 2606
+6 2607
+7 2607
+8 2606
+9 2607
+
+-- !count_string_1_256 --
+0 4682
+1 4681
+2 4681
+3 4681
+4 4681
+5 4681
+6 4681
+
+-- !count_string_nullable_1_256 --
+\N 4096
+0 4096
+1 4096
+2 4096
+3 4096
+4 4096
+5 4096
+6 4096
+
+-- !distinct_numeric_1_256 --
+32768 536854528 32768 1 1
+
+-- !distinct_composite_1_256 --
+32768 536854528 32768 429483623 4096
+
+-- !empty_1_256 --
+
+-- !string_4_0 --
+\N 4096 3276 53660880 8 32752
+0 4096 3277 53693647 7 32767
+1 4096 3277 53693641 1 32761
+2 4096 3277 53693642 2 32762
+3 4096 3277 53693643 3 32763
+4 4096 3277 53693644 4 32764
+5 4096 3276 53660880 12 32758
+6 4096 3277 53693646 6 32766
+
+-- !numeric_4_0 --
+0 2979 39040991
+1 2979 39049920
+10 2978 39032053
+2 2979 39026089
+3 2979 39035028
+4 2979 39043967
+5 2979 39052906
+6 2979 39061840
+7 2979 39038004
+8 2979 39046943
+9 2979 39055882
+
+-- !composite_4_0 --
+0 \N 1366 17886960
+0 0 1365 17886960
+0 1 1365 17886960
+0 2 1365 17886960
+0 3 1366 17919723
+0 4 1365 17886960
+0 5 1365 17886960
+0 6 1366 17919726
+1 \N 1365 17897880
+1 0 1366 17897887
+1 1 1366 17897881
+1 2 1365 17897880
+1 3 1365 17897880
+1 4 1366 17930644
+1 5 1365 17897880
+1 6 1365 17865120
+2 \N 1365 17876040
+2 0 1365 17908800
+2 1 1365 17908800
+2 2 1366 17908802
+2 3 1365 17876040
+2 4 1365 17876040
+2 5 1366 17876040
+2 6 1365 17908800
+
+-- !count_numeric_4_0 --
+0 2979
+1 2979
+10 2978
+2 2979
+3 2979
+4 2979
+5 2979
+6 2979
+7 2979
+8 2979
+9 2979
+
+-- !count_numeric_nullable_4_0 --
+\N 4096
+0 2606
+1 2607
+10 2606
+2 2606
+3 2607
+4 2607
+5 2606
+6 2607
+7 2607
+8 2606
+9 2607
+
+-- !count_string_4_0 --
+0 4682
+1 4681
+2 4681
+3 4681
+4 4681
+5 4681
+6 4681
+
+-- !count_string_nullable_4_0 --
+\N 4096
+0 4096
+1 4096
+2 4096
+3 4096
+4 4096
+5 4096
+6 4096
+
+-- !distinct_numeric_4_0 --
+32768 536854528 32768 1 1
+
+-- !distinct_composite_4_0 --
+32768 536854528 32768 429483623 4096
+
+-- !empty_4_0 --
+
+-- !string_4_1 --
+\N 4096 3276 53660880 8 32752
+0 4096 3277 53693647 7 32767
+1 4096 3277 53693641 1 32761
+2 4096 3277 53693642 2 32762
+3 4096 3277 53693643 3 32763
+4 4096 3277 53693644 4 32764
+5 4096 3276 53660880 12 32758
+6 4096 3277 53693646 6 32766
+
+-- !numeric_4_1 --
+0 2979 39040991
+1 2979 39049920
+10 2978 39032053
+2 2979 39026089
+3 2979 39035028
+4 2979 39043967
+5 2979 39052906
+6 2979 39061840
+7 2979 39038004
+8 2979 39046943
+9 2979 39055882
+
+-- !composite_4_1 --
+0 \N 1366 17886960
+0 0 1365 17886960
+0 1 1365 17886960
+0 2 1365 17886960
+0 3 1366 17919723
+0 4 1365 17886960
+0 5 1365 17886960
+0 6 1366 17919726
+1 \N 1365 17897880
+1 0 1366 17897887
+1 1 1366 17897881
+1 2 1365 17897880
+1 3 1365 17897880
+1 4 1366 17930644
+1 5 1365 17897880
+1 6 1365 17865120
+2 \N 1365 17876040
+2 0 1365 17908800
+2 1 1365 17908800
+2 2 1366 17908802
+2 3 1365 17876040
+2 4 1365 17876040
+2 5 1366 17876040
+2 6 1365 17908800
+
+-- !count_numeric_4_1 --
+0 2979
+1 2979
+10 2978
+2 2979
+3 2979
+4 2979
+5 2979
+6 2979
+7 2979
+8 2979
+9 2979
+
+-- !count_numeric_nullable_4_1 --
+\N 4096
+0 2606
+1 2607
+10 2606
+2 2606
+3 2607
+4 2607
+5 2606
+6 2607
+7 2607
+8 2606
+9 2607
+
+-- !count_string_4_1 --
+0 4682
+1 4681
+2 4681
+3 4681
+4 4681
+5 4681
+6 4681
+
+-- !count_string_nullable_4_1 --
+\N 4096
+0 4096
+1 4096
+2 4096
+3 4096
+4 4096
+5 4096
+6 4096
+
+-- !distinct_numeric_4_1 --
+32768 536854528 32768 1 1
+
+-- !distinct_composite_4_1 --
+32768 536854528 32768 429483623 4096
+
+-- !empty_4_1 --
+
+-- !string_4_4 --
+\N 4096 3276 53660880 8 32752
+0 4096 3277 53693647 7 32767
+1 4096 3277 53693641 1 32761
+2 4096 3277 53693642 2 32762
+3 4096 3277 53693643 3 32763
+4 4096 3277 53693644 4 32764
+5 4096 3276 53660880 12 32758
+6 4096 3277 53693646 6 32766
+
+-- !numeric_4_4 --
+0 2979 39040991
+1 2979 39049920
+10 2978 39032053
+2 2979 39026089
+3 2979 39035028
+4 2979 39043967
+5 2979 39052906
+6 2979 39061840
+7 2979 39038004
+8 2979 39046943
+9 2979 39055882
+
+-- !composite_4_4 --
+0 \N 1366 17886960
+0 0 1365 17886960
+0 1 1365 17886960
+0 2 1365 17886960
+0 3 1366 17919723
+0 4 1365 17886960
+0 5 1365 17886960
+0 6 1366 17919726
+1 \N 1365 17897880
+1 0 1366 17897887
+1 1 1366 17897881
+1 2 1365 17897880
+1 3 1365 17897880
+1 4 1366 17930644
+1 5 1365 17897880
+1 6 1365 17865120
+2 \N 1365 17876040
+2 0 1365 17908800
+2 1 1365 17908800
+2 2 1366 17908802
+2 3 1365 17876040
+2 4 1365 17876040
+2 5 1366 17876040
+2 6 1365 17908800
+
+-- !count_numeric_4_4 --
+0 2979
+1 2979
+10 2978
+2 2979
+3 2979
+4 2979
+5 2979
+6 2979
+7 2979
+8 2979
+9 2979
+
+-- !count_numeric_nullable_4_4 --
+\N 4096
+0 2606
+1 2607
+10 2606
+2 2606
+3 2607
+4 2607
+5 2606
+6 2607
+7 2607
+8 2606
+9 2607
+
+-- !count_string_4_4 --
+0 4682
+1 4681
+2 4681
+3 4681
+4 4681
+5 4681
+6 4681
+
+-- !count_string_nullable_4_4 --
+\N 4096
+0 4096
+1 4096
+2 4096
+3 4096
+4 4096
+5 4096
+6 4096
+
+-- !distinct_numeric_4_4 --
+32768 536854528 32768 1 1
+
+-- !distinct_composite_4_4 --
+32768 536854528 32768 429483623 4096
+
+-- !empty_4_4 --
+
+-- !string_4_16 --
+\N 4096 3276 53660880 8 32752
+0 4096 3277 53693647 7 32767
+1 4096 3277 53693641 1 32761
+2 4096 3277 53693642 2 32762
+3 4096 3277 53693643 3 32763
+4 4096 3277 53693644 4 32764
+5 4096 3276 53660880 12 32758
+6 4096 3277 53693646 6 32766
+
+-- !numeric_4_16 --
+0 2979 39040991
+1 2979 39049920
+10 2978 39032053
+2 2979 39026089
+3 2979 39035028
+4 2979 39043967
+5 2979 39052906
+6 2979 39061840
+7 2979 39038004
+8 2979 39046943
+9 2979 39055882
+
+-- !composite_4_16 --
+0 \N 1366 17886960
+0 0 1365 17886960
+0 1 1365 17886960
+0 2 1365 17886960
+0 3 1366 17919723
+0 4 1365 17886960
+0 5 1365 17886960
+0 6 1366 17919726
+1 \N 1365 17897880
+1 0 1366 17897887
+1 1 1366 17897881
+1 2 1365 17897880
+1 3 1365 17897880
+1 4 1366 17930644
+1 5 1365 17897880
+1 6 1365 17865120
+2 \N 1365 17876040
+2 0 1365 17908800
+2 1 1365 17908800
+2 2 1366 17908802
+2 3 1365 17876040
+2 4 1365 17876040
+2 5 1366 17876040
+2 6 1365 17908800
+
+-- !count_numeric_4_16 --
+0 2979
+1 2979
+10 2978
+2 2979
+3 2979
+4 2979
+5 2979
+6 2979
+7 2979
+8 2979
+9 2979
+
+-- !count_numeric_nullable_4_16 --
+\N 4096
+0 2606
+1 2607
+10 2606
+2 2606
+3 2607
+4 2607
+5 2606
+6 2607
+7 2607
+8 2606
+9 2607
+
+-- !count_string_4_16 --
+0 4682
+1 4681
+2 4681
+3 4681
+4 4681
+5 4681
+6 4681
+
+-- !count_string_nullable_4_16 --
+\N 4096
+0 4096
+1 4096
+2 4096
+3 4096
+4 4096
+5 4096
+6 4096
+
+-- !distinct_numeric_4_16 --
+32768 536854528 32768 1 1
+
+-- !distinct_composite_4_16 --
+32768 536854528 32768 429483623 4096
+
+-- !empty_4_16 --
+
+-- !string_4_64 --
+\N 4096 3276 53660880 8 32752
+0 4096 3277 53693647 7 32767
+1 4096 3277 53693641 1 32761
+2 4096 3277 53693642 2 32762
+3 4096 3277 53693643 3 32763
+4 4096 3277 53693644 4 32764
+5 4096 3276 53660880 12 32758
+6 4096 3277 53693646 6 32766
+
+-- !numeric_4_64 --
+0 2979 39040991
+1 2979 39049920
+10 2978 39032053
+2 2979 39026089
+3 2979 39035028
+4 2979 39043967
+5 2979 39052906
+6 2979 39061840
+7 2979 39038004
+8 2979 39046943
+9 2979 39055882
+
+-- !composite_4_64 --
+0 \N 1366 17886960
+0 0 1365 17886960
+0 1 1365 17886960
+0 2 1365 17886960
+0 3 1366 17919723
+0 4 1365 17886960
+0 5 1365 17886960
+0 6 1366 17919726
+1 \N 1365 17897880
+1 0 1366 17897887
+1 1 1366 17897881
+1 2 1365 17897880
+1 3 1365 17897880
+1 4 1366 17930644
+1 5 1365 17897880
+1 6 1365 17865120
+2 \N 1365 17876040
+2 0 1365 17908800
+2 1 1365 17908800
+2 2 1366 17908802
+2 3 1365 17876040
+2 4 1365 17876040
+2 5 1366 17876040
+2 6 1365 17908800
+
+-- !count_numeric_4_64 --
+0 2979
+1 2979
+10 2978
+2 2979
+3 2979
+4 2979
+5 2979
+6 2979
+7 2979
+8 2979
+9 2979
+
+-- !count_numeric_nullable_4_64 --
+\N 4096
+0 2606
+1 2607
+10 2606
+2 2606
+3 2607
+4 2607
+5 2606
+6 2607
+7 2607
+8 2606
+9 2607
+
+-- !count_string_4_64 --
+0 4682
+1 4681
+2 4681
+3 4681
+4 4681
+5 4681
+6 4681
+
+-- !count_string_nullable_4_64 --
+\N 4096
+0 4096
+1 4096
+2 4096
+3 4096
+4 4096
+5 4096
+6 4096
+
+-- !distinct_numeric_4_64 --
+32768 536854528 32768 1 1
+
+-- !distinct_composite_4_64 --
+32768 536854528 32768 429483623 4096
+
+-- !empty_4_64 --
+
+-- !string_4_256 --
+\N 4096 3276 53660880 8 32752
+0 4096 3277 53693647 7 32767
+1 4096 3277 53693641 1 32761
+2 4096 3277 53693642 2 32762
+3 4096 3277 53693643 3 32763
+4 4096 3277 53693644 4 32764
+5 4096 3276 53660880 12 32758
+6 4096 3277 53693646 6 32766
+
+-- !numeric_4_256 --
+0 2979 39040991
+1 2979 39049920
+10 2978 39032053
+2 2979 39026089
+3 2979 39035028
+4 2979 39043967
+5 2979 39052906
+6 2979 39061840
+7 2979 39038004
+8 2979 39046943
+9 2979 39055882
+
+-- !composite_4_256 --
+0 \N 1366 17886960
+0 0 1365 17886960
+0 1 1365 17886960
+0 2 1365 17886960
+0 3 1366 17919723
+0 4 1365 17886960
+0 5 1365 17886960
+0 6 1366 17919726
+1 \N 1365 17897880
+1 0 1366 17897887
+1 1 1366 17897881
+1 2 1365 17897880
+1 3 1365 17897880
+1 4 1366 17930644
+1 5 1365 17897880
+1 6 1365 17865120
+2 \N 1365 17876040
+2 0 1365 17908800
+2 1 1365 17908800
+2 2 1366 17908802
+2 3 1365 17876040
+2 4 1365 17876040
+2 5 1366 17876040
+2 6 1365 17908800
+
+-- !count_numeric_4_256 --
+0 2979
+1 2979
+10 2978
+2 2979
+3 2979
+4 2979
+5 2979
+6 2979
+7 2979
+8 2979
+9 2979
+
+-- !count_numeric_nullable_4_256 --
+\N 4096
+0 2606
+1 2607
+10 2606
+2 2606
+3 2607
+4 2607
+5 2606
+6 2607
+7 2607
+8 2606
+9 2607
+
+-- !count_string_4_256 --
+0 4682
+1 4681
+2 4681
+3 4681
+4 4681
+5 4681
+6 4681
+
+-- !count_string_nullable_4_256 --
+\N 4096
+0 4096
+1 4096
+2 4096
+3 4096
+4 4096
+5 4096
+6 4096
+
+-- !distinct_numeric_4_256 --
+32768 536854528 32768 1 1
+
+-- !distinct_composite_4_256 --
+32768 536854528 32768 429483623 4096
+
+-- !empty_4_256 --
diff --git
a/regression-test/suites/nereids_rules_p0/agg_strategy/bucketed_hash_agg_num_buckets.groovy
b/regression-test/suites/nereids_rules_p0/agg_strategy/bucketed_hash_agg_num_buckets.groovy
new file mode 100644
index 00000000000..ecef166d1ba
--- /dev/null
+++
b/regression-test/suites/nereids_rules_p0/agg_strategy/bucketed_hash_agg_num_buckets.groovy
@@ -0,0 +1,177 @@
+// 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.
+
+suite("bucketed_hash_agg_num_buckets") {
+ sql "set enable_sql_cache=false"
+ // Bucketed aggregation is disabled while spill is enabled, including in
fuzzy sessions.
+ sql "set enable_spill=false"
+ sql "set enable_force_spill=false"
+ sql "set enable_bucketed_hash_agg=false"
+ sql "set be_number_for_test=1"
+ sql "set agg_phase=1"
+ sql "set bucketed_agg_min_input_rows=0"
+ sql "set bucketed_agg_max_group_keys=0"
+ sql "set bucketed_agg_high_card_threshold=1.0"
+
+ sql "DROP TABLE IF EXISTS bucketed_agg_num_buckets_test"
+ sql """
+ CREATE TABLE bucketed_agg_num_buckets_test (
+ id INT NOT NULL,
+ distribution_key INT NOT NULL,
+ numeric_grp INT NOT NULL,
+ nullable_numeric_grp INT,
+ non_null_grp VARCHAR(32) NOT NULL,
+ grp VARCHAR(32),
+ val BIGINT
+ ) DUPLICATE KEY(id, distribution_key)
+ DISTRIBUTED BY HASH(distribution_key) BUCKETS 4
+ PROPERTIES('replication_num'='1')
+ """
+ // The distribution key is absent from every GROUP BY, preserving the
hash-distribute
+ // shape required by bucketed fusion even for the all-distinct id cases.
+ // Multiple input batches exercise both local table growth and merging
across sink instances.
+ sql """
+ INSERT INTO bucketed_agg_num_buckets_test
+ SELECT number, number % 13, number % 11, IF(number % 8 = 0, NULL,
number % 11),
+ CAST(number % 7 AS VARCHAR(32)),
+ IF(number % 8 = 0, NULL, CAST(number % 7 AS VARCHAR(32))),
+ IF(number % 5 = 0, NULL, number)
+ FROM numbers('number'='32768')
+ """
+
+ def queryCases = [
+ string: """
+ SELECT grp, COUNT(*), COUNT(val), SUM(val), MIN(val), MAX(val)
+ FROM bucketed_agg_num_buckets_test GROUP BY grp
+ """,
+ numeric: """
+ SELECT numeric_grp, COUNT(*), SUM(val)
+ FROM bucketed_agg_num_buckets_test GROUP BY numeric_grp
+ """,
+ composite: """
+ SELECT id % 3 AS k, grp, COUNT(*), SUM(val)
+ FROM bucketed_agg_num_buckets_test GROUP BY k, grp
+ """,
+ // COUNT(*) alone selects the inline-count path for supported key
types.
+ count_numeric: """
+ SELECT numeric_grp, COUNT(*)
+ FROM bucketed_agg_num_buckets_test GROUP BY numeric_grp
+ """,
+ count_numeric_nullable: """
+ SELECT nullable_numeric_grp, COUNT(*)
+ FROM bucketed_agg_num_buckets_test GROUP BY nullable_numeric_grp
+ """,
+ count_string: """
+ SELECT non_null_grp, COUNT(*)
+ FROM bucketed_agg_num_buckets_test GROUP BY non_null_grp
+ """,
+ count_string_nullable: """
+ SELECT grp, COUNT(*)
+ FROM bucketed_agg_num_buckets_test GROUP BY grp
+ """,
+ // Keep 32768 distinct keys in the tested inner aggregate. The outer
scalar
+ // aggregate reduces expected output to one row and cannot itself be
bucketed.
+ distinct_numeric: """
+ SELECT COUNT(*), SUM(k), SUM(cnt), MIN(cnt), MAX(cnt)
+ FROM (
+ SELECT id AS k, COUNT(*) AS cnt
+ FROM bucketed_agg_num_buckets_test GROUP BY id
+ ) t
+ """,
+ distinct_composite: """
+ SELECT COUNT(*), SUM(k), SUM(cnt), SUM(total),
+ SUM(IF(grp IS NULL, cnt, 0))
+ FROM (
+ SELECT id AS k, grp, COUNT(*) AS cnt, SUM(val) AS total
+ FROM bucketed_agg_num_buckets_test GROUP BY id, grp
+ ) t
+ """,
+ // Avoid a constant/range predicate that could remove the aggregate
from the plan.
+ empty: """
+ SELECT grp, COUNT(*), SUM(val) FROM bucketed_agg_num_buckets_test
+ WHERE MD5(CAST(id AS STRING)) = 'bucketed-empty' GROUP BY grp
+ """
+ ]
+
+ // Compute independent expected rows through regular aggregation once.
Golden generation
+ // must not accept a shared bucketed-aggregation bug as the expected
result.
+ sql "set parallel_pipeline_task_num=1"
+ def referenceResults = [:]
+ queryCases.each { kind, query ->
+ explain {
+ sql query
+ notContains "BUCKETED AGGREGATE"
+ }
+ referenceResults[kind] = order_sql(query)
+ }
+ String unlimitedCountQuery = """
+ SELECT id, COUNT(*) FROM bucketed_agg_num_buckets_test GROUP BY id
+ """
+ explain {
+ sql unlimitedCountQuery
+ notContains "BUCKETED AGGREGATE"
+ }
+ def limitReference = order_sql(unlimitedCountQuery).collectEntries { row ->
+ [(row[0]): row[1]]
+ }
+ sql "set enable_bucketed_hash_agg=true"
+
+ for (def parallelism : [1, 4]) {
+ sql "set parallel_pipeline_task_num=${parallelism}"
+ for (def numBuckets : [0, 1, 4, 16, 64, 256]) {
+ sql "set bucketed_agg_num_buckets=${numBuckets}"
+ String suffix = "${parallelism}_${numBuckets}"
+ queryCases.each { kind, query ->
+ explain {
+ sql query
+ contains "VBUCKETED AGGREGATE"
+ contains "buckets: ${numBuckets == 0 ? 'auto' :
numBuckets}"
+ }
+ assertEquals(referenceResults[kind], order_sql(query))
+ "order_qt_${kind}_${suffix}"(query)
+ }
+
+ // LimitAggToTopNAgg also adds TopN to an unordered LIMIT. Disable
it only
+ // for this query so the bucketed operator exercises early
termination.
+ // Any three groups are valid here; verify their invariants
without a golden key order.
+ String limitQuery = """
+ SELECT /*+ SET_VAR(push_topn_to_agg=false) */ id, COUNT(*)
+ FROM bucketed_agg_num_buckets_test GROUP BY id LIMIT 3
+ """
+ explain {
+ sql limitQuery
+ contains "VBUCKETED AGGREGATE"
+ contains "buckets: ${numBuckets == 0 ? 'auto' : numBuckets}"
+ }
+ def limitRows = sql limitQuery
+ assertEquals(3, limitRows.size())
+ assertEquals(3, limitRows.collect { it[0] }.toSet().size())
+ limitRows.each { row ->
+ long key = (row[0] as Number).longValue()
+ assertTrue(key >= 0 && key < 32768)
+ assertEquals(1L, (row[1] as Number).longValue())
+ assertEquals(limitReference[row[0]], row[1])
+ }
+ }
+ }
+ for (def invalid : [-1, 3, 255, 257, 512]) {
+ test {
+ sql "set bucketed_agg_num_buckets=${invalid}"
+ exception "bucketed_agg_num_buckets must be 0 or a power of two
between 1 and 256"
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
