[FLINK-40197][python] Support UDF bindings in DataFrame sql()

[fdolce]

What is the purpose of the change

pyflink.dataframe.sql() currently binds only DataFrames. This PR extends it to also bind UDFs created with pyflink.dataframe.udf, so a query can call a Python UDF the same way it references a DataFrame: either by its Python variable name via auto-bind, or under a chosen SQL name via an explicit keyword binding. UDFs are registered as temporary system functions for the duration of the call and dropped afterwards, mirroring how DataFrames are registered as temporary views.

@pf.udf
def add_one(value: int) -> int:
    return value + 1

pf.sql("SELECT add_one(a) AS a1 FROM df1")
pf.sql("SELECT inc(a) FROM src", auto_bind=False, src=df1, inc=add_one)

Brief change log

• sql() accepts _DataFrameUDFWrapper values in **bindings and picks them up from the caller's scope when auto_bind=True; other types still raise TypeError.
• UDFs are registered with create_temporary_system_function so they resolve by bare name independently of the current catalog/database, and are dropped in finally.
• Shadowing rules mirror the existing view rules: explicit bindings may shadow built-in and permanent catalog functions but raise ValueError on collision with an existing temporary function; auto-bound UDFs never shadow any existing function and are skipped with a warning instead. Name collision checks are case-insensitive, matching the function catalog.
• Registration of explicit bindings is all-or-nothing: if a later explicit binding is rejected, views/functions registered earlier in the same call are dropped before raising.
• UDF bindings do not take part in TableEnvironment resolution, which is still driven by the bound DataFrames only.
• Refactored the internals of sql.py into _register_views / _register_functions / _drop_views / _drop_functions helpers; updated docstrings and docs/reference/pyflink.dataframe/sql.rst.

Verifying this change

This change added tests and can be verified as follows:

• Extended flink-python/pyflink/dataframe/tests/test_sql.py with tests covering:
  • auto-bind of local and module-level UDFs, explicit bindings choosing the SQL name, explicit bindings taking precedence over auto-bind, auto_bind=False ignoring caller UDFs
  • case-insensitive function names, pandas UDFs, composing the result with the DataFrame API
  • cleanup of registered functions after a failing query, and rollback of earlier explicit bindings when a later one fails
  • collision handling: auto-bind warns and skips on collision with existing user-defined, built-in and permanent catalog functions; explicit bindings shadow built-in and permanent functions but raise on temporary (system and catalog) function collisions, case-insensitively
  • invalid SQL identifiers as explicit UDF names raise ValueError
  • UDF bindings do not influence TableEnvironment resolution
• Existing tests asserting that pyflink.table.udf objects are rejected / ignored were kept (renamed to *_table_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 (pyflink.dataframe.sql, @PublicEvolving, accepts UDFs in bindings; backwards compatible)
• The serializers: no
• The runtime per-record code paths (performance sensitive): no
• 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? docs (docs/reference/pyflink.dataframe/sql.rst) and the sql() docstring

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: Claude Code (Claude Fable 5.1)

[flinkbot]

CI report:

• 7e72270 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

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

[dianfu]

The registered names are assigned to views / functions only after each helper returns successfully. If an exception occurs after a partial auto-bind, the outer finally cannot see or clean up the objects that were already registered.

[fdolce]

It's true that the outer finally doesn't catch that, but both _register_views and _register_functions have their own internal try/catch to drop partially registered ones. If you prefer, I can move the responsibility of the drop up here, but as it is it should work correctly, or am I missing anything else?

[dianfu]

The internal cleanup only covers the explicit-binding loops. The auto-binding loops are outside those try/except blocks.

The same issue exists in _register_views(). Moving the registration tracker to the outer scope, or wrapping both explicit and auto registration in the helper cleanup block, would cover this path.

[dianfu]

The temporary functions cannot be derived by subtracting these two name sets. list_user_defined_functions() merges temporary system, temporary catalog, and permanent catalog functions, so the source information has already been lost.

For example, if both a permanent function f and a temporary catalog function f exist:

• user_defined == {"f"}
• _permanent_functions(t_env) == {"f"}
• temporary_functions == set()

An explicit f=my_udf binding is then allowed and registered as a temporary system function, which shadows the existing temporary function instead of raising the documented ValueError.

[fdolce]

Ah, yes, good catch, working on this

[dianfu]

The public type contract here does not match the return type of pf.udf().

sql() accepts _DataFrameUDFWrapper, while the pf.udf() overloads are declared to return Callable[..., Expression].

The documented usage works at runtime, but it will be rejected by static type checkers:

df = pf.from_dict({"a": [1]})

@pf.udf
def inc(value: int) -> int:
    return value + 1

result = pf.sql(
    "SELECT inc(a) FROM df",
    auto_bind=False,
    df=df,
    inc=inc,
)

At runtime, inc is an _DataFrameUDFWrapper. However, the pf.udf() overload declares it as Callable[..., Expression], which is incompatible with the _Binding annotation expected by pf.sql().

Do you think it makes sense to update the return type of py.udf from Callable[..., Expression] to _DataFrameUDFWrapper though it's not directly introduced by this PR?

[fdolce]

True. Mh, having pf.udf return _DataFrameUDFWrapper kind of makes that class a part of the public api though right? Let me see if I find any alternative solution here, I'll let you know

[dianfu]

Should we normalize the explicit names first and filter auto candidates using the normalized name set?

For the following example:

add_one = pf.udf(...)
pf.sql("SELECT ADD_ONE(a)", ADD_ONE=override)

add_one is not removed from auto_candidates because this comparison uses the original Python keys. The explicit function is registered first, and the auto-bound add_one then produces a misleading "function already exists" warning.

[fdolce]

True, working on this