This is an automated email from the ASF dual-hosted git repository.
HappenLee pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 40883594296 [fix](be) Reset quantile union state for empty window
frames (#69964)
40883594296 is described below
commit 4088359429618d31bee87b5f42e25c75935ee2af
Author: Jerry Hu <[email protected]>
AuthorDate: Tue Nov 3 11:46:56 2026 +0800
[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/aggregate_function_quantile_state.h | 5 +-
be/test/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(-)
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 <gtest/gtest.h>
+#include <cmath>
#include <limits>
#include <memory>
#include <optional>
#include <vector>
+#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<double>&
actual, const std::vector<d
}
}
+void expect_reset_quantile_state(size_t count, bool is_window) {
+ auto type = std::make_shared<DataTypeQuantileState>();
+ auto function = create_aggregate_function_quantile_state_union(
+ "quantile_union", {type}, type, false,
+ {.is_window_function = is_window, .column_names = {}});
+ std::unique_ptr<char[]> 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<DataTypeQuantileState>();
+ auto function = create_aggregate_function_quantile_state_union(
+ "quantile_union", {type}, type, false,
+ {.is_window_function = true, .column_names = {}});
+ std::unique_ptr<char[]> 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<AggregateFunctionQuantileStateUnionOp>;
+ auto& accumulator = reinterpret_cast<Data*>(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<DataTypeQuantileState>();
+ auto function = create_aggregate_function_quantile_state_union(
+ "quantile_union", {type}, type, false,
+ {.is_window_function = true, .column_names = {}});
+ std::unique_ptr<char[]> 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<DataTypeQuantileState>();
+ auto function = create_aggregate_function_quantile_state_union(
+ "quantile_union", {make_nullable(type)}, type, false,
+ {.is_window_function = true, .column_names = {}});
+ std::unique_ptr<char[]> 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<DataTypeQuantileState>();
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
+ """
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
