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..2254fecced3785 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); - sink_finished = std::make_unique[]>(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[]>(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 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/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); 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/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/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..ecef166d1bac05 --- /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" + } + } +}