From 10915037d22862bbe6c867c6146ea1765bade531 Mon Sep 17 00:00:00 2001 From: Hu Shenggang Date: Mon, 21 Sep 2026 15:24:36 +0800 Subject: [PATCH] [fix](be) Keep all samples when merging percentile states ### What problem does this PR solve? Issue Number: None Related PR: None Problem Summary: `Counts::merge` (the sample container behind `percentile`, `percentile_array` and `PERCENTILE_STATE/UNION/MERGE`) only moved the raw `_nums` of the source state into `_sorted_nums_vec`. Two in-memory merges were therefore wrong: 1. The source state was itself a merged state: its samples live only in `_sorted_nums_vec`, `_nums` is empty, so every sample was dropped. 2. The destination still held raw samples in `_nums`: `terminate` with a single sorted run overwrote `_nums` with that run, the multi-run path ignored `_nums`, `serialize` ignored `_sorted_nums_vec` when `_nums` was non-empty, and unsorted raw samples could be pushed as a "sorted" run. Bucketed hash aggregation merges per-instance hash-table states directly in the source operator, which hits both cases. For example, with rows `(shard, v) = (0,0),(1,0),(1,10)`, `agg_phase=1`, `enable_bucketed_hash_agg=true` and `parallel_pipeline_task_num=4`, SELECT PERCENTILE_MERGE(s) FROM ( SELECT shard, PERCENTILE_UNION(PERCENTILE_STATE(v, 0.625)) s FROM t GROUP BY shard) q; returned 0 or 6.25 non-deterministically instead of 2.5. Plain `percentile(v, q) ... GROUP BY` under bucketed aggregation lost samples too. Fix: `merge` now moves the raw samples of both sides (sorted if needed) into `_sorted_nums_vec` and takes over all sorted runs of the source. `serialize` and `terminate` fold any remaining raw samples into the sorted runs before merging them. The serialized format is unchanged. ### Release note Fix wrong and non-deterministic results of percentile / percentile_array / PERCENTILE_UNION when states are merged by bucketed hash aggregation. ### Check List (For Author) - Test: - Unit Test: PercentileUtilTest (3 new cases fail before the fix, all 16 pass after) - Regression test: query_p0/aggregate/percentile_bucketed_agg_merge (fails 3/3 runs on an unfixed BE, passes 3/3 with the fix; results match the non-bucketed controls) - Manual test: the reproduction above returns 2.5 in 8/8 runs - Behavior changed: Yes (percentile results under bucketed hash aggregation are now correct) - Does this need documentation: No Claude-Session: https://claude.ai/code/session_01Afs6xcnnuT7cUpnkiLEJ11 --- be/src/util/percentile_util.h | 36 +++++-- be/test/util/percentile_util_test.cpp | 84 +++++++++++++++++ .../percentile_bucketed_agg_merge.out | 23 +++++ .../percentile_bucketed_agg_merge.groovy | 93 +++++++++++++++++++ 4 files changed, 228 insertions(+), 8 deletions(-) create mode 100644 regression-test/data/query_p0/aggregate/percentile_bucketed_agg_merge.out create mode 100644 regression-test/suites/query_p0/aggregate/percentile_bucketed_agg_merge.groovy diff --git a/be/src/util/percentile_util.h b/be/src/util/percentile_util.h index 030e8df8069354..ba6716f75682e9 100644 --- a/be/src/util/percentile_util.h +++ b/be/src/util/percentile_util.h @@ -47,10 +47,15 @@ class Counts { public: Counts() = default; + // Moves all samples of `other` into this state. `other` may itself be a merged state whose + // samples live in `_sorted_nums_vec`, and either side may still hold unsorted raw samples. void merge(Counts* other) { - if (other != nullptr && !other->_nums.empty()) { - _sorted_nums_vec.emplace_back(std::move(other->_nums)); + _move_nums_to_sorted_vec(); + other->_move_nums_to_sorted_vec(); + for (auto& nums : other->_sorted_nums_vec) { + _sorted_nums_vec.emplace_back(std::move(nums)); } + other->_sorted_nums_vec.clear(); } void increment(Ty key, uint32_t i) { @@ -66,16 +71,16 @@ class Counts { void increment_batch(const PaddedPODArray& keys) { _nums.insert(keys.begin(), keys.end()); } void serialize(BufferWritable& buf) { - if (!_nums.empty()) { + if (_sorted_nums_vec.empty()) { pdqsort(_nums.begin(), _nums.end()); - size_t size = _nums.size(); - buf.write_binary(size); - buf.write(reinterpret_cast(_nums.data()), sizeof(Ty) * size); } else { - // convert _sorted_nums_vec to _nums and do seiralize again + // merge all sorted runs (including the raw samples) into `_nums` + _move_nums_to_sorted_vec(); _convert_sorted_num_vec_to_nums(); - serialize(buf); } + size_t size = _nums.size(); + buf.write_binary(size); + buf.write(reinterpret_cast(_nums.data()), sizeof(Ty) * size); } void unserialize(BufferReadable& buf) { @@ -87,9 +92,13 @@ class Counts { } double terminate(double quantile) { + if (!_sorted_nums_vec.empty()) { + _move_nums_to_sorted_vec(); + } if (_sorted_nums_vec.size() <= 1) { if (_sorted_nums_vec.size() == 1) { _nums = std::move(_sorted_nums_vec[0]); + _sorted_nums_vec.clear(); } if (_nums.empty()) { @@ -150,6 +159,17 @@ class Counts { auto operator<=>(const Node& other) const { return value <=> other.value; } }; + void _move_nums_to_sorted_vec() { + if (_nums.empty()) { + return; + } + if (!std::is_sorted(_nums.begin(), _nums.end())) { + pdqsort(_nums.begin(), _nums.end()); + } + _sorted_nums_vec.emplace_back(std::move(_nums)); + DCHECK(_nums.empty()); + } + void _convert_sorted_num_vec_to_nums() { size_t rows = 0; for (const auto& i : _sorted_nums_vec) { diff --git a/be/test/util/percentile_util_test.cpp b/be/test/util/percentile_util_test.cpp index 5f6126978e6fff..a00ae2f75a56e2 100644 --- a/be/test/util/percentile_util_test.cpp +++ b/be/test/util/percentile_util_test.cpp @@ -135,6 +135,90 @@ TEST_F(PercentileUtilTest, CountsSerializeMergedState) { EXPECT_DOUBLE_EQ(9.0, restored.terminate(1.0)); } +TEST_F(PercentileUtilTest, CountsMergeMergedStates) { + // Two merged states whose samples only live in sorted runs, e.g. partial states + // combined by different pipeline instances before the final merge. + Counts run0; + run0.increment(0); + Counts run1; + run1.increment(10); + run1.increment(0); + Counts left; + left.merge(&run0); + Counts right; + right.merge(&run1); + + Counts merged; + merged.merge(&left); + merged.merge(&right); + // samples are [0, 0, 10], (3 - 1) * 0.625 = 1.25 + EXPECT_DOUBLE_EQ(2.5, merged.terminate(0.625)); + + Counts merged_reverse; + Counts run2; + run2.increment(0); + Counts run3; + run3.increment(0); + run3.increment(10); + Counts left_reverse; + left_reverse.merge(&run2); + Counts right_reverse; + right_reverse.merge(&run3); + merged_reverse.merge(&right_reverse); + merged_reverse.merge(&left_reverse); + EXPECT_DOUBLE_EQ(2.5, merged_reverse.terminate(0.625)); +} + +TEST_F(PercentileUtilTest, CountsMergeRawStates) { + // Both sides hold unsorted raw samples, e.g. update states of different instances. + Counts dst; + dst.increment(9); + dst.increment(1); + Counts src; + src.increment(7); + src.increment(3); + src.increment(5); + dst.merge(&src); + // samples are [1, 3, 5, 7, 9] + EXPECT_DOUBLE_EQ(1.0, dst.terminate(0.0)); + EXPECT_DOUBLE_EQ(4.0, dst.terminate(0.375)); + EXPECT_DOUBLE_EQ(5.0, dst.terminate(0.5)); + EXPECT_DOUBLE_EQ(8.0, dst.terminate(0.875)); + EXPECT_DOUBLE_EQ(9.0, dst.terminate(1.0)); + + Counts single_run_dst; + single_run_dst.increment(4); + single_run_dst.increment(2); + Counts empty_src; + single_run_dst.merge(&empty_src); + EXPECT_DOUBLE_EQ(3.0, single_run_dst.terminate(0.5)); +} + +TEST_F(PercentileUtilTest, CountsSerializeMixedState) { + Counts src; + src.increment(8); + src.increment(2); + Counts merged; + merged.merge(&src); + // raw samples added after a merge must be kept together with the sorted runs + merged.increment(6); + merged.increment(4); + + auto col = ColumnString::create(); + BufferWritable writer(*col); + merged.serialize(writer); + writer.commit(); + + StringRef data(col->get_chars().data(), col->get_chars().size()); + BufferReadable reader(data); + Counts restored; + restored.unserialize(reader); + // samples are [2, 4, 6, 8] + EXPECT_DOUBLE_EQ(2.0, restored.terminate(0.0)); + EXPECT_DOUBLE_EQ(5.0, restored.terminate(0.5)); + EXPECT_DOUBLE_EQ(8.0, restored.terminate(1.0)); +} + TEST_F(PercentileUtilTest, CheckQuantileBoundary) { EXPECT_NO_THROW(check_quantile(0.0)); EXPECT_NO_THROW(check_quantile(0.5)); diff --git a/regression-test/data/query_p0/aggregate/percentile_bucketed_agg_merge.out b/regression-test/data/query_p0/aggregate/percentile_bucketed_agg_merge.out new file mode 100644 index 00000000000000..f176744d8037a2 --- /dev/null +++ b/regression-test/data/query_p0/aggregate/percentile_bucketed_agg_merge.out @@ -0,0 +1,23 @@ +-- This file is automatically generated. You should know what you did if you want to edit this +-- !union_merge -- +6.25 + +-- !union_merge_group -- +0 6 +1 6.375 + +-- !percentile_group -- +0 6 [0, 1, 5, 8] +1 6.375 [0, 2.25, 4.5, 10] + +-- !union_merge_no_bucketed -- +6.25 + +-- !union_merge_group_no_bucketed -- +0 6 +1 6.375 + +-- !percentile_group_no_bucketed -- +0 6 [0, 1, 5, 8] +1 6.375 [0, 2.25, 4.5, 10] + diff --git a/regression-test/suites/query_p0/aggregate/percentile_bucketed_agg_merge.groovy b/regression-test/suites/query_p0/aggregate/percentile_bucketed_agg_merge.groovy new file mode 100644 index 00000000000000..cfe27b6bca9cd4 --- /dev/null +++ b/regression-test/suites/query_p0/aggregate/percentile_bucketed_agg_merge.groovy @@ -0,0 +1,93 @@ +// 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. + +// Bucketed hash aggregation merges the per-instance states of the same group in the +// source operator. PERCENTILE states merged there must keep every sample of both sides. +suite("percentile_bucketed_agg_merge") { + sql "set experimental_enable_agg_state=true" + sql "set enable_bucketed_hash_agg=true" + sql "set be_number_for_test=1" + sql "set agg_phase=1" + sql "set parallel_pipeline_task_num=4" + 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 percentile_bucketed_agg_merge_t" + sql """ + CREATE TABLE percentile_bucketed_agg_merge_t ( + id INT NOT NULL, + shard INT NOT NULL, + v INT NOT NULL + ) + DUPLICATE KEY(id) + DISTRIBUTED BY HASH(id) BUCKETS 8 + PROPERTIES ('replication_num' = '1') + """ + sql """ + INSERT INTO percentile_bucketed_agg_merge_t VALUES + (1, 0, 0), (2, 1, 0), (4, 1, 10), + (5, 0, 7), (6, 1, 3), (7, 0, 1), (8, 1, 9), + (9, 0, 5), (10, 1, 2), (11, 0, 8), (12, 1, 6) + """ + + explain { + sql """ + SELECT shard, PERCENTILE_UNION(PERCENTILE_STATE(v, CAST(0.625 AS DOUBLE))) + FROM percentile_bucketed_agg_merge_t GROUP BY shard + """ + contains("BUCKETED AGGREGATE") + } + + // merged PERCENTILE_UNION states of the same shard come from different instances + order_qt_union_merge """ + SELECT PERCENTILE_MERGE(s) FROM ( + SELECT shard, PERCENTILE_UNION(PERCENTILE_STATE(v, CAST(0.625 AS DOUBLE))) s + FROM percentile_bucketed_agg_merge_t GROUP BY shard + ) q + """ + order_qt_union_merge_group """ + SELECT shard, PERCENTILE_MERGE(s) FROM ( + SELECT shard, PERCENTILE_UNION(PERCENTILE_STATE(v, CAST(0.625 AS DOUBLE))) s + FROM percentile_bucketed_agg_merge_t GROUP BY shard + ) q GROUP BY shard + """ + // raw (unmerged) update states of the same shard come from different instances + order_qt_percentile_group """ + SELECT shard, PERCENTILE(v, 0.625), PERCENTILE_ARRAY(v, [0, 0.25, 0.5, 1]) + FROM percentile_bucketed_agg_merge_t GROUP BY shard + """ + + // controls: the same queries without bucketed hash aggregation + sql "set enable_bucketed_hash_agg=false" + order_qt_union_merge_no_bucketed """ + SELECT PERCENTILE_MERGE(s) FROM ( + SELECT shard, PERCENTILE_UNION(PERCENTILE_STATE(v, CAST(0.625 AS DOUBLE))) s + FROM percentile_bucketed_agg_merge_t GROUP BY shard + ) q + """ + order_qt_union_merge_group_no_bucketed """ + SELECT shard, PERCENTILE_MERGE(s) FROM ( + SELECT shard, PERCENTILE_UNION(PERCENTILE_STATE(v, CAST(0.625 AS DOUBLE))) s + FROM percentile_bucketed_agg_merge_t GROUP BY shard + ) q GROUP BY shard + """ + order_qt_percentile_group_no_bucketed """ + SELECT shard, PERCENTILE(v, 0.625), PERCENTILE_ARRAY(v, [0, 0.25, 0.5, 1]) + FROM percentile_bucketed_agg_merge_t GROUP BY shard + """ +}