raoraoxiong commented on PR #28638: URL: https://github.com/apache/flink/pull/28638#issuecomment-5410230156
> @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. -- 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]
