fresh-borzoni commented on code in PR #28090:
URL: https://github.com/apache/flink/pull/28090#discussion_r4417661834
##########
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:
Still needed, FLINK-40841 and FLINK-40858 only changed
`StreamingJoinOperator` and its mini-batch variant, the multi-join operator
evaluates its conditions on the incoming row only. The multi-join ITs here
still return phantom rows on master. Added a comment explaining it.
I also found a gap in the delete-by-key check here: it was skipped when the
inputs had deletes by key but no updates, also fixed
--
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]
