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:
Intentional. Those checks only decide whether UPDATE_BEFORE can be skipped,
so with raw keys the worst case is a few extra messages, never wrong results.
I'd keep it for a separate PR.
--
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]
