[FLINK-40190][python] Add PyFlink DataFrame creation and conversion APIs by auroflow · Pull Request #28934 · apache/flink

auroflow · 2026-08-06T08:50:27Z

What is the purpose of the change

This pull request completes the data creation and conversion APIs for the PyFlink DataFrame API described in FLINK-40190.

It adds creation from pandas DataFrames, PyArrow tables, PyFlink Tables, dictionaries, records, and integer ranges. It also supports conversion back to Table and pandas, explicit column names, and event-time watermarks where applicable.

Brief change log

Add from_pandas(), from_arrow(), from_table(), and range(), and complete the existing from_dict() and from_records() creators.
Add DataFrame.to_table() and DataFrame.to_pandas().

Verifying this change

This change added tests and can be verified as follows:

Added plain unit tests for schema and watermark validation, record normalization, range behavior, Arrow batch splitting, and empty Arrow streams.
Added planner-backed tests for pandas and Arrow schema inference, positional renaming, typed empty inputs, watermark metadata, and Table identity.
Added wrapper tests for to_table() and to_pandas().
Added an integration smoke test covering from_arrow() → DataFrame transformation → to_pandas().

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: yes — Python in-memory and Arrow source serialization only; no state or persisted serialization format changes
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? PyDocs and public API docstrings

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

Yes (please specify the tool below)

Generated-by: Codex (GPT-5)

Add pandas, Arrow, Table, and range creators with strict schema and watermark validation. Add DataFrame conversion wrappers and schema-aware in-memory and Arrow source paths. Generated-by: Codex (GPT-5)

Align timezone-aware Arrow timestamps with existing Table API semantics, delegate pandas creation to the Arrow path, and add split-aware Arrow IPC serialization. Refine watermark and row helpers with focused tests. Generated-by: Codex (GPT-5)

Combine inferred-schema row conversion with DataFrame creation and let range use its known BIGINT schema directly. Generated-by: Codex (GPT-5)

flinkbot · 2026-08-06T09:05:30Z

CI report:

3c212a2 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

Reject ranges whose emitted values exceed signed BIGINT bounds before creating the underlying table. Cover valid boundary values and ascending and descending overflow cases. Generated-by: Codex (GPT-5)

Clarify that to_table does not execute a job, simplify the DataFrame creation and results reference pages, and exercise from_pandas through filtering, projection, and conversion in the existing pandas round-trip integration smoke test. Generated-by: Codex (GPT-5)

Inline the single-use Arrow IPC writer into TableEnvironment._from_arrow and remove its helper-specific tests. This also avoids the NamedTemporaryFile wrapper type mismatch reported by mypy. Generated-by: Codex (GPT-5)

MattBelle · 2026-08-10T15:52:11Z

