[FLINK-40434][python] Add flat_map support to the DataFrame API

[hsrain3]

What is the purpose of the change

This PR implements FLINK-40434 by adding DataFrame.flat_map for transforming each input row into zero or more output rows. It also introduces pf.udtf for reusable table function declarations. The transformation is lazy and produces only the UDTF output columns.

Brief change log

Add DataFrame.flat_map and pf.udtf, supporting Python callables and TableFunction instances or classes.
Support explicit output types and inference from iterable return annotations, preserving named output fields.
Provide dictionary input for plain callables and named Row input for UDTF declarations in both process and thread modes.
Add API documentation and examples.

Verifying this change

his change added tests and can be verified as follows:

• Added tests for declarations, type inference, validation, output normalization, lifecycle, serialization, and declaration reuse.
• Added bounded integration tests covering batch, streaming, and thread execution modes.
• Ran DataFrame UDTF tests, existing DataFrame UDF tests, and Table UDTF regression tests: 72 passed.
• Python lint, implementation type checking, and documentation build passed.

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, adds PublicEvolving DataFrame APIs
• The serializers: no
• The runtime per-record code paths (performance sensitive): yes, introduces input/output adaptation for the new flat_map operation
• 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, including API reference, Python docstrings, and usage examples

---

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

• Yes (please specify the tool below)

Generated-by: Codex(GPT-6)

[flinkbot]

CI report:

• e8e399e Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

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

[dianfu]

Could you move it to the end of "Core Operations" group?

[hsrain3]

Sure. Moved flat_map to the end of the “Core Operations” group.

[dianfu]

It's strange that when decorated with @pf.udtf, the input type is Row, otherwise, it's a dict. Why not always make it a dict?

[hsrain3]

Make sense. flat_map now consistently passes each input row as a dict.

[dianfu]

It would also be useful to demonstrate the explicit return_dtype path, the output column names, udfs defined with Callable Classes, Classes extending TableFunction, etc?

[dianfu]

It seems that it doesn't support callable class? It worth to support it also.

[hsrain3]

Agreed. Callable instances were already supported. Added support for passing callable classes directly to flat_map and pf.udtf.
Added examples for explicit return_dtype and output column names, plus callable class and TableFunction examples.

[dianfu]

Another problem worth to address: this change introduces the public pf.udtf declaration, but pyflink.dataframe.sql still recognizes only _DataFrameUDFWrapper:

• _BINDABLE_TYPES excludes _DataFrameUDTFWrapper;
• _get_udfs() does not collect UDTF declarations;
• function registration assumes the scalar-UDF-specific _table_udf_wrapper.

As a result, explicit binding currently fails before SQL planning:

source = pf.from_dict({"text": ["ab"]})

@pf.udtf(return_dtype=str)
def chars(text):
    yield from text

pf.sql(
    "SELECT ch FROM source, LATERAL TABLE(chars(text)) AS T(ch)",
    auto_bind=False,
    source=source,
    chars=chars,
)

[hsrain3]

Another problem worth to address: this change introduces the public pf.udtf declaration, but pyflink.dataframe.sql still recognizes only _DataFrameUDFWrapper:

• _BINDABLE_TYPES excludes _DataFrameUDTFWrapper;
• _get_udfs() does not collect UDTF declarations;
• function registration assumes the scalar-UDF-specific _table_udf_wrapper.

As a result, explicit binding currently fails before SQL planning:

source = pf.from_dict({"text": ["ab"]})

@pf.udtf(return_dtype=str)
def chars(text):
    yield from text

pf.sql(
    "SELECT ch FROM source, LATERAL TABLE(chars(text)) AS T(ch)",
    auto_bind=False,
    source=source,
    chars=chars,
)

@dianfu Thanks for catching this! Added UDTF support to explicit binding validation, auto-binding discovery, and function registration.

[dianfu]

@hsrain3 Thanks for the update. LGTM.