[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.

The duplication actually originates in the planner. RexProgram already shares structurally identical expressions through RexLocalRef, so SELECT udf(a), udf(a) reaches the physical node as a single shared call. It is RexProgram#expandLocalRef in the Python calc translation that expands that shared reference back into independent expression trees, which is what makes the same UDF get shipped to the worker once per occurrence.

This PR therefore addresses it with two logical optimizer rules, covering two scenarios:

1. Duplicates within a projection: identical deterministic calls in the projection (e.g. SELECT udf(a), udf(a)) are computed once and the shared result is projected back into the original positions.
2. Duplicates shared between condition and projection: after RemoteCalcSplitConditionRule splits a Calc with Python UDFs in its condition, projections reuse the 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.

Because the deduplication happens during logical optimization, neither CommonExecPythonCalc nor the projection codegen needs to be aware of it, and the result is visible in EXPLAIN and assertable in plan tests.

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 RemoteCalcProjectionCseRule, which splits a Calc containing duplicated deterministic remote calls into a bottom Calc computing each distinct call once plus a top Calc that is a pure RexInputRef projection restoring the original output schema
• Add RemoteCalcConditionProjectionCseRule, which rewrites a Calc projection to reference remote calls already computed by the Calc below it
• Add RemoteCalcCseUtil holding the shared reusability predicate, so both rules agree on which calls may safely share a single evaluation
• Register both rules in the stream and batch rule sets: CONDITION_PROJECTION_CSE after SPLIT_CONDITION, and PROJECTION_CSE after REWRITE_PROJECT once the projection has been normalized
• The rules are implemented against RemoteCallFinder, so they are not Python-specific and can be reused for other remote call types

Verifying this change

This change added tests and can be verified as follows:

• PythonCalcCseTest (+ plan XML): plan tests covering both rules — duplicates within a projection, duplicates with a forwarded field, duplicates shared between condition and projection, a shared call nested inside another UDF call, plus negative cases (distinct calls, different UDFs and non-deterministic calls 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, no plan change 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:

• 3318646 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?

[raoraoxiong]

@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?

@dianfu Thanks for the review — agreed, this belongs in the planner.

Looking into it, the duplication actually originates in the planner already.
RexProgram shares structurally identical expressions through RexLocalRef, so
SELECT udf(a), udf(a) arrives with a single shared call. It's
RexProgram#expandLocalRef in the Python calc translation that expands that
shared ref back into independent expression trees, which is what makes the same
UDF get shipped to the worker once per occurrence. The ExecNode-level
deduplication in this PR was really just restoring information the planner had
already computed.

So I'll rework this PR to do it as a logical rule instead: split a calc with
duplicated deterministic calls into a bottom calc computing each distinct call
once, plus a top calc that projects the shared result back into the original
positions. That keeps it alongside the existing Python rules, and means
CommonExecPythonCalc and the codegen expansion projection don't need to change
at all. It also makes the deduplication visible in EXPLAIN and assertable in
plan tests, which the current approach can't do since it happens after plan
output.

I'll push the reworked version shortly.

[raoraoxiong]

@flinkbot run azure