Group streaming rows into time windows

-- events in arrival order; the one that arrives fifth, with event time 00:05, is late
CREATE TEMP TABLE events AS SELECT * FROM (VALUES
  (1, TIMESTAMP '2026-03-01 00:01:00'), (2, TIMESTAMP '2026-03-01 00:04:00'), (3, TIMESTAMP '2026-03-01 00:08:00'),
  (4, TIMESTAMP '2026-03-01 00:12:00'), (5, TIMESTAMP '2026-03-01 00:05:00'), (6, TIMESTAMP '2026-03-01 00:11:00'),
  (7, TIMESTAMP '2026-03-01 00:30:00'), (8, TIMESTAMP '2026-03-01 00:34:00')
) t(arrival, ts);
-- the watermark before an event is the latest event time already seen, less 5 minutes of allowed lateness;
-- an event behind it is late and is left out of every window
CREATE TEMP TABLE on_time AS
SELECT arrival, ts FROM (
  SELECT *, max(ts) OVER (ORDER BY arrival ROWS BETWEEN UNBOUNDED PRECEDING AND 1 PRECEDING) - INTERVAL 5 MINUTE AS watermark FROM events)
WHERE watermark IS NULL OR ts >= watermark;
-- fixed (tumbling) windows of 10 minutes
CREATE TEMP TABLE fixed AS
SELECT time_bucket(INTERVAL 10 MINUTE, ts) AS window_start, count(*) AS events FROM on_time GROUP BY ALL;
-- sliding windows of 10 minutes every 5: an event belongs to the two windows that cover it
CREATE TEMP TABLE sliding AS
SELECT time_bucket(INTERVAL 5 MINUTE, ts) - k * INTERVAL 5 MINUTE AS window_start, count(*) AS events
FROM on_time, range(2) r(k) GROUP BY ALL;
-- session windows: a gap of more than 5 minutes between events closes the session
CREATE TEMP TABLE sessions AS
SELECT min(ts) AS window_start, max(ts) AS window_end, count(*) AS events
FROM (SELECT *, sum(starts_session) OVER (ORDER BY ts) AS session_id
      FROM (SELECT *, (lag(ts) OVER (ORDER BY ts) IS NULL OR ts - lag(ts) OVER (ORDER BY ts) > INTERVAL 5 MINUTE)::INT AS starts_session FROM on_time))
GROUP BY session_id;
-- two runs over one stream. State kept between them: the events in windows still open, and the watermark.
-- A window is emitted by the first run whose watermark has reached its end; an arrival behind the stored watermark is dropped.
CREATE TEMP TABLE state_events (ts TIMESTAMP);
CREATE TEMP TABLE state_watermark (watermark TIMESTAMP);
CREATE TEMP TABLE emitted (run INTEGER, window_start TIMESTAMP, events BIGINT);
CREATE TEMP TABLE run1_arrivals AS SELECT * FROM (VALUES
  (TIMESTAMP '2026-03-01 00:01:00'), (TIMESTAMP '2026-03-01 00:04:00'), (TIMESTAMP '2026-03-01 00:08:00'),
  (TIMESTAMP '2026-03-01 00:12:00'), (TIMESTAMP '2026-03-01 00:17:00')) t(ts);
CREATE TEMP TABLE run2_arrivals AS SELECT * FROM (VALUES
  (TIMESTAMP '2026-03-01 00:05:00'), (TIMESTAMP '2026-03-01 00:11:00'), (TIMESTAMP '2026-03-01 00:13:00'),
  (TIMESTAMP '2026-03-01 00:30:00'), (TIMESTAMP '2026-03-01 00:34:00')) t(ts);
-- run 1: nothing is stored yet, so nothing is late
INSERT INTO state_events SELECT ts FROM run1_arrivals;
INSERT INTO state_watermark SELECT max(ts) - INTERVAL 5 MINUTE FROM state_events;
INSERT INTO emitted
SELECT 1, time_bucket(INTERVAL 10 MINUTE, ts), count(*) FROM state_events
WHERE time_bucket(INTERVAL 10 MINUTE, ts) + INTERVAL 10 MINUTE <= (SELECT watermark FROM state_watermark) GROUP BY ALL;
DELETE FROM state_events WHERE time_bucket(INTERVAL 10 MINUTE, ts) + INTERVAL 10 MINUTE <= (SELECT watermark FROM state_watermark);
CREATE TEMP TABLE open_after_run1 AS SELECT ts FROM state_events;
-- run 2: read the stored watermark, drop the arrivals behind it, then move the watermark on and emit what has closed
CREATE TEMP TABLE run2_late AS SELECT ts FROM run2_arrivals WHERE ts < (SELECT watermark FROM state_watermark);
INSERT INTO state_events SELECT ts FROM run2_arrivals WHERE ts >= (SELECT watermark FROM state_watermark);
UPDATE state_watermark SET watermark = greatest(watermark, (SELECT max(ts) FROM state_events) - INTERVAL 5 MINUTE);
INSERT INTO emitted
SELECT 2, time_bucket(INTERVAL 10 MINUTE, ts), count(*) FROM state_events
WHERE time_bucket(INTERVAL 10 MINUTE, ts) + INTERVAL 10 MINUTE <= (SELECT watermark FROM state_watermark) GROUP BY ALL;
DELETE FROM state_events WHERE time_bucket(INTERVAL 10 MINUTE, ts) + INTERVAL 10 MINUTE <= (SELECT watermark FROM state_watermark);
SELECT * FROM fixed ORDER BY window_start;
SELECT * FROM sliding ORDER BY window_start;
SELECT * FROM sessions ORDER BY window_start;
SELECT * FROM emitted ORDER BY run, window_start;
SELECT CASE WHEN ((SELECT count(*) FROM on_time) = 7 AND (SELECT list(arrival ORDER BY arrival) FROM events WHERE arrival NOT IN (SELECT arrival FROM on_time)) = [5] AND (SELECT list((strftime(window_start, '%H:%M'), events) ORDER BY window_start) FROM fixed) = [('00:00', 3), ('00:10', 2), ('00:30', 2)] AND (SELECT list((strftime(window_start, '%H:%M'), events) ORDER BY window_start) FROM sliding) = [('23:55', 2), ('00:00', 3), ('00:05', 3), ('00:10', 2), ('00:25', 2), ('00:30', 2)] AND (SELECT list((strftime(window_start, '%H:%M'), strftime(window_end, '%H:%M'), events) ORDER BY window_start) FROM sessions) = [('00:01', '00:12', 5), ('00:30', '00:34', 2)] AND (SELECT list((run, strftime(window_start, '%H:%M'), events) ORDER BY run, window_start) FROM emitted) = [(1, '00:00', 3), (2, '00:10', 3)] AND (SELECT list(strftime(ts, '%H:%M') ORDER BY ts) FROM open_after_run1) = ['00:12', '00:17'] AND (SELECT list(strftime(ts, '%H:%M') ORDER BY ts) FROM run2_late) = ['00:05', '00:11'] AND (SELECT strftime(watermark, '%H:%M') FROM state_watermark) = '00:29' AND (SELECT list(strftime(ts, '%H:%M') ORDER BY ts) FROM state_events) = ['00:30', '00:34']) THEN 'as expected' ELSE error('the fixed, sliding and session windows of the on-time events were not the expected counts, or the late event was not left out, or the two runs did not emit window 00:00 then window 00:10 with the watermark carried over') END AS verdict;