dianfu commented on code in PR #29196:
URL: https://github.com/apache/flink/pull/29196#discussion_r4055963018
##########
flink-python/pyflink/dataframe/dataframe.py:
##########
@@ -622,6 +623,104 @@ def top_n(
distinct = drop_duplicates
unique = drop_duplicates
+ @PublicEvolving()
+ def explode(
+ self,
Review Comment:
Since this method accepts only one input column, the remaining parameters
are configuration options, what about make the other parameters keyword-only:
```
def explode(
self,
column,
*,
output_column=None,
ignore_empty_and_null=False,
)
```
This avoids ambiguous positional calls and leaves room for future options
without breaking callers.
##########
flink-python/pyflink/dataframe/dataframe.py:
##########
@@ -622,6 +623,104 @@ def top_n(
distinct = drop_duplicates
unique = drop_duplicates
+ @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 removed. Other input columns are
retained, followed by
+ the expanded fields. For a computed collection expression, all input
columns are retained.
+ MAP values yield key and value fields; ROW elements yield one field
per ROW field.
+ Empty and null collections produce a row with null output fields unless
+ ``ignore_empty_and_null`` is true.
+
+ :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.
+
+ Example::
+
+ >>> import pyflink.dataframe as pf
+ >>> df = pf.from_dict({"id": [1, 2], "tags": [["a", "b"], []]})
+ >>> result = df.explode("tags")
+ >>> result = df.explode("tags", "tag", ignore_empty_and_null=True)
+
+ .. 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]
+ if isinstance(data_type, MapType):
+ field_count = 2
+ elif isinstance(data_type, (ArrayType, MultisetType)):
+ element_type = data_type.element_type
+ field_count = len(element_type.fields) if isinstance(element_type,
RowType) else 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":
+ collection_name = columns.pop(resolved.getFieldIndex())
+ else:
+ 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")
+
+ projections = ["src." + _quote_identifier(name) for name in columns]
+ projections.extend("expanded." + _quote_identifier(name) for name in
output_names)
Review Comment:
The current implementation always appends expanded fields. What about
replacing a directly referenced column in place to follow the behavior of
engines such as pandas, Polars, etc.
For an input schema `(id, items, label)`:
- `explode("items", output_column="item")` should produce `(id, item,
label)`, not `(id, label, item)`.
- If `items` is a MAP or `ARRAY<ROW>` with outputs `["key", "value"]`, the
result should be `(id, key, value, label)`.
- For a computed expression such as `explode(array(col("id"), lit(0)),
output_column="item")`, there is no original column position to replace, so
retaining all input columns and appending the result as `(id, items, label,
item)` is reasonable.
##########
flink-python/pyflink/dataframe/tests/test_dataframe.py:
##########
@@ -2543,6 +2640,172 @@ def test_union_all_retains_duplicates(self):
)
+class DataFrameExplodeITTests(PyFlinkITTestCase):
Review Comment:
This adds 12 IT methods and, with the parameter loops, roughly 26
`collect()` executions. Since this API delegates execution to the existing SQL
UNNEST implementation, the ARRAY/MAP/MULTISET/ROW × batch/streaming × flag
matrix largely duplicates the existing UnnestITCase coverage. Note that ITCase
is expensive and should be avoid as much as possible.
Could we keep the DataFrame IT coverage focused on wrapper-specific
behavior: one batch test for the two `ignore_empty_and_null` modes, one
middle-position multi-field output, computed/direct-expression handling, one
streaming smoke test? Type resolution and validation can remain in the unit
tests.
Reducing the number of `collect()` calls, rather than only combining test
methods, should substantially lower the IT cost.
--
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]
