Files
greptimedb/tests/cases/standalone/common/flow/flow_advance_ttl.sql
discord9 27090496d1 test(flow): stabilize FLUSH_FLOW assertions after async source-table mirrors (#9230)
* test(flow): wait for async source-table mirrors before FLUSH_FLOW

Streaming flows execute inline in the flownode insert handler since
#8976, and source-table inserts are mirrored from the frontend as
detached tasks. Tests that INSERT then ADMIN FLUSH_FLOW then SELECT
intermittently miss rows on slow CI runners because the flush no longer
implies the mirror reached the flownode (flow_no_aggr lost row 'l',
flow_advance_ttl lost row '23' after the post-restart reinsert).

Add SQLNESS SLEEP 3s between those mirror inserts and the flush, the
established pattern in the flow suite, so the assertions are stable.

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

* test(flow): also guard first-section flushes in flow_advance_ttl

Local reproduction without sleeps failed in a window the previous
commit did not cover: the first INSERT (20,20,22) of each section is
followed immediately by ADMIN FLUSH_FLOW, and the distributed run
observed an empty sink on that SELECT (~1 in 8 iterations). Add the
same SLEEP 3s guard before those two flushes.

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

---------

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
2026-09-18 08:03:07 +00:00

184 lines
4.8 KiB
SQL

-- test ttl = instant
CREATE TABLE distinct_basic (
"number" INT,
ts TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY(number),
TIME INDEX(ts)
)WITH ('ttl' = 'instant');
-- request-local DISTINCT is supported in streaming mode
-- SQLNESS REPLACE id=\d+ id=REDACTED
CREATE FLOW test_distinct_basic SINK TO out_distinct_basic AS
SELECT
DISTINCT number as dis
FROM
distinct_basic;
-- instant-TTL sources reject non-stateless LIMIT plans
CREATE FLOW test_limit_instant_rejected SINK TO out_limit_instant_rejected AS
SELECT
number
FROM
distinct_basic
LIMIT 1;
-- flow_options should have a flow_type:streaming
-- since source table's ttl=instant and DISTINCT is request-local
SELECT flow_name, options FROM INFORMATION_SCHEMA.FLOWS;
SHOW CREATE TABLE distinct_basic;
-- SQLNESS REPLACE \d{4} REDACTED
SHOW CREATE TABLE out_distinct_basic;
-- SQLNESS SLEEP 3s
INSERT INTO
distinct_basic
VALUES
(20, "2021-07-01 00:00:00.200"),
(20, "2021-07-01 00:00:00.200"),
(22, "2021-07-01 00:00:00.600");
-- Mirror inserts reach the flownode asynchronously; wait before flushing.
-- SQLNESS SLEEP 3s
-- SQLNESS REPLACE (ADMIN\sFLUSH_FLOW\('\w+'\)\s+\|\n\+-+\+\n\|\s+)[0-9]+\s+\| $1 FLOW_FLUSHED |
ADMIN FLUSH_FLOW('test_distinct_basic');
SELECT
dis
FROM
out_distinct_basic;
SELECT number FROM distinct_basic;
-- SQLNESS SLEEP 6s
ADMIN FLUSH_TABLE('distinct_basic');
-- Recover the persisted streaming DISTINCT flow, then replan its first write
-- against an extended source schema without recreating the flow.
-- SQLNESS ARG restart=true
SELECT 1;
ALTER TABLE distinct_basic ADD COLUMN extra INT NULL;
INSERT INTO
distinct_basic (number, ts)
VALUES
(23, "2021-07-01 00:00:01.600"),
(23, "2021-07-01 00:00:01.600");
-- Mirror inserts reach the flownode asynchronously; wait before flushing.
-- SQLNESS SLEEP 3s
-- SQLNESS REPLACE (ADMIN\sFLUSH_FLOW\('\w+'\)\s+\|\n\+-+\+\n\|\s+)[0-9]+\s+\| $1 FLOW_FLUSHED |
ADMIN FLUSH_FLOW('test_distinct_basic');
SELECT
dis
FROM
out_distinct_basic;
SELECT number FROM distinct_basic;
DROP FLOW test_distinct_basic;
DROP TABLE distinct_basic;
DROP TABLE out_distinct_basic;
-- test ttl = instant with EVAL INTERVAL must be rejected
-- since the batching scheduler cannot read instant-TTL source tables and the
-- streaming engine cannot honor the schedule
CREATE TABLE distinct_basic (
"number" INT,
ts TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY(number),
TIME INDEX(ts)
)WITH ('ttl' = 'instant');
-- SQLNESS REPLACE id=\d+ id=REDACTED
CREATE FLOW test_distinct_basic SINK TO out_distinct_basic EVAL INTERVAL '1m' AS
SELECT
DISTINCT number as dis
FROM
distinct_basic;
SELECT count(*) FROM INFORMATION_SCHEMA.FLOWS WHERE flow_name = 'test_distinct_basic';
DROP TABLE distinct_basic;
-- test ttl = 5s (DISTINCT remains batching for persisted sources)
CREATE TABLE distinct_basic (
"number" INT,
ts TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY(number),
TIME INDEX(ts)
)WITH ('ttl' = '5s');
-- Without a schedule, persisted DISTINCT must reach batching validation,
-- not silently become request-local streaming.
CREATE FLOW test_distinct_persisted_unscheduled
SINK TO out_distinct_persisted_unscheduled AS
SELECT DISTINCT number AS dis FROM distinct_basic;
DROP FLOW IF EXISTS test_distinct_persisted_unscheduled;
DROP TABLE IF EXISTS out_distinct_persisted_unscheduled;
CREATE FLOW test_distinct_basic SINK TO out_distinct_basic EVAL INTERVAL '1m' AS
SELECT
DISTINCT number as dis
FROM
distinct_basic;
-- flow_options should have a flow_type:batching
-- persisted-source DISTINCT retains batching semantics
SELECT flow_name, options FROM INFORMATION_SCHEMA.FLOWS;
-- SQLNESS ARG restart=true
SELECT 1;
-- SQLNESS SLEEP 3s
INSERT INTO
distinct_basic
VALUES
(20, "2021-07-01 00:00:00.200"),
(20, "2021-07-01 00:00:00.200"),
(22, "2021-07-01 00:00:00.600");
-- Mirror inserts reach the flownode asynchronously; wait before flushing.
-- SQLNESS SLEEP 3s
-- SQLNESS REPLACE (ADMIN\sFLUSH_FLOW\('\w+'\)\s+\|\n\+-+\+\n\|\s+)[0-9]+\s+\| $1 FLOW_FLUSHED |
ADMIN FLUSH_FLOW('test_distinct_basic');
SHOW CREATE TABLE distinct_basic;
SHOW CREATE TABLE out_distinct_basic;
SELECT
dis
FROM
out_distinct_basic;
SELECT number FROM distinct_basic;
-- SQLNESS SLEEP 6s
ADMIN FLUSH_TABLE('distinct_basic');
INSERT INTO
distinct_basic
VALUES
(23, "2021-07-01 00:00:01.600");
-- Mirror inserts reach the flownode asynchronously; wait before flushing.
-- SQLNESS SLEEP 3s
-- SQLNESS REPLACE (ADMIN\sFLUSH_FLOW\('\w+'\)\s+\|\n\+-+\+\n\|\s+)[0-9]+\s+\| $1 FLOW_FLUSHED |
ADMIN FLUSH_FLOW('test_distinct_basic');
SELECT
dis
FROM
out_distinct_basic;
SELECT number FROM distinct_basic;
DROP FLOW test_distinct_basic;
DROP TABLE distinct_basic;
DROP TABLE out_distinct_basic;