mirror of
https://github.com/GreptimeTeam/greptimedb.git
synced 2026-09-09 15:02:19 +00:00
discord9
ad7b0ace64
* feat(flow): support eval schedule offsets Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * fix(flow): remove redundant schedule assertion Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * refactor(flow): trim eval offset compatibility scope Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * test(flow): trim eval offset edge coverage Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * docs(flow): trim eval offset comment noise Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * fix(flow): address eval offset review feedback Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * test(compat): cover Flow eval offset persistence Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> --------- Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
111 lines
4.9 KiB
Plaintext
111 lines
4.9 KiB
Plaintext
-- EVAL OFFSET: fixed UTC epoch phase (offset + k * interval), deterministic.
|
|
CREATE TABLE eval_offset_input (
|
|
ts TIMESTAMP(3) TIME INDEX,
|
|
series STRING,
|
|
v DOUBLE,
|
|
PRIMARY KEY(series)
|
|
);
|
|
|
|
Affected Rows: 0
|
|
|
|
-- Scheduled at odd seconds (1 + 2k): phase anchored to the Unix epoch.
|
|
CREATE FLOW eval_offset_phase_flow
|
|
SINK TO eval_offset_phase_sink
|
|
EVAL INTERVAL '2s'
|
|
EVAL OFFSET '1s'
|
|
AS
|
|
SELECT date_trunc('second', now()) AS ts, count(v) AS c
|
|
FROM eval_offset_input
|
|
GROUP BY date_trunc('second', now());
|
|
|
|
Affected Rows: 0
|
|
|
|
SHOW CREATE FLOW eval_offset_phase_flow;
|
|
|
|
+------------------------+------------------------------------------------------------------------------------------------------------------------+
|
|
| Flow | Create Flow |
|
|
+------------------------+------------------------------------------------------------------------------------------------------------------------+
|
|
| eval_offset_phase_flow | CREATE FLOW IF NOT EXISTS eval_offset_phase_flow |
|
|
| | SINK TO public.eval_offset_phase_sink |
|
|
| | EVAL INTERVAL '2 s' |
|
|
| | EVAL OFFSET '1 s' |
|
|
| | AS SELECT date_trunc('second', now()) AS ts, count(v) AS c FROM eval_offset_input GROUP BY date_trunc('second', now()) |
|
|
+------------------------+------------------------------------------------------------------------------------------------------------------------+
|
|
|
|
SELECT flow_definition FROM information_schema.flows WHERE flow_name = 'eval_offset_phase_flow';
|
|
|
|
+------------------------------------------------------------------------------------------------------------------------+
|
|
| flow_definition |
|
|
+------------------------------------------------------------------------------------------------------------------------+
|
|
| CREATE FLOW IF NOT EXISTS eval_offset_phase_flow |
|
|
| SINK TO public.eval_offset_phase_sink |
|
|
| EVAL INTERVAL '2 s' |
|
|
| EVAL OFFSET '1 s' |
|
|
| AS SELECT date_trunc('second', now()) AS ts, count(v) AS c FROM eval_offset_input GROUP BY date_trunc('second', now()) |
|
|
+------------------------------------------------------------------------------------------------------------------------+
|
|
|
|
INSERT INTO eval_offset_input VALUES
|
|
(now(), 'a', 1.0),
|
|
(now(), 'b', 2.0);
|
|
|
|
Affected Rows: 2
|
|
|
|
-- SQLNESS SLEEP 5s
|
|
-- Prove the offset phase: every row the offset flow wrote has an odd
|
|
-- second-of-minute (1 + 2k is always odd). This is deterministic regardless
|
|
-- of the exact wall-clock alignment of the evaluations.
|
|
SELECT
|
|
count(*) > 0 AS offset_flow_ran,
|
|
bool_and(date_part('second', ts)::BIGINT % 2 = 1) AS offset_flow_at_odd_seconds
|
|
FROM eval_offset_phase_sink;
|
|
|
|
+-----------------+----------------------------+
|
|
| offset_flow_ran | offset_flow_at_odd_seconds |
|
|
+-----------------+----------------------------+
|
|
| true | true |
|
|
+-----------------+----------------------------+
|
|
|
|
CREATE FLOW invalid_eval_offset_no_interval
|
|
SINK TO invalid_eval_offset_sink
|
|
EVAL OFFSET '1s'
|
|
AS SELECT 1;
|
|
|
|
Error: 1004(InvalidArguments), Invalid interval provided: EVAL OFFSET requires EVAL INTERVAL to be specified
|
|
|
|
CREATE FLOW invalid_eval_offset_too_large
|
|
SINK TO invalid_eval_offset_sink
|
|
EVAL INTERVAL '1s'
|
|
EVAL OFFSET '2s'
|
|
AS SELECT 1;
|
|
|
|
Error: 1004(InvalidArguments), Invalid interval provided: EVAL OFFSET must be in range [0, EVAL INTERVAL), got 2 seconds with EVAL INTERVAL 1 seconds
|
|
|
|
CREATE FLOW invalid_eval_offset_fractional
|
|
SINK TO invalid_eval_offset_sink
|
|
EVAL INTERVAL '1s'
|
|
EVAL OFFSET '0.5s'
|
|
AS SELECT 1;
|
|
|
|
Error: 1004(InvalidArguments), Invalid interval provided: EVAL OFFSET must be a whole number of seconds
|
|
|
|
CREATE FLOW invalid_eval_offset_fractional_interval
|
|
SINK TO invalid_eval_offset_sink
|
|
EVAL INTERVAL '1.5s'
|
|
EVAL OFFSET '1s'
|
|
AS SELECT 1;
|
|
|
|
Error: 1004(InvalidArguments), Invalid interval provided: EVAL INTERVAL must be a whole number of seconds when EVAL OFFSET is specified
|
|
|
|
DROP FLOW eval_offset_phase_flow;
|
|
|
|
Affected Rows: 0
|
|
|
|
DROP TABLE eval_offset_phase_sink;
|
|
|
|
Affected Rows: 0
|
|
|
|
DROP TABLE eval_offset_input;
|
|
|
|
Affected Rows: 0
|
|
|