auroflow opened a new pull request, #29652: URL: https://github.com/apache/flink/pull/29652
## What is the purpose of the change Mixing synchronous and asynchronous Python scalar UDFs can produce incorrectly ordered columns in both streaming and batch execution. For example, `select(sync_add(a), async_add(a))`, with input `1` and functions adding `1` and `2`, returns `(3, 2)` instead of `(2, 3)`. Different result types can also cause incorrect decoding or type-related exceptions. Calc splitting can produce projections that interleave UDF calls and input references, while Python operators assemble forwarded fields followed by UDF results. This change restores the logical projection order inside the Python operators without introducing an additional Calc. This closes [FLINK-41253](https://issues.apache.org/jira/browse/FLINK-41253). ## Brief change log - Map joined Python scalar output fields to their logical projection positions, retaining the existing collector for identity mappings. - Derive Python async UDF result types from actual call positions instead of assuming they occupy the trailing output columns. - Use the logical output type for the Python Calc transformation. - Add regression coverage for projection order, heterogeneous types, nulls, row kinds, nested async calls, and deduplication. ## Verifying this change This change added tests and can be verified as follows: - `PythonAsyncProjectionOrderTest` verifies that mixed projections require no additional reordering Calc and repeated async calls remain deduplicated. - `PythonScalarFunctionOperatorTestBase` verifies output types, nulls, and INSERT/DELETE row kinds for ordinary and Arrow Python scalar operators. - `test_mixed_projection_order` verifies mixed and nested calls, heterogeneous types, nulls, and repeated input references in streaming and batch modes, with operator chaining enabled and disabled. ## Does this pull request potentially affect one of the following parts: - Dependencies (does it add or upgrade a dependency): no - The public API, i.e., is any changed class annotated with `@Public(Evolving)`: no - The serializers: yes — corrects the types supplied for Python async result decoding and downstream serialization; no serialization format changes - The runtime per-record code paths (performance sensitive): yes - Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no - The S3 file system connector: no ## Documentation - Does this pull request introduce a new feature? no - If yes, how is the feature documented? not applicable --- ##### Was generative AI tooling used to co-author this PR? - [X] Yes (please specify the tool below) Generated-by: Codex (GPT-6) -- 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]
