dianfu commented on code in PR #28934:
URL: https://github.com/apache/flink/pull/28934#discussion_r3764890150
##########
flink-python/pyflink/dataframe/convert.py:
##########
@@ -241,6 +512,7 @@ def from_dict(
raise TypeError("data must be a mapping")
if not data:
raise ValueError("data must not be empty")
+ watermark_spec = _WatermarkSpec.parse(watermark)
Review Comment:
Add an example about watermark usage.
##########
flink-python/pyflink/dataframe/convert.py:
##########
@@ -120,10 +183,202 @@ def _validate_schema(schema: List[str]) -> None:
raise ValueError("schema field names must be unique")
+def _resolve_column_names(
+ input_names: Sequence[str], schema: Optional[List[str]]
+) -> List[str]:
+ column_names = list(input_names) if schema is None else schema
+ if (
+ schema is not None
+ and isinstance(schema, list)
+ and len(schema) != len(input_names)
+ ):
+ raise ValueError(
+ f"schema has {len(schema)} fields but data has "
+ f"{len(input_names)} columns"
+ )
+ _validate_schema(column_names)
+ return column_names
+
+
+def _infer_schema_and_create_dataframe(
+ rows: Sequence[Sequence[Any]],
+ column_names: List[str],
+ watermark: Optional[_WatermarkSpec] = None,
+) -> DataFrame:
+ row_type = _infer_schema_from_data(rows, names=column_names)
+ if watermark is None:
+ table_schema = None
+ else:
+ row_type = watermark.normalize_row_type(row_type)
+ table_schema = (
+ Schema.new_builder()
+ .from_row_data_type(row_type)
+ .watermark(*watermark)
+ .build()
+ )
+ converter = _create_converter(row_type)
+ verify_row = _create_type_verifier(row_type)
+ sql_rows = []
+ for row in rows:
+ row = converter(row)
+ verify_row(row)
+ sql_rows.append(row_type.to_sql_type(row))
+
+ table = get_or_create_table_environment()._from_elements(
+ sql_rows, row_type, table_schema
+ )
+ return DataFrame(table)
+
+
+@PublicEvolving()
+def from_table(table: Table) -> DataFrame:
+ """
+ Create a DataFrame that wraps a PyFlink Table.
+
+ :param table: Table to wrap without copying or converting it.
+ :return: A DataFrame backed by the exact supplied Table.
+ :raises TypeError: If ``table`` is not a :class:`~pyflink.table.Table`.
+
+ Example::
+
+ >>> import pyflink.dataframe as pf
+ >>> table = table_env.from_elements([(1, "Alice")], ["id", "name"])
+ >>> dataframe = pf.from_table(table)
+ >>> dataframe.to_table() is table
+ True
+
+ .. versionadded:: 2.4.0
+ """
+ if not isinstance(table, Table):
+ raise TypeError("table must be a pyflink.table.Table")
+ return DataFrame(table)
+
+
+@PublicEvolving()
+def from_pandas(
+ pdf: Any,
+ schema: Optional[List[str]] = None,
+ watermark: Optional[Tuple[str, str]] = None,
+) -> DataFrame:
+ """
+ Create a DataFrame from a pandas DataFrame.
+
+ Types are inferred from the Arrow representation of the pandas columns. An
explicit ``schema``
+ renames columns positionally and must contain exactly one unique,
non-empty name per input
+ column. Empty inputs are supported when their pandas dtypes can be
converted to Flink types.
+
+ ``watermark`` declares an event-time column and its SQL watermark
expression. The selected
+ column must have a timestamp-compatible type. Its precision is normalized
to milliseconds;
+ values with finer precision are truncated to ``TIMESTAMP(3)`` or
``TIMESTAMP_LTZ(3)``.
+
+ :param pdf: pandas DataFrame to convert.
+ :param schema: Optional list of positional result column names.
+ :param watermark: Optional ``(column, expression)`` watermark declaration.
+ :return: A DataFrame containing the pandas rows.
+ :raises TypeError: If the input, schema, watermark, or inferred types are
invalid.
+ :raises ValueError: If schema width or watermark column requirements are
not met.
+
+ Example::
+
+ >>> import pandas as pd
+ >>> import pyflink.dataframe as pf
+ >>> pdf = pd.DataFrame({"identifier": [1, 2], "name": ["Alice",
"Bob"]})
+ >>> dataframe = pf.from_pandas(pdf, schema=["id", "name"])
+ >>> events = pf.from_pandas(
+ ... pd.DataFrame({"ts": pd.to_datetime(["2026-01-01T00:00:00Z"])}),
+ ... watermark=("ts", "ts - INTERVAL '5' SECOND"),
+ ... )
+
+ .. versionadded:: 2.4.0
+ """
+ import pandas as pd
+
+ if not isinstance(pdf, pd.DataFrame):
+ raise TypeError(
+ f"data must be a pandas.DataFrame, but was {type(pdf).__name__}"
+ )
+
+ import pyarrow as pa
+
+ return from_arrow(
+ pa.Table.from_pandas(pdf, preserve_index=False),
+ schema=schema,
+ watermark=watermark,
+ )
+
+
+@PublicEvolving()
+def from_arrow(
+ table: Any,
+ schema: Optional[List[str]] = None,
+ watermark: Optional[Tuple[str, str]] = None,
+) -> DataFrame:
+ """
+ Create a DataFrame from a PyArrow Table without converting through pandas.
+
+ An explicit ``schema`` renames columns positionally and must contain
exactly one unique,
+ non-empty name per input column. Empty tables are supported when their
Arrow field types can be
+ converted to Flink types.
+
+ ``watermark`` declares an event-time column and its SQL watermark
expression. The selected
+ column must have a timestamp-compatible type. Its precision is normalized
to milliseconds;
+ values with finer precision are truncated to ``TIMESTAMP(3)`` or
``TIMESTAMP_LTZ(3)``.
+
+ :param table: PyArrow Table to convert.
+ :param schema: Optional list of positional result column names.
+ :param watermark: Optional ``(column, expression)`` watermark declaration.
+ :return: A DataFrame containing the Arrow rows.
+ :raises TypeError: If the input, schema, watermark, or inferred types are
invalid.
+ :raises ValueError: If schema width or watermark column requirements are
not met.
+
+ Example::
+
+ >>> import pyarrow as pa
+ >>> import pyflink.dataframe as pf
+ >>> table = pa.table({"id": [1, 2], "name": ["Alice", "Bob"]})
+ >>> dataframe = pf.from_arrow(table)
+ >>> events = pf.from_arrow(
+ ... pa.table({"ts": pa.array([0], type=pa.timestamp("ms"))}),
+ ... watermark=("ts", "ts - INTERVAL '5' SECOND"),
+ ... )
+
+ .. versionadded:: 2.4.0
+ """
+ import pyarrow as pa
+
+ if not isinstance(table, pa.Table):
+ raise TypeError(
+ f"data must be a pyarrow.Table, but was {type(table).__name__}"
+ )
+ watermark_spec = _WatermarkSpec.parse(watermark)
+ names = _resolve_column_names(table.column_names, schema)
+ row_type = RowType(
+ [
+ RowField(name, from_arrow_type(field.type, field.nullable))
Review Comment:
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.
##########
flink-python/pyflink/dataframe/convert.py:
##########
@@ -120,10 +183,202 @@ def _validate_schema(schema: List[str]) -> None:
raise ValueError("schema field names must be unique")
+def _resolve_column_names(
+ input_names: Sequence[str], schema: Optional[List[str]]
+) -> List[str]:
+ column_names = list(input_names) if schema is None else schema
+ if (
+ schema is not None
+ and isinstance(schema, list)
+ and len(schema) != len(input_names)
+ ):
+ raise ValueError(
+ f"schema has {len(schema)} fields but data has "
+ f"{len(input_names)} columns"
+ )
+ _validate_schema(column_names)
+ return column_names
+
+
+def _infer_schema_and_create_dataframe(
+ rows: Sequence[Sequence[Any]],
+ column_names: List[str],
+ watermark: Optional[_WatermarkSpec] = None,
+) -> DataFrame:
+ row_type = _infer_schema_from_data(rows, names=column_names)
+ if watermark is None:
+ table_schema = None
+ else:
+ row_type = watermark.normalize_row_type(row_type)
+ table_schema = (
+ Schema.new_builder()
+ .from_row_data_type(row_type)
+ .watermark(*watermark)
+ .build()
+ )
+ converter = _create_converter(row_type)
+ verify_row = _create_type_verifier(row_type)
+ sql_rows = []
+ for row in rows:
+ row = converter(row)
+ verify_row(row)
+ sql_rows.append(row_type.to_sql_type(row))
+
+ table = get_or_create_table_environment()._from_elements(
+ sql_rows, row_type, table_schema
+ )
+ return DataFrame(table)
+
+
+@PublicEvolving()
+def from_table(table: Table) -> DataFrame:
+ """
+ Create a DataFrame that wraps a PyFlink Table.
+
+ :param table: Table to wrap without copying or converting it.
+ :return: A DataFrame backed by the exact supplied Table.
+ :raises TypeError: If ``table`` is not a :class:`~pyflink.table.Table`.
+
+ Example::
+
+ >>> import pyflink.dataframe as pf
+ >>> table = table_env.from_elements([(1, "Alice")], ["id", "name"])
+ >>> dataframe = pf.from_table(table)
+ >>> dataframe.to_table() is table
+ True
+
+ .. versionadded:: 2.4.0
+ """
+ if not isinstance(table, Table):
+ raise TypeError("table must be a pyflink.table.Table")
+ return DataFrame(table)
+
+
+@PublicEvolving()
+def from_pandas(
+ pdf: Any,
+ schema: Optional[List[str]] = None,
+ watermark: Optional[Tuple[str, str]] = None,
+) -> DataFrame:
+ """
+ Create a DataFrame from a pandas DataFrame.
+
+ Types are inferred from the Arrow representation of the pandas columns. An
explicit ``schema``
+ renames columns positionally and must contain exactly one unique,
non-empty name per input
+ column. Empty inputs are supported when their pandas dtypes can be
converted to Flink types.
+
+ ``watermark`` declares an event-time column and its SQL watermark
expression. The selected
+ column must have a timestamp-compatible type. Its precision is normalized
to milliseconds;
+ values with finer precision are truncated to ``TIMESTAMP(3)`` or
``TIMESTAMP_LTZ(3)``.
+
+ :param pdf: pandas DataFrame to convert.
+ :param schema: Optional list of positional result column names.
+ :param watermark: Optional ``(column, expression)`` watermark declaration.
+ :return: A DataFrame containing the pandas rows.
+ :raises TypeError: If the input, schema, watermark, or inferred types are
invalid.
+ :raises ValueError: If schema width or watermark column requirements are
not met.
+
+ Example::
+
+ >>> import pandas as pd
+ >>> import pyflink.dataframe as pf
+ >>> pdf = pd.DataFrame({"identifier": [1, 2], "name": ["Alice",
"Bob"]})
+ >>> dataframe = pf.from_pandas(pdf, schema=["id", "name"])
+ >>> events = pf.from_pandas(
+ ... pd.DataFrame({"ts": pd.to_datetime(["2026-01-01T00:00:00Z"])}),
+ ... watermark=("ts", "ts - INTERVAL '5' SECOND"),
+ ... )
+
+ .. versionadded:: 2.4.0
+ """
+ import pandas as pd
+
+ if not isinstance(pdf, pd.DataFrame):
+ raise TypeError(
+ f"data must be a pandas.DataFrame, but was {type(pdf).__name__}"
+ )
+
+ import pyarrow as pa
+
+ return from_arrow(
+ pa.Table.from_pandas(pdf, preserve_index=False),
Review Comment:
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.
##########
flink-python/pyflink/dataframe/convert.py:
##########
@@ -120,10 +183,202 @@ def _validate_schema(schema: List[str]) -> None:
raise ValueError("schema field names must be unique")
+def _resolve_column_names(
+ input_names: Sequence[str], schema: Optional[List[str]]
+) -> List[str]:
+ column_names = list(input_names) if schema is None else schema
+ if (
+ schema is not None
+ and isinstance(schema, list)
+ and len(schema) != len(input_names)
+ ):
+ raise ValueError(
+ f"schema has {len(schema)} fields but data has "
+ f"{len(input_names)} columns"
+ )
+ _validate_schema(column_names)
+ return column_names
+
+
+def _infer_schema_and_create_dataframe(
+ rows: Sequence[Sequence[Any]],
+ column_names: List[str],
+ watermark: Optional[_WatermarkSpec] = None,
+) -> DataFrame:
+ row_type = _infer_schema_from_data(rows, names=column_names)
+ if watermark is None:
+ table_schema = None
+ else:
+ row_type = watermark.normalize_row_type(row_type)
+ table_schema = (
+ Schema.new_builder()
+ .from_row_data_type(row_type)
+ .watermark(*watermark)
+ .build()
+ )
+ converter = _create_converter(row_type)
+ verify_row = _create_type_verifier(row_type)
+ sql_rows = []
+ for row in rows:
+ row = converter(row)
+ verify_row(row)
+ sql_rows.append(row_type.to_sql_type(row))
+
+ table = get_or_create_table_environment()._from_elements(
+ sql_rows, row_type, table_schema
+ )
+ return DataFrame(table)
+
+
+@PublicEvolving()
+def from_table(table: Table) -> DataFrame:
+ """
+ Create a DataFrame that wraps a PyFlink Table.
+
+ :param table: Table to wrap without copying or converting it.
+ :return: A DataFrame backed by the exact supplied Table.
+ :raises TypeError: If ``table`` is not a :class:`~pyflink.table.Table`.
+
+ Example::
+
+ >>> import pyflink.dataframe as pf
+ >>> table = table_env.from_elements([(1, "Alice")], ["id", "name"])
+ >>> dataframe = pf.from_table(table)
+ >>> dataframe.to_table() is table
+ True
+
+ .. versionadded:: 2.4.0
+ """
+ if not isinstance(table, Table):
+ raise TypeError("table must be a pyflink.table.Table")
+ return DataFrame(table)
+
+
+@PublicEvolving()
+def from_pandas(
+ pdf: Any,
+ schema: Optional[List[str]] = None,
+ watermark: Optional[Tuple[str, str]] = None,
+) -> DataFrame:
+ """
+ Create a DataFrame from a pandas DataFrame.
+
+ Types are inferred from the Arrow representation of the pandas columns. An
explicit ``schema``
+ renames columns positionally and must contain exactly one unique,
non-empty name per input
+ column. Empty inputs are supported when their pandas dtypes can be
converted to Flink types.
+
+ ``watermark`` declares an event-time column and its SQL watermark
expression. The selected
+ column must have a timestamp-compatible type. Its precision is normalized
to milliseconds;
+ values with finer precision are truncated to ``TIMESTAMP(3)`` or
``TIMESTAMP_LTZ(3)``.
+
+ :param pdf: pandas DataFrame to convert.
+ :param schema: Optional list of positional result column names.
+ :param watermark: Optional ``(column, expression)`` watermark declaration.
+ :return: A DataFrame containing the pandas rows.
+ :raises TypeError: If the input, schema, watermark, or inferred types are
invalid.
+ :raises ValueError: If schema width or watermark column requirements are
not met.
+
+ Example::
+
+ >>> import pandas as pd
+ >>> import pyflink.dataframe as pf
+ >>> pdf = pd.DataFrame({"identifier": [1, 2], "name": ["Alice",
"Bob"]})
+ >>> dataframe = pf.from_pandas(pdf, schema=["id", "name"])
+ >>> events = pf.from_pandas(
+ ... pd.DataFrame({"ts": pd.to_datetime(["2026-01-01T00:00:00Z"])}),
+ ... watermark=("ts", "ts - INTERVAL '5' SECOND"),
+ ... )
+
+ .. versionadded:: 2.4.0
+ """
+ import pandas as pd
+
+ if not isinstance(pdf, pd.DataFrame):
+ raise TypeError(
+ f"data must be a pandas.DataFrame, but was {type(pdf).__name__}"
+ )
+
+ import pyarrow as pa
+
+ return from_arrow(
+ pa.Table.from_pandas(pdf, preserve_index=False),
+ schema=schema,
+ watermark=watermark,
+ )
+
+
+@PublicEvolving()
+def from_arrow(
+ table: Any,
+ schema: Optional[List[str]] = None,
+ watermark: Optional[Tuple[str, str]] = None,
+) -> DataFrame:
+ """
+ Create a DataFrame from a PyArrow Table without converting through pandas.
+
+ An explicit ``schema`` renames columns positionally and must contain
exactly one unique,
+ non-empty name per input column. Empty tables are supported when their
Arrow field types can be
+ converted to Flink types.
+
+ ``watermark`` declares an event-time column and its SQL watermark
expression. The selected
+ column must have a timestamp-compatible type. Its precision is normalized
to milliseconds;
+ values with finer precision are truncated to ``TIMESTAMP(3)`` or
``TIMESTAMP_LTZ(3)``.
+
+ :param table: PyArrow Table to convert.
+ :param schema: Optional list of positional result column names.
+ :param watermark: Optional ``(column, expression)`` watermark declaration.
+ :return: A DataFrame containing the Arrow rows.
+ :raises TypeError: If the input, schema, watermark, or inferred types are
invalid.
+ :raises ValueError: If schema width or watermark column requirements are
not met.
+
+ Example::
+
+ >>> import pyarrow as pa
+ >>> import pyflink.dataframe as pf
+ >>> table = pa.table({"id": [1, 2], "name": ["Alice", "Bob"]})
+ >>> dataframe = pf.from_arrow(table)
+ >>> events = pf.from_arrow(
+ ... pa.table({"ts": pa.array([0], type=pa.timestamp("ms"))}),
+ ... watermark=("ts", "ts - INTERVAL '5' SECOND"),
+ ... )
+
+ .. versionadded:: 2.4.0
+ """
+ import pyarrow as pa
+
+ if not isinstance(table, pa.Table):
+ raise TypeError(
+ f"data must be a pyarrow.Table, but was {type(table).__name__}"
+ )
+ watermark_spec = _WatermarkSpec.parse(watermark)
+ names = _resolve_column_names(table.column_names, schema)
+ row_type = RowType(
+ [
+ RowField(name, from_arrow_type(field.type, field.nullable))
+ for name, field in zip(names, table.schema)
+ ]
+ )
+ if watermark_spec is None:
+ table_schema = None
+ else:
+ row_type = watermark_spec.normalize_row_type(row_type)
+ table_schema = (
+ Schema.new_builder()
+ .from_row_data_type(row_type)
+ .watermark(*watermark_spec)
+ .build()
+ )
+ result = get_or_create_table_environment()._from_arrow(
Review Comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
