[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

@@ -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();
 }
 

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;

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; }
 

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(

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

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
+    """
+}