This is an automated email from the ASF dual-hosted git repository.
Mryange 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 92b46f325ee [fix](function) Fix window_funnel missing boundary events
when window overflows at datetime upper bound (#67823)
92b46f325ee is described below
commit 92b46f325eee7339ad659520142c01762302d4cc
Author: Mryange <[email protected]>
AuthorDate: Tue Sep 15 10:32:43 2026 +0800
[fix](function) Fix window_funnel missing boundary events when window
overflows at datetime upper bound (#67823)
For timestamps near the DATETIME upper bound (e.g. `9999-12-31
23:59:58`), adding the sliding window seconds in
`date_add_interval<SECOND>` overflows the date range. Both
`window_funnel` and `window_funnel_v2` ignored the `false` return value,
leaving the window end timestamp at the starting point, so the boundary
event inside the enlarged window was missed: `window=1` returned 2 while
`window=10` returned 1.
### Release note
None
### Check List (For Author)
- Test <!-- At least one of them must be included. -->
- [ ] Regression test
- [ ] Unit Test
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason <!-- Add your reason? -->
- Behavior changed:
- [ ] No.
- [ ] Yes. <!-- Explain the behavior change -->
- Does this need documentation?
- [ ] No.
- [ ] Yes. <!-- Add document PR link here. eg:
https://github.com/apache/doris-website/pull/1214 -->
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label <!-- Add branch pick label that this PR
should merge into -->
---
.../aggregate/aggregate_function_window_funnel.h | 20 +++----
.../aggregate_function_window_funnel_v2.h | 8 +--
be/test/exprs/aggregate/vec_window_funnel_test.cpp | 62 ++++++++++++++++++++++
.../exprs/aggregate/vec_window_funnel_v2_test.cpp | 62 ++++++++++++++++++++++
4 files changed, 134 insertions(+), 18 deletions(-)
diff --git a/be/src/exprs/aggregate/aggregate_function_window_funnel.h
b/be/src/exprs/aggregate/aggregate_function_window_funnel.h
index 2ecacd5b083..7b8dff2ff33 100644
--- a/be/src/exprs/aggregate/aggregate_function_window_funnel.h
+++ b/be/src/exprs/aggregate/aggregate_function_window_funnel.h
@@ -160,8 +160,8 @@ struct WindowFunnelState {
}
}
- bool _within_window(const DateValueType& base_timestamp, const
DateValueType& current_timestamp,
- const DateValueType& end_timestamp) const {
+ bool _within_window(const DateValueType& base_timestamp,
+ const DateValueType& current_timestamp) const {
if constexpr (T == TYPE_TIMESTAMP_NS) {
const auto elapsed_nanos =
static_cast<__int128>(current_timestamp.epoch_nanos()) -
base_timestamp.epoch_nanos();
@@ -169,14 +169,13 @@ struct WindowFunnelState {
static_cast<__int128>(window) *
TimeStampNsValue::NANOS_PER_SECOND;
return elapsed_nanos <= window_nanos;
}
- return current_timestamp <= end_timestamp;
+ return
static_cast<__int128>(current_timestamp.datetime_diff_in_microseconds(
+ base_timestamp)) <= static_cast<__int128>(window) *
1000000;
}
template <WindowFunnelMode WINDOW_FUNNEL_MODE>
int _match_event_list(size_t& start_row, size_t row_count) const {
int matched_count = 0;
- DateValueType end_timestamp;
-
if (window < 0) {
throw Exception(ErrorCode::INVALID_ARGUMENT,
"the sliding time window must be a positive
integer, but got: {}",
@@ -190,12 +189,6 @@ struct WindowFunnelState {
if (match_row < row_count) {
auto prev_timestamp = timestamp_data[match_row];
const auto first_timestamp = prev_timestamp;
- if constexpr (T != TYPE_TIMESTAMP_NS) {
- TimeInterval interval(SECOND, window, false);
- end_timestamp = first_timestamp;
- end_timestamp.template date_add_interval<SECOND>(interval);
- }
-
matched_count++;
column_idx++;
auto last_match_row = match_row;
@@ -205,7 +198,7 @@ struct WindowFunnelState {
if constexpr (WINDOW_FUNNEL_MODE == WindowFunnelMode::FIXED) {
if (event_data[match_row] == 1) {
auto current_timestamp = timestamp_data[match_row];
- if (_within_window(first_timestamp, current_timestamp,
end_timestamp)) {
+ if (_within_window(first_timestamp,
current_timestamp)) {
matched_count++;
continue;
}
@@ -215,8 +208,7 @@ struct WindowFunnelState {
match_row = simd::find_one(event_data.data(), match_row,
row_count);
if (match_row < row_count) {
auto current_timestamp = timestamp_data[match_row];
- bool is_matched =
- _within_window(first_timestamp, current_timestamp,
end_timestamp);
+ bool is_matched = _within_window(first_timestamp,
current_timestamp);
if (is_matched) {
if constexpr (WINDOW_FUNNEL_MODE ==
WindowFunnelMode::INCREASE) {
is_matched = current_timestamp > prev_timestamp;
diff --git a/be/src/exprs/aggregate/aggregate_function_window_funnel_v2.h
b/be/src/exprs/aggregate/aggregate_function_window_funnel_v2.h
index af2fac10e73..b32c7bae3a7 100644
--- a/be/src/exprs/aggregate/aggregate_function_window_funnel_v2.h
+++ b/be/src/exprs/aggregate/aggregate_function_window_funnel_v2.h
@@ -259,10 +259,10 @@ struct WindowFunnelStateV2 {
static_cast<__int128>(window) *
TimeStampNsValue::NANOS_PER_SECOND;
return elapsed_nanos <= window_nanos;
} else {
- DateValueType end_ts = _ts_from_int(base_ts);
- TimeInterval interval(SECOND, window, false);
- end_ts.template date_add_interval<SECOND>(interval);
- return current_ts <= end_ts.to_date_int_val();
+ const auto base = _ts_from_int(base_ts);
+ const auto current = _ts_from_int(current_ts);
+ return
static_cast<__int128>(current.datetime_diff_in_microseconds(base)) <=
+ static_cast<__int128>(window) * 1000000;
}
}
diff --git a/be/test/exprs/aggregate/vec_window_funnel_test.cpp
b/be/test/exprs/aggregate/vec_window_funnel_test.cpp
index ed87928eec2..6edc0db5ce4 100644
--- a/be/test/exprs/aggregate/vec_window_funnel_test.cpp
+++ b/be/test/exprs/aggregate/vec_window_funnel_test.cpp
@@ -145,6 +145,68 @@ TEST_F(VWindowFunnelTest, testEmpty) {
agg_function->destroy(place2);
}
+TEST_F(VWindowFunnelTest, testWindowOverflow) {
+ AggregateFunctionSimpleFactory factory =
AggregateFunctionSimpleFactory::instance();
+ DataTypes data_types = {std::make_shared<DataTypeInt64>(),
std::make_shared<DataTypeString>(),
+ std::make_shared<DataTypeDateTimeV2>(),
+ std::make_shared<DataTypeUInt8>(),
std::make_shared<DataTypeUInt8>()};
+ auto overflow_agg_function = factory.get("window_funnel", data_types,
nullptr, false,
+
BeExecVersionManager::get_newest_version());
+ ASSERT_NE(overflow_agg_function, nullptr);
+
+ auto column_mode = ColumnString::create();
+ column_mode->insert(Field::create_field<TYPE_STRING>("default"));
+ column_mode->insert(Field::create_field<TYPE_STRING>("default"));
+
+ auto column_timestamp = ColumnDateTimeV2::create();
+ for (const auto& second : {58, 59}) {
+ VecDateTimeValue time_value;
+ time_value.unchecked_set_time(9999, 12, 31, 23, 59, second);
+ auto dtv2 = time_value.to_datetime_v2();
+ column_timestamp->insert_data((char*)&dtv2, 0);
+ }
+
+ auto column_window = ColumnInt64::create();
+ column_window->insert(Field::create_field<TYPE_BIGINT>(10));
+ column_window->insert(Field::create_field<TYPE_BIGINT>(10));
+ auto column_event1 = ColumnUInt8::create();
+ column_event1->insert(Field::create_field<TYPE_BOOLEAN>(1));
+ column_event1->insert(Field::create_field<TYPE_BOOLEAN>(0));
+ auto column_event2 = ColumnUInt8::create();
+ column_event2->insert(Field::create_field<TYPE_BOOLEAN>(0));
+ column_event2->insert(Field::create_field<TYPE_BOOLEAN>(1));
+
+ std::unique_ptr<char[]> memory(new
char[overflow_agg_function->size_of_data()]);
+ AggregateDataPtr place = memory.get();
+ overflow_agg_function->create(place);
+ const IColumn* columns[] = {column_window.get(), column_mode.get(),
column_timestamp.get(),
+ column_event1.get(), column_event2.get()};
+ for (int row = 0; row < 2; ++row) {
+ overflow_agg_function->add(place, columns, row, arena);
+ }
+
+ ColumnInt32 result;
+ overflow_agg_function->insert_result_into(place, result);
+ EXPECT_EQ(result.get_element(0), 2);
+ overflow_agg_function->destroy(place);
+}
+
+TEST(VWindowFunnelDateTimeV2Test, PreservesMicrosecondWindowBoundary) {
+ WindowFunnelState<TYPE_DATETIMEV2> state(2);
+ state.window = 1;
+ state.window_funnel_mode = WindowFunnelMode::DEFAULT;
+ DateV2Value<DateTimeV2ValueType> base;
+ DateV2Value<DateTimeV2ValueType> exact;
+ DateV2Value<DateTimeV2ValueType> outside;
+ base.unchecked_set_time(9999, 12, 31, 23, 59, 58, 0);
+ exact.unchecked_set_time(9999, 12, 31, 23, 59, 59, 0);
+ outside.unchecked_set_time(9999, 12, 31, 23, 59, 59, 1);
+ EXPECT_TRUE(state._within_window(base, exact));
+ EXPECT_FALSE(state._within_window(base, outside));
+ state.window = std::numeric_limits<int64_t>::max();
+ EXPECT_TRUE(state._within_window(base, outside));
+}
+
TEST_F(VWindowFunnelTest, testSerialize) {
const int NUM_CONDS = 4;
auto column_mode = ColumnString::create();
diff --git a/be/test/exprs/aggregate/vec_window_funnel_v2_test.cpp
b/be/test/exprs/aggregate/vec_window_funnel_v2_test.cpp
index 9b907d153d1..bf25af9df8d 100644
--- a/be/test/exprs/aggregate/vec_window_funnel_v2_test.cpp
+++ b/be/test/exprs/aggregate/vec_window_funnel_v2_test.cpp
@@ -118,6 +118,68 @@ TEST_F(VWindowFunnelV2Test, testEmpty) {
agg_function->destroy(place2);
}
+TEST_F(VWindowFunnelV2Test, testWindowOverflow) {
+ AggregateFunctionSimpleFactory factory =
AggregateFunctionSimpleFactory::instance();
+ DataTypes data_types = {std::make_shared<DataTypeInt64>(),
std::make_shared<DataTypeString>(),
+ std::make_shared<DataTypeDateTimeV2>(),
+ std::make_shared<DataTypeUInt8>(),
std::make_shared<DataTypeUInt8>()};
+ auto overflow_agg_function = factory.get("window_funnel_v2", data_types,
nullptr, false,
+
BeExecVersionManager::get_newest_version());
+ ASSERT_NE(overflow_agg_function, nullptr);
+
+ auto column_mode = ColumnString::create();
+ column_mode->insert(Field::create_field<TYPE_STRING>("default"));
+ column_mode->insert(Field::create_field<TYPE_STRING>("default"));
+
+ auto column_timestamp = ColumnDateTimeV2::create();
+ for (const auto& second : {58, 59}) {
+ VecDateTimeValue time_value;
+ time_value.unchecked_set_time(9999, 12, 31, 23, 59, second);
+ auto dtv2 = time_value.to_datetime_v2();
+ column_timestamp->insert_data((char*)&dtv2, 0);
+ }
+
+ auto column_window = ColumnInt64::create();
+ column_window->insert(Field::create_field<TYPE_BIGINT>(10));
+ column_window->insert(Field::create_field<TYPE_BIGINT>(10));
+ auto column_event1 = ColumnUInt8::create();
+ column_event1->insert(Field::create_field<TYPE_BOOLEAN>(1));
+ column_event1->insert(Field::create_field<TYPE_BOOLEAN>(0));
+ auto column_event2 = ColumnUInt8::create();
+ column_event2->insert(Field::create_field<TYPE_BOOLEAN>(0));
+ column_event2->insert(Field::create_field<TYPE_BOOLEAN>(1));
+
+ std::unique_ptr<char[]> memory(new
char[overflow_agg_function->size_of_data()]);
+ AggregateDataPtr place = memory.get();
+ overflow_agg_function->create(place);
+ const IColumn* columns[] = {column_window.get(), column_mode.get(),
column_timestamp.get(),
+ column_event1.get(), column_event2.get()};
+ for (int row = 0; row < 2; ++row) {
+ overflow_agg_function->add(place, columns, row, arena);
+ }
+
+ ColumnInt32 result;
+ overflow_agg_function->insert_result_into(place, result);
+ EXPECT_EQ(result.get_element(0), 2);
+ overflow_agg_function->destroy(place);
+}
+
+TEST(VWindowFunnelV2DateTimeV2Test, PreservesMicrosecondWindowBoundary) {
+ WindowFunnelStateV2<TYPE_DATETIMEV2> state(2);
+ state.window = 1;
+ state.window_funnel_mode = WindowFunnelMode::DEFAULT;
+ DateV2Value<DateTimeV2ValueType> base;
+ DateV2Value<DateTimeV2ValueType> exact;
+ DateV2Value<DateTimeV2ValueType> outside;
+ base.unchecked_set_time(9999, 12, 31, 23, 59, 58, 0);
+ exact.unchecked_set_time(9999, 12, 31, 23, 59, 59, 0);
+ outside.unchecked_set_time(9999, 12, 31, 23, 59, 59, 1);
+ EXPECT_TRUE(state._within_window(base.to_date_int_val(),
exact.to_date_int_val()));
+ EXPECT_FALSE(state._within_window(base.to_date_int_val(),
outside.to_date_int_val()));
+ state.window = std::numeric_limits<int64_t>::max();
+ EXPECT_TRUE(state._within_window(base.to_date_int_val(),
outside.to_date_int_val()));
+}
+
TEST_F(VWindowFunnelV2Test, testSerialize) {
const int NUM_CONDS = 4;
auto column_mode = ColumnString::create();
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
