xuyangzhong commented on code in PR #27838:
URL: https://github.com/apache/flink/pull/27838#discussion_r2999335448
##########
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:
I have updated the code here and add an example in the method comment.
--
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]
