fresh-borzoni commented on code in PR #28090:
URL: https://github.com/apache/flink/pull/28090#discussion_r4417667149
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkChangelogModeInferenceProgram.scala:
##########
@@ -871,25 +909,34 @@ class FlinkChangelogModeInferenceProgram extends
FlinkOptimizeProgram[StreamOpti
}
case normalize: StreamPhysicalChangelogNormalize =>
- // changelog normalize currently only supports input only sending
UPDATE_AFTER
- val children = visitChildren(normalize,
UpdateKindTrait.ONLY_UPDATE_AFTER)
- // use requiredTrait as providedTrait,
- // because changelog normalize supports all kinds of UpdateKind
+ // Prefer UA-only from children, fall back to BEFORE_AND_AFTER when
the child can't
+ // drop UB (e.g. a CDC source with a non-upsert-key filter pushed
down). The runtime
+ // operator folds incoming UB/DELETE events into its state.
+ val inputModifyKindSet = getModifyKindSet(normalize.getInput)
Review Comment:
I believe it's already fixed, wdym?
--
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]
