Flink SQL Joins & Windows

## What a window TVF is A **window table-valued function (TVF)** takes a table and returns a table. In Flink SQL you call it in the `FROM` clause, and it assigns every input row to one or more windows by adding three columns: - `window_start` — the inclusive start of the assigned window. - `window_end` — the exclusive end of the window. - `window_time` — a time attribute equal to `window_end - 1ms`, which later time-based operations (another window TVF, an interval join) can use. The argument wrapped in `DESCRIPTOR(...)` names the column that carries time. In streaming mode it must be a **time attribute** (event time or processing time); after the TVF, that original column becomes an ordinary timestamp. A TVF on its own only labels rows. What turns it into a windowed computation is the operation on top: an aggregation, a Top-N, a join or a deduplication keyed on `window_start` and `window_end`. ## Writing the per-minute count ```sql SELECT window_start, window_end, campaign_id, COUNT(*) AS impressions FROM TUMBLE(TABLE impressions, DESCRIPTOR(impression_time), INTERVAL '1' MINUTE) GROUP BY window_start, window_end, campaign_id; ``` The planner recognises a **window aggregation** because `window_start` and `window_end` both appear in `GROUP BY`. Leave them out and it is no longer a window aggregation: grouping by `campaign_id` alone becomes an ordinary continuous `GROUP BY` that keeps a running count per campaign indefinitely and emits an update for every row. ## What the query emits A window aggregation behaves very differently from a continuous `GROUP BY`: - It emits **one final row per window and key**, not intermediate results. With an event-time attribute that happens when the watermark passes `window_end`. - Its output is **insert-only (append-only)**: a window's row is never retracted or updated, so a sink that accepts only inserts can take it. - It **purges the window's state** once the result is out, so it needs no idle-state TTL to stay bounded. - Flink's tuning guide states that mini-batch buffering is always on for window TVF aggregations, whatever `table.exec.mini-batch.enabled` says. How the watermark is generated, and what happens to rows behind it, is event-time material rather than a property of the TVF. ## The four window TVFs | TVF | Signature (Flink 2.3) | Windows per row | Typical use | |---|---|---|---| | `TUMBLE` | `TUMBLE(TABLE t, DESCRIPTOR(ts), size [, offset])` | exactly one | per-minute counts | | `HOP` | `HOP(TABLE t, DESCRIPTOR(ts), slide, size [, offset])` | several (size / slide) | trailing 10-minute totals every 5 minutes | | `CUMULATE` | `CUMULATE(TABLE t, DESCRIPTOR(ts), step, size)` | several, all sharing one start | since-midnight totals refreshed hourly | | `SESSION` | `SESSION(TABLE t [PARTITION BY k], DESCRIPTOR(ts), gap)` | one session | activity bursts per user | Two details trip people up. `HOP` takes the **slide before the size**, so `HOP(..., INTERVAL '5' MINUTES, INTERVAL '10' MINUTES)` means ten-minute windows starting every five minutes. And `SESSION` keys its sessions with `PARTITION BY` inside the call; without it, all rows share one global sequence of sessions. The session TVF is not supported in batch mode at 2.3. ## Why TVFs replaced the group-window functions Before window TVFs, Flink SQL wrote windows as grouping functions: `GROUP BY TUMBLE(impression_time, INTERVAL '1' MINUTE), campaign_id`, with `TUMBLE_START(...)` in the select list. Flink 2.3 still documents that syntax under a **deprecated** warning and recommends TVFs because: 1. The TVF form follows the SQL standard's polymorphic table functions and puts the window in `FROM`, where it composes like any other relation. 2. Only TVFs support **window Top-N**, **window joins** and **window deduplication**; group-window functions support aggregation and nothing else. 3. TVF aggregations get the optimisations in Flink's performance-tuning guide and support standard `GROUPING SETS`. ## Common mistakes - Grouping by the original time column instead of `window_start, window_end`. - Swapping `HOP`'s slide and size. - Forgetting that `window_end` is exclusive: a row stamped exactly 10:01:00 belongs to the window starting at 10:01, not to [10:00, 10:01). - Using `HOP` with a one-day size when the business wants a since-midnight running total; that is `CUMULATE`. - Pointing `DESCRIPTOR` at a plain `TIMESTAMP` column in streaming mode; it has to be a declared time attribute. - Expecting a window's row to be revised later; a window TVF aggregation emits it once.