[FLINK-40186][python] Introduce the PyFlink DataFrame API by auroflow · Pull Request #28797 · apache/flink

auroflow · 2026-07-22T05:21:07Z

What is the purpose of the change

This pull request introduces the foundation for a Pythonic DataFrame API in PyFlink.

Brief change log

Add the pyflink.dataframe package and its public evolving interfaces.
Add DataFrame creation from column-oriented dictionaries and row-oriented records, including mappings, named tuples, and general sequences.
Add column selection, filtering, column addition or replacement, item access, and result collection.
Add column and literal expressions, including explicitly typed and null literals.
Add configurable TableEnvironment lifecycle functions.
Add schema, record, expression, and environment validation.
Include the DataFrame package in packaging and static type checking.
Add unit and end-to-end tests for the introduced functionality.

Verifying this change

This change added tests and can be verified as follows:

Added tests for DataFrame creation, schema inference and validation, mappings, named tuples, and general sequence records.
Added tests for selection, filtering, callable expressions, column replacement, item access, typed literals, and result collection.
Added tests for TableEnvironment configuration and lifecycle handling.

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? PyDocs

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

Yes (please specify the tool below)

Generated-by: OpenAI Codex (GPT-5.6 Sol)

flinkbot · 2026-07-22T05:30:20Z

CI report:

07d0215 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

dianfu · 2026-07-23T08:29:37Z

+
+        self.assertEqual(result.collect(), [Row(1, "Alice", 31)])
+
+    def test_from_dict_respects_schema_order_and_subset(self):


It has introduced about 30 it cases. Please check if we could merge them. Besides, it's also worth to check if it's possible to convert part of them into unit tests.

I've converted most of them to unit tests that check the resulting DataFrame schema. I kept two IT cases to test the results of basic DataFrame workflows.

dianfu · 2026-07-23T08:37:20Z

+                        )
+            field_rows.append(tuple(field_values[name] for name in columns))
+        rows = field_rows
+    elif isinstance(first_record, Sequence):


The isinstance(record, Sequence) check also accepts str, bytes, and similar scalar values. For example: from_records(["ab", "cd"], schema=["x", "y"]) produces rows ("a", "b") and ("c", "d") which is not expected.

Makes sense! This could become a misuse pattern. I added validation for str, bytes, bytearray, and memoryview.

dianfu · 2026-07-23T08:40:18Z

