diff --git a/tests/cases/standalone/common/flow/flow_incremental_state_merge.result b/tests/cases/standalone/common/flow/flow_incremental_state_merge.result new file mode 100644 index 00000000000..9a9b23dfef3 --- /dev/null +++ b/tests/cases/standalone/common/flow/flow_incremental_state_merge.result @@ -0,0 +1,348 @@ +-- Incremental aggregate state merge over a partitioned source (two regions). +-- A WHERE-filtered flow mixes scalar aggregates (sum, count) with aggregate +-- state columns (hll, stddev_pop_state, uddsketch_state); each later delta in +-- the same time window must merge its state with the state already persisted +-- in the sink. Covers two nonempty deltas across both regions, one round whose +-- delta is empty because the flow's predicate drops the newly inserted row, +-- and a final nonempty delta checked against a direct source aggregation. +CREATE TABLE flow_incr_state_merge_input ( + host_id INT, + user_id BIGINT, + n INT, + v DOUBLE, + ts TIMESTAMP TIME INDEX, + PRIMARY KEY(host_id, user_id) +) +PARTITION ON COLUMNS (host_id) ( + host_id < 3, + host_id >= 3 +) +WITH ( + append_mode = 'true' +); + +Affected Rows: 0 + +CREATE FLOW flow_incr_state_merge SINK TO flow_incr_state_merge_sink +WITH (experimental_enable_incremental_read = 'true') +AS +SELECT + sum(n) AS total, + count(n) AS row_count, + hll(user_id) AS user_hll, + stddev_pop_state(v) AS v_stddev_state, + uddsketch_state(128, 0.01, v) AS v_sketch_state, + date_bin(INTERVAL '1 minute', ts, '2024-01-01 00:00:00') AS time_window +FROM + flow_incr_state_merge_input +WHERE + v > 0 +GROUP BY + time_window; + +Affected Rows: 0 + +-- ==== Phase 1: initial rows in the same time window, one per region ==== +INSERT INTO flow_incr_state_merge_input VALUES + (1, 1, 1, 1.0, '2024-01-01 00:00:00'), + (3, 2, 2, 2.0, '2024-01-01 00:00:10'); + +Affected Rows: 2 + +-- SQLNESS REPLACE (ADMIN\sFLUSH_FLOW\('\w+'\)\s+\|\n\+-+\+\n\|\s+)[0-9]+\s+\| $1 FLOW_FLUSHED | +ADMIN FLUSH_FLOW('flow_incr_state_merge'); + ++-------------------------------------------+ +| ADMIN FLUSH_FLOW('flow_incr_state_merge') | ++-------------------------------------------+ +| FLOW_FLUSHED | ++-------------------------------------------+ + +-- The first round seeds the sink with a full-snapshot aggregate state. +SELECT + total, + row_count, + time_window +FROM + flow_incr_state_merge_sink +ORDER BY + time_window; + ++-------+-----------+---------------------+ +| total | row_count | time_window | ++-------+-----------+---------------------+ +| 3 | 2 | 2024-01-01T00:00:00 | ++-------+-----------+---------------------+ + +-- Move the checkpointed source rows and the seeded sink state into SST, so +-- the next round has to merge a new delta with persisted state. +ADMIN FLUSH_TABLE('flow_incr_state_merge_input'); + ++--------------------------------------------------+ +| ADMIN FLUSH_TABLE('flow_incr_state_merge_input') | ++--------------------------------------------------+ +| 0 | ++--------------------------------------------------+ + +ADMIN FLUSH_TABLE('flow_incr_state_merge_sink'); + ++-------------------------------------------------+ +| ADMIN FLUSH_TABLE('flow_incr_state_merge_sink') | ++-------------------------------------------------+ +| 0 | ++-------------------------------------------------+ + +-- ==== Phase 2: first nonempty delta, same window, both regions ==== +INSERT INTO flow_incr_state_merge_input VALUES + (2, 3, 3, 3.0, '2024-01-01 00:00:20'), + (4, 4, 4, 4.0, '2024-01-01 00:00:40'); + +Affected Rows: 2 + +-- SQLNESS REPLACE (ADMIN\sFLUSH_FLOW\('\w+'\)\s+\|\n\+-+\+\n\|\s+)[0-9]+\s+\| $1 FLOW_FLUSHED | +ADMIN FLUSH_FLOW('flow_incr_state_merge'); + ++-------------------------------------------+ +| ADMIN FLUSH_FLOW('flow_incr_state_merge') | ++-------------------------------------------+ +| FLOW_FLUSHED | ++-------------------------------------------+ + +-- Scalars accumulate; each state column must merge with the persisted state. +SELECT + total, + row_count, + hll_count(hll_merge(user_hll)) AS users, + ROUND(stddev_pop_calc(stddev_pop_merge(v_stddev_state)), 6) AS v_stddev, + ROUND(uddsketch_calc(0.5, uddsketch_merge(128, 0.01, v_sketch_state)), 4) AS v_p50, + time_window +FROM + flow_incr_state_merge_sink +GROUP BY + total, + row_count, + time_window +ORDER BY + time_window; + ++-------+-----------+-------+----------+--------+---------------------+ +| total | row_count | users | v_stddev | v_p50 | time_window | ++-------+-----------+-------+----------+--------+---------------------+ +| 10 | 4 | 4 | 1.118034 | 2.9742 | 2024-01-01T00:00:00 | ++-------+-----------+-------+----------+--------+---------------------+ + +ADMIN FLUSH_TABLE('flow_incr_state_merge_input'); + ++--------------------------------------------------+ +| ADMIN FLUSH_TABLE('flow_incr_state_merge_input') | ++--------------------------------------------------+ +| 0 | ++--------------------------------------------------+ + +ADMIN FLUSH_TABLE('flow_incr_state_merge_sink'); + ++-------------------------------------------------+ +| ADMIN FLUSH_TABLE('flow_incr_state_merge_sink') | ++-------------------------------------------------+ +| 0 | ++-------------------------------------------------+ + +-- ==== Phase 3: filtered-out row must run one round with an empty delta ==== +-- Record the flow's last execution time before the excluded insert: the +-- comparison below proves the empty delta was actually evaluated. +CREATE TABLE flow_incr_state_merge_last_exec ( + marker STRING, + last_exec TIMESTAMP TIME INDEX +) WITH ( + append_mode = 'true' +); + +Affected Rows: 0 + +-- The background loop runs on its own interval; keep waiting until it has +-- executed at least once so information_schema.flows.last_execution_time is +-- non-null before snapshotting it. +-- SQLNESS SLEEP 8s +INSERT INTO flow_incr_state_merge_last_exec (marker, last_exec) +SELECT + 'before_empty_delta', + last_execution_time +FROM + information_schema.flows +WHERE + flow_name = 'flow_incr_state_merge'; + +Affected Rows: 1 + +-- The row's timestamp is inside the existing window, so the source write still +-- dirties that window; the flow's WHERE predicate then drops the row from the +-- delta, leaving the delta empty for the window. +INSERT INTO flow_incr_state_merge_input VALUES + (1, 9, 9, -1.0, '2024-01-01 00:00:30'); + +Affected Rows: 1 + +-- SQLNESS SLEEP 6s +-- SQLNESS REPLACE (ADMIN\sFLUSH_FLOW\('\w+'\)\s+\|\n\+-+\+\n\|\s+)[0-9]+\s+\| $1 FLOW_FLUSHED | +ADMIN FLUSH_FLOW('flow_incr_state_merge'); + ++-------------------------------------------+ +| ADMIN FLUSH_FLOW('flow_incr_state_merge') | ++-------------------------------------------+ +| FLOW_FLUSHED | ++-------------------------------------------+ + +-- An empty delta must leave the persisted sink states untouched. +SELECT + total, + row_count, + hll_count(hll_merge(user_hll)) AS users, + ROUND(stddev_pop_calc(stddev_pop_merge(v_stddev_state)), 6) AS v_stddev, + ROUND(uddsketch_calc(0.5, uddsketch_merge(128, 0.01, v_sketch_state)), 4) AS v_p50, + time_window +FROM + flow_incr_state_merge_sink +GROUP BY + total, + row_count, + time_window +ORDER BY + time_window; + ++-------+-----------+-------+----------+--------+---------------------+ +| total | row_count | users | v_stddev | v_p50 | time_window | ++-------+-----------+-------+----------+--------+---------------------+ +| 10 | 4 | 4 | 1.118034 | 2.9742 | 2024-01-01T00:00:00 | ++-------+-----------+-------+----------+--------+---------------------+ + +-- last_execution_time must have advanced past the mark taken above, proving +-- the dirty window triggered a new round even though its delta was empty. A +-- NULL comparison means the mark never advanced: fail explicitly instead of +-- silently treating it as "unchanged". +INSERT INTO flow_incr_state_merge_last_exec (marker, last_exec) +SELECT + 'after_empty_delta', + last_execution_time +FROM + information_schema.flows +WHERE + flow_name = 'flow_incr_state_merge'; + +Affected Rows: 1 + +SELECT + ( + SELECT + last_exec + FROM + flow_incr_state_merge_last_exec + WHERE + marker = 'after_empty_delta' + ) > ( + SELECT + last_exec + FROM + flow_incr_state_merge_last_exec + WHERE + marker = 'before_empty_delta' + ) AS empty_delta_executed; + ++----------------------+ +| empty_delta_executed | ++----------------------+ +| true | ++----------------------+ + +-- Flush the input before the final delta so each phase's delta is isolated in +-- its own round; the background loop may have already consumed the last +-- insert, which is fine — the values below only depend on the merge outcome. +ADMIN FLUSH_TABLE('flow_incr_state_merge_input'); + ++--------------------------------------------------+ +| ADMIN FLUSH_TABLE('flow_incr_state_merge_input') | ++--------------------------------------------------+ +| 0 | ++--------------------------------------------------+ + +ADMIN FLUSH_TABLE('flow_incr_state_merge_sink'); + ++-------------------------------------------------+ +| ADMIN FLUSH_TABLE('flow_incr_state_merge_sink') | ++-------------------------------------------------+ +| 0 | ++-------------------------------------------------+ + +-- ==== Phase 4: final nonempty delta, same window, both regions ==== +INSERT INTO flow_incr_state_merge_input VALUES + (1, 5, 5, 5.0, '2024-01-01 00:00:05'), + (3, 6, 6, 6.0, '2024-01-01 00:00:55'); + +Affected Rows: 2 + +-- SQLNESS REPLACE (ADMIN\sFLUSH_FLOW\('\w+'\)\s+\|\n\+-+\+\n\|\s+)[0-9]+\s+\| $1 FLOW_FLUSHED | +ADMIN FLUSH_FLOW('flow_incr_state_merge'); + ++-------------------------------------------+ +| ADMIN FLUSH_FLOW('flow_incr_state_merge') | ++-------------------------------------------+ +| FLOW_FLUSHED | ++-------------------------------------------+ + +-- Merge the final delta with the persisted states and compare the merged +-- result against a direct aggregation over the accepted source rows. +SELECT + merged.total = direct.total AS total_matches, + merged.row_count = direct.row_count AS count_matches, + merged.users = direct.users AS hll_matches, + abs(merged.v_stddev - direct.v_stddev) < 1e-9 AS stddev_matches, + ROUND(merged.v_p50, 4) = ROUND(direct.v_p50, 4) AS p50_matches +FROM + ( + SELECT + total, + row_count, + hll_count(hll_merge(user_hll)) AS users, + stddev_pop_calc(stddev_pop_merge(v_stddev_state)) AS v_stddev, + uddsketch_calc(0.5, uddsketch_merge(128, 0.01, v_sketch_state)) AS v_p50 + FROM + flow_incr_state_merge_sink + GROUP BY + total, + row_count + ) AS merged, + ( + SELECT + sum(n) AS total, + count(n) AS row_count, + hll_count(hll(user_id)) AS users, + stddev_pop(v) AS v_stddev, + uddsketch_calc(0.5, uddsketch_state(128, 0.01, v)) AS v_p50 + FROM + flow_incr_state_merge_input + WHERE + v > 0 + ) AS direct; + ++---------------+---------------+-------------+----------------+-------------+ +| total_matches | count_matches | hll_matches | stddev_matches | p50_matches | ++---------------+---------------+-------------+----------------+-------------+ +| true | true | true | true | true | ++---------------+---------------+-------------+----------------+-------------+ + +-- Clean up +DROP FLOW flow_incr_state_merge; + +Affected Rows: 0 + +DROP TABLE flow_incr_state_merge_input; + +Affected Rows: 0 + +DROP TABLE flow_incr_state_merge_sink; + +Affected Rows: 0 + +DROP TABLE flow_incr_state_merge_last_exec; + +Affected Rows: 0 + diff --git a/tests/cases/standalone/common/flow/flow_incremental_state_merge.sql b/tests/cases/standalone/common/flow/flow_incremental_state_merge.sql new file mode 100644 index 00000000000..a48b39e9fd2 --- /dev/null +++ b/tests/cases/standalone/common/flow/flow_incremental_state_merge.sql @@ -0,0 +1,232 @@ +-- Incremental aggregate state merge over a partitioned source (two regions). +-- A WHERE-filtered flow mixes scalar aggregates (sum, count) with aggregate +-- state columns (hll, stddev_pop_state, uddsketch_state); each later delta in +-- the same time window must merge its state with the state already persisted +-- in the sink. Covers two nonempty deltas across both regions, one round whose +-- delta is empty because the flow's predicate drops the newly inserted row, +-- and a final nonempty delta checked against a direct source aggregation. +CREATE TABLE flow_incr_state_merge_input ( + host_id INT, + user_id BIGINT, + n INT, + v DOUBLE, + ts TIMESTAMP TIME INDEX, + PRIMARY KEY(host_id, user_id) +) +PARTITION ON COLUMNS (host_id) ( + host_id < 3, + host_id >= 3 +) +WITH ( + append_mode = 'true' +); + +CREATE FLOW flow_incr_state_merge SINK TO flow_incr_state_merge_sink +WITH (experimental_enable_incremental_read = 'true') +AS +SELECT + sum(n) AS total, + count(n) AS row_count, + hll(user_id) AS user_hll, + stddev_pop_state(v) AS v_stddev_state, + uddsketch_state(128, 0.01, v) AS v_sketch_state, + date_bin(INTERVAL '1 minute', ts, '2024-01-01 00:00:00') AS time_window +FROM + flow_incr_state_merge_input +WHERE + v > 0 +GROUP BY + time_window; + +-- ==== Phase 1: initial rows in the same time window, one per region ==== +INSERT INTO flow_incr_state_merge_input VALUES + (1, 1, 1, 1.0, '2024-01-01 00:00:00'), + (3, 2, 2, 2.0, '2024-01-01 00:00:10'); + +-- SQLNESS REPLACE (ADMIN\sFLUSH_FLOW\('\w+'\)\s+\|\n\+-+\+\n\|\s+)[0-9]+\s+\| $1 FLOW_FLUSHED | +ADMIN FLUSH_FLOW('flow_incr_state_merge'); + +-- The first round seeds the sink with a full-snapshot aggregate state. +SELECT + total, + row_count, + time_window +FROM + flow_incr_state_merge_sink +ORDER BY + time_window; + +-- Move the checkpointed source rows and the seeded sink state into SST, so +-- the next round has to merge a new delta with persisted state. +ADMIN FLUSH_TABLE('flow_incr_state_merge_input'); + +ADMIN FLUSH_TABLE('flow_incr_state_merge_sink'); + +-- ==== Phase 2: first nonempty delta, same window, both regions ==== +INSERT INTO flow_incr_state_merge_input VALUES + (2, 3, 3, 3.0, '2024-01-01 00:00:20'), + (4, 4, 4, 4.0, '2024-01-01 00:00:40'); + +-- SQLNESS REPLACE (ADMIN\sFLUSH_FLOW\('\w+'\)\s+\|\n\+-+\+\n\|\s+)[0-9]+\s+\| $1 FLOW_FLUSHED | +ADMIN FLUSH_FLOW('flow_incr_state_merge'); + +-- Scalars accumulate; each state column must merge with the persisted state. +SELECT + total, + row_count, + hll_count(hll_merge(user_hll)) AS users, + ROUND(stddev_pop_calc(stddev_pop_merge(v_stddev_state)), 6) AS v_stddev, + ROUND(uddsketch_calc(0.5, uddsketch_merge(128, 0.01, v_sketch_state)), 4) AS v_p50, + time_window +FROM + flow_incr_state_merge_sink +GROUP BY + total, + row_count, + time_window +ORDER BY + time_window; + +ADMIN FLUSH_TABLE('flow_incr_state_merge_input'); + +ADMIN FLUSH_TABLE('flow_incr_state_merge_sink'); + +-- ==== Phase 3: filtered-out row must run one round with an empty delta ==== +-- Record the flow's last execution time before the excluded insert: the +-- comparison below proves the empty delta was actually evaluated. +CREATE TABLE flow_incr_state_merge_last_exec ( + marker STRING, + last_exec TIMESTAMP TIME INDEX +) WITH ( + append_mode = 'true' +); + +-- The background loop runs on its own interval; keep waiting until it has +-- executed at least once so information_schema.flows.last_execution_time is +-- non-null before snapshotting it. +-- SQLNESS SLEEP 8s +INSERT INTO flow_incr_state_merge_last_exec (marker, last_exec) +SELECT + 'before_empty_delta', + last_execution_time +FROM + information_schema.flows +WHERE + flow_name = 'flow_incr_state_merge'; + +-- The row's timestamp is inside the existing window, so the source write still +-- dirties that window; the flow's WHERE predicate then drops the row from the +-- delta, leaving the delta empty for the window. +INSERT INTO flow_incr_state_merge_input VALUES + (1, 9, 9, -1.0, '2024-01-01 00:00:30'); + +-- SQLNESS SLEEP 6s +-- SQLNESS REPLACE (ADMIN\sFLUSH_FLOW\('\w+'\)\s+\|\n\+-+\+\n\|\s+)[0-9]+\s+\| $1 FLOW_FLUSHED | +ADMIN FLUSH_FLOW('flow_incr_state_merge'); + +-- An empty delta must leave the persisted sink states untouched. +SELECT + total, + row_count, + hll_count(hll_merge(user_hll)) AS users, + ROUND(stddev_pop_calc(stddev_pop_merge(v_stddev_state)), 6) AS v_stddev, + ROUND(uddsketch_calc(0.5, uddsketch_merge(128, 0.01, v_sketch_state)), 4) AS v_p50, + time_window +FROM + flow_incr_state_merge_sink +GROUP BY + total, + row_count, + time_window +ORDER BY + time_window; + +-- last_execution_time must have advanced past the mark taken above, proving +-- the dirty window triggered a new round even though its delta was empty. A +-- NULL comparison means the mark never advanced: fail explicitly instead of +-- silently treating it as "unchanged". +INSERT INTO flow_incr_state_merge_last_exec (marker, last_exec) +SELECT + 'after_empty_delta', + last_execution_time +FROM + information_schema.flows +WHERE + flow_name = 'flow_incr_state_merge'; + +SELECT + ( + SELECT + last_exec + FROM + flow_incr_state_merge_last_exec + WHERE + marker = 'after_empty_delta' + ) > ( + SELECT + last_exec + FROM + flow_incr_state_merge_last_exec + WHERE + marker = 'before_empty_delta' + ) AS empty_delta_executed; + +-- Flush the input before the final delta so each phase's delta is isolated in +-- its own round; the background loop may have already consumed the last +-- insert, which is fine — the values below only depend on the merge outcome. + +ADMIN FLUSH_TABLE('flow_incr_state_merge_input'); + +ADMIN FLUSH_TABLE('flow_incr_state_merge_sink'); + +-- ==== Phase 4: final nonempty delta, same window, both regions ==== +INSERT INTO flow_incr_state_merge_input VALUES + (1, 5, 5, 5.0, '2024-01-01 00:00:05'), + (3, 6, 6, 6.0, '2024-01-01 00:00:55'); + +-- SQLNESS REPLACE (ADMIN\sFLUSH_FLOW\('\w+'\)\s+\|\n\+-+\+\n\|\s+)[0-9]+\s+\| $1 FLOW_FLUSHED | +ADMIN FLUSH_FLOW('flow_incr_state_merge'); + +-- Merge the final delta with the persisted states and compare the merged +-- result against a direct aggregation over the accepted source rows. +SELECT + merged.total = direct.total AS total_matches, + merged.row_count = direct.row_count AS count_matches, + merged.users = direct.users AS hll_matches, + abs(merged.v_stddev - direct.v_stddev) < 1e-9 AS stddev_matches, + ROUND(merged.v_p50, 4) = ROUND(direct.v_p50, 4) AS p50_matches +FROM + ( + SELECT + total, + row_count, + hll_count(hll_merge(user_hll)) AS users, + stddev_pop_calc(stddev_pop_merge(v_stddev_state)) AS v_stddev, + uddsketch_calc(0.5, uddsketch_merge(128, 0.01, v_sketch_state)) AS v_p50 + FROM + flow_incr_state_merge_sink + GROUP BY + total, + row_count + ) AS merged, + ( + SELECT + sum(n) AS total, + count(n) AS row_count, + hll_count(hll(user_id)) AS users, + stddev_pop(v) AS v_stddev, + uddsketch_calc(0.5, uddsketch_state(128, 0.01, v)) AS v_p50 + FROM + flow_incr_state_merge_input + WHERE + v > 0 + ) AS direct; + +-- Clean up +DROP FLOW flow_incr_state_merge; + +DROP TABLE flow_incr_state_merge_input; + +DROP TABLE flow_incr_state_merge_sink; + +DROP TABLE flow_incr_state_merge_last_exec;