auroflow opened a new pull request, #29029: URL: https://github.com/apache/flink/pull/29029
## What is the purpose of the change This pull request adds DataFrame-oriented scalar UDF support to the PyFlink DataFrame API. It introduces the public `pyflink.dataframe.udf` decorator for constructing single-column expressions used by operations such as `with_column`, `with_columns`, and `select`. The implementation builds on the existing PyFlink Table API scalar UDF execution paths without requiring Java, planner, or Table API changes. The supported execution modes are: - General synchronous scalar UDFs - General asynchronous scalar UDFs - Pandas-vectorized synchronous scalar UDFs The decorator supports Python functions, callable objects, callable classes, `ScalarFunction`, and `AsyncScalarFunction`. Zero-argument class declarations are initialized on the TaskManager, while configured instances are created on the client and serialized to the TaskManager. ## Brief change log - Added the public DataFrame `udf` decorator and `DataFrameUDFWrapper`. - Added expression-oriented scalar UDF binding over the existing Table API UDF implementation. - Added return-type inference from functions, `__call__`, and `eval`, including Python types, SQL type strings, and nested `TypedDict` results. - Added explicit and annotation-inferred pandas execution mode. - Added support for synchronous and asynchronous callable declarations. - Added TaskManager-side initialization and lifecycle handling for zero-argument callable, `ScalarFunction`, and `AsyncScalarFunction` classes. - Added recursive normalization for structured single-column results, including arrays, maps, and structs. - Added API documentation and declaration examples. ### Intentionally left out of this pull request This pull request intentionally implements only the first stage of DataFrame UDF support. The following capabilities are deferred to future pull requests: - Per-UDF `concurrency` and `batch_size` options. These will be addressed by [FLINK-40471](https://issues.apache.org/jira/browse/FLINK-40471). - Arrow-native vectorized UDF execution and `func_type="arrow"`. This will be addressed by [FLINK-40472](https://issues.apache.org/jira/browse/FLINK-40472). - Asynchronous pandas-vectorized UDF execution. ## Verifying this change This change added tests and can be verified as follows: - Added table-driven declaration tests covering: - Functions, callable objects, and class declarations - `ScalarFunction` and `AsyncScalarFunction` classes and instances - Return-type inference and structured results - General, asynchronous, and pandas mode selection - Determinism and function-name metadata - Invalid declarations and constructor validation - Added adapter tests covering: - TaskManager-side construction - `open` and `close` lifecycle delegation - Determinism validation - Initialization and binding failure cleanup - Synchronous and asynchronous invocation - Added a planner-backed test for expression binding and resolved output schemas. - Added one consolidated MiniCluster integration test with one `collect()` covering general synchronous, general asynchronous, pandas, structured, callable-class, and scalar-function-class UDFs. ## 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)`: yes - The serializers: no - The runtime per-record code paths (performance sensitive): yes, for the newly introduced DataFrame scalar UDF execution path - 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? yes - If yes, how is the feature documented? API reference documentation and Python docstrings with examples --- ##### Was generative AI tooling used to co-author this PR? - [X] Yes (please specify the tool below) Generated-by: OpenAI Codex (GPT-5) -- 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]
