[FLINK-39986][table-planner][python] Support Python UDF deduplication in projection and condition

[raoraoxiong]

What is the purpose of the change

This PR is the first of two PRs implementing common sub-expression elimination (CSE) for Python UDFs (FLINK-39986). Each invocation of a Python UDF involves cross-process communication between the JVM and the Python worker, so duplicated calls are significantly more expensive than duplicated Java expressions.

This PR deduplicates Python UDF calls on the JVM side only (no protocol or Python worker changes), covering two scenarios:

1. Top-level projection duplicates: identical deterministic calls in the projection (e.g. SELECT udf(a), udf(a)) are sent to the Python worker only once; a codegen expansion projection maps the deduplicated results back to the original output schema.
2. Condition-projection sharing: after RemoteCalcSplitConditionRule splits a Calc with Python UDFs in its condition, the new RemoteCalcConditionProjectionCseRule rewrites the top Calc so that projections reuse the Python UDF results already computed for the WHERE condition (e.g. SELECT udf(a) + 1 FROM T WHERE udf(a) > 0), including calls nested inside Java expressions.

Non-deterministic calls are never deduplicated and are always evaluated independently.

Deduplication of Python UDF calls nested inside other Python UDF calls (e.g. udf(udf(a))), which requires extending the JVM-to-worker protocol with result references, is addressed in the follow-up PR #28998.

Brief change log

• Add PythonCallDeduplicator and PythonCallCseResult for structural deduplication of top-level projection calls (deduplication by RexCall structural equivalence, which correctly distinguishes calls that differ only in return type)
• Append a CSE expansion projection in CommonExecPythonCalc when duplicated results need to be restored to the output schema
• Add ProjectionCodeGenerator#generateProjectionOperator for the codegen column-mapping projection operator
• Add RemoteCalcConditionProjectionCseRule and register it in stream and batch rule sets after SPLIT_CONDITION

Verifying this change

This change added tests and can be verified as follows:

• CommonExecPythonCalcCseTest: parameterized unit tests for the deduplication logic (deterministic/non-deterministic/mixed/return-type-sensitive cases) and the expansion projection detail name
• PythonCalcConditionCseTest (+ plan XML): plan tests verifying condition-projection deduplication, including negative cases (different UDFs are not deduplicated)
• test_python_local_ref_reuse in flink-python/pyflink/table/tests/test_udf.py: end-to-end tests verifying deterministic calls are reused and non-deterministic calls are not (UUID suffix in UDF output proves whether calls were deduplicated)

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): (no)
• The serializers: (no)
• The runtime per-record code paths (performance sensitive): (yes, Python UDF execution path; fewer UDF invocations per record when duplicates exist, a pass-through when there are no duplicates)
• 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? (not applicable — transparent optimization, no user-facing API change)

AI Usage Disclosure

• This contribution was developed with AI assistance (see ASF Generative Tooling Guidance)

Generated-by: Claude-4.6-Opus

[flinkbot]

CI report:

• 26333c9 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

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

[raoraoxiong]

hi @liuyongvs @snuyanzin ，please take a look.

[snuyanzin]

1. please follow contributors guidelines first, current PR is not ready to be reviewed. As an example: CI is not green
2. Please do not create new classes in scala
3. do not use deprecated Calcite api especially for new rules

[raoraoxiong]

@flinkbot run azure

[raoraoxiong]

@flinkbot re-run azure

[raoraoxiong]

1. please follow contributors guidelines first, current PR is not ready to be reviewed. As an example: CI is not green
2. Please do not create new classes in scala
3. do not use deprecated Calcite api especially for new rules

@snuyanzin Hi, I've update commits and fixed the CI failure, please take a look if you have a time, thanks

[raoraoxiong]

@snuyanzin @liuyongvs Hi, I've update commits and fixed the CI failure, please take a look if you have a time, thanks

[snuyanzin]

would be great to see feedback from someone with python expertise may be @dianfu

[dianfu]

@raoraoxiong Thanks for this work! The overall direction LGTM.

Could you split this PR to ease the review? For example, at least two PRs: one for cross-Calc reuse and another for the duplication inside one Python operator.

[dianfu]

Unnecessary changes. Please submit in a separate PR.

[raoraoxiong]

ok

[dianfu]

@raoraoxiong Hey, thanks for the update. Regarding to the latest PR, it addresses two problems, for problem: Top-level projection duplicates: identical deterministic calls in the projection (e.g. SELECT udf(a), udf(a)) are sent to the Python worker only once; a codegen expansion projection maps the deduplicated results back to the original output schema. have you considered addressing it in the planner?