fresh-borzoni commented on code in PR #28090:
URL: https://github.com/apache/flink/pull/28090#discussion_r4563273791
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkChangelogModeInferenceProgram.scala:
##########
@@ -1744,25 +1750,37 @@ class FlinkChangelogModeInferenceProgram extends
FlinkOptimizeProgram[StreamOpti
upsertKeys != null && upsertKeys.exists(!_.isEmpty)
}
+ /**
+ * Whether an inner lookup join drops an update whose row stops matching,
i.e. a condition reads
+ * columns outside the input's upsert key, or the lookup table is filtered
and the join key isn't
+ * on the upsert key.
+ */
+ private def isNonUpsertKeyLookupCondition(lookupJoin:
StreamPhysicalLookupJoin): Boolean = {
+ if (lookupJoin.joinType != JoinRelType.INNER) {
+ return false
+ }
+ val input = lookupJoin.getInput
+ // columns of the looked up row are never in the input's upsert key
+ val conditions = lookupJoin.finalPreFilterCondition ++
lookupJoin.finalRemainingCondition
+ conditions.exists(c => !isOnUpsertKey(input,
RelOptUtil.InputFinder.bits(c))) ||
+ (lookupJoin.calcOnTemporalTable.exists(_.getCondition != null) &&
Review Comment:
Agreed, a filtered lookup row is the same as a missing one, which
[FLINK-41443 (https://issues.apache.org/jira/browse/FLINK-41443) tracks.
Removed the check, and for the same reason join conditions now only count
the input's columns.
--
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]
