fresh-borzoni commented on code in PR #29653:
URL: https://github.com/apache/flink/pull/29653#discussion_r4424410388
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkChangelogModeInferenceProgram.scala:
##########
@@ -1713,15 +1710,16 @@ class FlinkChangelogModeInferenceProgram extends
FlinkOptimizeProgram[StreamOpti
}
/**
- * Whether this calc's own output still has an upsert key after its
projection. A DELETE_BY_KEY
- * tombstone passed through this calc must still carry a key in its output,
otherwise nothing
- * downstream would know what to delete.
+ * Whether this calc's own output still has an upsert key after its
projection, including one
+ * recovered through an equi-join equivalence. A DELETE_BY_KEY tombstone
passed through this calc
+ * must still carry a key in its output, otherwise nothing downstream would
know what to delete.
*
* This method is an extra safety net, in case downstream consumers don't
require an upsert key.
*/
private def hasOutputUpsertKey(calc: StreamPhysicalCalcBase): Boolean = {
val fmq =
FlinkRelMetadataQuery.reuseOrCreate(calc.getCluster.getMetadataQuery)
- val upsertKeys = fmq.getUpsertKeys(calc)
+ val allColumns = ImmutableBitSet.range(calc.getRowType.getFieldCount)
+ val upsertKeys = UpsertKeyEquivalenceUtil.upsertKeysWithin(calc,
allColumns, fmq)
Review Comment:
Yes, intentional as that key is only read by the materializer, which is only
added when no key fits the PK even with the equivalence, or with FORCE. The raw
key is a real key of the input, so it's the right one there.
--
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]
