-- 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