mirror of
https://github.com/GreptimeTeam/greptimedb.git
synced 2026-10-19 10:22:31 +00:00
* test(mito2): cover regex inverted index pruning
Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
* refactor(flow): execute streaming flows with DataFusion
Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
* refactor(flow): remove legacy streaming runtime
Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
* fix(flow): avoid retrying stateless sink inserts
Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
* fix(flow): align stateless writes with sink schema
Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
* fix(flow): reject stale stateless source schemas
Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
* fix(flow): validate stateless flow routing
Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
* Revert "test(mito2): cover regex inverted index pruning"
This reverts commit 79e96ac745.
Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
* fix(flow): preserve source timestamps in stateless flows
Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
* fix(flow): address stateless review feedback
Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
* fix(flow): recover stateless flows after schema changes
Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
* fix(flow): serialize schema rebuilds and validate retained sources
Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
* test(flow): verify streaming recovery and schema changes through SQL
Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
* fix(flow): cool down failed schema rebuilds
Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
* fix(flow): explain legacy aggregate recreation requirements
Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
* fix(flow): adapt stateless provider downcast to DataFusion 55
Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
---------
Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
176 lines
4.4 KiB
SQL
176 lines
4.4 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");
|
|
|
|
-- 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");
|
|
|
|
-- 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");
|
|
|
|
-- 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");
|
|
|
|
-- 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;
|