lincoln-lil commented on code in PR #27838:
URL: https://github.com/apache/flink/pull/27838#discussion_r2998826058
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkChangelogModeInferenceProgram.scala:
##########
@@ -1521,28 +1509,39 @@ class FlinkChangelogModeInferenceProgram extends
FlinkOptimizeProgram[StreamOpti
}
}
- //
-------------------------------------------------------------------------------------------
-
- private def areUpsertKeysDifferentFromPk(sink: StreamPhysicalSink) = {
- // if sink's pk(s) are not exactly match input changeLogUpsertKeys then
it will fallback
- // to beforeAndAfter mode for the correctness
- var upsertKeyDifferentFromPk: Boolean = false
- val sinkDefinedPks =
sink.contextResolvedTable.getResolvedSchema.getPrimaryKeyIndexes
-
- if (sinkDefinedPks.nonEmpty) {
- val sinkPks = ImmutableBitSet.of(sinkDefinedPks: _*)
- val fmq =
FlinkRelMetadataQuery.reuseOrCreate(sink.getCluster.getMetadataQuery)
- val changeLogUpsertKeys = fmq.getUpsertKeys(sink.getInput)
- // if input is UA only, primary key != upsert key (upsert key can be
null) we should
- // fallback to beforeAndAfter.
- // Notice: even sink pk(s) contains input upsert key we cannot
optimize to UA only,
- // this differs from batch job's unique key inference
- if (changeLogUpsertKeys == null ||
!changeLogUpsertKeys.exists(_.equals(sinkPks))) {
- upsertKeyDifferentFromPk = true
- }
- }
- upsertKeyDifferentFromPk
+ }
+
+ /**
+ * Check whether input's upsert keys differ from sink's primary keys,
considering immutable
+ * columns.
+ *
+ * <p>A sink pk is considered "covered" when there exists an upsert key `uk`
such that: <ul>
+ * <li>`uk` is a subset of sink pk (no extra columns that could cause key
collision)</li> <li>`uk`
+ * union immutable columns covers all sink pk columns (immutable columns
never change, so they
+ * effectively act as part of the key for upsert semantics)</li> </ul>
+ *
+ * <p>Notice: even if sink pk(s) contains input upsert key we cannot
optimize to UA only when the
+ * upsert key has columns outside sink pk, this differs from batch job's
unique key inference.
+ */
+ private def areUpsertKeysWithImmutableColsDifferentFromPk(sink:
StreamPhysicalSink): Boolean = {
Review Comment:
The word "Different" implies an equality comparison (the old method
`areUpsertKeysDifferentFromPk` indeed used `!exists(_.equals(sinkPks))`).
However, the new logic actually checks a subset/coverage relationship, so the
method name should more accurately reflect the "coverage"(satisfy) semantics.
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkChangelogModeInferenceProgram.scala:
##########
@@ -1521,28 +1509,39 @@ class FlinkChangelogModeInferenceProgram extends
FlinkOptimizeProgram[StreamOpti
}
}
- //
-------------------------------------------------------------------------------------------
-
- private def areUpsertKeysDifferentFromPk(sink: StreamPhysicalSink) = {
- // if sink's pk(s) are not exactly match input changeLogUpsertKeys then
it will fallback
- // to beforeAndAfter mode for the correctness
- var upsertKeyDifferentFromPk: Boolean = false
- val sinkDefinedPks =
sink.contextResolvedTable.getResolvedSchema.getPrimaryKeyIndexes
-
- if (sinkDefinedPks.nonEmpty) {
- val sinkPks = ImmutableBitSet.of(sinkDefinedPks: _*)
- val fmq =
FlinkRelMetadataQuery.reuseOrCreate(sink.getCluster.getMetadataQuery)
- val changeLogUpsertKeys = fmq.getUpsertKeys(sink.getInput)
- // if input is UA only, primary key != upsert key (upsert key can be
null) we should
- // fallback to beforeAndAfter.
- // Notice: even sink pk(s) contains input upsert key we cannot
optimize to UA only,
- // this differs from batch job's unique key inference
- if (changeLogUpsertKeys == null ||
!changeLogUpsertKeys.exists(_.equals(sinkPks))) {
- upsertKeyDifferentFromPk = true
- }
- }
- upsertKeyDifferentFromPk
+ }
+
+ /**
+ * Check whether input's upsert keys differ from sink's primary keys,
considering immutable
+ * columns.
+ *
+ * <p>A sink pk is considered "covered" when there exists an upsert key `uk`
such that: <ul>
+ * <li>`uk` is a subset of sink pk (no extra columns that could cause key
collision)</li> <li>`uk`
+ * union immutable columns covers all sink pk columns (immutable columns
never change, so they
+ * effectively act as part of the key for upsert semantics)</li> </ul>
+ *
+ * <p>Notice: even if sink pk(s) contains input upsert key we cannot
optimize to UA only when the
+ * upsert key has columns outside sink pk, this differs from batch job's
unique key inference.
+ */
+ private def areUpsertKeysWithImmutableColsDifferentFromPk(sink:
StreamPhysicalSink): Boolean = {
+ val sinkDefinedPks =
sink.contextResolvedTable.getResolvedSchema.getPrimaryKeyIndexes
+ if (sinkDefinedPks.isEmpty) {
+ return false
+ }
+ val sinkPks = ImmutableBitSet.of(sinkDefinedPks: _*)
+ val fmq =
FlinkRelMetadataQuery.reuseOrCreate(sink.getCluster.getMetadataQuery)
+ val changeLogUpsertKeys = fmq.getUpsertKeys(sink.getInput)
+ // if upsert key is null, we should fall back to beforeAndAfter
+ if (changeLogUpsertKeys == null) {
+ return true
}
+ val immutableCols = fmq.getImmutableColumns(sink.getInput)
+ val effectiveImmutableCols =
+ if (immutableCols != null) immutableCols else ImmutableBitSet.of()
+ // check: uk ⊆ sinkPks AND sinkPks ⊆ (uk ∪ immutableCols)
Review Comment:
The double negation might be readability issue here.
A minimal rewrite version(maybe slightly improvement or you can think of
other writings):
```scala
val sinkPkCovered = changeLogUpsertKeys.exists { uk =>
val ukWithinPk = sinkPks.contains(uk)
val pkFullyCovered = uk.union(effectiveImmutableCols).contains(sinkPks)
ukWithinPk && pkFullyCovered
}
!sinkPkCovered
```
--
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]
