polyzos commented on code in PR #28090:
URL: https://github.com/apache/flink/pull/28090#discussion_r4411277136
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkChangelogModeInferenceProgram.scala:
##########
@@ -963,27 +1010,31 @@ class FlinkChangelogModeInferenceProgram extends
FlinkOptimizeProgram[StreamOpti
case multiJoin: StreamPhysicalMultiJoin =>
val onlyAfterByParent = requiredUpdateTrait.updateKind ==
UpdateKind.ONLY_UPDATE_AFTER
- val children = multiJoin.getInputs.zipWithIndex.map {
- case (child, childOrdinal) =>
- val physicalChild = child.asInstanceOf[StreamPhysicalRel]
- val supportOnlyAfter =
multiJoin.inputUniqueKeyContainsCommonJoinKey(childOrdinal)
- val inputModifyKindSet = getModifyKindSet(physicalChild)
- if (onlyAfterByParent) {
- if (inputModifyKindSet.contains(ModifyKind.UPDATE) &&
!supportOnlyAfter) {
- // the parent requires only-after, however, the multi-join
doesn't support this for this input
- None
+ if (onlyAfterByParent &&
hasNonUpsertKeyMultiJoinCondition(multiJoin)) {
Review Comment:
Plain `StreamPhysicalJoin` dropped its analogous ONLY_UPDATE_AFTER guard
(per the last commit message, relying on FLINK-40841's runtime retraction).
Does `StreamExecMultiJoin` get the same retraction behavior, or is this guard
still needed here because multi-join doesn't share that fix yet? If it's the
latter, worth a one-line comment explaining the asymmetry.
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkChangelogModeInferenceProgram.scala:
##########
@@ -1422,27 +1473,34 @@ class FlinkChangelogModeInferenceProgram extends
FlinkOptimizeProgram[StreamOpti
createNewNode(process, Some(children), providedDeleteTrait)
case join: StreamPhysicalJoin =>
Review Comment:
The FLINK-40841 rationale for why plain joins don't need an
ONLY_UPDATE_AFTER guard (only this DELETE_BY_KEY one) currently lives only in
the commit message. Maybe add a short code comment here so future readers don't
have to dig through git history to understand the asymmetry.??
##########
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:
CDC-dedup + minibatch still fails to plan after this change. If that's
intentional, maybe mention this and add a release note?? since it's a behavior
change (previously planned, now errors).
--
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]
