[FLINK-39986][table-planner][python] Support full-tree CSE for nested Python UDF calls

[raoraoxiong]

What is the purpose of the change

This PR is the second of two PRs implementing common sub-expression elimination (CSE) for Python UDFs (FLINK-39986), building on top of #28638 (which covers top-level projection deduplication and condition-projection sharing on the JVM side).

This PR extends the deduplication to nested sub-expressions (full-tree CSE). Python UDF call trees are flattened in post-order and deduplicated by structural equivalence, enabling cross-subtree reuse: in SELECT udf1(x), udf2(udf1(x)), the inner udf1(x) is computed only once and its result is passed to udf2 by reference.

To achieve this, the JVM-to-worker protocol is extended: a nested call whose result is already computed is no longer re-executed inside the Python worker. Instead, the planner emits a ResultRef carrying the index of the pre-computed result, transferred via the new refIndex field in the protobuf Input message. The Python worker executes the deduplicated calls sequentially and resolves result references before invoking each UDF.

The protobuf change is backward compatible: inputs without refIndex follow the original nested-evaluation path.

Brief change log

• Extend PythonCallDeduplicator to flatten nested call trees (post-order, so child results are available before parents) and build a sub-expression cross-reference map (refMap); non-deterministic children are not flattened to prevent incorrect sharing
• Add PythonFunctionInfo.ResultRef for referencing pre-computed results and wire it through CommonPythonUtil / ProtoUtils
• Extend protobuf Input message with refIndex field
• Modify the Python worker (operations.py) to support sequential execution with result references
• Add a CSE annotation to the PythonCalc plan description showing top-level reuse relationships, e.g. (CSE: EXPR$2->EXPR$1)

Verifying this change

This change added tests and can be verified as follows:

• CommonExecPythonCalcCseTest#testRefMapResolvesStructurallyEqualSubExpressions: unit test for cross-subtree reference resolution
• PythonCalcConditionCseTest#testNestedUdfWithCseAnnotation (+ plan XML): plan test verifying nested deduplication and the CSE plan annotation
• flink-python/pyflink/fn_execution/tests/test_scalar_function_cse.py: Python worker tests for sequential execution with result references
• Nested-call cases in test_python_local_ref_reuse in flink-python/pyflink/table/tests/test_udf.py: end-to-end tests verifying nested deterministic calls are reused

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, protobuf change is a backward-compatible optional field)
• The runtime per-record code paths (performance sensitive): (yes, Python UDF execution path; nested duplicated calls are executed once per record instead of repeatedly)
• 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:

• c3647d5 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

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