This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new dad6cd65a25 branch-4.2: [fix](be) Reset quantile union state for empty
window frames (#70239)
dad6cd65a25 is described below
commit dad6cd65a2599e755f6a8e3b8260a6d97f72eb1e
Author: Jerry Hu <[email protected]>
AuthorDate: Thu Nov 5 17:20:15 2026 +0800
branch-4.2: [fix](be) Reset quantile union state for empty window frames
(#70239)
### What problem does this PR solve?
Issue Number: None
Related PR: #69964
A bounded ROWS frame can become empty after a nonempty frame. The
`quantile_union` aggregate previously reset only its first-input flag,
leaving samples from the prior frame in `QuantileState`. Empty and
NULL-only frames could consequently return stale quantiles.
### What is changed?
- Reset the aggregate's `QuantileState` together with its first-input
flag.
- Add `QuantileState::reset()` to discard samples while retaining
explicit-vector capacity; preserve `clear()`'s capacity-release
behavior.
- Add focused unit and SQL regression coverage for empty, NULL-only, and
repeated frames.
### Release note
Fix `quantile_union` over empty window frames so `quantile_percent`
returns NaN rather than values from an earlier frame.
### Test
- ASAN BE+FE build, including FE Checkstyle: passed.
- Focused ASAN BE unit tests: 7/7 passed.
- Quantile SQL regression suites: 3/3 passed.
- clang-format, check-format, and `git diff --check`: passed.
---
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 | 185 +++++++++++++++++++++
.../test_quantile_union_empty_window.out | 28 ++++
.../test_quantile_union_empty_window.groovy | 91 ++++++++++
6 files changed, 315 insertions(+), 2 deletions(-)
diff --git a/be/src/core/value/quantile_state.cpp
b/be/src/core/value/quantile_state.cpp
index 47df852e2ad..c338b6e16b5 100644
--- a/be/src/core/value/quantile_state.cpp
+++ b/be/src/core/value/quantile_state.cpp
@@ -425,10 +425,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 e85bd4ec40b..192f73537a1 100644
--- a/be/src/exprs/aggregate/aggregate_function_quantile_state.h
+++ b/be/src/exprs/aggregate/aggregate_function_quantile_state.h
@@ -84,7 +84,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 d8497cba9cc..ad41f5821c2 100644
--- a/be/test/exprs/aggregate/agg_percentile_test.cpp
+++ b/be/test/exprs/aggregate/agg_percentile_test.cpp
@@ -17,11 +17,17 @@
#include <gtest/gtest.h>
+#include <cmath>
#include <memory>
#include <vector>
+#include "core/arena.h"
#include "core/column/column_complex.h"
+#include "core/column/column_nullable.h"
+#include "core/column/column_vector.h"
+#include "core/data_type/data_type_nullable.h"
#include "core/data_type/data_type_quantilestate.h"
+#include "core/value/quantile_state.h"
#include "exprs/aggregate/aggregate_function_quantile_state.h"
#include "util/defer_op.h"
#include "util/tdigest.h"
@@ -43,8 +49,187 @@ std::vector<double> expected_quantiles(const
std::vector<double>& values,
return result;
}
+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]
