Understanding Window Table-Valued Functions (TVF) in Flink and Their Optimizations

Table-valued functions (TVF) return a table and are common in databases such as Oracle and SQL Server. In Flink 1.13 the community introduced window TVF to replace the old grouped‑window syntax.

Example: the classic 10‑second tumbling window aggregation in Flink SQL before 1.13 uses

SELECT TUMBLE_START(procTime, INTERVAL '10' SECONDS) AS window_start, merchandiseId, COUNT(1) AS sellCount FROM rtdw_dwd.kafka_order_done_log GROUP BY TUMBLE(procTime, INTERVAL '10' SECONDS), merchandiseId;

; after 1.13 it can be rewritten with

SELECT window_start, window_end, merchandiseId, COUNT(1) AS sellCount FROM TABLE(TUMBLE(TABLE rtdw_dwd.kafka_order_done_log, DESCRIPTOR(procTime), INTERVAL '10' SECONDS)) GROUP BY window_start, window_end, merchandiseId;

.

The design originates from a 2019 SIGMOD paper and follows the SQL‑2016 standard; Calcite added support for rolling and sliding window TVFs from version 1.25. Window TVF brings standardisation, easier implementation and extra features such as local‑global aggregation optimisation, distinct‑hotspot optimisation, Top‑N support and GROUPING SETS.

The physical plan transforms the logical TVF call into a StreamPhysicalWindowTableFunction, then into a StreamPhysicalWindowAggregate, and finally into an executable StreamExecWindowAggregate. Rules like ConverterRule: StreamPhysicalWindowTableFunctionRule, ConverterRule: StreamPhysicalWindowAggregateRule and PullUpWindowTableFunctionIntoWindowAggregateRule orchestrate this conversion.

Flink also introduces “sliced windows” to mitigate state explosion of fine‑grained sliding windows. A cumulative window is divided into slices, each slice being a rolling window whose intermediate results are reused. The CumulativeSliceAssigner adds an isIncremental() method to control incremental output.

Key components of the slicing implementation are WindowBuffer, WindowValueState and NamespaceAggsHandleFunction. The processor’s processElement method assigns slices, registers timers and buffers elements, while merge combines slice states back into the first slice.

A small improvement adds a boolean INCREMENTAL flag to the cumulative window TVF, allowing it to emit results only when the aggregation changes, reducing downstream pressure at the cost of extra state access.