This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new 1705f1ad8c2 branch-4.1: [fix](exec) Retain sliding window rows during
eviction #67274 (#67329)
1705f1ad8c2 is described below
commit 1705f1ad8c23ae615662aca904c40c0f6f9a2e65
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Sun Sep 6 05:57:16 2026 +0800
branch-4.1: [fix](exec) Retain sliding window rows during eviction #67274
(#67329)
Cherry-picked from #67274
Co-authored-by: Mryange <[email protected]>
---
be/src/exec/operator/analytic_sink_operator.cpp | 14 +++++
be/src/exec/operator/analytic_sink_operator.h | 9 +++-
.../exec/operator/analytic_sink_operator_test.cpp | 60 ++++++++++++++++++++--
3 files changed, 76 insertions(+), 7 deletions(-)
diff --git a/be/src/exec/operator/analytic_sink_operator.cpp
b/be/src/exec/operator/analytic_sink_operator.cpp
index 133b4ec9e60..5f9ba79efab 100644
--- a/be/src/exec/operator/analytic_sink_operator.cpp
+++ b/be/src/exec/operator/analytic_sink_operator.cpp
@@ -65,8 +65,10 @@ Status AnalyticSinkLocalState::init(RuntimeState* state,
LocalSinkStateInfo& inf
} else {
if (!p._has_window_start) {
_executor.get_next_impl =
&AnalyticSinkLocalState::_get_next_for_unbounded_rows;
+ _rows_window_type = RowsWindowType::UNBOUNDED_START;
} else {
_executor.get_next_impl =
&AnalyticSinkLocalState::_get_next_for_sliding_rows;
+ _rows_window_type = RowsWindowType::SLIDING;
}
_streaming_mode = true;
_support_incremental_calculate = (p._has_window_start &&
p._has_window_end);
@@ -845,6 +847,18 @@ void AnalyticSinkLocalState::_remove_unused_rows() {
if (idx < 0 || _input_block_first_row_positions[idx] <=
unused_rows_pos) {
return;
}
+ if (_rows_window_type != RowsWindowType::NONE) {
+ // Sliding frames need the outgoing row; unbounded-start frames
need the next unread row.
+ const int64_t earliest_required_row =
+ _rows_window_type == RowsWindowType::SLIDING
+ ? std::max(_partition_by_pose.start,
+ _current_row_position +
_rows_start_offset - 1)
+ : std::max(_partition_by_pose.start,
+ _current_row_position +
_rows_end_offset);
+ if (_have_removed_rows + earliest_required_row < unused_rows_pos) {
+ return;
+ }
+ }
} else {
if (_have_removed_rows + _partition_by_pose.start <= unused_rows_pos) {
return;
diff --git a/be/src/exec/operator/analytic_sink_operator.h
b/be/src/exec/operator/analytic_sink_operator.h
index a7466761e18..af37bae331c 100644
--- a/be/src/exec/operator/analytic_sink_operator.h
+++ b/be/src/exec/operator/analytic_sink_operator.h
@@ -20,6 +20,8 @@
#include <stdint.h>
+#include <algorithm>
+
#include "exec/operator/operator.h"
#include "exec/pipeline/dependency.h"
@@ -32,8 +34,8 @@ struct BoundaryPose {
int64_t end = 0;
bool is_ended = false;
void remove_unused_rows(int64_t cnt) {
- start -= cnt;
- end -= cnt;
+ start = std::max<int64_t>(0, start - cnt);
+ end = std::max<int64_t>(0, end - cnt);
}
};
@@ -73,6 +75,8 @@ public:
Status close(RuntimeState* state, Status exec_status) override;
private:
+ enum class RowsWindowType { NONE, UNBOUNDED_START, SLIDING };
+
friend class AnalyticSinkOperatorX;
Status _execute_impl(RuntimeState* state);
// over(partition by k1 order by k2 range|rows unbounded preceding and
unbounded following)
@@ -145,6 +149,7 @@ private:
std::vector<uint8_t> _use_null_result;
std::vector<uint8_t> _could_use_previous_result;
bool _streaming_mode = false;
+ RowsWindowType _rows_window_type = RowsWindowType::NONE;
bool _support_incremental_calculate = true;
bool _need_more_data = false;
int64_t _current_row_position = 0;
diff --git a/be/test/exec/operator/analytic_sink_operator_test.cpp
b/be/test/exec/operator/analytic_sink_operator_test.cpp
index b5e5787e8a6..ec3934540c6 100644
--- a/be/test/exec/operator/analytic_sink_operator_test.cpp
+++ b/be/test/exec/operator/analytic_sink_operator_test.cpp
@@ -83,6 +83,15 @@ private:
} // namespace
+TEST(BoundaryPoseTest, RemoveUnusedRowsKeepsPhysicalCoordinatesNonNegative) {
+ BoundaryPose pose {.start = 1, .end = 5};
+
+ pose.remove_unused_rows(2);
+
+ EXPECT_EQ(pose.start, 0);
+ EXPECT_EQ(pose.end, 3);
+}
+
struct AnalyticSinkOperatorTest : public ::testing::Test {
void Initialize(int batch_size) {
sink = std::make_unique<AnalyticSinkOperatorX>(&pool);
@@ -441,7 +450,48 @@ TEST_F(AnalyticSinkOperatorTest, AggFunction3) {
std::cout << "######### AggFunction with row_number test end #########" <<
std::endl;
}
-TEST_F(AnalyticSinkOperatorTest, AggFunction4) {
+TEST_F(AnalyticSinkOperatorTest,
UnboundedRowsRetainsNextUnreadRowDuringEviction) {
+ constexpr int batch_size = 2;
+ Initialize(batch_size);
+ create_operator(true, 1, "sum", {std::make_shared<DataTypeInt64>()},
+ std::make_shared<DataTypeInt64>());
+ sink->_agg_expr_ctxs.resize(1);
+ sink->_agg_expr_ctxs[0] =
+ MockSlotRef::create_mock_contexts(0,
std::make_shared<DataTypeInt64>());
+ TAnalyticWindow window;
+ window.type = TAnalyticWindowType::ROWS;
+ TAnalyticWindowBoundary window_end;
+ window_end.type = TAnalyticWindowBoundaryType::PRECEDING;
+ window_end.__set_rows_offset_value(5);
+ window.__set_window_end(window_end);
+ create_window_type(false, true, window);
+ create_local_state();
+
+ for (int row = 1; row <= 8; row += batch_size) {
+ Block block = ColumnHelper::create_block<DataTypeInt64>({row, row +
1});
+ auto status = sink->sink(state.get(), &block, row + batch_size > 8);
+ EXPECT_TRUE(status.ok()) << status.msg();
+ }
+
+ const std::vector<int64_t> expected_sums {0, 0, 0, 0, 0, 1, 3, 6};
+ for (int row = 1; row <= 8; row += batch_size) {
+ Block block = ColumnHelper::create_block<DataTypeInt64>({});
+ bool eos = false;
+ auto status = source->get_block(state.get(), &block, &eos);
+ EXPECT_TRUE(status.ok()) << status.msg();
+ EXPECT_TRUE(ColumnHelper::block_equal(
+ block, ColumnHelper::create_block<DataTypeInt64>(
+ {row, row + 1}, {expected_sums[row - 1],
expected_sums[row]})));
+ }
+
+ Block block = ColumnHelper::create_block<DataTypeInt64>({});
+ bool eos = false;
+ auto status = source->get_block(state.get(), &block, &eos);
+ EXPECT_TRUE(status.ok()) << status.msg();
+ EXPECT_TRUE(eos);
+}
+
+TEST_F(AnalyticSinkOperatorTest,
SlidingRowsSumRetainsOutgoingRowDuringEviction) {
int batch_size = 2;
Initialize(batch_size);
create_operator(true, 1, "sum", {std::make_shared<DataTypeInt64>()},
@@ -453,14 +503,14 @@ TEST_F(AnalyticSinkOperatorTest, AggFunction4) {
temp_window.type = TAnalyticWindowType::ROWS;
TAnalyticWindowBoundary window_start;
window_start.type = TAnalyticWindowBoundaryType::PRECEDING;
- window_start.__set_rows_offset_value(1);
+ window_start.__set_rows_offset_value(4);
temp_window.__set_window_start(window_start);
TAnalyticWindowBoundary window_end;
window_end.type = TAnalyticWindowBoundaryType::CURRENT_ROW;
temp_window.__set_window_end(window_end);
create_window_type(true, true, temp_window);
create_local_state();
- // test with row_number agg function and has window:
_get_next_for_unbounded_rows
+ // The frame is wider than one buffered block, so eviction must retain its
outgoing row.
auto sink_data = [&](int row_count, bool eos) {
std::vector<int64_t> data_vals;
@@ -504,7 +554,7 @@ TEST_F(AnalyticSinkOperatorTest, AggFunction4) {
{
int row_count = 0;
std::vector<int64_t> data_vals {0, 1, 2, 3, 4, 5, 6, 7, 8, 9};
- std::vector<int64_t> expect_vals {0, 1, 3, 5, 7, 9, 11, 13, 15, 17};
//sum
+ std::vector<int64_t> expect_vals {0, 1, 3, 6, 10, 15, 20, 25, 30, 35};
//sum
for (int i = 0; i < 5; i++) {
compare_block_result(row_count, data_vals, expect_vals);
row_count += batch_size;
@@ -516,7 +566,7 @@ TEST_F(AnalyticSinkOperatorTest, AggFunction4) {
EXPECT_EQ(block2.rows(), 0);
EXPECT_TRUE(eos2);
}
- std::cout << "######### AggFunction with row_number test end #########" <<
std::endl;
+ std::cout << "######### sliding rows sum eviction test end #########" <<
std::endl;
}
TEST_F(AnalyticSinkOperatorTest, AggFunction5) {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
