raoraoxiong opened a new pull request, #28998:
URL: https://github.com/apache/flink/pull/28998
## What is the purpose of the change
This pull request is the second step of Common Sub-expression Elimination
(CSE) for Python UDFs (FLINK-39986), built on top of #28638. It extends the
deduplication from identical top-level projection calls to:
1. **Full-tree CSE**: nested Python UDF sub-expressions (e.g., `SELECT
udf1(x), udf2(udf1(x))`) are flattened so that each unique sub-expression is
computed only once by the Python worker, and outer calls reference the
pre-computed intermediate result.
2. **Condition-projection deduplication**: Python UDF calls shared between
the WHERE condition and the SELECT projection (e.g., `SELECT udf1(x) FROM T
WHERE udf1(x) > 0`) are deduplicated by a new logical optimizer rule.
**Note**: this branch currently contains the commit of #28638 as its first
commit. It will be rebased onto master once #28638 is merged; only the second
commit is the subject of this review.
## Brief change log
- Add `PythonCallDeduplicator` and `PythonCallCseResult` for structural
deduplication with nested sub-expression flattening and reference maps
- Add `PythonFunctionInfo.ResultRef` for referencing pre-computed UDF
results by index
- Extend protobuf `Input` message with a `refIndex` field so a UDF input can
reference the result of a previously executed UDF in the same operator
- Modify the Python worker (`operations.py`) to execute UDFs sequentially
and resolve result references
- Add `RemoteCalcConditionProjectionCseRule` optimizer rule (placed
alongside the other Python-related rules in `plan/rules/logical`) and register
it in stream and batch logical rule sets
## Verifying this change
This change added tests and can be verified as follows:
- Extended unit tests in `CommonExecPythonCalcRefReuseTest` covering nested
sub-expression flattening, reference map construction, and non-deterministic
call handling
- Added plan tests in `PythonCalcConditionCseTest` verifying the new
optimizer rule rewrites condition/projection shared calls
- Added Python-side tests in `test_scalar_function_cse.py` covering
sequential execution with result references in the Python worker
- Added integration tests in `test_udf.py` verifying end-to-end correctness
of nested UDF calls and WHERE/SELECT shared calls
## 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 protobuf `Input` message is extended with an
optional field in a backward-compatible way)
- The runtime per-record code paths (performance sensitive): (yes, reduces
redundant Python UDF invocations in the per-record path)
- 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? (no)
- If yes, how is the feature documented? (not applicable)
## AI Usage Disclosure
- [x] This contribution was created with AI assistance
Generated-by: Claude-4.6-Opus
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
