[FLINK-40424][table][python] Support windowing TVFs in the DataFrame API by Timm0 · Pull Request #29070 · apache/flink

Timm0 · 2026-09-02T12:36:26Z

What is the purpose of the change

This change adds tumble, hop, cumulate, and session window methods to the DataFrame API and the internal Table API machinery they need. The window operation is built programmatically as a LogicalTableFunctionScan
rather than by generating SQL strings, so the DataFrame layer does not have to track Flink's SQL dialect. No new public Table API is introduced, since all added Table API types are @Internal.

Brief change log

Add tumble/hop/cumulate/session window methods to the PyFlink DataFrame API, with time-column resolution and day-time interval handling
Introduce internal WindowTableFunctionQueryOperation
Dispatch it through QueryOperationVisitor and QueryOperationDefaultVisitor and construct it via OperationTreeBuilder.windowTableFunction
Convert the operation to a LogicalTableFunctionScan programmatically in QueryOperationConverter
Add tests: api-java operation test, planner WindowTableFunctionRelNodeTest, QueryOperation serialization + semantic programs, and PyFlink unit + IT tests
Expose createTable/getOperationTreeBuilder on the Table API test steps and refactor _materialize into test_case_utils for stream test cases
Document the window methods in dataframe.rst

Verifying this change

Added api-java operation test
Added planner WindowTableFunctionRelNodeTest
Added QueryOperation serialization + semantic programs
Added PyFlink unit + IT tests

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): no
The public API, i.e., is any changed class annotated with @Public(Evolving): yes
The serializers: no
The runtime per-record code paths (performance sensitive): no
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
The S3 file system connector: no

Documentation

Does this pull request introduce a new feature? yes
If yes, how is the feature documented? docs flink-python/docs/reference/pyflink.dataframe/dataframe.rst + Python docstrings

Was generative AI tooling used to co-author this PR?

Yes (please specify the tool below)

Generated-by: Opus 4.8 (1M context)

flinkbot · 2026-09-02T12:41:53Z

CI report:

5c485e1 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

snuyanzin · 2026-09-03T16:00:59Z

+
+def _to_interval_expression(value: Union["datetime.timedelta", Expression]) -> Any:
+    if isinstance(value, datetime.timedelta):
+        millis = value // datetime.timedelta(milliseconds=1)


Do we intentionally drop subsecond part in python?

Correct me if I'm wrong but Flink supports ms resolution, while Python timedelta supports sub-ms. Therefore we are dropping any sub-ms here. One thing we could do here though is to throw a ValueError for non-zero sub-ms instead of silently dropping. If you think that this would be the right call here, then I'm happy to add a guard.

snuyanzin · 2026-09-03T16:16:02Z

+        final List<Column> columns = new ArrayList<>(inputSchema.getColumns());
+        columns.add(Column.physical("window_start", DataTypes.TIMESTAMP(3).notNull()));
+        columns.add(Column.physical("window_end", DataTypes.TIMESTAMP(3).notNull()));
+        columns.add(Column.physical("window_time", timeColumnType.notNull()));
+        return ResolvedSchema.of(columns);


It looks like it might lead to validation error if there are already columns with such names

You are right, but this is also current behavior. During conversion I have re-used SqlWindowTableFunction.inferRowType() which is the same path as SQL, and I have done some additional testing here and both paths throw a ValidationException in the planner. I can add an additional guard in SqlWindowTableFunction.inferRowType() if we want to have a unified error message.

- Add `tumble`/`hop`/`cumulate`/`session` window methods to the PyFlink `DataFrame` API, with time-column resolution and day-time interval handling - Introduce internal `WindowTableFunctionQueryOperation` - Dispatch it through `QueryOperationVisitor` and `QueryOperationDefaultVisitor` and construct it via `OperationTreeBuilder.windowTableFunction` - Convert the operation to a `LogicalTableFunctionScan` programmatically in `QueryOperationConverter` - Add tests: api-java operation test, planner `WindowTableFunctionRelNodeTest`, `QueryOperation` serialization + semantic programs, and PyFlink unit + IT tests - Expose `createTable`/`getOperationTreeBuilder` on the Table API test steps and refactor `_materialize` into `test_case_utils` for stream test cases - Document the window methods in `dataframe.rst`

Timm0 force-pushed the FLINK-40424 branch from 5d13295 to ab68303 Compare September 2, 2026 12:41

Timm0 marked this pull request as ready for review September 2, 2026 12:41

Timm0 force-pushed the FLINK-40424 branch from ab68303 to fce27eb Compare September 2, 2026 16:12