This is an automated email from the ASF dual-hosted git repository.
panxiaolei 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 a0b9e3784f9 [Bug](function) fix wrong result of window funnel v2 +
deduplication/fixed mode (#62043)
a0b9e3784f9 is described below
commit a0b9e3784f982deeaf43fd9da03bdbb598da3e57
Author: Pxl <[email protected]>
AuthorDate: Tue Apr 7 15:39:14 2026 +0800
[Bug](function) fix wrong result of window funnel v2 + deduplication/fixed
mode (#62043)
FIXED mode
修复前的问题：旧代码从前往后只扫描一遍，只维护一个链。遇到新的条件1就重启一个新链，丢弃旧链。这样可能错过从后面某个条件1开始能匹配更长链的情况。
FIXED mode 修复后的逻辑：
1. 找到所有满足条件1的行作为候选起点
2. 对每个起点，逐行往后检查：
- 下一行必须匹配当前期望的下一个条件（条件链严格递增）
- 如果下一行匹配了某个条件但不是期望的那个 → 链断裂，放弃这个起点
- 不匹配任何条件的行会被跳过（这是
4.1 的新行为，V1 会在此断裂）
3. 取所有起点中最长的链作为结果
DEDUP mode 修复前的问题：同样是单次扫描，一个链。无法正确处理"后面的起点能匹配更长链"的情况。
DEDUP mode 修复后的逻辑：
1. 找到所有满足条件1的行作为候选起点
2. 对每个起点，按顺序搜索下一个期望的条件：
- 找到匹配条件 K 的行后，检查中间经过的行是否有任何已匹配条件的重复出现
- 有重复 → 链断裂（这就是 deduplication 的含义：同一条件出现两次就打断链）
- 无重复 → 接受匹配，继续搜索下一个条件
3. 取所有起点中最长的链作为结果
两种模式的共同改变：旧代码只尝试一个起点（扫描一遍），新代码尝试每个可能的起点（多次扫描），取最优结果。这和 V1 的做法一致。
This pull request refactors and improves the implementation of the
window funnel aggregate function in `window_funnel_v2`, focusing on the
DEDUPLICATION and FIXED modes. The changes enhance correctness by more
closely matching the intended semantics, improve efficiency with
multi-pass algorithms, and add comprehensive test coverage for edge
cases and complex scenarios.
### Algorithmic and Semantic Improvements
* Rewrote the DEDUPLICATION mode (`_get_deduplication`) to use a
multi-pass, gap-based duplicate detection algorithm, accurately matching
V1 semantics by checking for duplicate events in the "gap" between
matched levels and pruning impossible chains early.
* Refactored the FIXED mode (`_get_fixed`) to a multi-pass, row-based
approach that aligns with documented V2 semantics, including the correct
handling of irrelevant events and level jumps, and added pruning for
efficiency.
### Code Simplification and Maintenance
* Removed legacy and now-unnecessary helper functions (`_promote_level`,
`_eliminate_chain`, and related timestamp tracking), streamlining the
codebase and reducing complexity.
[[1]](diffhunk://#diff-1a1c09dde1a5d97a9723ffebb33ddb27344131ac2031c986ce6866bf248c5971L396-R599)
[[2]](diffhunk://#diff-1a1c09dde1a5d97a9723ffebb33ddb27344131ac2031c986ce6866bf248c5971L532-L551)
* Added new helper methods (`_next_row_start`, `_row_start`,
`_row_contains_level`) to clarify row and event handling, making the
logic more readable and maintainable.
### Test Coverage
* Added a comprehensive set of regression tests for DEDUPLICATION and
FIXED modes, covering multi-event rows, duplicate detection, level
jumps, window boundaries, irrelevant event skipping, and various edge
cases, ensuring correctness and robustness of the new algorithms.
---
.../aggregate_function_window_funnel_v2.h | 266 +++++---
.../exprs/aggregate/vec_window_funnel_v2_test.cpp | 743 +++++++++++++++++++++
.../data/query_p0/aggregate/window_funnel_v2.out | 56 ++
.../query_p0/aggregate/window_funnel_v2.groovy | 412 ++++++++++++
4 files changed, 1377 insertions(+), 100 deletions(-)
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 8038c0bef47..421fe701dae 100644
--- a/be/src/exprs/aggregate/aggregate_function_window_funnel_v2.h
+++ b/be/src/exprs/aggregate/aggregate_function_window_funnel_v2.h
@@ -393,9 +393,7 @@ private:
if (matched) {
events_timestamp[event_idx] =
{events_timestamp[event_idx - 1].first_ts,
evt.timestamp, i};
- if (event_idx > curr_level) {
- curr_level = event_idx;
- }
+ curr_level = std::max(event_idx, curr_level);
if (event_idx + 1 == event_count) {
return event_count;
}
@@ -403,113 +401,201 @@ private:
}
}
- if (curr_level + 1 > max_level) {
- max_level = curr_level + 1;
- }
+ max_level = std::max(curr_level + 1, max_level);
}
return max_level;
}
- /// DEDUPLICATION mode: if a previously matched event level appears again,
- /// the current chain is terminated and max_level is updated.
- /// This preserves V1 semantics where duplicate events break the chain.
+ /// DEDUPLICATION mode (multi-pass, matching V1 semantics):
+ ///
+ /// V1 tries each E0-matching row as a chain starting point and searches
for
+ /// conditions one at a time. Before accepting a match for condition K, V1
checks
+ /// the "gap" (rows between the previous match and the current match) for
any
+ /// re-occurrence of already-matched conditions 0..K-1. If found, the
chain breaks.
+ ///
+ /// A single-pass approach cannot correctly implement this because V2's
greedy
+ /// multi-condition-per-row advancement creates premature higher-level
matches
+ /// that then conflict with later events. The multi-pass approach iterates
over
+ /// each event-0 occurrence, building chains one level at a time with
gap-based
+ /// duplicate detection.
int _get_deduplication() const {
- std::vector<TimestampPair> events_timestamp(event_count);
- int max_level = -1;
- int curr_level = -1;
+ int max_level = 0;
+ const size_t list_size = events_list.size();
- for (size_t i = 0; i < events_list.size(); ++i) {
- const auto& evt = events_list[i];
- int event_idx = get_event_idx(evt.event_idx) - 1;
+ for (size_t start = 0; start < list_size; ++start) {
+ // Remaining-events pruning: if remaining events can't beat
max_level, stop.
+ if (static_cast<int>(list_size - start) <= max_level) {
+ break;
+ }
- if (event_idx == 0) {
- // Duplicate of event 0: terminate current chain first
- if (events_timestamp[0].has_value()) {
- if (curr_level > max_level) {
- max_level = curr_level;
+ int start_event = get_event_idx(events_list[start].event_idx) - 1;
+ if (start_event != 0) {
+ continue;
+ }
+
+ // Build a chain from this event-0
+ UInt64 first_ts = events_list[start].timestamp;
+ int curr_level = 0;
+ size_t last_advance_idx = start;
+
+ for (size_t i = start + 1; i < list_size; ++i) {
+ int event_idx = get_event_idx(events_list[i].event_idx) - 1;
+
+ // Only search for the next expected level (matches V1's
sequential scan)
+ if (event_idx != curr_level + 1) {
+ continue;
+ }
+
+ // Must be from a different row than the last chain event
+ if (_is_same_row(last_advance_idx, i)) {
+ continue;
+ }
+
+ // Window check
+ if (!_within_window(first_ts, events_list[i].timestamp)) {
+ break;
+ }
+
+ // Gap-based dedup check (V1 semantics):
+ // Check if any event at a level below the target
(0..event_idx-1)
+ // fired in the gap between the last chain event's row and the
+ // current event's row.
+ size_t gap_start = _next_row_start(last_advance_idx + 1);
+ size_t gap_end = _row_start(i);
+
+ bool dup = false;
+ if (gap_start < gap_end) {
+ for (size_t j = gap_start; j < gap_end; ++j) {
+ int j_level = get_event_idx(events_list[j].event_idx)
- 1;
+ if (j_level < event_idx) {
+ dup = true;
+ break;
+ }
}
- _eliminate_chain(curr_level, events_timestamp);
}
- // Start a new chain
- events_timestamp[0] = {evt.timestamp, evt.timestamp, i};
- curr_level = 0;
- } else if (events_timestamp[event_idx].has_value()) {
- // Duplicate event detected: this level was already matched
- if (curr_level > max_level) {
- max_level = curr_level;
+
+ if (dup) {
+ break;
}
- // Eliminate current chain
- _eliminate_chain(curr_level, events_timestamp);
- } else if (events_timestamp[event_idx - 1].has_value() &&
- !_is_same_row(events_timestamp[event_idx -
1].last_list_idx, i)) {
- // Must be from a DIFFERENT row than the predecessor level
- if (_promote_level(events_timestamp, evt.timestamp, i,
event_idx, curr_level,
- false)) {
+
+ // Accept this match and advance the chain
+ curr_level = event_idx;
+ last_advance_idx = i;
+
+ if (curr_level + 1 == event_count) {
return event_count;
}
}
- }
- if (curr_level > max_level) {
- return curr_level + 1;
+ max_level = std::max(curr_level + 1, max_level);
}
- return max_level + 1;
+ return max_level;
}
- /// FIXED mode (StarRocks-style semantics): if a matched event appears
whose
- /// predecessor level has NOT been matched, the chain is broken (event
level jumped).
- /// Note: V2 semantics differ from V1. V1 checks physical row adjacency;
- /// V2 checks event level continuity (unmatched rows don't break the
chain).
+ /// FIXED mode (multi-pass, row-based).
+ ///
+ /// Row-by-row semantics similar to V1: after matching c_K at a row, the
NEXT
+ /// event-producing row must match c_{K+1} or the chain breaks.
+ ///
+ /// Difference from V1: rows matching no condition ("truly irrelevant") are
+ /// absent from events_list and implicitly skipped. V1 treats them as
+ /// non-matching and breaks the chain. This is the documented 4.1 behavior
+ /// change: irrelevant events no longer break the chain.
+ ///
+ /// Like V1, tries every c1 event as a potential chain starting point and
+ /// returns the maximum level reached across all attempts.
+ ///
+ /// Within each row, events are stored in descending condition-index order:
+ /// [c_max(cont=0), ..., c_min(cont=N)]. The first (non-continuation) event
+ /// carries the row's highest matched condition index. c1 (index 1 after
add,
+ /// 0 after -1) is always the last event of its row.
int _get_fixed() const {
- std::vector<TimestampPair> events_timestamp(event_count);
int max_level = -1;
- int curr_level = -1;
- bool first_event = false;
- for (size_t i = 0; i < events_list.size(); ++i) {
- const auto& evt = events_list[i];
- int event_idx = get_event_idx(evt.event_idx) - 1;
+ for (size_t start_i = 0; start_i < events_list.size(); ++start_i) {
+ // Only start from c1 events (condition 0 in 0-based)
+ if (get_event_idx(events_list[start_i].event_idx) != 1) continue;
- if (event_idx == 0) {
- // Save current chain before starting a new one
- if (events_timestamp[0].has_value()) {
- if (curr_level > max_level) {
- max_level = curr_level;
- }
- _eliminate_chain(curr_level, events_timestamp);
- }
- events_timestamp[0] = {evt.timestamp, evt.timestamp, i};
- curr_level = 0;
- first_event = true;
- } else if (first_event && !events_timestamp[event_idx -
1].has_value()) {
- // Event level jumped: predecessor was not matched
- if (curr_level >= 0) {
- if (curr_level > max_level) {
- max_level = curr_level;
- }
- _eliminate_chain(curr_level, events_timestamp);
+ UInt64 chain_start_ts = events_list[start_i].timestamp;
+ int curr_level = 0;
+
+ // c1 is always the last event of its row, so start_i+1 is
+ // the first event of the next row (or end of list).
+ size_t i = start_i + 1;
+
+ while (i < events_list.size() && curr_level + 1 < event_count) {
+ // Skip continuation events to reach the first event of the
next row.
+ i = _next_row_start(i);
+ if (i >= events_list.size()) break;
+
+ // Window check against the row's timestamp.
+ if (!_within_window(chain_start_ts, events_list[i].timestamp))
break;
+
+ int expected = curr_level + 1;
+ int highest = get_event_idx(events_list[i].event_idx) - 1;
+
+ if (highest < expected) {
+ // Row's highest matched condition is below expected →
break.
+ break;
}
- } else if (events_timestamp[event_idx - 1].has_value() &&
- !_is_same_row(events_timestamp[event_idx -
1].last_list_idx, i)) {
- // Must be from a DIFFERENT row than the predecessor level
- if (_promote_level(events_timestamp, evt.timestamp, i,
event_idx, curr_level,
- false)) {
- return event_count;
+
+ if (!_row_contains_level(i, expected)) {
+ // Row matches higher conditions but not the expected one
+ // (level jump) → break.
+ break;
}
+
+ curr_level = expected;
+
+ // Advance past this row to the start of the next one.
+ i = _next_row_start(i + 1);
+ }
+
+ if (curr_level > max_level) {
+ max_level = curr_level;
+ if (max_level + 1 == event_count) return event_count;
}
- }
- if (curr_level > max_level) {
- return curr_level + 1;
+ // Prune: remaining events cannot beat current best.
+ if (static_cast<int>(events_list.size() - start_i) <= max_level)
break;
}
+
return max_level + 1;
}
- /// Clear the current event chain back to the beginning.
- static void _eliminate_chain(int& curr_level, std::vector<TimestampPair>&
events_timestamp) {
- for (; curr_level >= 0; --curr_level) {
- events_timestamp[curr_level].reset();
+ /// Advance past continuation events starting at `i`, returning the index
of the
+ /// next non-continuation event (i.e., the first event of the next row), or
+ /// events_list.size() if no more rows remain.
+ size_t _next_row_start(size_t i) const {
+ while (i < events_list.size() &&
is_continuation(events_list[i].event_idx)) {
+ ++i;
+ }
+ return i;
+ }
+
+ /// Walk backward from `i` to find the first event of the row containing
`i`.
+ /// The first event of a row has no continuation flag set.
+ size_t _row_start(size_t i) const {
+ while (i > 0 && is_continuation(events_list[i].event_idx)) {
+ --i;
}
+ return i;
+ }
+
+ /// Check if the row whose first event is at `row_first` contains an event
+ /// matching the given 0-based `target_level`. The first event of a row
has the
+ /// highest matched condition index; subsequent continuation events have
lower indices.
+ bool _row_contains_level(size_t row_first, int target_level) const {
+ if (get_event_idx(events_list[row_first].event_idx) - 1 ==
target_level) {
+ return true;
+ }
+ for (size_t j = row_first + 1;
+ j < events_list.size() &&
is_continuation(events_list[j].event_idx); ++j) {
+ if (get_event_idx(events_list[j].event_idx) - 1 == target_level) {
+ return true;
+ }
+ }
+ return false;
}
/// Check if the event at `current_idx` in events_list is from the same
original row
@@ -529,26 +615,6 @@ private:
}
return true;
}
-
- /// Try to promote the chain to the next level.
- /// Returns true if we've matched all events (early termination).
- bool _promote_level(std::vector<TimestampPair>& events_timestamp, UInt64
ts, size_t list_idx,
- int event_idx, int& curr_level, bool increase_mode)
const {
- bool matched = _within_window(events_timestamp[event_idx -
1].first_ts, ts);
- if (increase_mode) {
- matched = matched && events_timestamp[event_idx - 1].last_ts < ts;
- }
- if (matched) {
- events_timestamp[event_idx] = {events_timestamp[event_idx -
1].first_ts, ts, list_idx};
- if (event_idx > curr_level) {
- curr_level = event_idx;
- }
- if (event_idx + 1 == event_count) {
- return true;
- }
- }
- return false;
- }
};
class AggregateFunctionWindowFunnelV2
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 6e82d430513..cbf6d0f9bf8 100644
--- a/be/test/exprs/aggregate/vec_window_funnel_v2_test.cpp
+++ b/be/test/exprs/aggregate/vec_window_funnel_v2_test.cpp
@@ -1164,4 +1164,747 @@ TEST_F(VWindowFunnelV2Test,
testIncreaseModeOldChainBetter) {
}
}
+// Test DEDUPLICATION mode: same-row multi-event matching should NOT break the
chain.
+// This is the exact scenario from the bug report where V1 returns 3 but V2
returned 2.
+//
+// 3 conditions, window=1 day (86400s), DEDUPLICATION mode
+// Row 0 (t=10:00): event1=T, event2=T, event3=F → matches E0+E1 on same row
+// Row 1 (t=11:00): event1=T, event2=T, event3=F → matches E0+E1 on same row
+// Row 2 (t=12:00): event1=T, event2=F, event3=T → matches E0+E2 on same row
+//
+// V2 stores events in reverse order per row: [E1, E0] for each multi-match
row.
+// After sort, events_list for row0: (t=10:00, E1-cont), (t=10:00, E0)
+// Bug: when processing E0 from row0 after E1 was already used to advance the
chain,
+// old V2 would break the chain. With the fix, E0 is recognized as same-row
and skipped.
+//
+// Expected chain: E0@row0 → E1@row1 → E2@row2 = 3
+TEST_F(VWindowFunnelV2Test, testDeduplicationSameRowMultiEvent) {
+ AggregateFunctionSimpleFactory factory =
AggregateFunctionSimpleFactory::instance();
+ DataTypes data_types_3 = {
+ std::make_shared<DataTypeInt64>(),
std::make_shared<DataTypeString>(),
+ std::make_shared<DataTypeDateTimeV2>(),
std::make_shared<DataTypeUInt8>(),
+ std::make_shared<DataTypeUInt8>(),
std::make_shared<DataTypeUInt8>()};
+ auto agg_func_3 = factory.get("window_funnel_v2", data_types_3, nullptr,
false,
+ BeExecVersionManager::get_newest_version());
+ ASSERT_NE(agg_func_3, nullptr);
+
+ const int NUM_ROWS = 3;
+ auto column_mode = ColumnString::create();
+ for (int i = 0; i < NUM_ROWS; i++) {
+ column_mode->insert(Field::create_field<TYPE_STRING>("deduplication"));
+ }
+ auto column_timestamp = ColumnDateTimeV2::create();
+ VecDateTimeValue tv0, tv1, tv2;
+ tv0.unchecked_set_time(2022, 3, 12, 10, 0, 0);
+ tv1.unchecked_set_time(2022, 3, 12, 11, 0, 0);
+ tv2.unchecked_set_time(2022, 3, 12, 12, 0, 0);
+ auto dtv2_0 = tv0.to_datetime_v2();
+ auto dtv2_1 = tv1.to_datetime_v2();
+ auto dtv2_2 = tv2.to_datetime_v2();
+ column_timestamp->insert_data((char*)&dtv2_0, 0);
+ column_timestamp->insert_data((char*)&dtv2_1, 0);
+ column_timestamp->insert_data((char*)&dtv2_2, 0);
+
+ // Row 0: event1=T, event2=T, event3=F (matches E0 and E1)
+ // Row 1: event1=T, event2=T, event3=F (matches E0 and E1)
+ // Row 2: event1=T, event2=F, event3=T (matches E0 and E2)
+ auto column_event1 = ColumnUInt8::create();
+ column_event1->insert(Field::create_field<TYPE_BOOLEAN>(1));
+ column_event1->insert(Field::create_field<TYPE_BOOLEAN>(1));
+ column_event1->insert(Field::create_field<TYPE_BOOLEAN>(1));
+
+ auto column_event2 = ColumnUInt8::create();
+ column_event2->insert(Field::create_field<TYPE_BOOLEAN>(1));
+ column_event2->insert(Field::create_field<TYPE_BOOLEAN>(1));
+ column_event2->insert(Field::create_field<TYPE_BOOLEAN>(0));
+
+ auto column_event3 = ColumnUInt8::create();
+ column_event3->insert(Field::create_field<TYPE_BOOLEAN>(0));
+ column_event3->insert(Field::create_field<TYPE_BOOLEAN>(0));
+ column_event3->insert(Field::create_field<TYPE_BOOLEAN>(1));
+
+ auto column_window = ColumnInt64::create();
+ for (int i = 0; i < NUM_ROWS; i++) {
+ column_window->insert(Field::create_field<TYPE_BIGINT>(86400));
+ }
+
+ std::unique_ptr<char[]> memory(new char[agg_func_3->size_of_data()]);
+ AggregateDataPtr place = memory.get();
+ agg_func_3->create(place);
+ const IColumn* column[6] = {column_window.get(), column_mode.get(),
column_timestamp.get(),
+ column_event1.get(), column_event2.get(),
column_event3.get()};
+ for (int i = 0; i < NUM_ROWS; i++) {
+ agg_func_3->add(place, column, i, arena);
+ }
+
+ ColumnInt32 column_result;
+ agg_func_3->insert_result_into(place, column_result);
+ // Chain: E0@row0 -> E1@row1 -> E2@row2 = 3
+ // Before fix: V2 returned 2 because same-row E0 broke the chain after E1
was used.
+ EXPECT_EQ(column_result.get_data()[0], 3);
+ agg_func_3->destroy(place);
+}
+
+// Test DEDUPLICATION mode: a true duplicate on a DIFFERENT row should still
break the chain.
+// This ensures the fix doesn't over-suppress chain breaks.
+//
+// 3 conditions, window=86400s, DEDUPLICATION mode
+// Row 0 (t=10:00): event1=T only
+// Row 1 (t=11:00): event2=T only
+// Row 2 (t=12:00): event1=T only ← true duplicate E0 from different row →
breaks chain
+// Row 3 (t=13:00): event3=T only
+//
+// Expected: 2 (chain E0@row0 → E1@row1, then E0@row2 breaks it; new chain
E0@row2 alone = 1)
+TEST_F(VWindowFunnelV2Test, testDeduplicationTrueDuplicateStillBreaks) {
+ AggregateFunctionSimpleFactory factory =
AggregateFunctionSimpleFactory::instance();
+ DataTypes data_types_3 = {
+ std::make_shared<DataTypeInt64>(),
std::make_shared<DataTypeString>(),
+ std::make_shared<DataTypeDateTimeV2>(),
std::make_shared<DataTypeUInt8>(),
+ std::make_shared<DataTypeUInt8>(),
std::make_shared<DataTypeUInt8>()};
+ auto agg_func_3 = factory.get("window_funnel_v2", data_types_3, nullptr,
false,
+ BeExecVersionManager::get_newest_version());
+ ASSERT_NE(agg_func_3, nullptr);
+
+ const int NUM_ROWS = 4;
+ auto column_mode = ColumnString::create();
+ for (int i = 0; i < NUM_ROWS; i++) {
+ column_mode->insert(Field::create_field<TYPE_STRING>("deduplication"));
+ }
+ auto column_timestamp = ColumnDateTimeV2::create();
+ VecDateTimeValue tv0, tv1, tv2, tv3;
+ tv0.unchecked_set_time(2022, 3, 12, 10, 0, 0);
+ tv1.unchecked_set_time(2022, 3, 12, 11, 0, 0);
+ tv2.unchecked_set_time(2022, 3, 12, 12, 0, 0);
+ tv3.unchecked_set_time(2022, 3, 12, 13, 0, 0);
+ auto dtv2_0 = tv0.to_datetime_v2();
+ auto dtv2_1 = tv1.to_datetime_v2();
+ auto dtv2_2 = tv2.to_datetime_v2();
+ auto dtv2_3 = tv3.to_datetime_v2();
+ column_timestamp->insert_data((char*)&dtv2_0, 0);
+ column_timestamp->insert_data((char*)&dtv2_1, 0);
+ column_timestamp->insert_data((char*)&dtv2_2, 0);
+ column_timestamp->insert_data((char*)&dtv2_3, 0);
+
+ // Row 0: event1=T only
+ // Row 1: event2=T only
+ // Row 2: event1=T only (true duplicate from different row)
+ // Row 3: event3=T only
+ auto column_event1 = ColumnUInt8::create();
+ column_event1->insert(Field::create_field<TYPE_BOOLEAN>(1));
+ column_event1->insert(Field::create_field<TYPE_BOOLEAN>(0));
+ 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));
+ column_event2->insert(Field::create_field<TYPE_BOOLEAN>(0));
+ column_event2->insert(Field::create_field<TYPE_BOOLEAN>(0));
+
+ auto column_event3 = ColumnUInt8::create();
+ column_event3->insert(Field::create_field<TYPE_BOOLEAN>(0));
+ column_event3->insert(Field::create_field<TYPE_BOOLEAN>(0));
+ column_event3->insert(Field::create_field<TYPE_BOOLEAN>(0));
+ column_event3->insert(Field::create_field<TYPE_BOOLEAN>(1));
+
+ auto column_window = ColumnInt64::create();
+ for (int i = 0; i < NUM_ROWS; i++) {
+ column_window->insert(Field::create_field<TYPE_BIGINT>(86400));
+ }
+
+ std::unique_ptr<char[]> memory(new char[agg_func_3->size_of_data()]);
+ AggregateDataPtr place = memory.get();
+ agg_func_3->create(place);
+ const IColumn* column[6] = {column_window.get(), column_mode.get(),
column_timestamp.get(),
+ column_event1.get(), column_event2.get(),
column_event3.get()};
+ for (int i = 0; i < NUM_ROWS; i++) {
+ agg_func_3->add(place, column, i, arena);
+ }
+
+ ColumnInt32 column_result;
+ agg_func_3->insert_result_into(place, column_result);
+ // Chain: E0@row0 -> E1@row1 (level=2), then E0@row2 is a TRUE duplicate
from a
+ // different row → breaks chain. New chain: E0@row2, no E1 → level=1.
+ // max(2, 1) = 2
+ EXPECT_EQ(column_result.get_data()[0], 2);
+ agg_func_3->destroy(place);
+}
+
+// Dedup mode: same-row E0 skip when chain is advanced by same-row events.
+// When a row matches all 3 conditions and its E2 advances the chain, E0 from
+// the same row should be skipped (not restart the chain) because the row
already
+// contributed to chain advancement.
+TEST_F(VWindowFunnelV2Test, testDeduplicationE0PendingRestart) {
+ AggregateFunctionSimpleFactory factory =
AggregateFunctionSimpleFactory::instance();
+ DataTypes data_types_3 = {
+ std::make_shared<DataTypeInt64>(),
std::make_shared<DataTypeString>(),
+ std::make_shared<DataTypeDateTimeV2>(),
std::make_shared<DataTypeUInt8>(),
+ std::make_shared<DataTypeUInt8>(),
std::make_shared<DataTypeUInt8>()};
+ auto agg_func_3 = factory.get("window_funnel_v2", data_types_3, nullptr,
false,
+ BeExecVersionManager::get_newest_version());
+ ASSERT_NE(agg_func_3, nullptr);
+
+ // Scenario: 3 conditions, c1/c2/c3.
+ // Row 0 (t=10:00): c1=T only → E0@R0, chain starts level=0
+ // Row 1 (t=11:00): c2=T only → E1@R1, chain advances level=1
+ // Row 2 (t=12:00): c1=T, c2=T, c3=T → E2(cont=0), E1(cont=1),
E0(cont=1)
+ // E2: predecessor E1 matched, not same row → promote level=2
+ // E1(cont=1): duplicate, same row as E2@R2 → skip
+ // E0(cont=1): same row as chain (E2@R2 at level 2) → skip
+ // Row 3 (t=13:00): c3=T only → E2: duplicate level=2,
different row → terminates chain
+ // Result: max(2+1, 0+1) = 3
+ const int NUM_ROWS = 4;
+ auto column_mode = ColumnString::create();
+ for (int i = 0; i < NUM_ROWS; i++) {
+ column_mode->insert(Field::create_field<TYPE_STRING>("deduplication"));
+ }
+ auto column_timestamp = ColumnDateTimeV2::create();
+ VecDateTimeValue tv0, tv1, tv2, tv3;
+ tv0.unchecked_set_time(2022, 3, 12, 10, 0, 0);
+ tv1.unchecked_set_time(2022, 3, 12, 11, 0, 0);
+ tv2.unchecked_set_time(2022, 3, 12, 12, 0, 0);
+ tv3.unchecked_set_time(2022, 3, 12, 13, 0, 0);
+ auto dtv2_0 = tv0.to_datetime_v2();
+ auto dtv2_1 = tv1.to_datetime_v2();
+ auto dtv2_2 = tv2.to_datetime_v2();
+ auto dtv2_3 = tv3.to_datetime_v2();
+ column_timestamp->insert_data((char*)&dtv2_0, 0);
+ column_timestamp->insert_data((char*)&dtv2_1, 0);
+ column_timestamp->insert_data((char*)&dtv2_2, 0);
+ column_timestamp->insert_data((char*)&dtv2_3, 0);
+
+ // Row 0: c1=T, c2=F, c3=F
+ // Row 1: c1=F, c2=T, c3=F
+ // Row 2: c1=T, c2=T, c3=T
+ // Row 3: c1=F, c2=F, c3=T
+ auto column_event1 = ColumnUInt8::create();
+ column_event1->insert(Field::create_field<TYPE_BOOLEAN>(1));
+ column_event1->insert(Field::create_field<TYPE_BOOLEAN>(0));
+ 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));
+ column_event2->insert(Field::create_field<TYPE_BOOLEAN>(1));
+ column_event2->insert(Field::create_field<TYPE_BOOLEAN>(0));
+
+ auto column_event3 = ColumnUInt8::create();
+ column_event3->insert(Field::create_field<TYPE_BOOLEAN>(0));
+ column_event3->insert(Field::create_field<TYPE_BOOLEAN>(0));
+ column_event3->insert(Field::create_field<TYPE_BOOLEAN>(1));
+ column_event3->insert(Field::create_field<TYPE_BOOLEAN>(1));
+
+ auto column_window = ColumnInt64::create();
+ for (int i = 0; i < NUM_ROWS; i++) {
+ column_window->insert(Field::create_field<TYPE_BIGINT>(86400));
+ }
+
+ std::unique_ptr<char[]> memory(new char[agg_func_3->size_of_data()]);
+ AggregateDataPtr place = memory.get();
+ agg_func_3->create(place);
+ const IColumn* column[6] = {column_window.get(), column_mode.get(),
column_timestamp.get(),
+ column_event1.get(), column_event2.get(),
column_event3.get()};
+ for (int i = 0; i < NUM_ROWS; i++) {
+ agg_func_3->add(place, column, i, arena);
+ }
+
+ ColumnInt32 column_result;
+ agg_func_3->insert_result_into(place, column_result);
+ // Chain reaches level 2 (3 steps matched: E0→E1→E2) before E2@R3
duplicate terminates it.
+ EXPECT_EQ(column_result.get_data()[0], 3);
+ agg_func_3->destroy(place);
+}
+
+// Dedup mode: E0 restart after chain terminated by same-row duplicate.
+// When a row matches c1+c2, and c2 is a duplicate that terminates the chain,
+// the E0 from the same row should start a new chain because _eliminate_chain()
+// clears events_timestamp, so _is_same_row_as_chain won't trigger.
+TEST_F(VWindowFunnelV2Test, testDeduplicationE0PendingRestartApplied) {
+ AggregateFunctionSimpleFactory factory =
AggregateFunctionSimpleFactory::instance();
+ DataTypes data_types_3 = {
+ std::make_shared<DataTypeInt64>(),
std::make_shared<DataTypeString>(),
+ std::make_shared<DataTypeDateTimeV2>(),
std::make_shared<DataTypeUInt8>(),
+ std::make_shared<DataTypeUInt8>(),
std::make_shared<DataTypeUInt8>()};
+ auto agg_func_3 = factory.get("window_funnel_v2", data_types_3, nullptr,
false,
+ BeExecVersionManager::get_newest_version());
+ ASSERT_NE(agg_func_3, nullptr);
+
+ // Scenario: Row matches c1+c2 where c2 is a duplicate → chain terminated
by duplicate.
+ // The pending E0 from same row should restart.
+ // Row 0 (t=10:00): c1=T only → chain level=0
+ // Row 1 (t=11:00): c2=T only → chain level=1
+ // Row 2 (t=12:00): c1=T, c2=T → E1(cont=0), E0(cont=1)
+ // E1(cont=0): duplicate! R2 ≠ R0, R2 ≠ R1 → NOT same row as chain →
terminate.
+ // E0(cont=1): _is_same_row_as_chain → but chain just eliminated!
events_timestamp[0] cleared.
+ // Actually after eliminate, events_timestamp[0] has no value, so the
_is_same_row_as_chain
+ // check at E0 won't enter the if branch. E0 directly starts new chain.
+ // Row 3 (t=13:00): c2=T only → chain level=1
+ // Row 4 (t=14:00): c3=T only → chain level=2
+ // Result: max(1+1, 2+1) = 3
+ const int NUM_ROWS = 5;
+ auto column_mode = ColumnString::create();
+ for (int i = 0; i < NUM_ROWS; i++) {
+ column_mode->insert(Field::create_field<TYPE_STRING>("deduplication"));
+ }
+ auto column_timestamp = ColumnDateTimeV2::create();
+ VecDateTimeValue tv0, tv1, tv2, tv3, tv4;
+ tv0.unchecked_set_time(2022, 3, 12, 10, 0, 0);
+ tv1.unchecked_set_time(2022, 3, 12, 11, 0, 0);
+ tv2.unchecked_set_time(2022, 3, 12, 12, 0, 0);
+ tv3.unchecked_set_time(2022, 3, 12, 13, 0, 0);
+ tv4.unchecked_set_time(2022, 3, 12, 14, 0, 0);
+ auto dtv2_0 = tv0.to_datetime_v2();
+ auto dtv2_1 = tv1.to_datetime_v2();
+ auto dtv2_2 = tv2.to_datetime_v2();
+ auto dtv2_3 = tv3.to_datetime_v2();
+ auto dtv2_4 = tv4.to_datetime_v2();
+ column_timestamp->insert_data((char*)&dtv2_0, 0);
+ column_timestamp->insert_data((char*)&dtv2_1, 0);
+ column_timestamp->insert_data((char*)&dtv2_2, 0);
+ column_timestamp->insert_data((char*)&dtv2_3, 0);
+ column_timestamp->insert_data((char*)&dtv2_4, 0);
+
+ // Row 0: c1=T, c2=F, c3=F
+ // Row 1: c1=F, c2=T, c3=F
+ // Row 2: c1=T, c2=T, c3=F
+ // Row 3: c1=F, c2=T, c3=F
+ // Row 4: c1=F, c2=F, c3=T
+ auto column_event1 = ColumnUInt8::create();
+ column_event1->insert(Field::create_field<TYPE_BOOLEAN>(1));
+ column_event1->insert(Field::create_field<TYPE_BOOLEAN>(0));
+ column_event1->insert(Field::create_field<TYPE_BOOLEAN>(1));
+ column_event1->insert(Field::create_field<TYPE_BOOLEAN>(0));
+ 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));
+ column_event2->insert(Field::create_field<TYPE_BOOLEAN>(1));
+ column_event2->insert(Field::create_field<TYPE_BOOLEAN>(1));
+ column_event2->insert(Field::create_field<TYPE_BOOLEAN>(0));
+
+ auto column_event3 = ColumnUInt8::create();
+ column_event3->insert(Field::create_field<TYPE_BOOLEAN>(0));
+ column_event3->insert(Field::create_field<TYPE_BOOLEAN>(0));
+ column_event3->insert(Field::create_field<TYPE_BOOLEAN>(0));
+ column_event3->insert(Field::create_field<TYPE_BOOLEAN>(0));
+ column_event3->insert(Field::create_field<TYPE_BOOLEAN>(1));
+
+ auto column_window = ColumnInt64::create();
+ for (int i = 0; i < NUM_ROWS; i++) {
+ column_window->insert(Field::create_field<TYPE_BIGINT>(86400));
+ }
+
+ std::unique_ptr<char[]> memory(new char[agg_func_3->size_of_data()]);
+ AggregateDataPtr place = memory.get();
+ agg_func_3->create(place);
+ const IColumn* column[6] = {column_window.get(), column_mode.get(),
column_timestamp.get(),
+ column_event1.get(), column_event2.get(),
column_event3.get()};
+ for (int i = 0; i < NUM_ROWS; i++) {
+ agg_func_3->add(place, column, i, arena);
+ }
+
+ ColumnInt32 column_result;
+ agg_func_3->insert_result_into(place, column_result);
+ // First chain: E0@R0→E1@R1 (level=2). E1@R2 duplicate terminates.
+ // E0@R2 restarts. New chain: E0@R2→E1@R3→E2@R4 (level=3).
+ // max(2, 3) = 3.
+ EXPECT_EQ(column_result.get_data()[0], 3);
+ agg_func_3->destroy(place);
+}
+
+// Fixed mode: same-row multi-condition level jump fix.
+// When a row matches both a high-index condition (c3) and a low-index
condition (c2),
+// the high-index event was processed first and triggered a spurious level
jump before
+// the low-index event could advance the chain.
+TEST_F(VWindowFunnelV2Test, testFixedSameRowMultiCondLevelJump) {
+ // Use 3 conditions: c1=event1, c2=event2, c3=event3
+ AggregateFunctionSimpleFactory factory =
AggregateFunctionSimpleFactory::instance();
+ DataTypes data_types_3 = {
+ std::make_shared<DataTypeInt64>(),
std::make_shared<DataTypeString>(),
+ std::make_shared<DataTypeDateTimeV2>(),
std::make_shared<DataTypeUInt8>(),
+ std::make_shared<DataTypeUInt8>(),
std::make_shared<DataTypeUInt8>()};
+ auto agg_func_3 = factory.get("window_funnel_v2", data_types_3, nullptr,
false,
+ BeExecVersionManager::get_newest_version());
+ ASSERT_NE(agg_func_3, nullptr);
+
+ const int NUM_ROWS = 2;
+ auto column_mode = ColumnString::create();
+ for (int i = 0; i < NUM_ROWS; i++) {
+ column_mode->insert(Field::create_field<TYPE_STRING>("fixed"));
+ }
+ auto column_timestamp = ColumnDateTimeV2::create();
+ VecDateTimeValue tv0, tv1;
+ tv0.unchecked_set_time(2020, 1, 1, 0, 0, 0);
+ tv1.unchecked_set_time(2020, 1, 2, 0, 0, 0);
+ auto dtv2_0 = tv0.to_datetime_v2();
+ auto dtv2_1 = tv1.to_datetime_v2();
+ column_timestamp->insert_data((char*)&dtv2_0, 0);
+ column_timestamp->insert_data((char*)&dtv2_1, 0);
+
+ // Row 0: c1=T, c2=T, c3=T (all match — starts chain via c1)
+ // Row 1: c1=F, c2=T, c3=T (c3 would trigger level jump before c2 can
advance)
+ 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>(1));
+ column_event2->insert(Field::create_field<TYPE_BOOLEAN>(1));
+
+ auto column_event3 = ColumnUInt8::create();
+ column_event3->insert(Field::create_field<TYPE_BOOLEAN>(1));
+ column_event3->insert(Field::create_field<TYPE_BOOLEAN>(1));
+
+ auto column_window = ColumnInt64::create();
+ for (int i = 0; i < NUM_ROWS; i++) {
+ column_window->insert(Field::create_field<TYPE_BIGINT>(86400));
+ }
+
+ std::unique_ptr<char[]> memory(new char[agg_func_3->size_of_data()]);
+ AggregateDataPtr place = memory.get();
+ agg_func_3->create(place);
+ const IColumn* column[6] = {column_window.get(), column_mode.get(),
column_timestamp.get(),
+ column_event1.get(), column_event2.get(),
column_event3.get()};
+ for (int i = 0; i < NUM_ROWS; i++) {
+ agg_func_3->add(place, column, i, arena);
+ }
+
+ ColumnInt32 column_result;
+ agg_func_3->insert_result_into(place, column_result);
+ // Before fix: c3 from row 1 triggered level jump (c2 predecessor not
matched) → result 1.
+ // After fix: same-row look-ahead finds c2 can advance → skip level jump →
result 2.
+ EXPECT_EQ(column_result.get_data()[0], 2);
+ agg_func_3->destroy(place);
+}
+
+// Fixed mode: same-row duplicate should NOT suppress a level-jump break.
+// Counterexample from review: E0@r0, E1@r1, (E3,E1)@r2, E2@r3, E3@r4.
+// E1@r2 is a duplicate of an already-matched level (curr_level=2, sr_idx=1 ≤
curr_level),
+// so it must not be treated as an advancement. The chain should break at
E3@r2.
+TEST_F(VWindowFunnelV2Test, testFixedSameRowDuplicateDoesNotSuppressJump) {
+ AggregateFunctionSimpleFactory factory =
AggregateFunctionSimpleFactory::instance();
+ // 4 conditions: window, mode, timestamp, c1, c2, c3, c4
+ DataTypes data_types_4 = {
+ std::make_shared<DataTypeInt64>(),
std::make_shared<DataTypeString>(),
+ std::make_shared<DataTypeDateTimeV2>(),
std::make_shared<DataTypeUInt8>(),
+ std::make_shared<DataTypeUInt8>(),
std::make_shared<DataTypeUInt8>(),
+ std::make_shared<DataTypeUInt8>()};
+ auto agg_func = factory.get("window_funnel_v2", data_types_4, nullptr,
false,
+ BeExecVersionManager::get_newest_version());
+ ASSERT_NE(agg_func, nullptr);
+
+ const int NUM_ROWS = 5;
+ auto column_mode = ColumnString::create();
+ for (int i = 0; i < NUM_ROWS; i++) {
+ column_mode->insert(Field::create_field<TYPE_STRING>("fixed"));
+ }
+
+ // Timestamps: r0=t0, r1=t1, r2=t2, r3=t3, r4=t4 (all within window)
+ auto column_timestamp = ColumnDateTimeV2::create();
+ for (int i = 0; i < NUM_ROWS; i++) {
+ VecDateTimeValue tv;
+ tv.unchecked_set_time(2020, 1, 1 + i, 0, 0, 0);
+ auto dtv2 = tv.to_datetime_v2();
+ column_timestamp->insert_data((char*)&dtv2, 0);
+ }
+
+ // r0: c1=T, c2=F, c3=F, c4=F → E0
+ // r1: c1=F, c2=T, c3=F, c4=F → E1
+ // r2: c1=F, c2=T, c3=F, c4=T → E1(dup)+E3(jump)
+ // r3: c1=F, c2=F, c3=T, c4=F → E2
+ // r4: c1=F, c2=F, c3=F, c4=T → E3
+ auto column_c1 = ColumnUInt8::create();
+ for (int v : {1, 0, 0, 0, 0})
column_c1->insert(Field::create_field<TYPE_BOOLEAN>(v));
+
+ auto column_c2 = ColumnUInt8::create();
+ for (int v : {0, 1, 1, 0, 0})
column_c2->insert(Field::create_field<TYPE_BOOLEAN>(v));
+
+ auto column_c3 = ColumnUInt8::create();
+ for (int v : {0, 0, 0, 1, 0})
column_c3->insert(Field::create_field<TYPE_BOOLEAN>(v));
+
+ auto column_c4 = ColumnUInt8::create();
+ for (int v : {0, 0, 1, 0, 1})
column_c4->insert(Field::create_field<TYPE_BOOLEAN>(v));
+
+ auto column_window = ColumnInt64::create();
+ for (int i = 0; i < NUM_ROWS; i++) {
+ column_window->insert(Field::create_field<TYPE_BIGINT>(86400 * 10));
+ }
+
+ std::unique_ptr<char[]> memory(new char[agg_func->size_of_data()]);
+ AggregateDataPtr place = memory.get();
+ agg_func->create(place);
+ const IColumn* column[7] = {column_window.get(), column_mode.get(),
column_timestamp.get(),
+ column_c1.get(), column_c2.get(),
column_c3.get(),
+ column_c4.get()};
+ for (int i = 0; i < NUM_ROWS; i++) {
+ agg_func->add(place, column, i, arena);
+ }
+
+ ColumnInt32 column_result;
+ agg_func->insert_result_into(place, column_result);
+ // Chain: E0@r0 → E1@r1 → E3@r2 triggers level jump. E1@r2 is a duplicate
+ // (sr_idx=1 ≤ curr_level=2), so it must NOT suppress the break. Result =
2.
+ EXPECT_EQ(column_result.get_data()[0], 2);
+ agg_func->destroy(place);
+}
+
+// Fixed mode: when a row matches multiple conditions including E0 (c1),
+// the E0 continuation should NOT restart the chain if a same-row event
+// already advanced it. This emulates V1's row-based semantics where each
+// row is tested against exactly one expected condition.
+TEST_F(VWindowFunnelV2Test, testFixedContinuationE0DoesNotRestartChain) {
+ AggregateFunctionSimpleFactory factory =
AggregateFunctionSimpleFactory::instance();
+ DataTypes data_types_3 = {
+ std::make_shared<DataTypeInt64>(),
std::make_shared<DataTypeString>(),
+ std::make_shared<DataTypeDateTimeV2>(),
std::make_shared<DataTypeUInt8>(),
+ std::make_shared<DataTypeUInt8>(),
std::make_shared<DataTypeUInt8>()};
+ auto agg_func_3 = factory.get("window_funnel_v2", data_types_3, nullptr,
false,
+ BeExecVersionManager::get_newest_version());
+ ASSERT_NE(agg_func_3, nullptr);
+
+ // 3 rows, 3 conditions (c1, c2, c3):
+ // r0: c1=T, c2=F, c3=F -> E0 starts chain
+ // r1: c1=T, c2=T, c3=F -> c2 advances chain, c1(cont) should be
skipped
+ // r2: c1=F, c2=F, c3=T -> c3 advances chain to level 2
+ // Expected result: 3 (c1@r0 -> c2@r1 -> c3@r2)
+ // Without E0-skip fix, c1@r1(cont) would restart chain, and result would
be 2.
+ const int NUM_ROWS = 3;
+ auto column_mode = ColumnString::create();
+ for (int i = 0; i < NUM_ROWS; i++) {
+ column_mode->insert(Field::create_field<TYPE_STRING>("fixed"));
+ }
+ auto column_timestamp = ColumnDateTimeV2::create();
+ VecDateTimeValue tv;
+ for (int i = 0; i < NUM_ROWS; i++) {
+ tv.unchecked_set_time(2020, 1, i + 1, 0, 0, 0);
+ auto dtv2 = tv.to_datetime_v2();
+ column_timestamp->insert_data((char*)&dtv2, 0);
+ }
+
+ auto column_window = ColumnInt64::create();
+ for (int i = 0; i < NUM_ROWS; i++) {
+ column_window->insert(Field::create_field<TYPE_BIGINT>(86400 * 10));
+ }
+
+ // c1: r0=T, r1=T, r2=F
+ auto column_c1 = ColumnUInt8::create();
+ column_c1->insert(Field::create_field<TYPE_BOOLEAN>(1));
+ column_c1->insert(Field::create_field<TYPE_BOOLEAN>(1));
+ column_c1->insert(Field::create_field<TYPE_BOOLEAN>(0));
+
+ // c2: r0=F, r1=T, r2=F
+ auto column_c2 = ColumnUInt8::create();
+ column_c2->insert(Field::create_field<TYPE_BOOLEAN>(0));
+ column_c2->insert(Field::create_field<TYPE_BOOLEAN>(1));
+ column_c2->insert(Field::create_field<TYPE_BOOLEAN>(0));
+
+ // c3: r0=F, r1=F, r2=T
+ auto column_c3 = ColumnUInt8::create();
+ column_c3->insert(Field::create_field<TYPE_BOOLEAN>(0));
+ column_c3->insert(Field::create_field<TYPE_BOOLEAN>(0));
+ column_c3->insert(Field::create_field<TYPE_BOOLEAN>(1));
+
+ std::unique_ptr<char[]> memory(new char[agg_func_3->size_of_data()]);
+ AggregateDataPtr place = memory.get();
+ agg_func_3->create(place);
+ const IColumn* columns[6] = {column_window.get(), column_mode.get(),
column_timestamp.get(),
+ column_c1.get(), column_c2.get(),
column_c3.get()};
+ for (int i = 0; i < NUM_ROWS; i++) {
+ agg_func_3->add(place, columns, i, arena);
+ }
+
+ ColumnInt32 column_result;
+ agg_func_3->insert_result_into(place, column_result);
+ EXPECT_EQ(column_result.get_data()[0], 3);
+ agg_func_3->destroy(place);
+}
+
+// Dedup mode multi-pass: V1 reaches level 4 by starting from a later E0 and
skipping
+// over rows that match higher-level conditions. The old single-pass V2 broke
at level 3
+// because it greedily promoted from multi-condition rows, creating premature
duplicates.
+// 7 conditions, 15 valid rows (after null filtering). Window ~35 years
(effectively unlimited).
+// V1 chain: c1@R5(pk=8) → c2@R9(pk=6) → c3@R10(pk=12) → c4@R12(pk=5), then c1
dup at R13 → level=4.
+TEST_F(VWindowFunnelV2Test, testDeduplicationMultiPassGapCheck) {
+ AggregateFunctionSimpleFactory factory =
AggregateFunctionSimpleFactory::instance();
+ // 7 conditions
+ DataTypes data_types_7 = {
+ std::make_shared<DataTypeInt64>(),
std::make_shared<DataTypeString>(),
+ std::make_shared<DataTypeDateTimeV2>(),
std::make_shared<DataTypeUInt8>(),
+ std::make_shared<DataTypeUInt8>(),
std::make_shared<DataTypeUInt8>(),
+ std::make_shared<DataTypeUInt8>(),
std::make_shared<DataTypeUInt8>(),
+ std::make_shared<DataTypeUInt8>(),
std::make_shared<DataTypeUInt8>()};
+ auto agg_func = factory.get("window_funnel_v2", data_types_7, nullptr,
false,
+ BeExecVersionManager::get_newest_version());
+ ASSERT_NE(agg_func, nullptr);
+
+ // 15 rows sorted by timestamp (null rows excluded at SQL level).
+ // Conditions: c1=val<4, c2=val<=2, c3=val!=6, c4=val<=4,
c5=date='2023-12-13',
+ // c6=val<=5, c7=val<=8
+ struct RowData {
+ int year, month, day;
+ int val;
+ bool date_is_2023_12_13;
+ };
+ // clang-format off
+ RowData rows[] = {
+ {2000, 3, 19, 3, false}, // pk=10
+ {2000, 4, 3, 8, false}, // pk=9
+ {2001,10, 12, 3, false}, // pk=2
+ {2002,11, 15, 8, false}, // pk=0
+ {2003, 9, 10, 0, false}, // pk=19
+ {2006, 6, 6, 0, false}, // pk=8 ← V1 best E0 start
+ {2008, 7, 3, 8, false}, // pk=3
+ {2009, 3, 3, 9, false}, // pk=11
+ {2009, 5, 16, 8, true}, // pk=16
+ {2009,11, 3, 0, true}, // pk=6 ← all 7 conditions match
+ {2011, 6, 25, 7, true}, // pk=12
+ {2014, 5, 16, 6, false}, // pk=14
+ {2014,11, 2, 0, false}, // pk=5
+ {2016, 1, 8, 3, false}, // pk=13
+ {2017, 1, 3, 8, true}, // pk=17
+ };
+ // clang-format on
+ const int NUM_ROWS = 15;
+
+ auto column_mode = ColumnString::create();
+ for (int i = 0; i < NUM_ROWS; i++) {
+ column_mode->insert(Field::create_field<TYPE_STRING>("deduplication"));
+ }
+
+ auto column_timestamp = ColumnDateTimeV2::create();
+ for (int i = 0; i < NUM_ROWS; i++) {
+ VecDateTimeValue tv;
+ tv.unchecked_set_time(rows[i].year, rows[i].month, rows[i].day, 0, 0,
0);
+ auto dtv2 = tv.to_datetime_v2();
+ column_timestamp->insert_data((char*)&dtv2, 0);
+ }
+
+ auto column_window = ColumnInt64::create();
+ for (int i = 0; i < NUM_ROWS; i++) {
+ column_window->insert(Field::create_field<TYPE_BIGINT>(1119906808));
+ }
+
+ // c1: val < 4
+ auto col_c1 = ColumnUInt8::create();
+ for (int i = 0; i < NUM_ROWS; i++)
+ col_c1->insert(Field::create_field<TYPE_BOOLEAN>(rows[i].val < 4 ? 1 :
0));
+ // c2: val <= 2
+ auto col_c2 = ColumnUInt8::create();
+ for (int i = 0; i < NUM_ROWS; i++)
+ col_c2->insert(Field::create_field<TYPE_BOOLEAN>(rows[i].val <= 2 ? 1
: 0));
+ // c3: val != 6
+ auto col_c3 = ColumnUInt8::create();
+ for (int i = 0; i < NUM_ROWS; i++)
+ col_c3->insert(Field::create_field<TYPE_BOOLEAN>(rows[i].val != 6 ? 1
: 0));
+ // c4: val <= 4
+ auto col_c4 = ColumnUInt8::create();
+ for (int i = 0; i < NUM_ROWS; i++)
+ col_c4->insert(Field::create_field<TYPE_BOOLEAN>(rows[i].val <= 4 ? 1
: 0));
+ // c5: date = '2023-12-13'
+ auto col_c5 = ColumnUInt8::create();
+ for (int i = 0; i < NUM_ROWS; i++)
+
col_c5->insert(Field::create_field<TYPE_BOOLEAN>(rows[i].date_is_2023_12_13 ? 1
: 0));
+ // c6: val <= 5
+ auto col_c6 = ColumnUInt8::create();
+ for (int i = 0; i < NUM_ROWS; i++)
+ col_c6->insert(Field::create_field<TYPE_BOOLEAN>(rows[i].val <= 5 ? 1
: 0));
+ // c7: val <= 8
+ auto col_c7 = ColumnUInt8::create();
+ for (int i = 0; i < NUM_ROWS; i++)
+ col_c7->insert(Field::create_field<TYPE_BOOLEAN>(rows[i].val <= 8 ? 1
: 0));
+
+ std::unique_ptr<char[]> memory(new char[agg_func->size_of_data()]);
+ AggregateDataPtr place = memory.get();
+ agg_func->create(place);
+ const IColumn* column[10] = {column_window.get(), column_mode.get(),
column_timestamp.get(),
+ col_c1.get(), col_c2.get(),
col_c3.get(),
+ col_c4.get(), col_c5.get(),
col_c6.get(),
+ col_c7.get()};
+ for (int i = 0; i < NUM_ROWS; i++) {
+ agg_func->add(place, column, i, arena);
+ }
+
+ ColumnInt32 column_result;
+ agg_func->insert_result_into(place, column_result);
+ // V1 = 4: c1@R5(pk=8) → c2@R9(pk=6) → c3@R10(pk=12) → c4@R12(pk=5)
+ // c1 dup at R13(pk=13) breaks the chain before c5 can be matched.
+ EXPECT_EQ(column_result.get_data()[0], 4);
+ agg_func->destroy(place);
+}
+
+// FIXED mode "relevant-but-wrong row" break: when the next row after a matched
+// condition matches some condition but NOT the expected next one, the chain
must
+// break. V1 checks consecutive rows, so a c2-only row after c1→c2 breaks when
+// expecting c3. Before the fix, V2 silently re-promoted c2 and kept searching.
+TEST_F(VWindowFunnelV2Test, testFixedRelevantButWrongRowBreaksChain) {
+ AggregateFunctionSimpleFactory factory =
AggregateFunctionSimpleFactory::instance();
+ DataTypes data_types_3 = {
+ std::make_shared<DataTypeInt64>(),
std::make_shared<DataTypeString>(),
+ std::make_shared<DataTypeDateTimeV2>(),
std::make_shared<DataTypeUInt8>(),
+ std::make_shared<DataTypeUInt8>(),
std::make_shared<DataTypeUInt8>()};
+ auto agg_func_3 = factory.get("window_funnel_v2", data_types_3, nullptr,
false,
+ BeExecVersionManager::get_newest_version());
+ ASSERT_NE(agg_func_3, nullptr);
+
+ // 4 rows, 3 conditions (c1, c2, c3):
+ // r0: c1=T, c2=F, c3=F → E0, chain level 0
+ // r1: c1=F, c2=T, c3=F → c2 advances chain to level 1
+ // r2: c1=F, c2=T, c3=F → c2 re-matches level 1 (expected c3). Chain
BREAKS.
+ // r3: c1=F, c2=F, c3=T → c3 could advance but chain is already broken
+ // Expected: 2 (c1@r0 → c2@r1, break at r2)
+ const int NUM_ROWS = 4;
+ auto column_mode = ColumnString::create();
+ for (int i = 0; i < NUM_ROWS; i++) {
+ column_mode->insert(Field::create_field<TYPE_STRING>("fixed"));
+ }
+ auto column_timestamp = ColumnDateTimeV2::create();
+ VecDateTimeValue tv;
+ for (int i = 0; i < NUM_ROWS; i++) {
+ tv.unchecked_set_time(2020, 1, i + 1, 0, 0, 0);
+ auto dtv2 = tv.to_datetime_v2();
+ column_timestamp->insert_data((char*)&dtv2, 0);
+ }
+ auto column_window = ColumnInt64::create();
+ for (int i = 0; i < NUM_ROWS; i++) {
+ column_window->insert(Field::create_field<TYPE_BIGINT>(86400 * 10));
+ }
+
+ // c1: r0=T, r1=F, r2=F, r3=F
+ auto column_c1 = ColumnUInt8::create();
+ column_c1->insert(Field::create_field<TYPE_BOOLEAN>(1));
+ column_c1->insert(Field::create_field<TYPE_BOOLEAN>(0));
+ column_c1->insert(Field::create_field<TYPE_BOOLEAN>(0));
+ column_c1->insert(Field::create_field<TYPE_BOOLEAN>(0));
+
+ // c2: r0=F, r1=T, r2=T, r3=F
+ auto column_c2 = ColumnUInt8::create();
+ column_c2->insert(Field::create_field<TYPE_BOOLEAN>(0));
+ column_c2->insert(Field::create_field<TYPE_BOOLEAN>(1));
+ column_c2->insert(Field::create_field<TYPE_BOOLEAN>(1));
+ column_c2->insert(Field::create_field<TYPE_BOOLEAN>(0));
+
+ // c3: r0=F, r1=F, r2=F, r3=T
+ auto column_c3 = ColumnUInt8::create();
+ column_c3->insert(Field::create_field<TYPE_BOOLEAN>(0));
+ column_c3->insert(Field::create_field<TYPE_BOOLEAN>(0));
+ column_c3->insert(Field::create_field<TYPE_BOOLEAN>(0));
+ column_c3->insert(Field::create_field<TYPE_BOOLEAN>(1));
+
+ std::unique_ptr<char[]> memory(new char[agg_func_3->size_of_data()]);
+ AggregateDataPtr place = memory.get();
+ agg_func_3->create(place);
+ const IColumn* columns[6] = {column_window.get(), column_mode.get(),
column_timestamp.get(),
+ column_c1.get(), column_c2.get(),
column_c3.get()};
+ for (int i = 0; i < NUM_ROWS; i++) {
+ agg_func_3->add(place, columns, i, arena);
+ }
+
+ ColumnInt32 column_result;
+ agg_func_3->insert_result_into(place, column_result);
+ EXPECT_EQ(column_result.get_data()[0], 2);
+ agg_func_3->destroy(place);
+}
+
} // namespace doris
diff --git a/regression-test/data/query_p0/aggregate/window_funnel_v2.out
b/regression-test/data/query_p0/aggregate/window_funnel_v2.out
index 5d91239ad73..83ff120a218 100644
--- a/regression-test/data/query_p0/aggregate/window_funnel_v2.out
+++ b/regression-test/data/query_p0/aggregate/window_funnel_v2.out
@@ -86,3 +86,59 @@
-- !v2_increase_old_chain_better --
2
+-- !v2_dedup_same_row_multi_event --
+3
+
+-- !v2_dedup_true_dup_breaks --
+2
+
+-- !v2_fixed_same_row_level_jump --
+2
+
+-- !v2_fixed_multi_cond_complex --
+2
+
+-- !v2_fixed_e0_skip --
+3
+
+-- !v2_fixed_edge_single_row_all_match --
+1
+
+-- !v2_fixed_edge_next_row_all --
+2
+
+-- !v2_fixed_edge_e1e2_no_e0 --
+2
+
+-- !v2_fixed_edge_4cond_progressive --
+3
+
+-- !v2_fixed_edge_both_rows_multi --
+2
+
+-- !v2_fixed_edge_full_chain --
+4
+
+-- !v2_fixed_edge_level_jump_break --
+1
+
+-- !v2_fixed_edge_e0_restart_diff_row --
+2
+
+-- !v2_fixed_edge_window_boundary --
+1
+
+-- !v2_fixed_edge_multi_restart_best --
+3
+
+-- !v2_fixed_edge_many_e0_restarts --
+3
+
+-- !v2_fixed_edge_irrelevant_skip --
+3
+
+-- !v2_fixed_edge_relevant_but_wrong --
+2
+
+-- !v2_fixed_edge_relevant_wrong_no_restart --
+2
diff --git a/regression-test/suites/query_p0/aggregate/window_funnel_v2.groovy
b/regression-test/suites/query_p0/aggregate/window_funnel_v2.groovy
index e8157c23543..10aa142c842 100644
--- a/regression-test/suites/query_p0/aggregate/window_funnel_v2.groovy
+++ b/regression-test/suites/query_p0/aggregate/window_funnel_v2.groovy
@@ -486,5 +486,417 @@ suite("window_funnel_v2") {
from windowfunnel_v2_test t;
"""
+ // ==================== DEDUPLICATION mode: same-row multi-event bug fix
====================
+ // Regression test for the bug where deduplication mode incorrectly breaks
the chain
+ // when a single row matches multiple conditions including E0.
+ // V1 returns 3, V2 was returning 2 before the fix.
sql """ DROP TABLE IF EXISTS windowfunnel_v2_test """
+ sql """
+ CREATE TABLE IF NOT EXISTS windowfunnel_v2_test (
+ xwho varchar(50) NULL COMMENT 'xwho',
+ xwhen datetimev2(3) COMMENT 'xwhen',
+ xwhat int NULL COMMENT 'xwhat'
+ )
+ DUPLICATE KEY(xwho)
+ DISTRIBUTED BY HASH(xwho) BUCKETS 3
+ PROPERTIES (
+ "replication_num" = "1"
+ );
+ """
+ // Case 1: Rows matching both E0+E1 or E0+E2 simultaneously.
+ // Row 0 (t=10:00): xwhat=1 → matches E0 (xwhat=1) and E1 (xwhat!=2)
+ // Row 1 (t=11:00): xwhat=1 → matches E0 (xwhat=1) and E1 (xwhat!=2)
+ // Row 2 (t=12:00): xwhat=3 → matches E0 (xwhat=1 false, but see below),
E1 (xwhat!=2) and E2 (xwhat=3)
+ // Using conditions: xwhat=1, xwhat!=2, xwhat=3
+ // Row 0: E0=T(1=1), E1=T(1!=2), E2=F(1!=3) → same-row E0+E1
+ // Row 1: E0=T(1=1), E1=T(1!=2), E2=F(1!=3) → same-row E0+E1
+ // Row 2: E0=F(3!=1), E1=T(3!=2), E2=T(3=3) → same-row E1+E2
+ // Expected chain: E0@row0 → E1@row1 → E2@row2 = 3
+ sql "INSERT into windowfunnel_v2_test (xwho, xwhen, xwhat) VALUES('1',
'2022-03-12 10:00:00.000', 1)"
+ sql "INSERT INTO windowfunnel_v2_test (xwho, xwhen, xwhat) VALUES('1',
'2022-03-12 11:00:00.000', 1)"
+ sql "INSERT INTO windowfunnel_v2_test (xwho, xwhen, xwhat) VALUES('1',
'2022-03-12 12:00:00.000', 3)"
+ order_qt_v2_dedup_same_row_multi_event """
+ select
+ window_funnel(
+ 86400,
+ 'deduplication',
+ t.xwhen,
+ t.xwhat = 1,
+ t.xwhat != 2,
+ t.xwhat = 3
+ ) AS level
+ from windowfunnel_v2_test t;
+ """
+
+ // Case 2: True duplicate from a different row should still break the
chain.
+ sql """ truncate table windowfunnel_v2_test; """
+ sql "INSERT into windowfunnel_v2_test (xwho, xwhen, xwhat) VALUES('1',
'2022-03-12 10:00:00.000', 1)"
+ sql "INSERT INTO windowfunnel_v2_test (xwho, xwhen, xwhat) VALUES('1',
'2022-03-12 11:00:00.000', 2)"
+ sql "INSERT INTO windowfunnel_v2_test (xwho, xwhen, xwhat) VALUES('1',
'2022-03-12 12:00:00.000', 1)"
+ sql "INSERT INTO windowfunnel_v2_test (xwho, xwhen, xwhat) VALUES('1',
'2022-03-12 13:00:00.000', 3)"
+ order_qt_v2_dedup_true_dup_breaks """
+ select
+ window_funnel(
+ 86400,
+ 'deduplication',
+ t.xwhen,
+ t.xwhat = 1,
+ t.xwhat = 2,
+ t.xwhat = 3
+ ) AS level
+ from windowfunnel_v2_test t;
+ """
+
+ sql """ DROP TABLE IF EXISTS windowfunnel_v2_test """
+
+ // ==================== FIXED mode: same-row multi-condition level jump
fix ====================
+ // When a row matches both a high-index condition (e.g., c3) and a
low-index condition (e.g., c2),
+ // the high-index event was processed first (descending order) and
triggered a spurious level jump
+ // before the low-index event could advance the chain. This test verifies
the fix.
+
+ // Minimal reproduction: 3 conditions, 2 rows
+ sql """ DROP TABLE IF EXISTS windowfunnel_v2_fixed_multi_cond """
+ sql """
+ CREATE TABLE windowfunnel_v2_fixed_multi_cond (
+ ts datetime,
+ val int
+ ) DUPLICATE KEY(ts)
+ DISTRIBUTED BY HASH(ts) BUCKETS 3
+ PROPERTIES ("replication_num" = "1");
+ """
+ sql """
+ INSERT INTO windowfunnel_v2_fixed_multi_cond VALUES
+ ('2020-01-01 00:00:00', 3),
+ ('2020-01-02 00:00:00', 1);
+ """
+ // Row 1: val=3 -> c1(>2)=T, c2(<5)=T, c3(>0)=T (starts chain)
+ // Row 2: val=1 -> c1=F, c2(<5)=T, c3(>0)=T
+ // Bug: E2(c3) from row 2 triggered level jump before E1(c2) could advance.
+ // Fix: same-row look-ahead detects E1 can advance, skips level jump.
Result=2.
+ order_qt_v2_fixed_same_row_level_jump """
+ SELECT window_funnel(86400, 'fixed', ts, val > 2, val < 5, val > 0) AS
level
+ FROM windowfunnel_v2_fixed_multi_cond;
+ """
+
+ // 4 conditions with overlapping column-based predicates
+ sql """ DROP TABLE IF EXISTS windowfunnel_v2_fixed_multi_cond """
+ sql """
+ CREATE TABLE windowfunnel_v2_fixed_multi_cond (
+ col_date date NULL,
+ col_int int NULL,
+ col_datetime datetime NULL,
+ pk int
+ ) DUPLICATE KEY(col_date, col_int)
+ DISTRIBUTED BY HASH(pk) BUCKETS 10
+ PROPERTIES ("replication_num" = "1");
+ """
+ sql """
+ INSERT INTO windowfunnel_v2_fixed_multi_cond (pk, col_int, col_date,
col_datetime) VALUES
+ (0,2,'2011-03-24','2009-09-17
17:25:39'),(1,null,'2023-12-10','2001-09-19 20:24:22'),
+ (2,1,null,'2003-04-04 02:59:52'),(3,8,'2023-12-16','2004-02-13
03:59:09'),
+ (4,5,'2023-12-18','2015-09-09
23:02:51'),(5,1,'2023-12-14','2001-10-21 07:04:03'),
+ (6,3,'2023-12-10','2009-04-07
10:55:27'),(7,null,'2023-12-14','2018-11-02 09:18:56'),
+ (8,1,'2023-12-11','2015-05-03
17:34:12'),(9,1,'2023-12-12','2010-12-25 06:21:52'),
+ (10,3,null,'2008-06-02 22:49:15'),(11,0,'2023-12-12','2010-11-13
22:32:47'),
+ (12,1,'2023-12-16','2019-04-28
20:36:03'),(13,null,null,'2019-01-01 21:24:32'),
+ (14,3,'2023-12-15','2006-08-17 21:47:52'),(15,3,null,'2015-12-23
05:58:29'),
+ (16,5,'2013-09-15','2018-08-20
22:15:56'),(17,5,'2023-12-11','2017-01-17 11:34:56'),
+ (18,7,'2023-12-18','2016-01-16
13:00:40'),(19,null,'2023-12-13','2003-03-16 01:28:28'),
+ (20,0,'2023-12-16','2006-08-28
22:48:19'),(21,8,'2013-10-28','2007-01-19 07:52:53'),
+ (22,3,'2023-12-17','2001-06-05
18:32:05'),(23,2,'2023-12-17','2001-05-15 22:39:45'),
+ (24,null,'2023-12-18','2015-02-06
18:44:55'),(25,null,'2023-12-11','2004-02-22 19:46:22'),
+ (26,0,'2023-12-17','2001-05-20
05:25:26'),(27,7,'2023-12-15','2015-05-07 15:56:20'),
+ (28,0,'2023-12-16','2016-02-10
04:26:23'),(29,0,'2023-12-18','2019-09-04 05:15:31'),
+ (30,1,'2023-12-11','2000-12-08
10:19:58'),(31,9,'2023-12-14','2004-07-11 03:51:22'),
+ (32,5,'2023-12-16','2013-11-18 16:32:18'),(33,6,null,'2000-11-13
05:31:42'),
+ (34,7,'2023-12-18','2013-01-02
22:46:46'),(35,4,'2023-12-12','2008-10-23 09:22:02'),
+ (36,2,'2023-12-16','2006-01-18
13:41:57'),(37,5,'2023-12-14','2008-05-27 21:09:03'),
+ (38,2,'2023-12-15','2007-01-15
01:30:56'),(39,1,'2023-12-18','2003-04-10 16:23:13'),
+ (40,null,'2023-12-18','2018-03-25
20:34:39'),(41,4,'2023-12-15','2007-10-06 00:49:06'),
+ (42,7,'2023-12-11','2015-11-11
23:53:16'),(43,null,'2023-12-10','2002-05-26 11:17:20'),
+ (44,6,'2023-12-14','2003-04-11
05:05:55'),(45,null,'2014-08-07','2000-10-02 00:55:06'),
+ (46,2,'2023-12-16','2018-08-26
22:48:02'),(47,2,'2023-12-11','2004-12-15 08:38:08'),
+ (48,5,'2019-01-19','2000-11-09
19:21:35'),(49,6,'2023-12-12','2010-11-21 05:54:00');
+ """
+ // c1: col_int>2, c2: col_date<'2023-12-12', c3: col_int=5, c4:
col_date>'2009-10-14'
+ // Many rows match both c2 and c4 (e.g., col_date='2023-12-11'). In the
buggy code,
+ // c4's event triggered a level jump before c2 could advance. Expected: 2.
+ order_qt_v2_fixed_multi_cond_complex """
+ SELECT WINDOW_FUNNEL(1736123071, 'fixed', col_datetime,
+ col_int > 2, col_date < '2023-12-12', col_int = 5, col_date >
'2009-10-14')
+ FROM windowfunnel_v2_fixed_multi_cond;
+ """
+
+ sql """ DROP TABLE IF EXISTS windowfunnel_v2_fixed_multi_cond """
+
+ // ==================== FIXED mode: continuation E0 should not restart
chain ====================
+ // When a row matches both the expected next condition (e.g., c2) and c1,
the c2 event
+ // advances the chain while c1 (stored as continuation) should NOT restart
a new chain.
+ // This matches V1's row-based semantics.
+
+ sql """ DROP TABLE IF EXISTS windowfunnel_v2_fixed_e0_skip """
+ sql """
+ CREATE TABLE windowfunnel_v2_fixed_e0_skip (
+ ts datetime,
+ val int
+ ) DUPLICATE KEY(ts)
+ DISTRIBUTED BY HASH(ts) BUCKETS 3
+ PROPERTIES ("replication_num" = "1");
+ """
+ sql """
+ INSERT INTO windowfunnel_v2_fixed_e0_skip VALUES
+ ('2020-01-01 00:00:00', 1),
+ ('2020-01-02 00:00:00', 3),
+ ('2020-01-03 00:00:00', -1);
+ """
+ // r0: val=1 -> c1(val>0)=T, c2(val>2)=F, c3(val<0)=F. Starts chain.
+ // r1: val=3 -> c1=T, c2=T, c3=F. c2 advances chain; c1(cont) should NOT
restart.
+ // r2: val=-1 -> c1=F, c2=F, c3=T. c3 advances to level 2.
+ // Expected: 3 (c1@r0 -> c2@r1 -> c3@r2)
+ order_qt_v2_fixed_e0_skip """
+ SELECT window_funnel(864000, 'fixed', ts, val > 0, val > 2, val < 0)
AS level
+ FROM windowfunnel_v2_fixed_e0_skip;
+ """
+
+ sql """ DROP TABLE IF EXISTS windowfunnel_v2_fixed_e0_skip """
+
+ // ==================== FIXED mode: comprehensive same-row multi-condition
edge cases ====================
+
+ // Test: single row matching all conditions → only E0 counts (no next row
for c2)
+ sql """ DROP TABLE IF EXISTS windowfunnel_v2_fixed_edge """
+ sql """
+ CREATE TABLE windowfunnel_v2_fixed_edge (ts datetime, val int)
+ DUPLICATE KEY(ts) DISTRIBUTED BY HASH(ts) BUCKETS 3
+ PROPERTIES ("replication_num" = "1");
+ """
+ sql """ INSERT INTO windowfunnel_v2_fixed_edge VALUES ('2020-01-01
00:00:00', 3) """
+ order_qt_v2_fixed_edge_single_row_all_match """
+ SELECT window_funnel(864000, 'fixed', ts, val>=0, val>=1, val>=2) FROM
windowfunnel_v2_fixed_edge;
+ """
+ sql """ DROP TABLE IF EXISTS windowfunnel_v2_fixed_edge """
+
+ // Test: R0:E0, R1:E0+E1+E2 (next row matches all conditions)
+ sql """ DROP TABLE IF EXISTS windowfunnel_v2_fixed_edge """
+ sql """
+ CREATE TABLE windowfunnel_v2_fixed_edge (ts datetime, val int)
+ DUPLICATE KEY(ts) DISTRIBUTED BY HASH(ts) BUCKETS 3
+ PROPERTIES ("replication_num" = "1");
+ """
+ sql """
+ INSERT INTO windowfunnel_v2_fixed_edge VALUES
+ ('2020-01-01 00:00:00', 0), ('2020-01-02 00:00:00', 3);
+ """
+ order_qt_v2_fixed_edge_next_row_all """
+ SELECT window_funnel(864000, 'fixed', ts, val>=0, val>=1, val>=2) FROM
windowfunnel_v2_fixed_edge;
+ """
+ sql """ DROP TABLE IF EXISTS windowfunnel_v2_fixed_edge """
+
+ // Test: R0:E0, R1:E1+E2 (not E0), R2:E2 only
+ // c3 at R2 doesn't match → chain stops at level 1
+ sql """ DROP TABLE IF EXISTS windowfunnel_v2_fixed_edge """
+ sql """
+ CREATE TABLE windowfunnel_v2_fixed_edge (ts datetime, val int)
+ DUPLICATE KEY(ts) DISTRIBUTED BY HASH(ts) BUCKETS 3
+ PROPERTIES ("replication_num" = "1");
+ """
+ sql """
+ INSERT INTO windowfunnel_v2_fixed_edge VALUES
+ ('2020-01-01 00:00:00', 1), ('2020-01-02 00:00:00', 3),
('2020-01-03 00:00:00', 2);
+ """
+ order_qt_v2_fixed_edge_e1e2_no_e0 """
+ SELECT window_funnel(864000, 'fixed', ts, val=1, val>1, val>2) FROM
windowfunnel_v2_fixed_edge;
+ """
+ sql """ DROP TABLE IF EXISTS windowfunnel_v2_fixed_edge """
+
+ // Test: 4 conditions progressive (R0:c1, R1:c1+c2+c3, R2:all)
+ sql """ DROP TABLE IF EXISTS windowfunnel_v2_fixed_edge """
+ sql """
+ CREATE TABLE windowfunnel_v2_fixed_edge (ts datetime, val int)
+ DUPLICATE KEY(ts) DISTRIBUTED BY HASH(ts) BUCKETS 3
+ PROPERTIES ("replication_num" = "1");
+ """
+ sql """
+ INSERT INTO windowfunnel_v2_fixed_edge VALUES
+ ('2020-01-01 00:00:00', 0), ('2020-01-02 00:00:00', 2),
('2020-01-03 00:00:00', 3);
+ """
+ order_qt_v2_fixed_edge_4cond_progressive """
+ SELECT window_funnel(864000, 'fixed', ts, val>=0, val>=1, val>=2,
val>=3) FROM windowfunnel_v2_fixed_edge;
+ """
+ sql """ DROP TABLE IF EXISTS windowfunnel_v2_fixed_edge """
+
+ // Test: R0:E0+E1, R1:E0+E1+E2 (both rows match c1+c2)
+ sql """ DROP TABLE IF EXISTS windowfunnel_v2_fixed_edge """
+ sql """
+ CREATE TABLE windowfunnel_v2_fixed_edge (ts datetime, val int)
+ DUPLICATE KEY(ts) DISTRIBUTED BY HASH(ts) BUCKETS 3
+ PROPERTIES ("replication_num" = "1");
+ """
+ sql """
+ INSERT INTO windowfunnel_v2_fixed_edge VALUES
+ ('2020-01-01 00:00:00', 1), ('2020-01-02 00:00:00', 2);
+ """
+ order_qt_v2_fixed_edge_both_rows_multi """
+ SELECT window_funnel(864000, 'fixed', ts, val>=1, val>=1, val>=2) FROM
windowfunnel_v2_fixed_edge;
+ """
+ sql """ DROP TABLE IF EXISTS windowfunnel_v2_fixed_edge """
+
+ // Test: 4 rows, 4 conditions, each row adds one more matching condition
+ sql """ DROP TABLE IF EXISTS windowfunnel_v2_fixed_edge """
+ sql """
+ CREATE TABLE windowfunnel_v2_fixed_edge (ts datetime, val int)
+ DUPLICATE KEY(ts) DISTRIBUTED BY HASH(ts) BUCKETS 3
+ PROPERTIES ("replication_num" = "1");
+ """
+ sql """
+ INSERT INTO windowfunnel_v2_fixed_edge VALUES
+ ('2020-01-01 00:00:00', 0), ('2020-01-02 00:00:00', 1),
+ ('2020-01-03 00:00:00', 2), ('2020-01-04 00:00:00', 3);
+ """
+ order_qt_v2_fixed_edge_full_chain """
+ SELECT window_funnel(864000, 'fixed', ts, val>=0, val>=1, val>=2,
val>=3) FROM windowfunnel_v2_fixed_edge;
+ """
+ sql """ DROP TABLE IF EXISTS windowfunnel_v2_fixed_edge """
+
+ // Test: R0:E0, R1:only E2 (c2 not matched) → level jump breaks chain
+ sql """ DROP TABLE IF EXISTS windowfunnel_v2_fixed_edge """
+ sql """
+ CREATE TABLE windowfunnel_v2_fixed_edge (ts datetime, val int)
+ DUPLICATE KEY(ts) DISTRIBUTED BY HASH(ts) BUCKETS 3
+ PROPERTIES ("replication_num" = "1");
+ """
+ sql """
+ INSERT INTO windowfunnel_v2_fixed_edge VALUES
+ ('2020-01-01 00:00:00', 1), ('2020-01-02 00:00:00', 5);
+ """
+ order_qt_v2_fixed_edge_level_jump_break """
+ SELECT window_funnel(864000, 'fixed', ts, val=1, val=2, val>2) FROM
windowfunnel_v2_fixed_edge;
+ """
+ sql """ DROP TABLE IF EXISTS windowfunnel_v2_fixed_edge """
+
+ // Test: E0 restart from different row should work normally
+ sql """ DROP TABLE IF EXISTS windowfunnel_v2_fixed_edge """
+ sql """
+ CREATE TABLE windowfunnel_v2_fixed_edge (ts datetime, val int)
+ DUPLICATE KEY(ts) DISTRIBUTED BY HASH(ts) BUCKETS 3
+ PROPERTIES ("replication_num" = "1");
+ """
+ sql """
+ INSERT INTO windowfunnel_v2_fixed_edge VALUES
+ ('2020-01-01 00:00:00', 1), ('2020-01-02 00:00:00', 5),
+ ('2020-01-03 00:00:00', 1), ('2020-01-04 00:00:00', 2);
+ """
+ order_qt_v2_fixed_edge_e0_restart_diff_row """
+ SELECT window_funnel(864000, 'fixed', ts, val>0, val=2) FROM
windowfunnel_v2_fixed_edge;
+ """
+ sql """ DROP TABLE IF EXISTS windowfunnel_v2_fixed_edge """
+
+ // Test: Window boundary with multi-condition same-row (exceeds 1s window)
+ sql """ DROP TABLE IF EXISTS windowfunnel_v2_fixed_edge """
+ sql """
+ CREATE TABLE windowfunnel_v2_fixed_edge (ts datetime, val int)
+ DUPLICATE KEY(ts) DISTRIBUTED BY HASH(ts) BUCKETS 3
+ PROPERTIES ("replication_num" = "1");
+ """
+ sql """
+ INSERT INTO windowfunnel_v2_fixed_edge VALUES
+ ('2020-01-01 00:00:00', 0), ('2020-01-01 00:00:02', 2);
+ """
+ order_qt_v2_fixed_edge_window_boundary """
+ SELECT window_funnel(1, 'fixed', ts, val>=0, val>=1, val>=2) FROM
windowfunnel_v2_fixed_edge;
+ """
+ sql """ DROP TABLE IF EXISTS windowfunnel_v2_fixed_edge """
+
+ // Test: Multiple restarts find best chain
+ // c1@R0→c2@R1 then R2 has c2 again (breaks). c1@R1→c2@R2→c3@R3 succeeds →
level 3.
+ sql """ DROP TABLE IF EXISTS windowfunnel_v2_fixed_edge """
+ sql """
+ CREATE TABLE windowfunnel_v2_fixed_edge (ts datetime, val int)
+ DUPLICATE KEY(ts) DISTRIBUTED BY HASH(ts) BUCKETS 3
+ PROPERTIES ("replication_num" = "1");
+ """
+ sql """
+ INSERT INTO windowfunnel_v2_fixed_edge VALUES
+ ('2020-01-01 00:00:00', 1), ('2020-01-02 00:00:00', 2),
+ ('2020-01-03 00:00:00', 2), ('2020-01-04 00:00:00', 3);
+ """
+ order_qt_v2_fixed_edge_multi_restart_best """
+ SELECT window_funnel(864000, 'fixed', ts, val>0, val>1, val>2) FROM
windowfunnel_v2_fixed_edge;
+ """
+ sql """ DROP TABLE IF EXISTS windowfunnel_v2_fixed_edge """
+
+ // Test: Multiple restarts with many E0-only rows
+ sql """ DROP TABLE IF EXISTS windowfunnel_v2_fixed_edge """
+ sql """
+ CREATE TABLE windowfunnel_v2_fixed_edge (ts datetime, val int)
+ DUPLICATE KEY(ts) DISTRIBUTED BY HASH(ts) BUCKETS 3
+ PROPERTIES ("replication_num" = "1");
+ """
+ sql """
+ INSERT INTO windowfunnel_v2_fixed_edge VALUES
+ ('2020-01-01 00:00:00', 1), ('2020-01-02 00:00:00', 1),
+ ('2020-01-03 00:00:00', 1), ('2020-01-04 00:00:00', 2),
+ ('2020-01-05 00:00:00', 3);
+ """
+ order_qt_v2_fixed_edge_many_e0_restarts """
+ SELECT window_funnel(864000, 'fixed', ts, val>0, val>1, val>2) FROM
windowfunnel_v2_fixed_edge;
+ """
+ sql """ DROP TABLE IF EXISTS windowfunnel_v2_fixed_edge """
+
+ // Test: Irrelevant events are skipped in V2 FIXED mode (4.1+ behavior)
+ sql """ DROP TABLE IF EXISTS windowfunnel_v2_fixed_edge """
+ sql """
+ CREATE TABLE windowfunnel_v2_fixed_edge (ts datetime, val int)
+ DUPLICATE KEY(ts) DISTRIBUTED BY HASH(ts) BUCKETS 3
+ PROPERTIES ("replication_num" = "1");
+ """
+ sql """
+ INSERT INTO windowfunnel_v2_fixed_edge VALUES
+ ('2020-01-01 00:00:00', 1), ('2020-01-02 00:00:00', 0),
+ ('2020-01-03 00:00:00', 2), ('2020-01-04 00:00:00', 3);
+ """
+ // R1(val=0) matches NO condition → V2 skips it → E0@R0→E1@R2→E2@R3 = 3
+ order_qt_v2_fixed_edge_irrelevant_skip """
+ SELECT window_funnel(864000, 'fixed', ts, val=1, val=2, val=3) FROM
windowfunnel_v2_fixed_edge;
+ """
+ sql """ DROP TABLE IF EXISTS windowfunnel_v2_fixed_edge """
+
+ // Test: "Relevant-but-wrong row" breaks chain in FIXED mode.
+ // R2 matches c2 (relevant) but not c3 (expected) → chain breaks at R2.
+ sql """ DROP TABLE IF EXISTS windowfunnel_v2_fixed_edge """
+ sql """
+ CREATE TABLE windowfunnel_v2_fixed_edge (ts datetime, val int)
+ DUPLICATE KEY(ts) DISTRIBUTED BY HASH(ts) BUCKETS 3
+ PROPERTIES ("replication_num" = "1");
+ """
+ sql """
+ INSERT INTO windowfunnel_v2_fixed_edge VALUES
+ ('2020-01-01 00:00:00', 1), ('2020-01-02 00:00:00', 2),
+ ('2020-01-03 00:00:00', 2), ('2020-01-04 00:00:00', 3);
+ """
+ // c1@R0→c2@R1, R2 has c2 again (val=2, not c3=val=3) → breaks chain.
+ // No other row matches c1(val=1), so no multi-pass recovery. Expected: 2.
+ order_qt_v2_fixed_edge_relevant_but_wrong """
+ SELECT window_funnel(864000, 'fixed', ts, val=1, val=2, val=3) FROM
windowfunnel_v2_fixed_edge;
+ """
+ sql """ DROP TABLE IF EXISTS windowfunnel_v2_fixed_edge """
+
+ // Test: "Relevant-but-wrong row" with no better starting point → breaks
chain.
+ // Only c1 at R0. R1 has c2, R2 has c2 again (not c3). No later c1 to
restart.
+ sql """ DROP TABLE IF EXISTS windowfunnel_v2_fixed_edge """
+ sql """
+ CREATE TABLE windowfunnel_v2_fixed_edge (ts datetime, val int)
+ DUPLICATE KEY(ts) DISTRIBUTED BY HASH(ts) BUCKETS 3
+ PROPERTIES ("replication_num" = "1");
+ """
+ sql """
+ INSERT INTO windowfunnel_v2_fixed_edge VALUES
+ ('2020-01-01 00:00:00', 10), ('2020-01-02 00:00:00', 5),
+ ('2020-01-03 00:00:00', 5), ('2020-01-04 00:00:00', 20);
+ """
+ // c1: val>8 (only 10,20). c2: val<8 (only 5). c3: val>15 (only 20).
+ // c1@R0→c2@R1, R2 has c2 (not c3) → break. c1@R3: no more rows. Best=2.
+ order_qt_v2_fixed_edge_relevant_wrong_no_restart """
+ SELECT window_funnel(864000, 'fixed', ts, val>8, val<8, val>15) FROM
windowfunnel_v2_fixed_edge;
+ """
+ sql """ DROP TABLE IF EXISTS windowfunnel_v2_fixed_edge """
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
