[FLINK-40409][runtime] CDC YAML supports inline Python UDF definition

[yuxiqian]

This PR adds support for defining inline Python UDFs in YAML pipeline jobs:

transform:
  - source-table: db.users
    projection: ID, py_normalize(EMAIL) AS EMAIL_NORM, py_double(AGE) AS DOUBLED

pipeline:
  user-defined-function:
    - name: py_normalize
      python-code: |
        def eval(value: str) -> str:
            return None if value is None else value.strip().lower()
      python-executable: /usr/bin/python3

    - name: py_double
      python-code: |
        def eval(value: int) -> int:
            return None if value is None else value * 2

The wrapper itself is implemented as a Java UDF as well. No changes are made in the existing framework except the following:

• Runtime UDF binding names are slightly changed to allow defining multiple UDFs with the same class.
• Added an overload function for UserDefinedFunction#getReturnType to pass extra context info.

---

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

• Yes (please specify the tool below)

Generated-by: Codex (gpt-5.6-sol)

[yuxiqian]

Kindly ping @lvyanquan @haruki-830 if you're interested.

[Copilot]

Pull request overview

Adds inline Python UDF definitions to YAML pipelines using Pemja.

Changes:

• Adds Python UDF execution and return-type inference.
• Extends YAML parsing, UDF context, and runtime bindings.
• Adds packaging, CI, tests, and bilingual documentation.

Reviewed changes

Copilot reviewed 25 out of 25 changed files in this pull request and generated 4 comments.

Show a summary per file

File	Description
pom.xml	Registers the Python module.
flink-cdc-runtime/.../TransformParserTest.java	Updates generated UDF bindings.
flink-cdc-runtime/.../UserDefinedFunctionDescriptorTest.java	Tests option-aware return types.
flink-cdc-runtime/.../JaninoCompiler.java	Generates name-based UDF calls.
flink-cdc-runtime/.../UserDefinedFunctionDescriptor.java	Supplies UDF configuration context.
flink-cdc-runtime/.../TransformExpressionCompiler.java	Declares name-based evaluator parameters.
flink-cdc-python/.../PythonUdfSignatureTest.java	Tests signature inference errors.
flink-cdc-python/.../PemjaTestSupport.java	Configures Pemja tests.
flink-cdc-python/.../PythonUdfTest.java	Tests execution and dependencies.
flink-cdc-python/.../signature.py	Parses Python return annotations.
flink-cdc-python/.../PythonUdfSignature.java	Maps annotations to CDC types.
flink-cdc-python/.../PythonUdf.java	Implements Python-backed UDFs.
flink-cdc-python/pom.xml	Defines and shades Pemja dependency.
flink-cdc-e2e-tests/.../unexpected.yaml	Updates accepted YAML keys.
flink-cdc-e2e-tests/.../python_udf_test.sql	Adds database fixtures.
flink-cdc-e2e-tests/.../PythonUdfE2eITCase.java	Tests Python UDF pipelines end to end.
flink-cdc-dist/pom.xml	Packages Python support.
flink-cdc-common/.../UserDefinedFunction.java	Adds context-aware return typing.
flink-cdc-cli/.../pipeline-definition-with-python-udf.yaml	Adds parser fixture.
flink-cdc-cli/.../YamlPipelineDefinitionParserTest.java	Tests Python YAML validation.
flink-cdc-cli/.../YamlPipelineDefinitionParser.java	Parses inline Python UDF fields.
docs/content/.../transform.md	Documents Python UDFs in English.
docs/content.zh/.../transform.md	Documents Python UDFs in Chinese.
.github/workflows/modules.py	Adds the module to core CI.
.github/workflows/flink_cdc_base.yml	Installs Python and Pemja in CI.

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[lvyanquan]

Hi, @yuxiqian.
The issues previously raised by Copilot may be less likely to occur in our Pipeline use case.
However, resource cleanup is worth further attention because we need to extract Python files, and Flink
does not centrally clean up the java.io.tmpdir directory. We should ensure that no temporary files are
left behind.

[yuxiqian]

Fixed. But we may have no chance to clean-up resources if native error occurs and JVM crashes.

As java.io.tmpdir usually points to /tmp where files are not written to disks, it might be acceptable.

[lvyanquan]

It would be better if you could create a jira issue to track this as we may include this feature in the release notes.

[lvyanquan]

+1.