From 4088359429618d31bee87b5f42e25c75935ee2af Mon Sep 17 00:00:00 2001 From: Jerry Hu Date: Tue, 3 Nov 2026 11:46:56 +0800 Subject: [PATCH] [fix](be) Reset quantile union state for empty window frames (#69964) ### What problem does this PR solve? Issue Number: N/A Related PR: N/A Problem Summary: A bounded ROWS window can become empty after a nonempty frame. The `quantile_union` reset previously restored only the first-input flag and retained the preceding `QuantileState`. Without a subsequent value to merge, an empty or NULL-only frame returned the preceding frame's quantiles. Reset both the first-input flag and the stored samples before each frame. The logical reset retains explicit sample-vector capacity for reuse, while `clear()` preserves its storage-release behavior. ### Release note Fix `quantile_union` over empty window frames: `quantile_percent` now returns NaN instead of values from the preceding frame. ### Check List (For Author) - Test: ASAN BE unit tests (47 passed); ASAN BE+FE build (including FE Checkstyle); three isolated SQL regression suites (1 suite passed each); clang-format, format check, build hygiene, changed-line clang-tidy (no warnings), and diff check. No performance benchmark was run. - Behavior changed: Yes (empty window frames no longer inherit the preceding quantile state). - Does this need documentation: No. --- be/src/core/value/quantile_state.cpp | 6 +- be/src/core/value/quantile_state.h | 2 + .../aggregate_function_quantile_state.h | 5 +- .../exprs/aggregate/agg_percentile_test.cpp | 183 ++++++++++++++++++ .../test_quantile_union_empty_window.out | 28 +++ .../test_quantile_union_empty_window.groovy | 91 +++++++++ 6 files changed, 313 insertions(+), 2 deletions(-) create mode 100644 regression-test/data/query_p0/sql_functions/quantile_state_functions/test_quantile_union_empty_window.out create mode 100644 regression-test/suites/query_p0/sql_functions/quantile_state_functions/test_quantile_union_empty_window.groovy diff --git a/be/src/core/value/quantile_state.cpp b/be/src/core/value/quantile_state.cpp index c73ce277c2a..d432a44e352 100644 --- a/be/src/core/value/quantile_state.cpp +++ b/be/src/core/value/quantile_state.cpp @@ -424,10 +424,14 @@ void QuantileState::add_value(const double& value) { } } -void QuantileState::clear() { +void QuantileState::reset() { _type = EMPTY; _tdigest_ptr.reset(); _explicit_data.clear(); +} + +void QuantileState::clear() { + reset(); _explicit_data.shrink_to_fit(); } diff --git a/be/src/core/value/quantile_state.h b/be/src/core/value/quantile_state.h index fc540848c26..82e842a010c 100644 --- a/be/src/core/value/quantile_state.h +++ b/be/src/core/value/quantile_state.h @@ -60,6 +60,8 @@ public: size_t serialize(uint8_t* dst) const; void merge(const QuantileState& other); void add_value(const double& value); + // Clear samples while retaining explicit capacity for the next aggregate frame. + void reset(); void clear(); bool is_valid(const Slice& slice); size_t get_serialized_size() const; diff --git a/be/src/exprs/aggregate/aggregate_function_quantile_state.h b/be/src/exprs/aggregate/aggregate_function_quantile_state.h index 7f5f09710bd..2b43653f6bb 100644 --- a/be/src/exprs/aggregate/aggregate_function_quantile_state.h +++ b/be/src/exprs/aggregate/aggregate_function_quantile_state.h @@ -83,7 +83,10 @@ struct AggregateFunctionQuantileStateData { void read(BufferReadable& buf) { DataTypeQuantileState::deserialize_as_stream(value, buf); } - void reset() { is_first = true; } + void reset() { + is_first = true; + value.reset(); + } DataType& get() { return value; } diff --git a/be/test/exprs/aggregate/agg_percentile_test.cpp b/be/test/exprs/aggregate/agg_percentile_test.cpp index 7a1030e837c..3912d543b23 100644 --- a/be/test/exprs/aggregate/agg_percentile_test.cpp +++ b/be/test/exprs/aggregate/agg_percentile_test.cpp @@ -17,11 +17,13 @@ #include +#include #include #include #include #include +#include "core/arena.h" #include "core/column/column_array.h" #include "core/column/column_complex.h" #include "core/column/column_nullable.h" @@ -32,9 +34,11 @@ #include "core/data_type/data_type_number.h" #include "core/data_type/data_type_quantilestate.h" #include "core/string_buffer.hpp" +#include "core/value/quantile_state.h" #include "exprs/aggregate/aggregate_function_percentile.h" #include "exprs/aggregate/aggregate_function_quantile_state.h" #include "exprs/aggregate/aggregate_function_simple_factory.h" +#include "util/defer_op.h" #include "util/tdigest.h" namespace doris { @@ -111,8 +115,187 @@ void expect_results_equal(const std::vector& actual, const std::vector(); + auto function = create_aggregate_function_quantile_state_union( + "quantile_union", {type}, type, false, + {.is_window_function = is_window, .column_names = {}}); + std::unique_ptr memory(new char[function->size_of_data()]); + auto* place = memory.get(); + function->create(place); + Defer destroy([&] { function->destroy(place); }); + Arena arena; + auto input = ColumnQuantileState::create(); + QuantileState seed(10000); + for (size_t i = 0; i < count; ++i) { + seed.add_value(20); + } + input->insert_value(seed); + QuantileState next(2048); + next.add_value(30); + input->insert_value(next); + const IColumn* columns[] = {input.get()}; + + auto results = ColumnQuantileState::create(); + function->insert_result_into(place, *results); + function->add(place, columns, 0, arena); + function->insert_result_into(place, *results); + function->reset(place); + function->insert_result_into(place, *results); + function->reset(place); + function->insert_result_into(place, *results); + function->add(place, columns, 1, arena); + function->insert_result_into(place, *results); + + ASSERT_EQ(5, results->size()); + for (double percentile : {0.0, 0.5, 1.0}) { + EXPECT_TRUE(std::isnan(results->get_element(0).get_value_by_percentile(percentile))); + EXPECT_DOUBLE_EQ(20, results->get_element(1).get_value_by_percentile(percentile)); + EXPECT_TRUE(std::isnan(results->get_element(2).get_value_by_percentile(percentile))); + EXPECT_TRUE(std::isnan(results->get_element(3).get_value_by_percentile(percentile))); + EXPECT_DOUBLE_EQ(30, results->get_element(4).get_value_by_percentile(percentile)); + EXPECT_DOUBLE_EQ(20, input->get_element(0).get_value_by_percentile(percentile)); + } + EXPECT_EQ(sizeof(float) + 1, results->get_element(2).get_serialized_size()); + EXPECT_EQ(sizeof(float) + 1, results->get_element(3).get_serialized_size()); +} + +void append_explicit_frame(const AggregateFunctionPtr& function, AggregateDataPtr place, + const IColumn** columns, size_t row, size_t frame_size, Arena& arena, + ColumnQuantileState& results, QuantileState& accumulator) { + for (size_t sample = 0; sample < frame_size; ++sample) { + function->add(place, columns, row, arena); + } + ASSERT_EQ(EXPLICIT, accumulator._type); + const size_t capacity = accumulator._explicit_data.capacity(); + ASSERT_GE(capacity, frame_size); + function->insert_result_into(place, results); + function->reset(place); + EXPECT_EQ(EMPTY, accumulator._type); + EXPECT_TRUE(accumulator._explicit_data.empty()); + // Bounded ROWS frames reset on every row; the warmed vector must survive each reset. + EXPECT_EQ(capacity, accumulator._explicit_data.capacity()); + function->insert_result_into(place, results); + function->add(place, columns, row, arena); + EXPECT_EQ(capacity, accumulator._explicit_data.capacity()); + function->reset(place); +} + +void expect_repeated_explicit_frames_reuse_capacity(size_t frame_size) { + auto type = std::make_shared(); + auto function = create_aggregate_function_quantile_state_union( + "quantile_union", {type}, type, false, + {.is_window_function = true, .column_names = {}}); + std::unique_ptr memory(new char[function->size_of_data()]); + auto* place = memory.get(); + function->create(place); + Defer destroy([&] { function->destroy(place); }); + Arena arena; + auto input = ColumnQuantileState::create(); + for (double value : {20.0, 30.0}) { + QuantileState state; + state.add_value(value); + input->insert_value(state); + } + const IColumn* columns[] = {input.get()}; + using Data = AggregateFunctionQuantileStateData; + auto& accumulator = reinterpret_cast(place)->value; + + auto results = ColumnQuantileState::create(); + for (size_t frame = 0; frame < 3; ++frame) { + append_explicit_frame(function, place, columns, frame % 2, frame_size, arena, *results, + accumulator); + } + ASSERT_EQ(6, results->size()); + for (size_t frame = 0; frame < 3; ++frame) { + const double expected = frame % 2 == 0 ? 20 : 30; + EXPECT_DOUBLE_EQ(expected, results->get_element(2 * frame).get_value_by_percentile(0.5)); + EXPECT_TRUE(std::isnan(results->get_element(2 * frame + 1).get_value_by_percentile(0.5))); + } +} + } // namespace +TEST(AggregateFunctionQuantileStateTest, EmptyWindowAfterNonemptyWindow) { + auto type = std::make_shared(); + auto function = create_aggregate_function_quantile_state_union( + "quantile_union", {type}, type, false, + {.is_window_function = true, .column_names = {}}); + std::unique_ptr memory(new char[function->size_of_data()]); + auto* place = memory.get(); + function->create(place); + Defer destroy([&] { function->destroy(place); }); + Arena arena; + auto input = ColumnQuantileState::create(); + QuantileState state; + state.add_value(20); + input->insert_value(state); + const IColumn* columns[] = {input.get()}; + + auto results = ColumnQuantileState::create(); + function->insert_result_into(place, *results); + function->add(place, columns, 0, arena); + function->insert_result_into(place, *results); + function->reset(place); + function->insert_result_into(place, *results); + + ASSERT_EQ(3, results->size()); + for (double percentile : {0.0, 0.5, 1.0}) { + EXPECT_TRUE(std::isnan(results->get_element(0).get_value_by_percentile(percentile))); + EXPECT_DOUBLE_EQ(20, results->get_element(1).get_value_by_percentile(percentile)); + EXPECT_TRUE(std::isnan(results->get_element(2).get_value_by_percentile(percentile))); + } +} + +TEST(AggregateFunctionQuantileStateTest, ResetClearsAllRepresentations) { + for (bool is_window : {false, true}) { + for (size_t count : {size_t(1), size_t(3), size_t(QUANTILE_STATE_EXPLICIT_NUM + 1)}) { + expect_reset_quantile_state(count, is_window); + } + } +} + +TEST(AggregateFunctionQuantileStateTest, RepeatedExplicitFramesReuseCapacity) { + for (size_t frame_size : {size_t(2), size_t(3), size_t(QUANTILE_STATE_EXPLICIT_NUM)}) { + expect_repeated_explicit_frames_reuse_capacity(frame_size); + } +} + +TEST(AggregateFunctionQuantileStateTest, NullOnlyWindowAfterResetIsEmpty) { + auto type = std::make_shared(); + auto function = create_aggregate_function_quantile_state_union( + "quantile_union", {make_nullable(type)}, type, false, + {.is_window_function = true, .column_names = {}}); + std::unique_ptr memory(new char[function->size_of_data()]); + auto* place = memory.get(); + function->create(place); + Defer destroy([&] { function->destroy(place); }); + Arena arena; + auto nested = ColumnQuantileState::create(); + QuantileState state; + state.add_value(20); + nested->insert_value(state); + nested->insert_default(); + auto null_map = ColumnUInt8::create(); + null_map->insert_value(0); + null_map->insert_value(1); + auto input = ColumnNullable::create(std::move(nested), std::move(null_map)); + const IColumn* columns[] = {input.get()}; + + auto results = ColumnQuantileState::create(); + function->add(place, columns, 0, arena); + function->insert_result_into(place, *results); + function->reset(place); + function->add(place, columns, 1, arena); + function->insert_result_into(place, *results); + + ASSERT_EQ(2, results->size()); + for (double percentile : {0.0, 0.5, 1.0}) { + EXPECT_DOUBLE_EQ(20, results->get_element(0).get_value_by_percentile(percentile)); + EXPECT_TRUE(std::isnan(results->get_element(1).get_value_by_percentile(percentile))); + } +} + TEST(AggregateFunctionQuantileStateTest, GrowingWindowKeepsResultsCompact) { auto type = std::make_shared(); auto function = create_aggregate_function_quantile_state_union( diff --git a/regression-test/data/query_p0/sql_functions/quantile_state_functions/test_quantile_union_empty_window.out b/regression-test/data/query_p0/sql_functions/quantile_state_functions/test_quantile_union_empty_window.out new file mode 100644 index 00000000000..8fd70eee474 --- /dev/null +++ b/regression-test/data/query_p0/sql_functions/quantile_state_functions/test_quantile_union_empty_window.out @@ -0,0 +1,28 @@ +-- This file is automatically generated. You should know what you did if you want to edit this +-- !initial_empty -- +0 NaN NaN NaN + +-- !nonempty_then_empty -- +0 20 20 20 +1 NaN NaN NaN + +-- !consecutive_nonempty_then_empty -- +0 20 20 20 +1 30 30 30 +2 NaN NaN NaN + +-- !multi_value_frame_then_empty -- +0 20 25 30 +1 30 30 30 +2 NaN NaN NaN + +-- !current_row_control -- +0 10 10 10 +1 20 20 20 +2 30 30 30 + +-- !partition_reset -- +0 0 NaN NaN NaN +0 1 10 10 10 +1 0 NaN NaN NaN +1 1 30 30 30 diff --git a/regression-test/suites/query_p0/sql_functions/quantile_state_functions/test_quantile_union_empty_window.groovy b/regression-test/suites/query_p0/sql_functions/quantile_state_functions/test_quantile_union_empty_window.groovy new file mode 100644 index 00000000000..3bed8233f84 --- /dev/null +++ b/regression-test/suites/query_p0/sql_functions/quantile_state_functions/test_quantile_union_empty_window.groovy @@ -0,0 +1,91 @@ +// 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("test_quantile_union_empty_window") { + qt_initial_empty """ + SELECT id, quantile_percent(q, 0) AS q_min, + quantile_percent(q, 0.5) AS q_median, quantile_percent(q, 1) AS q_max + FROM ( + SELECT number AS id, + quantile_union(to_quantile_state(10 + 10 * number, 2048)) OVER ( + ORDER BY number ROWS BETWEEN 1 FOLLOWING AND 1 FOLLOWING + ) AS q + FROM numbers("number" = "1") + ) w ORDER BY id + """ + + qt_nonempty_then_empty """ + SELECT id, quantile_percent(q, 0) AS q_min, + quantile_percent(q, 0.5) AS q_median, quantile_percent(q, 1) AS q_max + FROM ( + SELECT number AS id, + quantile_union(to_quantile_state(10 + 10 * number, 2048)) OVER ( + ORDER BY number ROWS BETWEEN 1 FOLLOWING AND 1 FOLLOWING + ) AS q + FROM numbers("number" = "2") + ) w ORDER BY id + """ + + qt_consecutive_nonempty_then_empty """ + SELECT id, quantile_percent(q, 0) AS q_min, + quantile_percent(q, 0.5) AS q_median, quantile_percent(q, 1) AS q_max + FROM ( + SELECT number AS id, + quantile_union(to_quantile_state(10 + 10 * number, 2048)) OVER ( + ORDER BY number ROWS BETWEEN 1 FOLLOWING AND 1 FOLLOWING + ) AS q + FROM numbers("number" = "3") + ) w ORDER BY id + """ + + qt_multi_value_frame_then_empty """ + SELECT id, quantile_percent(q, 0) AS q_min, + quantile_percent(q, 0.5) AS q_median, quantile_percent(q, 1) AS q_max + FROM ( + SELECT number AS id, + quantile_union(to_quantile_state(10 + 10 * number, 2048)) OVER ( + ORDER BY number ROWS BETWEEN 1 FOLLOWING AND 2 FOLLOWING + ) AS q + FROM numbers("number" = "3") + ) w ORDER BY id + """ + + qt_current_row_control """ + SELECT id, quantile_percent(q, 0) AS q_min, + quantile_percent(q, 0.5) AS q_median, quantile_percent(q, 1) AS q_max + FROM ( + SELECT number AS id, + quantile_union(to_quantile_state(10 + 10 * number, 2048)) OVER ( + ORDER BY number ROWS BETWEEN CURRENT ROW AND CURRENT ROW + ) AS q + FROM numbers("number" = "3") + ) w ORDER BY id + """ + + qt_partition_reset """ + SELECT p, id, quantile_percent(q, 0) AS q_min, + quantile_percent(q, 0.5) AS q_median, quantile_percent(q, 1) AS q_max + FROM ( + SELECT number DIV 2 AS p, number % 2 AS id, + quantile_union(to_quantile_state(10 + 10 * number, 2048)) OVER ( + PARTITION BY number DIV 2 ORDER BY number % 2 + ROWS BETWEEN 1 PRECEDING AND 1 PRECEDING + ) AS q + FROM numbers("number" = "4") + ) w ORDER BY p, id + """ +}