+
+    .. versionadded:: 2.4.0
+    """
+    if not isinstance(data, Mapping):


The column validation also accepts strings because they implement Sequence:
from_dict({"name": "Bob"}) would create three rows containing "B", "o", and "b".

Added validation

dianfu · 2026-07-23T09:00:36Z

+
+
+@PublicEvolving()
+def set_table_environment(t_env: Optional["StreamTableEnvironment"]) -> None:


validate the type of the given t_env

dianfu · 2026-07-23T09:04:46Z

+                for name in columns:
+                    if name not in field_values:
+                        raise ValueError(
+                            "schema field %r is not present in records" % name


Since validation is performed per record, the message should identify the failing index consistently.

I included the failing index in the error message.

MattBelle · 2026-07-22T18:59:09Z

+__all__ = ["from_dict", "from_records"]
+
+
+class _SequenceRowIterable:


Why are we creating custom iterables? This class looks like it could be replaced with a simple:

rows = (tuple(record) for record in sequence_rows)

on line 178

I intended to use them to make the rows reusable, because the Table API layer iterates the input twice, first during schema inference, then during conversion. A bare generator expression is exhausted by the first pass and produces an empty table.

But I do agree the custom iterables are rather confusing. I changed both with list comprehensions, because these conversion methods are mainly for local debugging with small inputs, so we could afford some extra allocations in favor of readability here. WDYT?

Yeah I'm completely fine with list comprehensions. I assume the bulk of memory will be in the data objects themselves and since we aren't copying those this should just increase the memory consumption by a paltry 8 bytes * len(rows). Thanks for the extra context!

MattBelle · 2026-07-22T19:00:24Z

+        return (tuple(record) for record in self._records)
+
+
+class _ColumnRowIterable:


Same comment as _SequenceRowIterable except this class also doesn't appear to be used at all

It was used by from_dict. I've changed it to a list comprehension.

Looks good!

MattBelle · 2026-07-22T19:13:48Z

+
+    first_record = data[0]
+    rows: Iterable[Sequence[Any]]
+    named_tuple_records = _is_named_tuple_record(first_record)


nit: can we change the name from named_tuple_records to is_named_tuple_record? The current naming implies a collection of tuple records rather than a boolean

I removed it as part of the enum refactor.

Looks good!

MattBelle · 2026-07-22T19:24:35Z

+
+
+def _to_field_mapping(
+    record: Any, named_tuple_records: bool


Is the purpose of the named_tuple_records boolean enforcement of standardization? If so I feel like a name change would help clarify things. Something like:

def _to_field_mapping(record: Any, enforce_named_tuple: bool): ...

without seeing usage this kinda reads like redundant operations as is.

Yes, that was the intent, I changed the name according to your other comment.

Looks good!

MattBelle · 2026-07-22T20:24:10Z

+
+
+def _to_field_mapping(
+    record: Any, named_tuple_records: bool


I feel like the type of record could be described in a more readable fashion with an enum:

from enum import Enum class RecordType(Enum) NamedTuple = 'named_tuple' Mapping = 'mapping' Sequence = 'sequence' @classmethod def from_record(cls, record: Any) -> 'RecordType': if isinstance(record, tuple) and isinstance(getattr(record, "_fields", None), tuple): return cls.NamedTuple elif isinstance(record, Mapping): return cls.Mapping elif isinstance(record, Sequence): return cls.Sequence else: raise TypeError("records must be mappings or sequences")

this makes the supported types more obvious and encourages variable names that I think would better encompass intent.

Current usage could look like:

def from_records( data: Sequence[Union[Sequence[Any], Mapping[str, Any]]], schema: Optional[List[str]] = None, ) -> DataFrame: ... expected_record_type = RecordType.from_record(first_record) ... if expected_record_type in [RecordType.Mapping, RecordType.NamedTuple]: first_fields = _to_field_mapping(first_record, expected_record_type) ... else: # We know it must be a RecordType.Sequence since we are enforcing this on creation ...

def _to_field_mapping(record: Any, expected_type: RecordType) -> Mapping[str, Any]: record_type = RecordType.from_record(record) if expected_type == RecordType.NamedTuple: if record_type != expected_type: raise TypeError("records must all be named tuples") ... elif record_type != RecordType.Mapping: raise TypeError("records must all be mappings") # We should probably handle a sequence type here for completeness return cast(Mapping[str, Any], record)

looking forward this would also work really well with structural pattern matching once the lib drops support for python 3.9:

match RecordType.from_record(record): case RecordType.NamedTuple | RecordType.Mapping: # Do something case RecordType.Sequence: # Do something else case _: raise TypeError('unsupported record type')

Good idea! This is much clearer. I've refactored this part.

Looks good!

MattBelle · 2026-07-23T13:46:50Z

+        field_rows: List[Sequence[Any]] = []
+        for index, record in enumerate(data):
+            field_values = _to_field_mapping(record, named_tuple_records)
+            if schema is None and set(field_values.keys()) != expected_fields:


we have already validated that columns has unique fields. Do we expect the fields to have identical orderings? If so I think we can get rid of expected_fields and just do:

list(field_values.keys()) != columns

Yes, I was hoping to support non-identical orderings, so I'd prefer keeping it.

Fair enough. I'm fine with this then.

MattBelle · 2026-07-23T15:40:34Z

+    from pyflink.table import StreamTableEnvironment
+
+
+_global_table_environment: Optional["StreamTableEnvironment"] = None


Might be slightly out of scope, but thoughts on making this a ContextVar? It wouldn't make much difference with the current planned API, but with the addition of a function like:

from contextlib import contextmanager @contextmanager def table_environment(t_env): global _global_table_environment with _global_table_environment.set(t_env): yield _global_table_environment.get()

users would be able to selectively change their environments without worrying about mutating global state:

my_environment = ... set_table_environment(my_environment) assert get_table_environment() == my_environment # Start a temp environment temp_environment = ... with table_environment(temp_environment): assert get_table_environment() == temp_environment assert get_table_environment() == my_environment

this would better support async or multithreaded environments

Maybe we could do it in another PR? In my understanding, the global context management here is intended to abstract the concept of Flink's table environment away to make the DataFrame API cleaner, since most users use one single environment in a Flink job. We could do this once we find concrete multi-environment use cases

Fair. Just figured I'd throw out the suggestion for consideration.

Generated-by: OpenAI Codex (GPT-5.6 Sol)

Generated-by: OpenAI Codex (GPT-5)

MattBelle · 2026-07-24T13:09:38Z

+        raise ValueError("schema field names must be unique")
+
+
+def _validate_record_type(


this feels like it should live under the _RecordType enum like:

class _RecordType(Enum): ... def validate(self, record: Any): try: record_type = _RecordType.from_record(record) except TypeError: record_type = None if record_type is self: return # Treat named tuples as tuples when validating sequence records. if ( self is _RecordType.SEQUENCE and record_type is _RecordType.NAMED_TUPLE ): return raise TypeError(f'record must be a {self.value.replace("_", " ")}')

doing this completely eliminates the error class of "unsupported expected record type"

MattBelle · 2026-07-24T13:26:59Z

+    ):
+        return
+    if expected_record_type is _RecordType.NAMED_TUPLE:
+        raise TypeError("record at index %d must be a named tuple" % index)


make these f-strings. They are faster and easier to read than printf style formatting. printf style format strings are only really useful in log messages when you want to lazily interpolate strings on levels that may not be relevant at runtime

MattBelle · 2026-07-24T13:31:37Z

+            record, _SCALAR_SEQUENCE_TYPES
+        ):
+            return cls.SEQUENCE
+        raise TypeError


I feel like you should give this an error message like:

raise TypeError( "records must be a mapping or a sequence of values, " "such as a list or tuple" )

MattBelle · 2026-07-24T13:43:46Z

+    try:
+        expected_record_type = _RecordType.from_record(first_record)
+    except TypeError:
+        raise TypeError(


it'd simplify code complexity if you stacked the error messages. Assuming you took my above suggestion, you could change this to:

try: expected_record_type = _RecordType.from_record(first_record) except TypeError error: raise TypeError('invalid record at index 0') from error

and the users would get the same information, but without you needing to pass around the index everywhere.

This strategy could also be applied to the validation method I suggested like:

for index, record in enumerate(data): try: expected_record.validate(record) except TypeError as error: raise TypeError(f'invalid record at index {index}') from error

This should read better, result in more uniform error messages, and be less annoying to maintain in the future.

MattBelle · 2026-07-24T13:46:18Z

+            _validate_record_type(record, expected_record_type, index)
+            if len(record) != len(columns):
+                raise ValueError(
+                    "record at index %d has %d values but schema has %d fields"


use f-strings for exception messages

MattBelle · 2026-07-24T14:01:39Z

+
+    first_record = data[0]
+    rows: Iterable[Sequence[Any]]
+    named_tuple_records = _is_named_tuple_record(first_record)


Looks good!

MattBelle · 2026-07-24T14:03:43Z

+
+
+def _to_field_mapping(
+    record: Any, named_tuple_records: bool


Looks good!

MattBelle · 2026-07-24T14:08:24Z

+
+
+def _to_field_mapping(
+    record: Any, named_tuple_records: bool


Looks good!

MattBelle · 2026-07-24T14:10:19Z

+        field_rows: List[Sequence[Any]] = []
+        for index, record in enumerate(data):
+            field_values = _to_field_mapping(record, named_tuple_records)
+            if schema is None and set(field_values.keys()) != expected_fields:


Fair enough. I'm fine with this then.

MattBelle · 2026-07-24T14:13:17Z

+    from pyflink.table import StreamTableEnvironment
+
+
+_global_table_environment: Optional["StreamTableEnvironment"] = None


Fair. Just figured I'd throw out the suggestion for consideration.

auroflow added 4 commits July 22, 2026 11:56

[FLINK-40186][python] Introduce PyFlink DataFrame API

a12eb91

Generated-by: OpenAI Codex (GPT-5.6 Sol)

[FLINK-40186][python] Harden PyFlink DataFrame validation

365f659

Generated-by: OpenAI Codex (GPT-5.6 Sol)

[FLINK-40186][python] Align PyFlink DataFrame behavior

6b85a02

Generated-by: OpenAI Codex (GPT-5.6 Sol)

[FLINK-40186][python] Document and test PyFlink DataFrame APIs

9684ffc

Generated-by: OpenAI Codex (GPT-5.6 Sol)

dianfu changed the title ~~[FLINK-40185][python] Introduce the PyFlink DataFrame API~~ [FLINK-40186][python] Introduce the PyFlink DataFrame API Jul 22, 2026

auroflow force-pushed the auroflow/flink-40185-dataframe branch from 1b7cd38 to 9684ffc Compare July 22, 2026 05:41

dianfu reviewed Jul 23, 2026

View reviewed changes

MattBelle reviewed Jul 23, 2026

View reviewed changes

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

auroflow and others added 3 commits July 24, 2026 14:11

[FLINK-40186][python] Address comments

5bd7031

Generated-by: OpenAI Codex (GPT-5.6 Sol)

[FLINK-40186][python] Simplify DataFrame row conversion

51b25d9

Generated-by: OpenAI Codex (GPT-5)

[FLINK-40186][python] Clarify DataFrame record type handling

07d0215

Generated-by: OpenAI Codex (GPT-5)

MattBelle reviewed Jul 24, 2026

View reviewed changes


		self.assertEqual(result.collect(), [Row(1, "Alice", 31)])

		def test_from_dict_respects_schema_order_and_subset(self):



		@PublicEvolving()
		def set_table_environment(t_env: Optional["StreamTableEnvironment"]) -> None:

		__all__ = ["from_dict", "from_records"]


		class _SequenceRowIterable:

		return (tuple(record) for record in self._records)


		class _ColumnRowIterable:



		def _to_field_mapping(
		record: Any, named_tuple_records: bool

		from pyflink.table import StreamTableEnvironment


		_global_table_environment: Optional["StreamTableEnvironment"] = None

		raise ValueError("schema field names must be unique")


		def _validate_record_type(

Uh oh!

Conversation

auroflow commented Jul 22, 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 Jul 22, 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!

dianfu Jul 23, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

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!

auroflow Jul 24, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

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!

auroflow Jul 24, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

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!

flinkbot commented Jul 22, 2026 •

edited

Loading

dianfu Jul 23, 2026 •

edited

Loading

auroflow Jul 24, 2026 •

edited

Loading

auroflow Jul 24, 2026 •

edited

Loading