This is an automated email from the ASF dual-hosted git repository.
dianfu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 51f54b34c7d [FLINK-40186][python] Introduce the basic functionality of
Python DataFrame API
51f54b34c7d is described below
commit 51f54b34c7dd7f296c0513dbee63431617139e0c
Author: auroflow <[email protected]>
AuthorDate: Tue Jul 21 23:50:43 2026 +0800
[FLINK-40186][python] Introduce the basic functionality of Python DataFrame
API
Generated-by: OpenAI Codex (GPT-5.6 Sol)
This closes #28797.
---
flink-python/dev/integration_test.sh | 3 +
flink-python/docs/reference/index.rst | 1 +
.../{index.rst => pyflink.dataframe/creation.rst} | 28 +-
.../docs/reference/pyflink.dataframe/dataframe.rst | 78 +++
.../{index.rst => pyflink.dataframe/datatype.rst} | 26 +-
.../environment.rst} | 20 +-
.../reference/{ => pyflink.dataframe}/index.rst | 17 +-
flink-python/pyflink/dataframe/__init__.py | 60 +++
flink-python/pyflink/dataframe/context.py | 103 ++++
flink-python/pyflink/dataframe/convert.py | 270 ++++++++++
flink-python/pyflink/dataframe/dataframe.py | 350 +++++++++++++
flink-python/pyflink/dataframe/datatype.py | 85 +++
flink-python/pyflink/dataframe/tests/__init__.py | 17 +
.../pyflink/dataframe/tests/test_context.py | 93 ++++
.../pyflink/dataframe/tests/test_convert.py | 228 ++++++++
.../pyflink/dataframe/tests/test_dataframe.py | 575 +++++++++++++++++++++
.../pyflink/dataframe/tests/test_datatype.py | 65 +++
flink-python/pyflink/testing/test_case_utils.py | 50 ++
flink-python/setup.py | 1 +
flink-python/tox.ini | 2 +-
20 files changed, 2040 insertions(+), 32 deletions(-)
diff --git a/flink-python/dev/integration_test.sh
b/flink-python/dev/integration_test.sh
index c587a79f123..ba7ceb47aea 100755
--- a/flink-python/dev/integration_test.sh
+++ b/flink-python/dev/integration_test.sh
@@ -34,6 +34,9 @@ function test_all_modules() {
# test datastream module
test_module "datastream"
+ # test dataframe module
+ test_module "dataframe"
+
# test fn_execution module
test_module "fn_execution"
diff --git a/flink-python/docs/reference/index.rst
b/flink-python/docs/reference/index.rst
index 9776d64718f..d8915461795 100644
--- a/flink-python/docs/reference/index.rst
+++ b/flink-python/docs/reference/index.rst
@@ -23,6 +23,7 @@ API Reference
.. toctree::
:maxdepth: 2
+ pyflink.dataframe/index
pyflink.table/index
pyflink.datastream/index
pyflink.common/index
diff --git a/flink-python/docs/reference/index.rst
b/flink-python/docs/reference/pyflink.dataframe/creation.rst
similarity index 66%
copy from flink-python/docs/reference/index.rst
copy to flink-python/docs/reference/pyflink.dataframe/creation.rst
index 9776d64718f..51692dab6bc 100644
--- a/flink-python/docs/reference/index.rst
+++ b/flink-python/docs/reference/pyflink.dataframe/creation.rst
@@ -16,13 +16,25 @@
limitations under the License.
################################################################################
-=============
-API Reference
-=============
+==================
+DataFrame Creation
+==================
-.. toctree::
- :maxdepth: 2
+Functions for creating DataFrames from row-oriented or column-oriented Python
data.
- pyflink.table/index
- pyflink.datastream/index
- pyflink.common/index
+Example::
+
+ >>> import pyflink.dataframe as pf
+ >>> users = pf.from_records([
+ ... {"id": 1, "name": "Alice"},
+ ... {"id": 2, "name": "Bob"},
+ ... ])
+ >>> users = pf.from_dict({"id": [1, 2], "name": ["Alice", "Bob"]})
+
+.. currentmodule:: pyflink.dataframe
+
+.. autosummary::
+ :toctree: api/
+
+ from_records
+ from_dict
diff --git a/flink-python/docs/reference/pyflink.dataframe/dataframe.rst
b/flink-python/docs/reference/pyflink.dataframe/dataframe.rst
new file mode 100644
index 00000000000..2caa7503eba
--- /dev/null
+++ b/flink-python/docs/reference/pyflink.dataframe/dataframe.rst
@@ -0,0 +1,78 @@
+..
################################################################################
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+
################################################################################
+
+=========
+DataFrame
+=========
+
+A DataFrame provides a Pythonic interface for composing data transformations.
+Transformation methods return new DataFrames and support fluent chaining.
+
+Example::
+
+ >>> import pyflink.dataframe as pf
+ >>> df = pf.from_dict({"id": [1, 2], "name": ["a", "b"]})
+ >>> result = df.select("id", "name") \
+ ... .with_column("id_doubled", pf.col("id") * 2) \
+ ... .filter(pf.col("id") > 0)
+
+DataFrame
+---------
+
+.. currentmodule:: pyflink.dataframe
+
+.. autosummary::
+ :toctree: api/
+
+ DataFrame
+
+Transformations
+---------------
+
+.. currentmodule:: pyflink.dataframe
+
+.. autosummary::
+ :toctree: api/
+
+ DataFrame.select
+ DataFrame.with_column
+ DataFrame.filter
+ DataFrame.__getitem__
+
+Results
+-------
+
+.. currentmodule:: pyflink.dataframe
+
+.. autosummary::
+ :toctree: api/
+
+ DataFrame.collect
+
+Expressions
+-----------
+
+Functions for constructing column references and literal expressions.
+
+.. currentmodule:: pyflink.dataframe
+
+.. autosummary::
+ :toctree: api/
+
+ col
+ lit
diff --git a/flink-python/docs/reference/index.rst
b/flink-python/docs/reference/pyflink.dataframe/datatype.rst
similarity index 70%
copy from flink-python/docs/reference/index.rst
copy to flink-python/docs/reference/pyflink.dataframe/datatype.rst
index 9776d64718f..9d9b19b5cf5 100644
--- a/flink-python/docs/reference/index.rst
+++ b/flink-python/docs/reference/pyflink.dataframe/datatype.rst
@@ -16,13 +16,23 @@
limitations under the License.
################################################################################
-=============
-API Reference
-=============
+==========
+Data Types
+==========
-.. toctree::
- :maxdepth: 2
+Data types describe the logical types of values used by DataFrame expressions
and operations.
- pyflink.table/index
- pyflink.datastream/index
- pyflink.common/index
+Example::
+
+ >>> import pyflink.dataframe as pf
+ >>> identifier = pf.lit(42, pf.DataType.int64())
+ >>> name = pf.lit("Alice", pf.DataType.string())
+
+.. currentmodule:: pyflink.dataframe
+
+.. autosummary::
+ :toctree: api/
+
+ DataType
+ DataType.int64
+ DataType.string
diff --git a/flink-python/docs/reference/index.rst
b/flink-python/docs/reference/pyflink.dataframe/environment.rst
similarity index 75%
copy from flink-python/docs/reference/index.rst
copy to flink-python/docs/reference/pyflink.dataframe/environment.rst
index 9776d64718f..e0e4d08f925 100644
--- a/flink-python/docs/reference/index.rst
+++ b/flink-python/docs/reference/pyflink.dataframe/environment.rst
@@ -16,13 +16,17 @@
limitations under the License.
################################################################################
-=============
-API Reference
-=============
+=====================
+Execution Environment
+=====================
-.. toctree::
- :maxdepth: 2
+Functions for setting, inspecting, and creating the environment used by
DataFrame operations.
- pyflink.table/index
- pyflink.datastream/index
- pyflink.common/index
+.. currentmodule:: pyflink.dataframe
+
+.. autosummary::
+ :toctree: api/
+
+ set_table_environment
+ get_table_environment
+ get_or_create_table_environment
diff --git a/flink-python/docs/reference/index.rst
b/flink-python/docs/reference/pyflink.dataframe/index.rst
similarity index 83%
copy from flink-python/docs/reference/index.rst
copy to flink-python/docs/reference/pyflink.dataframe/index.rst
index 9776d64718f..765a2e5b884 100644
--- a/flink-python/docs/reference/index.rst
+++ b/flink-python/docs/reference/pyflink.dataframe/index.rst
@@ -16,13 +16,16 @@
limitations under the License.
################################################################################
-=============
-API Reference
-=============
+==================
+PyFlink DataFrame
+==================
+
+This page gives an overview of all public PyFlink DataFrame APIs.
.. toctree::
- :maxdepth: 2
+ :maxdepth: 1
- pyflink.table/index
- pyflink.datastream/index
- pyflink.common/index
+ dataframe
+ creation
+ datatype
+ environment
diff --git a/flink-python/pyflink/dataframe/__init__.py
b/flink-python/pyflink/dataframe/__init__.py
new file mode 100644
index 00000000000..8ad43bcbbe2
--- /dev/null
+++ b/flink-python/pyflink/dataframe/__init__.py
@@ -0,0 +1,60 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+
+"""
+A DataFrame API for PyFlink.
+
+It provides a fluent interface for composing and executing data
transformations.
+
+Example::
+
+ >>> import pyflink.dataframe as pf
+ >>> df = pf.from_records(
+ ... [(1, "Alice", 30), (2, "Bob", 17)],
+ ... schema=["id", "name", "age"],
+ ... )
+ >>> result = (
+ ... df.filter(pf.col("age") >= 18)
+ ... .with_column("age_next_year", pf.col("age") + 1)
+ ... .select("id", "name", "age_next_year")
+ ... )
+ >>> for row in result.collect():
+ ... print(row)
+ <Row(1, 'Alice', 31)>
+"""
+
+from pyflink.dataframe.convert import from_dict, from_records
+from pyflink.dataframe.context import (
+ get_or_create_table_environment,
+ get_table_environment,
+ set_table_environment,
+)
+from pyflink.dataframe.dataframe import DataFrame, col, lit
+from pyflink.dataframe.datatype import DataType
+
+__all__ = [
+ "DataFrame",
+ "DataType",
+ "col",
+ "lit",
+ "from_dict",
+ "from_records",
+ "set_table_environment",
+ "get_table_environment",
+ "get_or_create_table_environment",
+]
diff --git a/flink-python/pyflink/dataframe/context.py
b/flink-python/pyflink/dataframe/context.py
new file mode 100644
index 00000000000..031c2995108
--- /dev/null
+++ b/flink-python/pyflink/dataframe/context.py
@@ -0,0 +1,103 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+
+from typing import Optional
+
+from pyflink.table import StreamTableEnvironment, TableEnvironment
+from pyflink.util.api_stability_decorators import PublicEvolving
+
+__all__ = [
+ "set_table_environment",
+ "get_table_environment",
+ "get_or_create_table_environment",
+]
+
+_global_table_environment: Optional[TableEnvironment] = None
+
+
+@PublicEvolving()
+def set_table_environment(t_env: Optional[TableEnvironment]) -> None:
+ """
+ Set the environment used by DataFrame operations.
+
+ :param t_env: Environment to use, or ``None`` to clear it.
+ :raises TypeError: If ``t_env`` is neither a :class:`TableEnvironment` nor
``None``.
+
+ Example::
+
+ >>> import pyflink.dataframe as pf
+ >>> pf.set_table_environment(None)
+ >>> pf.get_table_environment() is None
+ True
+
+ .. versionadded:: 2.4.0
+ """
+ global _global_table_environment
+ if t_env is not None and not isinstance(t_env, TableEnvironment):
+ raise TypeError("t_env must be a TableEnvironment or None")
+ _global_table_environment = t_env
+
+
+@PublicEvolving()
+def get_table_environment() -> Optional[TableEnvironment]:
+ """
+ Return the environment used by DataFrame operations, if one is configured.
+
+ :return: The configured environment, or ``None``.
+
+ Example::
+
+ >>> import pyflink.dataframe as pf
+ >>> pf.set_table_environment(None)
+ >>> pf.get_table_environment() is None
+ True
+
+ .. versionadded:: 2.4.0
+ """
+ return _global_table_environment
+
+
+@PublicEvolving()
+def get_or_create_table_environment() -> TableEnvironment:
+ """
+ Return the configured environment, creating one when necessary.
+
+ The created environment is retained for subsequent DataFrame operations
and calls to
+ :func:`get_table_environment`.
+
+ :return: The configured or newly created environment.
+
+ Example::
+
+ >>> import pyflink.dataframe as pf
+ >>> pf.set_table_environment(None)
+ >>> environment = pf.get_or_create_table_environment()
+ >>> pf.get_table_environment() is environment
+ True
+
+ .. versionadded:: 2.4.0
+ """
+ global _global_table_environment
+
+ if _global_table_environment is None:
+ from pyflink.datastream import StreamExecutionEnvironment
+
+ stream_environment =
StreamExecutionEnvironment.get_execution_environment()
+ _global_table_environment =
StreamTableEnvironment.create(stream_environment)
+
+ return _global_table_environment
diff --git a/flink-python/pyflink/dataframe/convert.py
b/flink-python/pyflink/dataframe/convert.py
new file mode 100644
index 00000000000..9a78fa65ee3
--- /dev/null
+++ b/flink-python/pyflink/dataframe/convert.py
@@ -0,0 +1,270 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+
+from enum import Enum
+from typing import (
+ Any,
+ Collection,
+ List,
+ Mapping,
+ Optional,
+ Sequence,
+ Tuple,
+ Union,
+ cast,
+)
+
+from pyflink.dataframe.context import get_or_create_table_environment
+from pyflink.dataframe.dataframe import DataFrame
+from pyflink.util.api_stability_decorators import PublicEvolving
+
+__all__ = ["from_dict", "from_records"]
+
+_SCALAR_SEQUENCE_TYPES = (str, bytes, bytearray, memoryview)
+
+
+class _RecordType(Enum):
+ NAMED_TUPLE = "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.NAMED_TUPLE
+ if isinstance(record, Mapping):
+ return cls.MAPPING
+ if isinstance(record, Sequence) and not isinstance(
+ record, _SCALAR_SEQUENCE_TYPES
+ ):
+ return cls.SEQUENCE
+ raise TypeError(
+ "record must be a mapping or a sequence of values, "
+ "such as a list or tuple"
+ )
+
+ def validate(self, record: Any) -> None:
+ try:
+ record_type = self.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('_', ' ')}")
+
+ def field_names(self, record: Any) -> Collection[str]:
+ if self is _RecordType.NAMED_TUPLE:
+ return cast(Tuple[str, ...], getattr(record, "_fields"))
+ if self is _RecordType.MAPPING:
+ return cast(Mapping[str, Any], record).keys()
+ raise TypeError("sequence records do not have named fields")
+
+ def normalize_record(
+ self,
+ record: Any,
+ schema: List[str],
+ require_exact_fields: bool,
+ ) -> Tuple[Any, ...]:
+ if self is _RecordType.SEQUENCE:
+ row = tuple(record)
+ if require_exact_fields and len(row) != len(schema):
+ raise ValueError(
+ f"record has {len(row)} values but schema has
{len(schema)} fields"
+ )
+ return row
+ record_fields = self.field_names(record)
+ for name in schema:
+ if name not in record_fields:
+ raise ValueError(f"record is missing schema field {name!r}")
+ if require_exact_fields and len(record_fields) != len(schema):
+ extra_fields = [name for name in record_fields if name not in
schema]
+ raise ValueError(
+ f"record has fields not present in schema: {extra_fields!r}"
+ )
+ if self is _RecordType.MAPPING:
+ field_values = cast(Mapping[str, Any], record)
+ return tuple(field_values[name] for name in schema)
+ return tuple(getattr(record, name) for name in schema)
+
+
+def _validate_schema(schema: List[str]) -> None:
+ if not isinstance(schema, list) or any(not isinstance(name, str) for name
in schema):
+ raise TypeError("schema must be a list of strings")
+ if not schema:
+ raise ValueError("schema must not be empty")
+ if any(not name for name in schema):
+ raise ValueError("schema field names must not be empty")
+ if len(set(schema)) != len(schema):
+ raise ValueError("schema field names must be unique")
+
+
+@PublicEvolving()
+def from_records(
+ data: Sequence[Union[Sequence[Any], Mapping[str, Any]]],
+ schema: Optional[List[str]] = None,
+) -> DataFrame:
+ """
+ Create a DataFrame from row-oriented records.
+
+ For mapping and named tuple records with an explicit ``schema``, every
record must contain all
+ schema fields; other fields are ignored. When ``schema`` is omitted, the
keys or fields from
+ the first record are used as the schema and every record must have exactly
those fields.
+
+ For other sequence records, every record must have the same number of
values. A ``schema`` is
+ required to provide the field names.
+
+ Field types are inferred from the record values.
+
+ :param data: Non-empty sequence of mapping or sequence records.
+ :param schema: Optional non-empty list of field names.
+ :return: A DataFrame containing the records.
+ :raises TypeError: If a record or schema has an invalid type.
+ :raises ValueError: If data or schema is empty, schema field names are
invalid, a required
+ schema is omitted, a required field is absent, inferred record fields
differ, or record
+ widths differ.
+
+ Example::
+
+ >>> import pyflink.dataframe as pf
+ >>> users = pf.from_records([
+ ... {"id": 1, "name": "Alice"},
+ ... {"id": 2, "name": "Bob"},
+ ... ])
+ >>> users = pf.from_records(
+ ... [(1, "Alice"), (2, "Bob")], schema=["id", "name"]
+ ... )
+ >>> from typing import NamedTuple
+ >>> class User(NamedTuple):
+ ... id: int
+ ... name: str
+ >>> users = pf.from_records([User(1, "Alice"), User(2, "Bob")])
+ >>> selected_users = pf.from_records(
+ ... [User(1, "Alice")], schema=["name", "id"]
+ ... )
+
+ .. versionadded:: 2.4.0
+ """
+ if not isinstance(data, Sequence) or isinstance(data,
_SCALAR_SEQUENCE_TYPES):
+ raise TypeError(
+ "data must be a sequence of records, such as a list or tuple"
+ )
+ if not data:
+ raise ValueError("data must not be empty")
+
+ first_record = data[0]
+ try:
+ expected_record_type = _RecordType.from_record(first_record)
+ except TypeError as error:
+ raise TypeError("invalid record at index 0") from error
+
+ if expected_record_type is _RecordType.SEQUENCE:
+ if schema is None:
+ raise ValueError("schema is required for sequence records")
+ require_exact_fields = True
+ elif schema is None:
+ schema = list(expected_record_type.field_names(first_record))
+ require_exact_fields = True
+ else:
+ require_exact_fields = False
+
+ _validate_schema(schema)
+ rows: List[Sequence[Any]] = []
+ for index, record in enumerate(data):
+ try:
+ if index > 0:
+ expected_record_type.validate(record)
+ row = expected_record_type.normalize_record(
+ record, schema, require_exact_fields
+ )
+ except TypeError as error:
+ raise TypeError(f"invalid record at index {index}") from error
+ except ValueError as error:
+ raise ValueError(f"invalid record at index {index}") from error
+ rows.append(row)
+
+ return DataFrame(
+ get_or_create_table_environment().from_elements(rows, schema)
+ )
+
+
+@PublicEvolving()
+def from_dict(
+ data: Mapping[str, Sequence[Any]], schema: Optional[List[str]] = None
+) -> DataFrame:
+ """
+ Create a DataFrame from a column-oriented dictionary.
+
+ All selected columns must contain the same non-zero number of values.
``schema`` can select a
+ subset of columns and controls their order. If omitted, dictionary
insertion order is used.
+
+ :param data: Non-empty mapping of column names to value sequences.
+ :param schema: Optional non-empty list of selected column names.
+ :return: A DataFrame containing the selected columns.
+ :raises TypeError: If ``data`` is not a mapping, or the selected schema or
a selected column
+ value has an invalid type.
+ :raises ValueError: If the input is empty, schema field names are invalid,
selected column
+ lengths differ, or a selected column is missing.
+
+ Example::
+
+ >>> import pyflink.dataframe as pf
+ >>> users = pf.from_dict(
+ ... {"name": ["Alice", "Bob"], "id": [1, 2]},
+ ... schema=["id", "name"],
+ ... )
+
+ .. versionadded:: 2.4.0
+ """
+ if not isinstance(data, Mapping):
+ raise TypeError("data must be a mapping")
+ if not data:
+ raise ValueError("data must not be empty")
+ if schema is None:
+ schema = list(data.keys())
+ _validate_schema(schema)
+ for name in schema:
+ if name not in data:
+ raise ValueError(f"column {name!r} is not present in data")
+ values = data[name]
+ if not isinstance(values, Sequence) or isinstance(
+ values, _SCALAR_SEQUENCE_TYPES
+ ):
+ raise TypeError(
+ f"column {name!r} values must be a sequence, "
+ "such as a list or tuple"
+ )
+ lengths = {name: len(data[name]) for name in schema}
+ if len(set(lengths.values())) != 1:
+ raise ValueError("columns must have equal lengths")
+ row_count = next(iter(lengths.values()))
+ if row_count == 0:
+ raise ValueError("data must contain at least one row")
+ rows = [
+ tuple(data[name][row_index] for name in schema)
+ for row_index in range(row_count)
+ ]
+ return DataFrame(
+ get_or_create_table_environment().from_elements(rows, schema)
+ )
diff --git a/flink-python/pyflink/dataframe/dataframe.py
b/flink-python/pyflink/dataframe/dataframe.py
new file mode 100644
index 00000000000..66c2020ee4d
--- /dev/null
+++ b/flink-python/pyflink/dataframe/dataframe.py
@@ -0,0 +1,350 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+
+from typing import Any, Callable, List, Optional, Tuple, Union, overload
+
+from pyflink.common import Row
+from pyflink.dataframe.datatype import DataType
+from pyflink.table.expression import Expression
+from pyflink.table.expressions import (
+ and_,
+ call_sql,
+ col as table_col,
+ lit as table_lit,
+)
+from pyflink.table.table import Table
+from pyflink.table.types import DataTypes as TableDataTypes
+from pyflink.util.api_stability_decorators import PublicEvolving
+
+__all__ = ["DataFrame", "col", "lit"]
+
+
+@PublicEvolving()
+def col(name: str) -> Expression:
+ """
+ Create a column reference expression.
+
+ :param name: Name of the referenced column.
+ :return: An expression referencing the column.
+
+ Example::
+
+ >>> import pyflink.dataframe as pf
+ >>> df = pf.from_records([{"id": 1, "name": "Alice"}])
+ >>> result = df.select(pf.col("name"))
+
+ .. versionadded:: 2.4.0
+ """
+ return table_col(name)
+
+
+@PublicEvolving()
+def lit(value: Any, data_type: Optional[DataType] = None) -> Expression:
+ """
+ Create a literal expression.
+
+ The data type is inferred from ``value`` when ``data_type`` is omitted.
Otherwise, the
+ declared data type is applied during literal construction.
+
+ :param value: Literal value.
+ :param data_type: Optional data type for the literal.
+ :return: A literal expression.
+ :raises TypeError: If ``data_type`` is not a :class:`DataType`.
+
+ Example::
+
+ >>> import pyflink.dataframe as pf
+ >>> df = pf.from_records([{"id": 1}])
+ >>> result = df.select("id", status=pf.lit("active"))
+
+ .. versionadded:: 2.4.0
+ """
+ if data_type is None:
+ return table_lit(value)
+ if not isinstance(data_type, DataType):
+ raise TypeError("data_type must be a pyflink.dataframe.DataType")
+ table_data_type = data_type._to_table_data_type()
+ if value is None:
+ return table_lit(value, table_data_type)
+ if (
+ table_data_type.nullable() == TableDataTypes.BIGINT()
+ and isinstance(value, int)
+ and not isinstance(value, bool)
+ and -(1 << 31) <= value < (1 << 31)
+ ):
+ # Py4J sends Python integers in this range as java.lang.Integer, but a
typed BIGINT
+ # literal requires java.lang.Long. Match BIGINT independently of its
nullability, then
+ # cast a typed INT literal to the originally declared BIGINT type.
+ return table_lit(value,
TableDataTypes.INT().not_null()).cast(table_data_type)
+ return table_lit(value, table_data_type.not_null())
+
+
+@PublicEvolving()
+class DataFrame:
+ """
+ A modern DataFrame API for PyFlink.
+
+ DataFrame provides a Pythonic interface for data transformations. It
supports fluent chaining
+ of operations and provides a familiar DataFrame-style API.
+
+ Example::
+
+ >>> import pyflink.dataframe as pf
+ >>> df = pf.from_dict({"id": [1, 2], "name": ["a", "b"]})
+ >>> result = df.select("id", "name") \\
+ ... .with_column("id_doubled", pf.col("id") * 2) \\
+ ... .filter(pf.col("id") > 0)
+
+ .. versionadded:: 2.4.0
+ """
+
+ def __init__(self, table: Table):
+ self._table = table
+
+ @PublicEvolving()
+ def filter(
+ self,
+ *predicates: Union[
+ Expression, str, Callable[["DataFrame"], Expression]
+ ],
+ **constraints: Any,
+ ) -> "DataFrame":
+ """
+ Keep rows that satisfy every predicate and equality constraint.
+
+ Predicates may be boolean expressions, SQL expression strings, or
callables that receive
+ this DataFrame and return a boolean expression. A constraint value of
``None`` selects
+ rows where the corresponding column is null.
+
+ :param predicates: Conditions used to test each row.
+ :param constraints: Values keyed by the column names that must equal
them.
+ :return: A new filtered DataFrame.
+ :raises TypeError: If a predicate has an unsupported type or callable
result.
+ :raises ValueError: If no predicates or constraints are provided.
+
+ Example::
+
+ >>> import pyflink.dataframe as pf
+ >>> df = pf.from_records([
+ ... {"name": "Alice", "age": 30, "status": "active"},
+ ... {"name": "Bob", "age": 17, "status": "active"},
+ ... ])
+ >>> adults = df.filter(pf.col("age") >= 18, status="active")
+ >>> adults = df.filter(lambda current: current["age"] >= 18)
+ >>> missing_status = df.filter(status=None)
+
+ .. versionadded:: 2.4.0
+ """
+ if not predicates and not constraints:
+ raise ValueError(
+ "filter() requires at least one predicate or equality
constraint"
+ )
+
+ conditions: List[Expression] = []
+ for predicate in predicates:
+ if isinstance(predicate, str):
+ conditions.append(call_sql(predicate))
+ elif isinstance(predicate, Expression):
+ conditions.append(predicate)
+ elif callable(predicate) and not isinstance(predicate, type):
+ condition = predicate(self)
+ if not isinstance(condition, Expression):
+ raise TypeError(
+ "filter() callable predicates must return an
Expression"
+ )
+ conditions.append(condition)
+ else:
+ raise TypeError(
+ "predicate must be an Expression, SQL string, or callable"
+ )
+ for name, value in constraints.items():
+ column = table_col(name)
+ conditions.append(column.is_null if value is None else column ==
table_lit(value))
+
+ condition = conditions[0] if len(conditions) == 1 else
and_(*conditions)
+ return DataFrame(self._table.filter(condition))
+
+ @PublicEvolving()
+ def with_column(
+ self,
+ name: str,
+ expr: Union[Expression, Callable[["DataFrame"], Expression]],
+ ) -> "DataFrame":
+ """
+ Add a column, or replace an existing column with the same name.
+
+ ``expr`` may be an expression or a callable that receives this
DataFrame and returns an
+ expression.
+
+ :param name: Name of the added or replaced column.
+ :param expr: Expression or callable used to compute the column value.
+ :return: A new DataFrame with the requested column.
+ :raises TypeError: If ``name`` is not a string or ``expr`` does not
produce an expression.
+
+ Example::
+
+ >>> import pyflink.dataframe as pf
+ >>> df = pf.from_records([{"left": 1, "right": 2}])
+ >>> result = df.with_column(
+ ... "total", lambda current: current["left"] + current["right"]
+ ... )
+
+ .. versionadded:: 2.4.0
+ """
+ if not isinstance(name, str):
+ raise TypeError("name must be a string")
+ if isinstance(expr, Expression):
+ expression = expr
+ elif callable(expr) and not isinstance(expr, type):
+ expression = expr(self)
+ else:
+ raise TypeError("expr must be an Expression")
+ if not isinstance(expression, Expression):
+ raise TypeError("expr must be an Expression")
+ return
DataFrame(self._table.add_or_replace_columns(expression.alias(name)))
+
+ @PublicEvolving()
+ def select(
+ self,
+ *columns: Union[
+ str,
+ Expression,
+ List[Union[str, Expression]],
+ Tuple[Union[str, Expression], ...],
+ ],
+ **projections: Expression,
+ ) -> "DataFrame":
+ """
+ Select columns and compute named projections.
+
+ Column names and expressions are included in the supplied order. A
list or tuple may be
+ used to group column names and expressions. Named projections are
appended after the
+ positional columns.
+
+ :param columns: Column names and expressions to select.
+ :param projections: Expressions keyed by their result column names.
+ :return: A new DataFrame containing the selected columns and
projections.
+ :raises TypeError: If a column or projection is not a supported value.
+
+ Example::
+
+ >>> import pyflink.dataframe as pf
+ >>> df = pf.from_records([{"id": 1, "name": "Alice"}])
+ >>> result = df.select(
+ ... ("name", "id"), doubled=pf.col("id") * 2
+ ... )
+
+ .. versionadded:: 2.4.0
+ """
+ expressions: List[Expression] = []
+ for column in columns:
+ values = column if isinstance(column, (list, tuple)) else [column]
+ for value in values:
+ if isinstance(value, str):
+ expressions.append(table_col(value))
+ elif isinstance(value, Expression):
+ expressions.append(value)
+ else:
+ raise TypeError(
+ "columns must be strings, expressions, or lists or
tuples of them"
+ )
+
+ for name, projection in projections.items():
+ if not isinstance(projection, Expression):
+ raise TypeError("projections must be expressions")
+ expressions.append(projection.alias(name))
+
+ return DataFrame(self._table.select(*expressions))
+
+ @overload
+ def __getitem__(self, key: str) -> Expression:
+ ...
+
+ @overload
+ def __getitem__(
+ self, key: List[Union[str, Expression]]
+ ) -> "DataFrame":
+ ...
+
+ @overload
+ def __getitem__(
+ self, key: Tuple[Union[str, Expression], ...]
+ ) -> "DataFrame":
+ ...
+
+ @overload
+ def __getitem__(self, key: Expression) -> "DataFrame":
+ ...
+
+ @PublicEvolving()
+ def __getitem__(
+ self,
+ key: Union[
+ str,
+ List[Union[str, Expression]],
+ Tuple[Union[str, Expression], ...],
+ Expression,
+ ],
+ ) -> Union["DataFrame", Expression]:
+ """
+ Select a column, select multiple columns, or filter rows.
+
+ A string returns its column expression, a list or tuple returns a
DataFrame containing the
+ listed columns, and a boolean expression returns a filtered DataFrame.
+
+ :param key: Column name, list or tuple of columns, or boolean
expression.
+ :return: A column expression or a new DataFrame.
+ :raises TypeError: If ``key`` has an unsupported type.
+
+ Example::
+
+ >>> import pyflink.dataframe as pf
+ >>> df = pf.from_records([{"id": 1, "name": "Alice"}])
+ >>> identifier = df["id"]
+ >>> selected = df[("id", "name")]
+ >>> filtered = df[df["id"] > 0]
+
+ .. versionadded:: 2.4.0
+ """
+ if isinstance(key, str):
+ return table_col(key)
+ if isinstance(key, (list, tuple)):
+ return self.select(key)
+ if isinstance(key, Expression):
+ return self.filter(key)
+ raise TypeError("key must be a string, list, tuple, or Expression")
+
+ @PublicEvolving()
+ def collect(self) -> List[Row]:
+ """
+ Execute this DataFrame and return all rows.
+
+ The result iterator is always closed before this method returns or
propagates an error.
+
+ :return: All result rows in collection order.
+
+ Example::
+
+ >>> import pyflink.dataframe as pf
+ >>> df = pf.from_records([{"id": 1}, {"id": 2}])
+ >>> rows = df.collect()
+
+ .. versionadded:: 2.4.0
+ """
+ with self._table.execute().collect() as rows:
+ return list(rows)
diff --git a/flink-python/pyflink/dataframe/datatype.py
b/flink-python/pyflink/dataframe/datatype.py
new file mode 100644
index 00000000000..c22f4ba6d68
--- /dev/null
+++ b/flink-python/pyflink/dataframe/datatype.py
@@ -0,0 +1,85 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+
+from pyflink.table.types import DataType as TableDataType, DataTypes
+from pyflink.util.api_stability_decorators import PublicEvolving
+
+__all__ = ["DataType"]
+
+
+@PublicEvolving()
+class DataType:
+ """
+ Describes the logical data type of a value in the DataFrame API.
+
+ Data types declare the types of values used by DataFrame expressions and
operations.
+
+ Example::
+
+ >>> import pyflink.dataframe as pf
+ >>> integer_type = pf.DataType.int64()
+ >>> string_type = pf.DataType.string()
+
+ .. versionadded:: 2.4.0
+ """
+
+ def __init__(self, table_data_type: TableDataType):
+ self._table_data_type = table_data_type
+
+ @PublicEvolving()
+ def __eq__(self, other: object) -> bool:
+ if not isinstance(other, DataType):
+ return False
+ return self._table_data_type == other._table_data_type
+
+ @PublicEvolving()
+ def __hash__(self) -> int:
+ return hash(str(self._table_data_type))
+
+ @classmethod
+ @PublicEvolving()
+ def int64(cls) -> "DataType":
+ """
+ Create a 64-bit integer type.
+
+ Example::
+
+ >>> import pyflink.dataframe as pf
+ >>> expression = pf.lit(42, pf.DataType.int64())
+
+ .. versionadded:: 2.4.0
+ """
+ return cls(DataTypes.BIGINT())
+
+ @classmethod
+ @PublicEvolving()
+ def string(cls) -> "DataType":
+ """
+ Create a variable-length string type.
+
+ Example::
+
+ >>> import pyflink.dataframe as pf
+ >>> expression = pf.lit("Alice", pf.DataType.string())
+
+ .. versionadded:: 2.4.0
+ """
+ return cls(DataTypes.STRING())
+
+ def _to_table_data_type(self) -> TableDataType:
+ return self._table_data_type
diff --git a/flink-python/pyflink/dataframe/tests/__init__.py
b/flink-python/pyflink/dataframe/tests/__init__.py
new file mode 100644
index 00000000000..65b48d4d79b
--- /dev/null
+++ b/flink-python/pyflink/dataframe/tests/__init__.py
@@ -0,0 +1,17 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
diff --git a/flink-python/pyflink/dataframe/tests/test_context.py
b/flink-python/pyflink/dataframe/tests/test_context.py
new file mode 100644
index 00000000000..21f1f28ef56
--- /dev/null
+++ b/flink-python/pyflink/dataframe/tests/test_context.py
@@ -0,0 +1,93 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+
+import unittest
+from typing import get_type_hints, Optional
+
+import pyflink.dataframe as pf
+from pyflink.table import EnvironmentSettings, TableEnvironment
+from pyflink.testing.test_case_utils import PyFlinkUTTestCase
+
+
+class TableEnvironmentContextValidationTests(unittest.TestCase):
+ def test_public_type_hints_are_resolvable(self):
+ self.assertEqual(
+ get_type_hints(pf.set_table_environment),
+ {
+ "t_env": Optional[TableEnvironment],
+ "return": type(None),
+ },
+ )
+ self.assertEqual(
+ get_type_hints(pf.get_table_environment),
+ {"return": Optional[TableEnvironment]},
+ )
+ self.assertEqual(
+ get_type_hints(pf.get_or_create_table_environment),
+ {"return": TableEnvironment},
+ )
+
+ def test_set_environment_rejects_invalid_type_without_changing_state(self):
+ previous_environment = pf.get_table_environment()
+ self.addCleanup(pf.set_table_environment, previous_environment)
+
+ with self.assertRaisesRegex(
+ TypeError, "t_env must be a TableEnvironment or None"
+ ):
+ pf.set_table_environment(object())
+
+ self.assertIs(pf.get_table_environment(), previous_environment)
+
+
+class TableEnvironmentContextTests(PyFlinkUTTestCase):
+ def setUp(self):
+ super().setUp()
+ previous_environment = pf.get_table_environment()
+ self.addCleanup(pf.set_table_environment, previous_environment)
+
+ def test_set_environment_makes_it_retrievable(self):
+ pf.set_table_environment(self.t_env)
+
+ self.assertIs(pf.get_table_environment(), self.t_env)
+
+ def test_set_batch_table_environment_makes_it_retrievable(self):
+ batch_environment = TableEnvironment.create(
+ EnvironmentSettings.in_batch_mode()
+ )
+
+ pf.set_table_environment(batch_environment)
+
+ self.assertIs(pf.get_table_environment(), batch_environment)
+
+ def test_set_none_clears_the_environment(self):
+ pf.set_table_environment(self.t_env)
+
+ pf.set_table_environment(None)
+
+ self.assertIsNone(pf.get_table_environment())
+
+ def test_created_environment_is_retrievable(self):
+ pf.set_table_environment(None)
+
+ created_environment = pf.get_or_create_table_environment()
+
+ self.assertIs(created_environment, pf.get_table_environment())
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/flink-python/pyflink/dataframe/tests/test_convert.py
b/flink-python/pyflink/dataframe/tests/test_convert.py
new file mode 100644
index 00000000000..f08ede67ab7
--- /dev/null
+++ b/flink-python/pyflink/dataframe/tests/test_convert.py
@@ -0,0 +1,228 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+
+import unittest
+from typing import NamedTuple
+
+import pyflink.dataframe as pf
+
+
+class _Point(NamedTuple):
+ x: int
+ y: str
+
+
+class _OtherPoint(NamedTuple):
+ x: int
+ z: str
+
+
+class FromRecordsTests(unittest.TestCase):
+ def test_rejects_scalar_sequence_data(self):
+ for data in ["ab", b"ab", bytearray(b"ab"), memoryview(b"ab")]:
+ with self.subTest(data_type=type(data)):
+ with self.assertRaisesRegex(
+ TypeError,
+ "data must be a sequence of records, such as a list or
tuple",
+ ):
+ pf.from_records(data, schema=["value"])
+
+ def test_rejects_empty_data(self):
+ with self.assertRaisesRegex(ValueError, "data must not be empty"):
+ pf.from_records([], schema=["id"])
+
+ def test_rejects_empty_schema(self):
+ with self.assertRaisesRegex(ValueError, "schema must not be empty"):
+ pf.from_records([(1,)], schema=[])
+
+ def test_rejects_schema_that_is_not_a_list_of_strings(self):
+ for schema in [0, False, "", (), ("id",), [1]]:
+ with self.subTest(schema=schema):
+ with self.assertRaisesRegex(
+ TypeError, "schema must be a list of strings"
+ ):
+ pf.from_records([(1,)], schema=schema)
+
+ def test_rejects_empty_schema_field_name(self):
+ with self.assertRaisesRegex(ValueError, "schema field names must not
be empty"):
+ pf.from_records([(1,)], schema=[""])
+
+ def test_rejects_duplicate_schema_field_names(self):
+ with self.assertRaisesRegex(ValueError, "schema field names must be
unique"):
+ pf.from_records([(1, 2)], schema=["id", "id"])
+
+ def test_requires_schema_for_sequence_records(self):
+ with self.assertRaisesRegex(ValueError, "schema is required for
sequence records"):
+ pf.from_records([(1,)])
+
+ def test_rejects_unsupported_record_type(self):
+ with self.assertRaises(TypeError) as error:
+ pf.from_records([1], schema=["id"])
+ self.assertEqual(str(error.exception), "invalid record at index 0")
+ self.assertEqual(
+ str(error.exception.__cause__),
+ "record must be a mapping or a sequence of values, "
+ "such as a list or tuple",
+ )
+
+ def test_rejects_scalar_sequence_records(self):
+ for value in ["ab", b"ab", bytearray(b"ab"), memoryview(b"ab")]:
+ for index, records, cause in [
+ (
+ 0,
+ [value],
+ "record must be a mapping or a sequence of values, "
+ "such as a list or tuple",
+ ),
+ (1, [(1, 2), value], "record must be a sequence"),
+ ]:
+ with self.subTest(value_type=type(value), index=index):
+ with self.assertRaises(TypeError) as error:
+ pf.from_records(records, schema=["left", "right"])
+ self.assertEqual(
+ str(error.exception), f"invalid record at index
{index}"
+ )
+ self.assertEqual(str(error.exception.__cause__), cause)
+
+ def test_rejects_record_with_wrong_arity(self):
+ with self.assertRaises(ValueError) as error:
+ pf.from_records([(1, "Alice"), (2,)], schema=["id", "name"])
+ self.assertEqual(str(error.exception), "invalid record at index 1")
+ self.assertEqual(
+ str(error.exception.__cause__),
+ "record has 1 values but schema has 2 fields",
+ )
+
+ def test_rejects_mapping_records_with_different_keys(self):
+ records_with_different_keys = [
+ (
+ [{"a": 1}, {"a": 2, "b": 3}],
+ "record has fields not present in schema: ['b']",
+ ),
+ (
+ [{"a": 1, "b": 2}, {"a": 3}],
+ "record is missing schema field 'b'",
+ ),
+ ]
+ for records, cause in records_with_different_keys:
+ with self.subTest(records=records):
+ with self.assertRaises(ValueError) as error:
+ pf.from_records(records)
+ self.assertEqual(str(error.exception), "invalid record at
index 1")
+ self.assertEqual(str(error.exception.__cause__), cause)
+
+ def test_rejects_mixed_mapping_and_named_tuple_records_with_index(self):
+ invalid_records = [
+ (
+ [{"id": 1}, (2,)],
+ ["id"],
+ "record must be a mapping",
+ ),
+ (
+ [_Point(1, "a"), {"x": 2, "y": "b"}],
+ ["x", "y"],
+ "record must be a named tuple",
+ ),
+ ]
+ for records, schema, cause in invalid_records:
+ with self.subTest(records=records):
+ with self.assertRaises(TypeError) as error:
+ pf.from_records(records, schema=schema)
+ self.assertEqual(str(error.exception), "invalid record at
index 1")
+ self.assertEqual(str(error.exception.__cause__), cause)
+
+ def test_rejects_schema_that_renames_named_tuple_fields(self):
+ with self.assertRaises(ValueError) as error:
+ pf.from_records([_Point(1, "a")], schema=["a", "b"])
+ self.assertEqual(str(error.exception), "invalid record at index 0")
+ self.assertEqual(
+ str(error.exception.__cause__), "record is missing schema field
'a'"
+ )
+
+ def test_rejects_different_inferred_named_tuple_fields(self):
+ with self.assertRaises(ValueError) as error:
+ pf.from_records([_Point(1, "a"), _OtherPoint(2, "b")])
+ self.assertEqual(str(error.exception), "invalid record at index 1")
+ self.assertEqual(
+ str(error.exception.__cause__),
+ "record is missing schema field 'y'",
+ )
+
+ def test_rejects_schema_field_missing_from_mapping_records(self):
+ with self.assertRaises(ValueError) as error:
+ pf.from_records(
+ [{"id": 1, "name": "Alice"}, {"id": 2}],
+ schema=["id", "name"],
+ )
+ self.assertEqual(str(error.exception), "invalid record at index 1")
+ self.assertEqual(
+ str(error.exception.__cause__), "record is missing schema field
'name'"
+ )
+
+
+class FromDictTests(unittest.TestCase):
+ def test_rejects_data_that_is_not_a_mapping(self):
+ for data in [[], [("id", [1])]]:
+ with self.subTest(data=data):
+ with self.assertRaisesRegex(TypeError, "data must be a
mapping"):
+ pf.from_dict(data)
+
+ def test_rejects_scalar_sequence_column_values(self):
+ for values in ["ab", b"ab", bytearray(b"ab"), memoryview(b"ab")]:
+ with self.subTest(value_type=type(values)):
+ with self.assertRaisesRegex(
+ TypeError,
+ "column 'value' values must be a sequence, "
+ "such as a list or tuple",
+ ):
+ pf.from_dict({"value": values})
+
+ def test_rejects_empty_data(self):
+ with self.assertRaisesRegex(ValueError, "data must not be empty"):
+ pf.from_dict({})
+
+ def test_rejects_zero_rows(self):
+ with self.assertRaisesRegex(ValueError, "data must contain at least
one row"):
+ pf.from_dict({"id": []})
+
+ def test_rejects_columns_with_different_lengths(self):
+ with self.assertRaisesRegex(ValueError, "columns must have equal
lengths"):
+ pf.from_dict({"id": [1, 2], "name": ["Alice"]})
+
+ def test_rejects_schema_column_missing_from_data(self):
+ with self.assertRaisesRegex(ValueError, "column 'name' is not present
in data"):
+ pf.from_dict({"id": [1]}, schema=["id", "name"])
+
+ def test_rejects_schema_that_is_not_a_list_of_strings(self):
+ for schema in [0, False, "", (), ("id",), [1]]:
+ with self.subTest(schema=schema):
+ with self.assertRaisesRegex(
+ TypeError, "schema must be a list of strings"
+ ):
+ pf.from_dict({"id": [1]}, schema=schema)
+
+ def test_rejects_empty_schema_field_name(self):
+ with self.assertRaisesRegex(ValueError, "schema field names must not
be empty"):
+ pf.from_dict({"id": [1]}, schema=[""])
+
+ def test_rejects_duplicate_schema_field_names(self):
+ with self.assertRaisesRegex(ValueError, "schema field names must be
unique"):
+ pf.from_dict({"id": [1]}, schema=["id", "id"])
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/flink-python/pyflink/dataframe/tests/test_dataframe.py
b/flink-python/pyflink/dataframe/tests/test_dataframe.py
new file mode 100644
index 00000000000..98714e9fbb6
--- /dev/null
+++ b/flink-python/pyflink/dataframe/tests/test_dataframe.py
@@ -0,0 +1,575 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+
+import unittest
+from typing import NamedTuple
+
+import pyflink.dataframe as pf
+from py4j.protocol import Py4JJavaError
+from pyflink.common import Row
+from pyflink.table import (
+ DataTypes as TableDataTypes,
+ EnvironmentSettings,
+ TableEnvironment,
+)
+from pyflink.table.expression import Expression
+from pyflink.testing.test_case_utils import (
+ PyFlinkDataFrameUTTestCase,
+ PyFlinkITTestCase,
+ PyFlinkStreamDataFrameTestCase,
+)
+
+
+class _Point(NamedTuple):
+ x: int
+ y: str
+
+
+class _CloseableIterator:
+ def __init__(self, values=None, error=None):
+ self._values = iter(values or [])
+ self._error = error
+ self.closed = False
+
+ def __iter__(self):
+ return self
+
+ def __next__(self):
+ if self._error is not None:
+ raise self._error
+ return next(self._values)
+
+ def __enter__(self):
+ return self
+
+ def __exit__(self, exc_type, exc_value, traceback):
+ self.closed = True
+
+
+class _TableResult:
+ def __init__(self, iterator):
+ self._iterator = iterator
+
+ def collect(self):
+ return self._iterator
+
+
+class _Table:
+ def __init__(self, iterator):
+ self._iterator = iterator
+
+ def execute(self):
+ return _TableResult(self._iterator)
+
+
+class DataFrameCollectTests(unittest.TestCase):
+ def test_collect_returns_all_rows_and_closes_iterator(self):
+ iterator = _CloseableIterator([Row(1, "Alice")])
+ dataframe = pf.DataFrame(_Table(iterator))
+
+ self.assertEqual(dataframe.collect(), [Row(1, "Alice")])
+ self.assertTrue(iterator.closed)
+
+ def test_collect_closes_iterator_when_iteration_fails(self):
+ iterator = _CloseableIterator(error=RuntimeError("iteration failed"))
+ dataframe = pf.DataFrame(_Table(iterator))
+
+ with self.assertRaisesRegex(RuntimeError, "iteration failed"):
+ dataframe.collect()
+
+ self.assertTrue(iterator.closed)
+
+
+class DataFrameCreationTests(PyFlinkDataFrameUTTestCase):
+ def test_from_dict_uses_insertion_order_without_schema(self):
+ dataframe = pf.from_dict({"name": ["Alice"], "id": [1]})
+
+ self.assert_dataframe_schema(
+ dataframe,
+ ["name", "id"],
+ [TableDataTypes.STRING(), TableDataTypes.BIGINT()],
+ )
+
+ def test_from_dict_respects_explicit_schema_order_and_subset(self):
+ dataframe = pf.from_dict(
+ {
+ "name": ["Alice"],
+ "ignored": ["x"],
+ "id": [1],
+ },
+ schema=["id", "name"],
+ )
+
+ self.assert_dataframe_schema(
+ dataframe,
+ ["id", "name"],
+ [TableDataTypes.BIGINT(), TableDataTypes.STRING()],
+ )
+
+ def test_from_records_accepts_list_records(self):
+ dataframe = pf.from_records(
+ [[1, "Alice"], [2, "Bob"]],
+ schema=["id", "name"],
+ )
+
+ self.assert_dataframe_schema(
+ dataframe,
+ ["id", "name"],
+ [TableDataTypes.BIGINT(), TableDataTypes.STRING()],
+ )
+
+ def test_from_records_accepts_general_sequence_records(self):
+ dataframe = pf.from_records(
+ [range(2), range(2, 4)],
+ schema=["left", "right"],
+ )
+
+ self.assert_dataframe_schema(
+ dataframe,
+ ["left", "right"],
+ [TableDataTypes.BIGINT(), TableDataTypes.BIGINT()],
+ )
+
+ def test_from_records_infers_mapping_schema(self):
+ dataframe = pf.from_records(
+ [{"name": "Alice", "id": 1}, {"name": "Bob", "id": 2}]
+ )
+
+ self.assert_dataframe_schema(
+ dataframe,
+ ["name", "id"],
+ [TableDataTypes.STRING(), TableDataTypes.BIGINT()],
+ )
+
+ def test_from_records_selects_mapping_fields_with_explicit_schema(self):
+ dataframe = pf.from_records(
+ [
+ {"name": "Alice", "id": 1, "ignored": "x"},
+ {"name": "Bob", "id": 2, "ignored": "y"},
+ ],
+ schema=["id", "name"],
+ )
+
+ self.assert_dataframe_schema(
+ dataframe,
+ ["id", "name"],
+ [TableDataTypes.BIGINT(), TableDataTypes.STRING()],
+ )
+
+ def test_from_records_infers_named_tuple_schema(self):
+ dataframe = pf.from_records([_Point(1, "a"), _Point(2, "b")])
+
+ self.assert_dataframe_schema(
+ dataframe,
+ ["x", "y"],
+ [TableDataTypes.BIGINT(), TableDataTypes.STRING()],
+ )
+
+ def
test_from_records_selects_named_tuple_fields_with_explicit_schema(self):
+ dataframe = pf.from_records(
+ [_Point(1, "a"), _Point(2, "b")],
+ schema=["y", "x"],
+ )
+
+ self.assert_dataframe_schema(
+ dataframe,
+ ["y", "x"],
+ [TableDataTypes.STRING(), TableDataTypes.BIGINT()],
+ )
+
+
+class DataFrameSelectTests(PyFlinkDataFrameUTTestCase):
+ def setUp(self):
+ super().setUp()
+ self.dataframe = pf.from_records(
+ [(1, "Alice"), (2, "Bob")],
+ schema=["id", "name"],
+ )
+
+ def
test_select_accepts_names_expressions_lists_and_named_projections(self):
+ result = self.dataframe.select(
+ ["name"],
+ pf.col("id"),
+ doubled=pf.col("id") * 2,
+ )
+
+ self.assert_dataframe_schema(
+ result,
+ ["name", "id", "doubled"],
+ [
+ TableDataTypes.STRING(),
+ TableDataTypes.BIGINT(),
+ TableDataTypes.BIGINT(),
+ ],
+ )
+
+ def test_select_accepts_tuple_column_group(self):
+ result = self.dataframe.select(("name", "id"))
+
+ self.assert_dataframe_schema(
+ result,
+ ["name", "id"],
+ [TableDataTypes.STRING(), TableDataTypes.BIGINT()],
+ )
+
+ def test_select_rejects_non_string_column(self):
+ with self.assertRaisesRegex(TypeError, "columns must be strings"):
+ self.dataframe.select(42)
+
+ def test_select_rejects_non_expression_projection(self):
+ with self.assertRaisesRegex(TypeError, "projections must be
expressions"):
+ self.dataframe.select(answer=42)
+
+
+class DataFrameWithColumnTests(PyFlinkDataFrameUTTestCase):
+ def setUp(self):
+ super().setUp()
+ self.dataframe = pf.from_records(
+ [(1, "Alice", 30)],
+ schema=["id", "name", "age"],
+ )
+
+ def test_with_column_adds_callable_result(self):
+ result = self.dataframe.with_column(
+ "age_next_year",
+ lambda current: current["age"] + 1,
+ )
+
+ self.assert_dataframe_schema(
+ result,
+ ["id", "name", "age", "age_next_year"],
+ [
+ TableDataTypes.BIGINT(),
+ TableDataTypes.STRING(),
+ TableDataTypes.BIGINT(),
+ TableDataTypes.BIGINT(),
+ ],
+ )
+
+ def test_with_column_replaces_existing_column(self):
+ result = self.dataframe.with_column("age", pf.col("age") + 1)
+
+ self.assert_dataframe_schema(
+ result,
+ ["id", "name", "age"],
+ [
+ TableDataTypes.BIGINT(),
+ TableDataTypes.STRING(),
+ TableDataTypes.BIGINT(),
+ ],
+ )
+
+ def test_with_column_rejects_non_expression(self):
+ with self.assertRaisesRegex(TypeError, "expr must be an Expression"):
+ self.dataframe.with_column("answer", 42)
+
+ def test_with_column_rejects_callable_returning_non_expression(self):
+ with self.assertRaisesRegex(TypeError, "expr must be an Expression"):
+ self.dataframe.with_column("answer", lambda df: 42)
+
+ def test_with_column_rejects_expression_class(self):
+ with self.assertRaisesRegex(TypeError, "expr must be an Expression"):
+ self.dataframe.with_column("answer", Expression)
+
+ def test_with_column_rejects_non_string_name(self):
+ with self.assertRaisesRegex(TypeError, "name must be a string"):
+ self.dataframe.with_column(42, object())
+
+
+class DataFrameFilterTests(PyFlinkDataFrameUTTestCase):
+ def setUp(self):
+ super().setUp()
+ self.dataframe = pf.from_records(
+ [(1, 0.9, "NYC", None), (2, 0.95, "SF", "Paris")],
+ schema=["id", "score", "city", "destination"],
+ )
+
+ def assert_filter_schema(self, dataframe):
+ self.assert_dataframe_schema(
+ dataframe,
+ ["id", "score", "city", "destination"],
+ [
+ TableDataTypes.BIGINT(),
+ TableDataTypes.DOUBLE(),
+ TableDataTypes.STRING(),
+ TableDataTypes.STRING(),
+ ],
+ )
+
+ def test_filter_accepts_expression(self):
+ self.assert_filter_schema(self.dataframe.filter(pf.col("id") > 0))
+
+ def test_filter_accepts_multiple_predicates_and_constraints(self):
+ result = self.dataframe.filter(
+ pf.col("id") > 0,
+ pf.col("score") >= 0.8,
+ city="NYC",
+ )
+
+ self.assert_filter_schema(result)
+
+ def test_filter_accepts_none_constraint(self):
+ self.assert_filter_schema(self.dataframe.filter(destination=None))
+
+ def test_filter_accepts_sql_string_predicate(self):
+ self.assert_filter_schema(self.dataframe.filter("id > 0"))
+
+ def test_filter_accepts_callable_predicate(self):
+ self.assert_filter_schema(
+ self.dataframe.filter(lambda current: current["id"] > 0)
+ )
+
+ def test_filter_rejects_non_expression(self):
+ with self.assertRaisesRegex(TypeError, "predicate must be an
Expression"):
+ self.dataframe.filter(True)
+
+ def test_filter_requires_a_condition(self):
+ with self.assertRaisesRegex(ValueError, "requires at least one
predicate"):
+ self.dataframe.filter()
+
+ def test_filter_rejects_callable_returning_non_expression(self):
+ with self.assertRaisesRegex(
+ TypeError, "callable predicates must return an Expression"
+ ):
+ self.dataframe.filter(lambda df: True)
+
+ def test_filter_rejects_expression_class(self):
+ with self.assertRaisesRegex(TypeError, "predicate must be an
Expression"):
+ self.dataframe.filter(Expression)
+
+
+class DataFrameGetItemTests(PyFlinkDataFrameUTTestCase):
+ def setUp(self):
+ super().setUp()
+ self.dataframe = pf.from_records(
+ [(1, "Alice"), (2, "Bob")],
+ schema=["id", "name"],
+ )
+
+ def test_getitem_returns_expression_for_column_name(self):
+ self.assertIsInstance(self.dataframe["id"], Expression)
+
+ def test_getitem_selects_list_projection(self):
+ result = self.dataframe[["name", "id"]]
+
+ self.assert_dataframe_schema(
+ result,
+ ["name", "id"],
+ [TableDataTypes.STRING(), TableDataTypes.BIGINT()],
+ )
+
+ def test_getitem_selects_tuple_projection(self):
+ result = self.dataframe[("name", "id")]
+
+ self.assert_dataframe_schema(
+ result,
+ ["name", "id"],
+ [TableDataTypes.STRING(), TableDataTypes.BIGINT()],
+ )
+
+ def test_getitem_filters_with_expression(self):
+ result = self.dataframe[self.dataframe["id"] > 0]
+
+ self.assert_dataframe_schema(
+ result,
+ ["id", "name"],
+ [TableDataTypes.BIGINT(), TableDataTypes.STRING()],
+ )
+
+ def test_getitem_rejects_unsupported_key(self):
+ with self.assertRaisesRegex(TypeError, "key must be a string, list"):
+ self.dataframe[42]
+
+
+class DataFrameLiteralTests(PyFlinkDataFrameUTTestCase):
+ def setUp(self):
+ super().setUp()
+ self.dataframe = pf.from_records([(1,)], schema=["id"])
+
+ def test_lit_supports_inferred_and_explicit_types(self):
+ result = self.dataframe.select(
+ inferred_int=pf.lit(2),
+ inferred_string=pf.lit("x"),
+ explicit_int=pf.lit(3, pf.DataType.int64()),
+ explicit_large_int=pf.lit(1 << 40, pf.DataType.int64()),
+ explicit_string=pf.lit("y", pf.DataType.string()),
+ )
+
+ self.assert_dataframe_schema(
+ result,
+ [
+ "inferred_int",
+ "inferred_string",
+ "explicit_int",
+ "explicit_large_int",
+ "explicit_string",
+ ],
+ [
+ TableDataTypes.INT().not_null(),
+ TableDataTypes.CHAR(1).not_null(),
+ TableDataTypes.BIGINT().not_null(),
+ TableDataTypes.BIGINT().not_null(),
+ TableDataTypes.STRING().not_null(),
+ ],
+ )
+
+ def test_lit_supports_explicitly_typed_nulls(self):
+ result = self.dataframe.select(
+ null_int=pf.lit(None, pf.DataType.int64()),
+ null_string=pf.lit(None, pf.DataType.string()),
+ )
+
+ self.assert_dataframe_schema(
+ result,
+ ["null_int", "null_string"],
+ [TableDataTypes.BIGINT(), TableDataTypes.STRING()],
+ )
+
+ def test_lit_supports_small_int_for_non_nullable_bigint(self):
+ non_nullable_bigint = pf.DataType(TableDataTypes.BIGINT().not_null())
+ result = self.dataframe.select(value=pf.lit(3, non_nullable_bigint))
+
+ self.assert_dataframe_schema(
+ result,
+ ["value"],
+ [TableDataTypes.BIGINT().not_null()],
+ )
+
+ def test_lit_rejects_values_incompatible_with_explicit_type(self):
+ incompatible_values = [
+ (3.14, pf.DataType.int64()),
+ ("abc", pf.DataType.int64()),
+ (42, pf.DataType.string()),
+ ]
+ for value, data_type in incompatible_values:
+ with self.subTest(value=value, data_type=data_type):
+ with self.assertRaises(Py4JJavaError):
+ pf.lit(value, data_type)
+
+ def test_lit_rejects_non_dataframe_data_type(self):
+ with self.assertRaisesRegex(
+ TypeError, "data_type must be a pyflink.dataframe.DataType"
+ ):
+ pf.lit(1, object())
+
+
+class DataFrameITTests(PyFlinkStreamDataFrameTestCase):
+ def test_from_records(self):
+ dataframe = pf.from_records(
+ [(1, "Alice"), (2, "Bob")],
+ schema=["id", "name"],
+ )
+
+ self.assertEqual(
+ dataframe.collect(),
+ [Row(1, "Alice"), Row(2, "Bob")],
+ )
+
+ def test_basic_functionality(self):
+ df = pf.from_dict(
+ {
+ "name": [
+ "expression",
+ "Alice",
+ "sql",
+ "constraint",
+ "null_constraint",
+ "callable",
+ ],
+ "ignored": ["unused"],
+ "id": [0, 1, 2, 3, 4, 6],
+ "age": [20, 30, 40, 50, 60, 70],
+ "score": [0.95, 0.95, 0.7, 0.95, 0.95, 0.95],
+ "city": ["SF", "SF", "SF", "NYC", "SF", "SF"],
+ "destination": [None, None, None, None, "Paris", None],
+ },
+ schema=["id", "name", "age", "score", "city", "destination"],
+ )
+
+ result = (
+ df[df["id"] > 0]
+ .filter(
+ "score >= 0.9",
+ lambda current: current["id"] < 6,
+ city="SF",
+ destination=None,
+ )
+ .with_column(
+ "age_next_year",
+ lambda current: current["age"] + 1,
+ )
+ .with_column("age", pf.col("age") + 1)
+ .select(
+ "id",
+ "name",
+ "age",
+ age_next_year=pf.col("age_next_year"),
+ inferred_int=pf.lit(2),
+ inferred_string=pf.lit("x"),
+ explicit_int=pf.lit(3, pf.DataType.int64()),
+ explicit_large_int=pf.lit(1 << 40, pf.DataType.int64()),
+ explicit_string=pf.lit("y", pf.DataType.string()),
+ null_int=pf.lit(None, pf.DataType.int64()),
+ null_string=pf.lit(None, pf.DataType.string()),
+ non_nullable_int=pf.lit(
+ 3,
+ pf.DataType(TableDataTypes.BIGINT().not_null()),
+ ),
+ )[
+ (
+ "name",
+ "id",
+ "age",
+ "age_next_year",
+ "inferred_int",
+ "inferred_string",
+ "explicit_int",
+ "explicit_large_int",
+ "explicit_string",
+ "null_int",
+ "null_string",
+ "non_nullable_int",
+ )
+ ]
+ )
+
+ self.assertEqual(
+ result.collect(),
+ [Row("Alice", 1, 31, 31, 2, "x", 3, 1 << 40, "y", None, None, 3)],
+ )
+
+
+class DataFrameBatchITTests(PyFlinkITTestCase):
+ def setUp(self):
+ previous_environment = pf.get_table_environment()
+ self.addCleanup(pf.set_table_environment, previous_environment)
+ self.t_env =
TableEnvironment.create(EnvironmentSettings.in_batch_mode())
+
+ def test_from_records_with_batch_table_environment(self):
+ pf.set_table_environment(self.t_env)
+
+ result = pf.from_records(
+ [(1, "Alice"), (2, "Bob")],
+ schema=["id", "name"],
+ ).filter(pf.col("id") > 1)
+
+ self.assertEqual(result.collect(), [Row(2, "Bob")])
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/flink-python/pyflink/dataframe/tests/test_datatype.py
b/flink-python/pyflink/dataframe/tests/test_datatype.py
new file mode 100644
index 00000000000..756bba371ea
--- /dev/null
+++ b/flink-python/pyflink/dataframe/tests/test_datatype.py
@@ -0,0 +1,65 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+
+import unittest
+
+import pyflink.dataframe as pf
+from pyflink.table import DataTypes
+from pyflink.util.api_stability_decorators import PublicEvolving
+
+
+class DataTypeTests(unittest.TestCase):
+ def test_public_factory_surface(self):
+ public_methods = {
+ name for name in dir(pf.DataType) if not name.startswith("_")
+ }
+
+ self.assertEqual(public_methods, {"int64", "string"})
+
+ def test_int64_maps_to_table_bigint(self):
+ self.assertEqual(pf.DataType.int64()._to_table_data_type(),
DataTypes.BIGINT())
+
+ def test_string_maps_to_table_string(self):
+ self.assertEqual(pf.DataType.string()._to_table_data_type(),
DataTypes.STRING())
+
+ def test_logically_equal_types_compare_and_hash_equally(self):
+ first_int = pf.DataType.int64()
+ second_int = pf.DataType.int64()
+ string = pf.DataType.string()
+
+ self.assertEqual(first_int, second_int)
+ self.assertNotEqual(first_int, string)
+ self.assertEqual(len({first_int, second_int, string}), 2)
+
+ def test_equality_and_hash_are_public_evolving(self):
+ for method in [pf.DataType.__eq__, pf.DataType.__hash__]:
+ with self.subTest(method=method.__name__):
+ self.assertIn(
+ PublicEvolving,
+ getattr(method, "__stability_decorators", set()),
+ )
+
+ def test_nullability_modifiers_are_not_exposed(self):
+ for data_type in [pf.DataType.int64(), pf.DataType.string()]:
+ with self.subTest(data_type=data_type):
+ self.assertFalse(hasattr(data_type, "not_null"))
+ self.assertFalse(hasattr(data_type, "nullable"))
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/flink-python/pyflink/testing/test_case_utils.py
b/flink-python/pyflink/testing/test_case_utils.py
index 2c6c754bf5e..bc547bd8d36 100644
--- a/flink-python/pyflink/testing/test_case_utils.py
+++ b/flink-python/pyflink/testing/test_case_utils.py
@@ -155,6 +155,35 @@ class PyFlinkUTTestCase(PyFlinkTestCase):
self.t_env.get_config().set("python.fn-execution.bundle.size", "1")
+class PyFlinkDataFrameUTTestCase(PyFlinkUTTestCase):
+ """Base class for planner-backed DataFrame interface tests."""
+
+ def setUp(self) -> None:
+ from pyflink.dataframe import get_table_environment,
set_table_environment
+
+ super().setUp()
+ previous_environment = get_table_environment()
+ self.addCleanup(set_table_environment, previous_environment)
+ set_table_environment(self.t_env)
+
+ def assert_dataframe_schema(
+ self,
+ dataframe,
+ expected_column_names,
+ expected_column_data_types=None,
+ ):
+ resolved_schema = dataframe._table.get_resolved_schema()
+ self.assertEqual(
+ list(resolved_schema.get_column_names()),
+ expected_column_names,
+ )
+ if expected_column_data_types is not None:
+ self.assertEqual(
+ resolved_schema.get_column_data_types(),
+ expected_column_data_types,
+ )
+
+
class PyFlinkStreamTableTestCase(PyFlinkITTestCase):
"""
Base class for table stream tests.
@@ -169,6 +198,27 @@ class PyFlinkStreamTableTestCase(PyFlinkITTestCase):
cls.t_env.get_config().set("python.fn-execution.bundle.size", "1")
+class PyFlinkStreamDataFrameTestCase(PyFlinkStreamTableTestCase):
+ """Base class for DataFrame streaming tests."""
+
+ @classmethod
+ def setUpClass(cls):
+ from pyflink.dataframe import get_table_environment,
set_table_environment
+
+ cls._previous_table_environment = get_table_environment()
+ super(PyFlinkStreamDataFrameTestCase, cls).setUpClass()
+ set_table_environment(cls.t_env)
+
+ @classmethod
+ def tearDownClass(cls):
+ from pyflink.dataframe import set_table_environment
+
+ try:
+ super(PyFlinkStreamDataFrameTestCase, cls).tearDownClass()
+ finally:
+ set_table_environment(cls._previous_table_environment)
+
+
class PyFlinkBatchTableTestCase(PyFlinkITTestCase):
"""
Base class for table batch tests.
diff --git a/flink-python/setup.py b/flink-python/setup.py
index 9354d422a5b..40a04a70718 100644
--- a/flink-python/setup.py
+++ b/flink-python/setup.py
@@ -277,6 +277,7 @@ try:
scripts.append("pyflink/find_flink_home.py")
PACKAGES = ['pyflink',
+ 'pyflink.dataframe',
'pyflink.table',
'pyflink.util',
'pyflink.datastream',
diff --git a/flink-python/tox.ini b/flink-python/tox.ini
index 6f57a22dc26..ed09d9f4373 100644
--- a/flink-python/tox.ini
+++ b/flink-python/tox.ini
@@ -46,7 +46,7 @@ max-line-length=100
exclude=.tox/*,dev/*,lib/*,target/*,build/*,dist/*,pyflink/shell.py,.eggs/*,pyflink/fn_execution/tests/process_mode_test_data.py,pyflink/fn_execution/*_pb2.py*,pyflink/examples/table/basic_operations.py
[mypy]
-files=pyflink/common/*.py,pyflink/table/*.py,pyflink/datastream/*.py,pyflink/metrics/*.py
+files=pyflink/common/*.py,pyflink/table/*.py,pyflink/datastream/*.py,pyflink/metrics/*.py,pyflink/dataframe/*.py
ignore_missing_imports = True
strict_optional=False
