beetle0915 opened a new pull request, #29457: URL: https://github.com/apache/flink/pull/29457
## What is the purpose of the change Implement [FLINK-40856](https://issues.apache.org/jira/browse/FLINK-40856) by extending `DataFrame.explode()` to expand multiple ARRAY columns or row-wise expressions together by position. Unlike chaining single-column `explode()` calls, this does not produce a Cartesian product. The implementation follows the updated JIRA semantics: all arrays must have the same effective length in each input row, with empty and null arrays treated as length zero. Unequal lengths fail at execution time, regardless of `ignore_empty_and_null`. ## Brief change log - Accept a non-empty list of inputs; retain the existing single-input behavior for one-element lists. - Add an internal Java table function that validates all lengths before emitting one-based ordinals, including when the first array is empty or null. Build the expansion using a lateral join and indexed ARRAY access. - Replace directly referenced columns in their original positions and append computed-expression outputs. Preserve ARRAY<ROW> elements as a single ROW column. - Validate input types, duplicate direct inputs, output-name counts and name conflicts. - Add Java unit tests and Python schema, validation, batch and streaming tests. Combine successful flag variants into one `collect()` per mode to limit integration-test jobs. - Update the DataFrame reference and method docstring. ## Verifying this change Verified locally on the current PR head using Java 17 and Python 3.12: - `./mvnw -pl flink-python -Dtest=ExplodeOrdinalityFunctionTest test`: 4 tests passed; Checkstyle and Spotless passed. - `python -m pytest flink-python/pyflink/dataframe/tests/test_dataframe.py -q`: 298 tests passed, including the existing single-input tests and new batch/stream multi-input cases. - Flake8 with `flink-python/tox.ini`, mypy for `dataframe.py`, and `git diff --check` passed. Full-repository `./mvnw clean verify` and community CI have not been validated. This PR is a draft pending that validation and the FLIP-591 update noted below. ## Does this pull request potentially affect one of the following parts: - Dependencies: no. - The public API: yes, extends the `@PublicEvolving` DataFrame `explode()` input argument to accept lists. - The serializers: no. - The runtime per-record code paths: yes, adds length validation and ordinal generation for multi-input expansion only; existing single-input expansion is unchanged. - Deployment or recovery: no. - The S3 file system connector: no. ## Documentation - New feature: yes. - Documented in the DataFrame reference and method docstring. - FLINK-40856 also requests updating [FLIP-591](https://cwiki.apache.org/confluence/spaces/FLINK/pages/430409184/FLIP-591+Introducing+Python+DataFrame+API+in+PyFlink) with the API extension and semantics. That Confluence update is still pending and is not included in this repository change. --- ##### Was generative AI tooling used to co-author this PR? - [X] Yes (please specify the tool below) Generated-by: OpenAI Codex CLI 0.160.0 -- 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]
