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 17d58a40a85 [FLINK-40435][python] Add explode support to DataFrame API
17d58a40a85 is described below
commit 17d58a40a85c97eb48c335225d231d28b5ec4269
Author: beetle0915 <[email protected]>
AuthorDate: Tue Sep 15 23:39:17 2026 +0800
[FLINK-40435][python] Add explode support to DataFrame API
This closes #29196.
---
.../docs/reference/pyflink.dataframe/dataframe.rst | 1 +
flink-python/pyflink/dataframe/dataframe.py | 175 ++++++++++++++++
.../pyflink/dataframe/tests/test_dataframe.py | 220 +++++++++++++++++++++
3 files changed, 396 insertions(+)
diff --git a/flink-python/docs/reference/pyflink.dataframe/dataframe.rst
b/flink-python/docs/reference/pyflink.dataframe/dataframe.rst
index 9c65ebdef80..13eef577d3e 100644
--- a/flink-python/docs/reference/pyflink.dataframe/dataframe.rst
+++ b/flink-python/docs/reference/pyflink.dataframe/dataframe.rst
@@ -66,6 +66,7 @@ Transformations
DataFrame.rename
DataFrame.filter
DataFrame.where
+ DataFrame.explode
DataFrame.drop_duplicates
DataFrame.distinct
DataFrame.unique
diff --git a/flink-python/pyflink/dataframe/dataframe.py
b/flink-python/pyflink/dataframe/dataframe.py
index 119d48b06b6..a3635f876b4 100644
--- a/flink-python/pyflink/dataframe/dataframe.py
+++ b/flink-python/pyflink/dataframe/dataframe.py
@@ -50,6 +50,7 @@ from pyflink.table.expressions import (
lit as table_lit,
)
from pyflink.table.table import Table
+from pyflink.table.types import ArrayType, MapType, MultisetType, RowType
from pyflink.util.api_stability_decorators import PublicEvolving
__all__ = ["DataFrame", "GroupedDataFrame", "col", "lit"]
@@ -711,6 +712,180 @@ class DataFrame:
# Table UDTFs expose positional field names, so restore the declared
names.
return DataFrame(table.alias(output_columns[0], *output_columns[1:]))
+ @PublicEvolving()
+ def explode(
+ self,
+ column: Union[str, Expression],
+ *,
+ output_column: Optional[Union[str, List[str]]] = None,
+ ignore_empty_and_null: bool = False,
+ ) -> "DataFrame":
+ """
+ Expand an ARRAY, MAP, or MULTISET into rows, preserving duplicate
occurrences.
+
+ A referenced input column is replaced in place by the expanded
element. For a computed
+ collection expression, all input columns are retained and the expanded
element is
+ appended. MAP values yield key and value fields. ROW elements remain a
single ROW column;
+ use :attr:`pyflink.table.Expression.flatten` in a subsequent
:meth:`select` to expand its
+ fields. Empty and null collections produce a row with null output
fields unless
+ ``ignore_empty_and_null`` is true.
+
+ .. warning::
+ Due to FLINK-40658, actual null ROW elements in an ARRAY are not
handled correctly
+ and may be silently dropped or cause execution to fail. This
limitation is independent
+ of ``ignore_empty_and_null``. A non-null ROW whose fields are all
null is not affected.
+
+ :param column: Collection column name or row-wise expression to expand.
+ :param output_column: Output name or list of names. Required for
multiple fields;
+ a single field defaults to the selected column name. Names must be
unique and must
+ not conflict with retained input columns.
+ :param ignore_empty_and_null: Whether to drop rows with empty or null
collections.
+ :return: A new DataFrame with the expanded rows.
+ :raises TypeError: If an argument has an unsupported type or the input
is not a collection.
+ :raises ValueError: If the expression is not row-wise, selects
multiple columns,
+ or output names are invalid.
+
+ Examples:
+
+ Explode an ARRAY of scalar values::
+
+ >>> import pyflink.dataframe as pf
+ >>> df = pf.from_dict({"id": [1, 2], "tags": [["a", "b"], []]})
+ >>> result = df.explode(
+ ... "tags", output_column="tag", ignore_empty_and_null=True)
+
+ Explode an ARRAY of ROW values, then expand the ROW fields explicitly::
+
+ >>> import pyflink.dataframe as pf
+ >>> from typing import NamedTuple
+ >>> class Item(NamedTuple):
+ ... name: str
+ ... quantity: int
+ >>> orders = pf.from_records(
+ ... [(1, [Item("apple", 2)])], schema=["id", "items"])
+ >>> exploded = orders.explode("items", ignore_empty_and_null=True)
+ >>> flattened = exploded.select("id", pf.col("items").flatten)
+ >>> flattened.columns
+ ['id', 'items$name', 'items$quantity']
+
+ .. versionadded:: 2.4.0
+ """
+ if not isinstance(column, (str, Expression)):
+ raise TypeError("column must be a column name or expression")
+ if not isinstance(ignore_empty_and_null, bool):
+ raise TypeError("ignore_empty_and_null must be a boolean")
+ if output_column is not None and not isinstance(output_column, (str,
list)):
+ raise TypeError("output_column must be a string or list of
strings")
+
+ expression = table_col(column) if isinstance(column, str) else column
+ selected = self._table.select(expression)
+ schema = selected.get_resolved_schema()
+ if len(schema.get_column_names()) != 1:
+ raise ValueError("column must select a single column")
+ projection = selected._j_table.getQueryOperation()
+ # Aggregates insert an intermediate operation whose field indexes
refer to its result.
+ if not
projection.getChildren().get(0).equals(self._table._j_table.getQueryOperation()):
+ raise ValueError("column must be a row-wise expression, not an
aggregation")
+
+ data_type = schema.get_column_data_types()[0]
+ element_row_type = None
+ element_row_type_sql = None
+ if isinstance(data_type, MapType):
+ field_count = 2
+ elif isinstance(data_type, (ArrayType, MultisetType)):
+ element_type = data_type.element_type
+ element_row_type = element_type if isinstance(element_type,
RowType) else None
+ if element_row_type is not None:
+ collection_type =
schema._j_resolved_schema.getColumnDataTypes().get(0)
+ element_row_type_sql = (
+ collection_type.getChildren()
+ .get(0)
+ .getLogicalType()
+ .copy(True)
+ .asSerializableString()
+ )
+ field_count = 1
+ else:
+ raise TypeError("column must have an ARRAY, MAP, or MULTISET type")
+
+ if output_column is None:
+ if field_count != 1:
+ raise ValueError("output_column is required for multiple
output fields")
+ output_names = [schema.get_column_names()[0]]
+ else:
+ output_names = [output_column] if isinstance(output_column, str)
else output_column
+ if not all(isinstance(name, str) for name in output_names):
+ raise TypeError("output_column must contain only strings")
+ if len(output_names) != field_count:
+ raise ValueError("output_column must contain %d name(s)" %
field_count)
+ if any(not name for name in output_names) or len(set(output_names)) !=
field_count:
+ raise ValueError("output_column names must be non-empty and
unique")
+
+ table = self._table
+ columns = list(table.get_resolved_schema().get_column_names())
+ resolved = projection.getProjectList().get(0)
+ # Resolve the input field by index so an alias does not hide the
column to remove.
+ if resolved.getClass().getSimpleName() == "FieldReferenceExpression":
+ output_index = resolved.getFieldIndex()
+ collection_name = columns.pop(output_index)
+ else:
+ output_index = len(columns)
+ taken = set(columns) | set(output_names)
+ collection_name = _unique_name("__pf_explode", taken)
+ table = table.add_columns(expression.alias(collection_name))
+ if set(output_names).intersection(columns):
+ raise ValueError("output_column names conflict with retained input
columns")
+
+ unnest_output_names = output_names
+ ordinality_name = None
+ if element_row_type is not None:
+ # SQL UNNEST expands ROW fields. Reassemble them below, using
ordinality to
+ # distinguish an outer-join placeholder from a real ROW whose
fields are all null.
+ taken = set(columns) | set(output_names) | {collection_name}
+ unnest_output_names = []
+ for index in range(len(element_row_type.fields)):
+ name = _unique_name("__pf_explode_field_%d" % index, taken)
+ taken.add(name)
+ unnest_output_names.append(name)
+ if not ignore_empty_and_null:
+ ordinality_name = _unique_name("__pf_explode_ordinality",
taken)
+
+ if element_row_type is None:
+ explode_projection = [
+ "expanded." + _quote_identifier(name) for name in
unnest_output_names
+ ]
+ else:
+ row_value = "CAST(ROW(%s) AS %s)" % (
+ ", ".join(
+ "expanded." + _quote_identifier(name) for name in
unnest_output_names
+ ),
+ element_row_type_sql,
+ )
+ if ordinality_name is not None:
+ row_value = "CASE WHEN expanded.%s IS NULL THEN CAST(NULL AS
%s) ELSE %s END" % (
+ _quote_identifier(ordinality_name),
+ element_row_type_sql,
+ row_value,
+ )
+ explode_projection = [row_value + " AS " +
_quote_identifier(output_names[0])]
+
+ projections = ["src." + _quote_identifier(name) for name in columns]
+ projections[output_index:output_index] = explode_projection
+ query = "SELECT %s FROM %s AS src %s UNNEST(src.%s)%s AS
expanded(%s)%s" % (
+ ", ".join(projections),
+ _quote_identifier(str(table)),
+ "CROSS JOIN" if ignore_empty_and_null else "LEFT JOIN",
+ _quote_identifier(collection_name),
+ " WITH ORDINALITY" if ordinality_name is not None else "",
+ ", ".join(
+ _quote_identifier(name)
+ for name in unnest_output_names
+ + ([ordinality_name] if ordinality_name is not None else [])
+ ),
+ "" if ignore_empty_and_null else " ON TRUE",
+ )
+ return DataFrame(table._t_env.sql_query(query))
+
# ======================== Filtering & Ordering ========================
@PublicEvolving()
diff --git a/flink-python/pyflink/dataframe/tests/test_dataframe.py
b/flink-python/pyflink/dataframe/tests/test_dataframe.py
index 510e5af1529..f7be076ca5a 100644
--- a/flink-python/pyflink/dataframe/tests/test_dataframe.py
+++ b/flink-python/pyflink/dataframe/tests/test_dataframe.py
@@ -38,6 +38,7 @@ from pyflink.dataframe.datatype import DataType
from pyflink.table import (
DataTypes as TableDataTypes,
EnvironmentSettings,
+ Table,
TableEnvironment,
TableSchema,
)
@@ -332,6 +333,134 @@ class
DataFrameSetOperationTests(PyFlinkDataFrameUTTestCase):
getattr(left, method)(right)
+class DataFrameExplodeTests(PyFlinkDataFrameUTTestCase):
+ def setUp(self):
+ super().setUp()
+ self.df = pf.from_table(self.t_env.sql_query(
+ "SELECT 1 AS id, ARRAY[1, 2] AS items, 'A' AS label"))
+
+ def test_explode_is_lazy_and_preserves_input(self):
+ table = self.df.to_table()
+ with patch.object(Table, "execute") as execute:
+ result = self.df.explode("items")
+ execute.assert_not_called()
+ self.assertIsNot(result, self.df)
+ self.assertIs(result.to_table()._t_env, self.t_env)
+ self.assertIs(self.df.to_table(), table)
+ self.assert_dataframe_schema(self.df, ["id", "items", "label"])
+ self.assert_dataframe_schema(result, ["id", "items", "label"], [
+ TableDataTypes.INT().not_null(), TableDataTypes.INT(),
+ TableDataTypes.CHAR(1).not_null(),
+ ])
+
+ def test_explode_supports_column_expressions_and_aliases(self):
+ for column, output, expected in [
+ ("items", None, "items"),
+ (pf.col("items"), None, "items"),
+ (self.df["items"], "item", "item"),
+ (pf.col("items").alias("renamed"), None, "renamed"),
+ ("items", ["item"], "item"),
+ ]:
+ with self.subTest(column=str(column), output=output):
+ self.assert_dataframe_schema(
+ self.df.explode(column, output_column=output), ["id",
expected, "label"])
+
+ def test_explode_supports_computed_collection_expressions(self):
+ from pyflink.table.expressions import array
+
+ result = self.df.explode(
+ array(pf.col("id"), pf.lit(2)), output_column="value")
+ self.assert_dataframe_schema(result, ["id", "items", "label", "value"])
+
+ def test_explode_quotes_identifiers(self):
+ df = pf.from_table(self.t_env.sql_query(
+ "SELECT 1 AS `select`, ARRAY[2] AS `a``b`, 3 AS `value`"))
+
+ result = df.explode(pf.col("a`b"), output_column="value` name")
+
+ self.assert_dataframe_schema(result, ["select", "value` name",
"value"])
+
+ def test_explode_resolves_collection_output_types(self):
+ item_type = TableDataTypes.ROW([
+ TableDataTypes.FIELD("number", TableDataTypes.INT()),
+ TableDataTypes.FIELD("text", TableDataTypes.STRING()),
+ ]).not_null()
+ single_field_item_type = TableDataTypes.ROW([
+ TableDataTypes.FIELD("number", TableDataTypes.INT()),
+ ]).not_null()
+ for sql, output, ignore, names, types in [
+ ("SELECT 1 AS id, MAP['a', 1] AS items, 'x' AS label",
+ ["key", "value"], False, ["id", "key", "value", "label"],
+ [TableDataTypes.INT().not_null(), TableDataTypes.CHAR(1),
+ TableDataTypes.INT(), TableDataTypes.CHAR(1).not_null()]),
+ ("SELECT ARRAY[CAST(ROW(1, 'a') AS ROW<number INT, text STRING>)]
AS items",
+ "item", True, ["item"], [item_type]),
+ ("SELECT ARRAY[CAST(ROW(1) AS ROW<number INT>)] AS items",
+ None, True, ["items"], [single_field_item_type]),
+ ("SELECT MULTISET[1, 1, 2] AS items", None, False, ["items"],
+ [TableDataTypes.INT()]),
+ ]:
+ with self.subTest(sql=sql):
+ df = pf.from_table(self.t_env.sql_query(sql))
+ self.assert_dataframe_schema(
+ df.explode(
+ "items",
+ output_column=output,
+ ignore_empty_and_null=ignore,
+ ),
+ names,
+ types,
+ )
+
+ def test_explode_rejects_invalid_arguments(self):
+ for column in [None, 1, ["items"]]:
+ with self.subTest(column=column):
+ with self.assertRaisesRegex(TypeError, "column"):
+ self.df.explode(column)
+ for flag in [None, 1, "true"]:
+ with self.subTest(flag=flag):
+ with self.assertRaisesRegex(TypeError,
"ignore_empty_and_null"):
+ self.df.explode("items", ignore_empty_and_null=flag)
+ for output in [1, ("item",), [1]]:
+ with self.subTest(output=output):
+ with self.assertRaisesRegex(TypeError, "output_column"):
+ self.df.explode("items", output_column=output)
+ with self.assertRaisesRegex(TypeError, "positional argument"):
+ self.df.explode("items", "item")
+
+ def test_explode_rejects_missing_and_non_collection_columns(self):
+ for column in ["missing", pf.col("missing")]:
+ with self.subTest(column=str(column)):
+ with self.assertRaisesRegex(Py4JJavaError, "missing"):
+ self.df.explode(column)
+ with self.assertRaisesRegex(TypeError, "ARRAY, MAP, or MULTISET"):
+ self.df.explode("id")
+ with self.assertRaisesRegex(ValueError, "single column"):
+ self.df.explode(pf.col("*"))
+
+ def test_explode_rejects_aggregate_expressions(self):
+ df = pf.from_table(self.t_env.sql_query("SELECT ARRAY[9] AS items, 1
AS id"))
+ with self.assertRaisesRegex(ValueError, "row-wise"):
+ df.explode(pf.col("id").collect, output_column="value")
+
+ def test_explode_rejects_invalid_output_names(self):
+ for output in [[], ["a", "b"], "", [""]]:
+ with self.subTest(output=output):
+ with self.assertRaisesRegex(ValueError, "output_column"):
+ self.df.explode("items", output_column=output)
+ with self.assertRaisesRegex(ValueError, "conflict"):
+ self.df.explode("items", output_column="id")
+ df = pf.from_table(self.t_env.sql_query("SELECT MAP['a', 1] AS items"))
+ for output in [None, "item", ["item"], ["item", "item"]]:
+ with self.subTest(output=output):
+ with self.assertRaisesRegex(ValueError, "output_column"):
+ df.explode("items", output_column=output)
+ df = pf.from_table(self.t_env.sql_query(
+ "SELECT ARRAY[CAST(ROW(1, 'a') AS ROW<number INT, text STRING>)]
AS items"))
+ with self.assertRaisesRegex(ValueError, "output_column"):
+ df.explode("items", output_column=["number", "text"])
+
+
class DataFrameSortingTests(PyFlinkDataFrameUTTestCase):
def setUp(self):
super().setUp()
@@ -2686,6 +2815,97 @@ class
DataFrameSetOperationStreamITTests(PyFlinkStreamDataFrameTestCase):
)
+class DataFrameExplodeITTests(PyFlinkITTestCase):
+ def setUp(self):
+ self.t_env =
TableEnvironment.create(EnvironmentSettings.in_batch_mode())
+
+ def test_explode_direct_column_with_both_empty_collection_modes(self):
+ df = pf.from_table(self.t_env.from_elements(
+ [(1, [2, 2, None], "a"), (2, [], "b"), (3, None, "c")],
+ TableDataTypes.ROW([
+ TableDataTypes.FIELD("id", TableDataTypes.INT()),
+ TableDataTypes.FIELD("items",
TableDataTypes.ARRAY(TableDataTypes.INT())),
+ TableDataTypes.FIELD("label", TableDataTypes.STRING()),
+ ])))
+ for ignore in [False, True]:
+ with self.subTest(ignore=ignore):
+ expected = [Row(1, 2, "a"), Row(1, 2, "a"), Row(1, None, "a")]
+ if not ignore:
+ expected += [Row(2, None, "b"), Row(3, None, "c")]
+ self.assertCountEqual(
+ df.explode("items",
ignore_empty_and_null=ignore).collect(), expected)
+
+ def test_explode_preserves_row_elements_and_distinguishes_outer_null(self):
+ df = pf.from_table(self.t_env.from_elements(
+ [(1, [Row(2, "a"), Row(3, "b")], "x")],
+ TableDataTypes.ROW([
+ TableDataTypes.FIELD("id", TableDataTypes.INT()),
+ TableDataTypes.FIELD("items",
TableDataTypes.ARRAY(TableDataTypes.ROW([
+ TableDataTypes.FIELD("number", TableDataTypes.INT()),
+ TableDataTypes.FIELD("text", TableDataTypes.STRING()),
+ ]))),
+ TableDataTypes.FIELD("label", TableDataTypes.STRING()),
+ ])))
+ exploded = df.explode("items", ignore_empty_and_null=True)
+ self.assertEqual(exploded.columns, ["id", "items", "label"])
+ flattened = exploded.select("id", pf.col("items").flatten, "label")
+ self.assertEqual(
+ flattened.columns, ["id", "items$number", "items$text", "label"])
+ self.assertCountEqual(
+ flattened.collect(),
+ [
+ Row(1, 2, "a", "x"),
+ Row(1, 3, "b", "x"),
+ ],
+ )
+
+ multiset_df = pf.from_table(self.t_env.sql_query(
+ "SELECT id, COLLECT(item) AS items, MAX(label) AS label FROM "
+ "(VALUES "
+ "(1, CAST(ROW(2, 3) AS ROW<number INT, code INT>), 9), "
+ "(1, CAST(ROW(2, 3) AS ROW<number INT, code INT>), 9), "
+ "(1, CAST(ROW(CAST(NULL AS INT), CAST(NULL AS INT)) "
+ "AS ROW<number INT, code INT>), 9)) AS T(id, item, label) GROUP BY
id "
+ "UNION ALL SELECT 3, "
+ "CAST(NULL AS ROW<number INT, code INT> MULTISET), 8"))
+ distinguished = multiset_df.explode("items").select(
+ "id",
+ pf.col("items").is_null.alias("item_is_null"),
+ pf.col("items").get("number").alias("number"),
+ pf.col("items").get("code").alias("code"),
+ "label",
+ )
+ self.assertCountEqual(
+ distinguished.collect(),
+ [
+ Row(1, False, 2, 3, 9),
+ Row(1, False, 2, 3, 9),
+ Row(1, False, None, None, 9),
+ Row(3, True, None, None, 8),
+ ],
+ )
+
+ def test_explode_computed_expression_appends_output(self):
+ from pyflink.table.expressions import array
+
+ df = pf.from_table(self.t_env.sql_query("SELECT 1 AS __pf_explode"))
+ result = df.explode(
+ array(pf.col("__pf_explode"), pf.lit(2)).alias("pair"),
output_column="value")
+ self.assertEqual(result.columns, ["__pf_explode", "value"])
+ self.assertCountEqual(result.collect(), [Row(1, 1), Row(1, 2)])
+
+
+class DataFrameExplodeStreamITTests(PyFlinkStreamDataFrameTestCase):
+ def test_explode_in_streaming_mode(self):
+ df = pf.from_table(self.t_env.from_elements(
+ [(1, [2, 3])],
+ TableDataTypes.ROW([
+ TableDataTypes.FIELD("id", TableDataTypes.INT()),
+ TableDataTypes.FIELD("items",
TableDataTypes.ARRAY(TableDataTypes.INT())),
+ ])))
+ self.assertCountEqual(df.explode("items").collect(), [Row(1, 2),
Row(1, 3)])
+
+
class DataFrameWindowITTests(PyFlinkStreamDataFrameTestCase):
@classmethod
def setUpClass(cls):
