# Group streaming rows into time windows Shape / table A variant of [Calculate across rows](https://meridian.online/operations/calculate-across-rows). Assign rows of a stream to fixed, sliding or session windows of event time, so later steps emit one result per window. Page: ## Status Runs as a SQL step. arcform carries it as a `sql:` step today. ## The DuckDB step assign events to fixed, sliding and session windows of event time, closing them on a watermark carried across two runs ```sql -- 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; ``` The statement builds its own data. It needs nothing installed beyond DuckDB itself. ## What other tools call it Meridian's classification; an independent blind rating agreed on 95.7% of 324 rows (variant, version 3). - Duckle: [`Tumbling Window`](https://github.com/slothflowlabs/duckle/blob/f85f8960edfb6efbf4cb89d58a7b32e16df0e867/frontend/src/workflow-ui/palette-data.ts) (read 2026-09-25) - Apache Hop: [`Beam timestamp`](https://hop.apache.org/manual/latest/pipeline/transforms/beamtimestamp.html) (read 2026-09-27) - Apache Hop: [`Beam window`](https://hop.apache.org/manual/latest/pipeline/transforms/beamwindow.html) (read 2026-09-27) - Talend: [`tWindow`](https://help.qlik.com/talend/en-US/components/8.0/processing/processing-component) (read 2026-09-27)