+    return column_names
+
+
+def _parse_watermark(


Can we move this under the _WaterMarkSpec class? Like:

class Watermark: @classmethod def parse(cls, ....): .... return cls(....)

Agreed. I moved it under _WaterMarkSpec.

MattBelle · 2026-08-10T15:54:09Z

+    return _WatermarkSpec(*watermark)
+
+
+def _normalize_watermark_row_type(row_type: RowType, watermark: _WatermarkSpec) -> RowType:


Can we move this under the _WaterMarkSpec class? Like:

class Watermark: def normalize_row_type(self, row_type: RowType) -> RowType: ...

Moved under _WaterMarkSpec.

MattBelle · 2026-08-10T16:29:18Z

+    table_schema = (
+        Schema.new_builder()
+        .from_row_data_type(row_type)
+        .watermark(watermark.column, watermark.expression)


since the watermark is a NamedTuple, you can simplify the call by unpacking it like:

schema.watermark(*watermark)

I feel like this both reads better and is more concise.

Agreed, I changed this to unpacking.

MattBelle · 2026-08-10T16:35:27Z

+    return RowType(fields, row_type._nullable)
+
+
+def _resolve_watermark_schema(


This helper function feels shallow and with only 2 uses a little unecessary. Thoughts on inlining the logic at the call sites like:

if watermark: row_type = watermark.normalize_row_type(row_type) table_schema = ... else: table_schema = None

this feels a easier to follow to me

I inlined this helper function.

MattBelle · 2026-08-10T16:53:40Z

since _WatermarkSpec is a NamedTuple you should add a quick test to verify unpacking order. This would help guard against errors caused by someone changing the internal field order or values. In particular, this would enable us to be confident that:

schema.watermark(*watermark)

won't quietly break in the future.

I added a test here to verify unpacking order.

Encapsulate watermark parsing and row-type normalization, inline schema construction, and verify watermark unpacking order. Generated-by: Codex (GPT-5)

dianfu · 2026-08-12T08:27:48Z

+    names = _resolve_column_names(table.column_names, schema)
+    row_type = RowType(
+        [
+            RowField(name, from_arrow_type(field.type, field.nullable))


from_arrow_type() maps timezone-bearing Arrow timestamps to TIMESTAMP, and _from_arrow then casts away the timezone. With Asia/Shanghai, 2026-01-01 00:00+08:00 round-trips as naïve 2025-12-31 16:00, which also changes downstream event-time behavior.

dianfu · 2026-08-12T08:32:30Z

+    import pyarrow as pa
+
+    return from_arrow(
+        pa.Table.from_pandas(pdf, preserve_index=False),


pa.Table.from_pandas rejects duplicate pandas column names before from_arrow can apply the supplied positional schema. Consequently a DataFrame with columns ['x', 'x'] cannot be converted even with schema=['left', 'right'], contrary to the positional-renaming contract. Assign temporary unique names or convert columns by position before constructing the Arrow table.

dianfu · 2026-08-12T08:41:31Z

+            .watermark(*watermark_spec)
+            .build()
+        )
+    result = get_or_create_table_environment()._from_arrow(


Arrow null fields become Flink NullType here, producing a DataFrame whose schema resolves but whose execution always fails with FlinkTypeFactory's 'null type is reserved' ValidationException. This occurs for common inputs such as pd.DataFrame({'x': [None]}), and the name-only schema offers no escape hatch. Reject these fields during creation or support an explicit typed schema.

dianfu · 2026-08-12T08:45:30Z

        raise TypeError("data must be a mapping")
    if not data:
        raise ValueError("data must not be empty")
+    watermark_spec = _WatermarkSpec.parse(watermark)


Add an example about watermark usage.

auroflow added 3 commits August 6, 2026 17:04

[FLINK-40190][python] Add DataFrame creation and conversion APIs

e55ecd5

Add pandas, Arrow, Table, and range creators with strict schema and watermark validation. Add DataFrame conversion wrappers and schema-aware in-memory and Arrow source paths. Generated-by: Codex (GPT-5)

[FLINK-40190][python] Simplify DataFrame row creation

03c6eb8

Combine inferred-schema row conversion with DataFrame creation and let range use its known BIGINT schema directly. Generated-by: Codex (GPT-5)

[FLINK-40190][python] Validate DataFrame range BIGINT bounds

654b9ca

Reject ranges whose emitted values exceed signed BIGINT bounds before creating the underlying table. Cover valid boundary values and ascending and descending overflow cases. Generated-by: Codex (GPT-5)

auroflow force-pushed the codex/flink-40190-dataframe-conversion branch from f53446b to 654b9ca Compare August 6, 2026 09:40

auroflow force-pushed the codex/flink-40190-dataframe-conversion branch from 66e5281 to e693442 Compare August 7, 2026 03:02

auroflow marked this pull request as ready for review August 7, 2026 03:09

[FLINK-40190][python] Inline Arrow table serialization

9208ecc

Inline the single-use Arrow IPC writer into TableEnvironment._from_arrow and remove its helper-specific tests. This also avoids the NamedTemporaryFile wrapper type mismatch reported by mypy. Generated-by: Codex (GPT-5)

MattBelle reviewed Aug 10, 2026

View reviewed changes

github-actions Bot added the community-reviewed PR has been reviewed by the community. label Aug 11, 2026

[FLINK-40190][python] Refine watermark handling

3c212a2

Encapsulate watermark parsing and row-type normalization, inline schema construction, and verify watermark unpacking order. Generated-by: Codex (GPT-5)

dianfu reviewed Aug 12, 2026

View reviewed changes

		return _WatermarkSpec(*watermark)


		def _normalize_watermark_row_type(row_type: RowType, watermark: _WatermarkSpec) -> RowType:

		return RowType(fields, row_type._nullable)


		def _resolve_watermark_schema(

Uh oh!

Conversation

auroflow commented Aug 6, 2026

What is the purpose of the change

Brief change log

Verifying this change

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

Documentation

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

Uh oh!

flinkbot commented Aug 6, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

flinkbot commented Aug 6, 2026 •

edited

Loading