[FLINK-40431][python] Add scalar UDF support to DataFrame API

[auroflow]

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.
• Arrow-native vectorized UDF execution and func_type="arrow". This will be addressed by 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?

• Yes (please specify the tool below)

Generated-by: OpenAI Codex (GPT-5)

[flinkbot]

CI report:

• be3a2a4 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[dianfu]

Could you add some examples in DataFrame.with_column on how to use UDF in it?

[auroflow]

Sure, I have added an example using UDF in the docstring of with_column.

[dianfu]

It can silently convert a valid property-backed field into NULL. For the following example:

class Score:
    def __init__(self, raw):
        self.raw = raw

    @property
    def doubled(self):
        return self.raw * 2


@pf.udf(
    return_dtype=pf.DataType.struct({
        "doubled": pf.DataType.int64(),
    })
)
def calculate(value):
    return Score(value)

For input value 3, the expected output should be Row(doubled=6), the actual result is Row(doubled=None)

[auroflow]

Thanks for catching this! Now structured object results evaluate fields with getattr() first, so property backed fields are preserved.

[dianfu]

It makes inspect.signature() follow the wrong object for class-based UDFs. For example:

class Add:
    def __call__(self, value: int, offset: int = 1) -> int:
        return value + offset

add = pf.udf(Add)
inspect.signature(add)  # currently: (), expected: (value, offset=1)

The same issue is more visible for a ScalarFunction instance: source.source is not callable, so inspect.signature(pf.udf(MyScalarFunction())) raises TypeError, even though the actual invocation target is eval. This can produce incorrect signatures in help(), IDEs, and documentation/introspection tooling.

Could we set signature from source.inspection_target, removing the implicit self/cls parameter for class declarations, or otherwise avoid advertising source.source as the wrapped callable?

[auroflow]

Good catch! I fixed this issue by setting an explicit __signature__ from the correct invocation target, so the result in your example is correct.

[dianfu]

The result type is constant, but _normalize_user_value() performs dynamic type dispatch for every record and every nested element. Primitive results also pay three isinstance checks even though no conversion is needed. _normalize_user_value will be executed for each result and so it's performance is very important.

Could we pre-build a type-specific normalizer in _bind_func()—identity for primitive types and recursively composed converters for Row/Array/Map—to avoid repeated dispatch on this per-record hot path?

[auroflow]

Makes sense. I replaced _normalize_user_value() with _create_result_normalizer(), which recursively constructs the Row, Array, and Map normalizers once.

[dianfu]

Do we really need to expose DataFrameUDFWrapper?

[auroflow]

I agree that DataFrameUDFWrapper does not need to be exposed. Accordingly, I changed the return type of udf() to Callable[..., Expression], which slightly deviates from our original design, but provides a simpler public API without the implementation detail. Let me know if you are happy with it!