From e2d4c6228663149235f13faf991b8c1e679a62ae Mon Sep 17 00:00:00 2001 From: happenlee Date: Thu, 22 Oct 2026 12:06:49 +0800 Subject: [PATCH 1/4] [improvement](be) Adapt bucket counts for bucketed aggregation ### What problem does this PR solve? Issue Number: N/A Related PR: N/A Problem Summary: Bucketed aggregation always allocates 256 hash tables, even for serial execution, and its bucket-routing bits overlap phmap probe-index bits as tables grow. Add a validated and forwarded bucket-count session variable, capture it in an optional plan field, choose one bucket for a single sink and source, and use high mixed-hash bits for routing. Reuse batch insertion for one bucket, avoid unused inline-count storage and skip single-sink merging. Add configuration, concurrency, plan-translation and SQL regression coverage. ### Release note Add bucketed_agg_num_buckets: 0 automatically chooses one bucket for a single sink and source and 256 otherwise; explicit values must be powers of two in [1, 256]. Missing Thrift fields retain 256 buckets. SQL aggregation semantics are unchanged. ### Check List (For Author) - Test: FE unit tests (35 passed); clang-format 16; FE Checkstyle; BE build hygiene; git diff --check; independent audit of 120 supplied golden sections. BE unit tests blocked by missing third-party libraries; regression execution and complete clang-tidy analysis remain pending. - Behavior changed: Yes (automatic bucket counts and per-session override; unchanged query results) - Does this need documentation: Yes (session-variable description included; website documentation PR pending) --- .../bucketed_aggregation_sink_operator.cpp | 143 +- .../bucketed_aggregation_sink_operator.h | 10 +- .../bucketed_aggregation_source_operator.cpp | 19 +- be/src/exec/pipeline/dependency.cpp | 33 +- be/src/exec/pipeline/dependency.h | 23 +- .../bucketed_agg_shared_state_test.cpp | 91 ++ .../translator/PhysicalPlanTranslator.java | 3 +- .../planner/BucketedAggregationNode.java | 7 +- .../org/apache/doris/qe/SessionVariable.java | 23 +- .../BucketedAggregateTranslatorTest.java | 48 + .../qe/SessionVariableBucketedAggTest.java | 65 + gensrc/thrift/PlanNodes.thrift | 3 + .../bucketed_hash_agg_num_buckets.out | 1236 +++++++++++++++++ .../bucketed_hash_agg_num_buckets.groovy | 174 +++ 14 files changed, 1791 insertions(+), 87 deletions(-) create mode 100644 fe/fe-core/src/test/java/org/apache/doris/qe/SessionVariableBucketedAggTest.java create mode 100644 regression-test/data/nereids_rules_p0/agg_strategy/bucketed_hash_agg_num_buckets.out create mode 100644 regression-test/suites/nereids_rules_p0/agg_strategy/bucketed_hash_agg_num_buckets.groovy diff --git a/be/src/exec/operator/bucketed_aggregation_sink_operator.cpp b/be/src/exec/operator/bucketed_aggregation_sink_operator.cpp index 9c76ec7e8a7800..2471735a80ff88 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(_num_buckets)); + COUNTER_SET(ADD_COUNTER(custom_profile(), "NumSinkInstances", TUnit::UNIT), + static_cast(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(); // 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(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((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(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(_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( - 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::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 1cd6a16fbe6799..6096f2e1963d24 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 { public: @@ -59,8 +59,9 @@ class BucketedAggSinkLocalState : public PipelineXSinkLocalStateper_instance_data[_instance_idx]. - std::vector _bucket_agg_data; // [256] + std::vector _bucket_agg_data; Arena* _arena = nullptr; /// Per-instance clones of probe expression contexts. Required because @@ -83,6 +84,7 @@ class BucketedAggSinkLocalState : public PipelineXSinkLocalState { public: @@ -136,6 +138,8 @@ class BucketedAggSinkOperatorX final : public DataSinkOperatorX _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 44221f14a4c84b..b1d20022bd83db 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>) { 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( 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 85ca4f0c21b5c8..60291ebea4fb2f 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()) { - bucket_agg_data.resize(BUCKETED_AGG_NUM_BUCKETS); +BucketedAggSharedState::PerInstanceData::PerInstanceData(int num_buckets) + : arena(std::make_unique()) { + bucket_agg_data.resize(num_buckets); for (auto& p : bucket_agg_data) { p = std::make_unique(); } @@ -261,16 +262,36 @@ BucketedAggSharedState::~BucketedAggSharedState() { } Status BucketedAggSharedState::init_instances(int num_instances, - const std::function& metadata_init) { + const std::function& 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); + 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[]>(num_instances); 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 714b9c10a52f0a..63f0bf50a1a4cb 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 @@ struct BucketedAggSharedState : public BasicSharedState { ~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 for string keys instead of StringHashMap. std::vector> bucket_agg_data; std::unique_ptr 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 @@ struct BucketedAggSharedState : public BasicSharedState { std::vector 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 num_sinks_finished = 0; @@ -526,7 +534,8 @@ struct BucketedAggSharedState : public BasicSharedState { /// 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& metadata_init); + Status init_instances(int num_instances, const std::function& metadata_init, + int requested_num_buckets = BUCKETED_AGG_NUM_BUCKETS); private: std::once_flag _init_once; 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 39ecaba7a67871..35370ab06d8dc3 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 +#include + #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 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 35e271b703a255..e648f33356013c 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 @@ private PlanFragment visitBucketedFusion( 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 5c36ead71fe575..354e354e97d36b 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 @@ 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 @@ protected void toThrift(TPlanNode msg) { bucketedAggNode.setAggregateFunctions(aggregateFunctions); bucketedAggNode.setTupleId(aggInfo.getOutputTupleId().asInt()); bucketedAggNode.setNeedFinalize(needsFinalize); + bucketedAggNode.setNumBuckets(numBuckets); msg.bucketed_agg_node = bucketedAggNode; } @@ -76,6 +79,8 @@ protected void toThrift(TPlanNode msg) { 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 542ddbc5174092..82225563965708 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 @@ -1041,6 +1041,7 @@ public static double getHotValueThreshold() { 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"; @@ -3110,9 +3111,16 @@ public void setSkewRewriteAggBucketNum(int num) { + "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; @@ -6452,6 +6460,19 @@ public void checkSkewRewriteJoinSaltExplodeFactor(String factorStr) { } } + 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 40b241a0b1af1d..760ef0f83ca225 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.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 @@ private String explain(Planner planner) { .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 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/qe/SessionVariableBucketedAggTest.java b/fe/fe-core/src/test/java/org/apache/doris/qe/SessionVariableBucketedAggTest.java new file mode 100644 index 00000000000000..864fe250fac4d1 --- /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 a3e997ddeb6b79..37517ab9a9f6ec 100644 --- a/gensrc/thrift/PlanNodes.thrift +++ b/gensrc/thrift/PlanNodes.thrift @@ -1319,6 +1319,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 00000000000000..0ec53f56edb27b --- /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 00000000000000..67b49797c11eaa --- /dev/null +++ b/regression-test/suites/nereids_rules_p0/agg_strategy/bucketed_hash_agg_num_buckets.groovy @@ -0,0 +1,174 @@ +// 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" + 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" + } + } +} From 2ea21cef7faab6191d85ea7ae2ca370f90ed33ef Mon Sep 17 00:00:00 2001 From: happenlee Date: Sun, 25 Oct 2026 02:31:06 +0800 Subject: [PATCH 2/4] [test](fe) Fix bucketed aggregation test prerequisites ### What problem does this PR solve? Issue Number: N/A Related PR: #69425 Problem Summary: The subquery translation unit test constructs a translator context without a session, so reading the configured bucket count throws a null pointer exception. Reuse the planner's CascadesContext, following existing translation tests. The bucket-count regression suite also requires a bucketed plan while fuzzy sessions may enable spill, which disables that plan. Explicitly disable both spill flags before exercising the bucket-count matrix. ### Release note None ### Check List (For Author) - Test: Unit Test: 50 tests passed across AnalyzeWhereSubqueryTest, BucketedAggregateTranslatorTest and SessionVariableBucketedAggTest; no failures, errors or skips. Runtime regression remains for CI. - Behavior changed: No; test setup only. - Does this need documentation: No. --- .../nereids/rules/analysis/AnalyzeWhereSubqueryTest.java | 5 +++-- .../agg_strategy/bucketed_hash_agg_num_buckets.groovy | 3 +++ 2 files changed, 6 insertions(+), 2 deletions(-) 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 0695b05ef700de..bc8748ea7c8c9e 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 void testTranslateCase() throws Exception { 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/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 index 67b49797c11eaa..ecef166d1bac05 100644 --- 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 @@ -17,6 +17,9 @@ 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" From 93f2199b2b4c4694fb2ad45e11780c1f423dfd27 Mon Sep 17 00:00:00 2001 From: happenlee Date: Sun, 25 Oct 2026 17:11:07 +0800 Subject: [PATCH 3/4] [test](be) Make bucketed merge failure injection deterministic ### What problem does this PR solve? Issue Number: N/A Related PR: #69425 Problem Summary: The merge-cleanup test calls get_block once and expects an injected exception on a duplicate key. Its original input also contains unique keys, so a changed bucket order can output a unique-key bucket before reaching the merge that throws. Use identical keys in both sinks so every non-empty bucket exercises duplicate merging before output. The test still verifies that all six aggregate states remain available for cleanup after the exception. ### Release note None ### Check List (For Author) - Test: Manual test: reviewed bucket traversal and the failure-before-transfer invariant; clang-format 16, check-format and git diff --check passed. - Unit Test: attempted run-be-ut.sh -j 48 --run --filter=BucketedAggOperatorTest.*; configuration completed, but the build stopped because local libsimdutf.a is missing. Tests did not execute locally and require CI validation. - Behavior changed: No; unit-test data only. - Does this need documentation: No. --- be/test/exec/operator/agg_operator_test.cpp | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/be/test/exec/operator/agg_operator_test.cpp b/be/test/exec/operator/agg_operator_test.cpp index f8e866eb4ce903..35fdd05cc34679 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); From 576b4676128a51a6f773adb903f3af730e1973fd Mon Sep 17 00:00:00 2001 From: happenlee Date: Mon, 26 Oct 2026 12:12:00 +0800 Subject: [PATCH 4/4] [fix](be) Use constructed instance count for bucketed sink flags ### What problem does this PR solve? Issue Number: N/A Related PR: #69425 Problem Summary: The performance pipeline fails during GCC 15 Release compilation with -Werror=alloc-size-larger-than after bucketed aggregation switches per-instance initialization from resize to reserve and emplace_back. GCC infers a potentially negative signed instance count at the completion-flag array allocation. Use the populated instance vector size for that allocation, preserving its length while avoiding the signed size conversion. Extracted initialization code reproduces the original diagnostic and compiles cleanly after the change with GCC 15.1.0, -O3 and -Werror. ### Release note None ### Check List (For Author) - Test: Manual test: GCC 15.1.0 extracted-code compile reproducer before/after; repository clang-format 16 formatting and check; BE build hygiene; git diff --check. BucketedAggSharedStateTest build attempted with run-be-ut.sh -j 48, blocked by missing local libsimdutf.a; tests did not execute. clang-tidy attempted but local compilation resources were incomplete. - Behavior changed: No - Does this need documentation: No --- be/src/exec/pipeline/dependency.cpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/be/src/exec/pipeline/dependency.cpp b/be/src/exec/pipeline/dependency.cpp index 60291ebea4fb2f..2254fecced3785 100644 --- a/be/src/exec/pipeline/dependency.cpp +++ b/be/src/exec/pipeline/dependency.cpp @@ -286,7 +286,7 @@ Status BucketedAggSharedState::init_instances(int num_instances, for (int i = 0; i < num_instances; ++i) { per_instance_data.emplace_back(num_buckets); } - sink_finished = std::make_unique[]>(num_instances); + sink_finished = std::make_unique[]>(per_instance_data.size()); for (int i = 0; i < num_instances; ++i) { sink_finished[i].store(false, std::memory_order_relaxed); }