pnowojski commented on code in PR #26051:
URL: https://github.com/apache/flink/pull/26051#discussion_r1946606727
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalRank.scala:
##########
@@ -109,28 +109,34 @@ class StreamPhysicalRank(
.item("select", getRowType.getFieldNames.mkString(", "))
}
- private def getDeduplicateDescription(isRowtime: Boolean, isLastRow:
Boolean): String = {
+ private def getDeduplicateDescription(
+ isRowtime: Boolean,
+ isLastRow: Boolean,
+ insertOnly: Boolean): String = {
val fieldNames = getRowType.getFieldNames
val orderString = if (isRowtime) "ROWTIME" else "PROCTIME"
val keep = if (isLastRow) "LastRow" else "FirstRow"
- s"Deduplicate(keep=[$keep],
key=[${partitionKey.toArray.map(fieldNames.get).mkString(", ")}],
order=[$orderString])"
+ s"Deduplicate(keep=[$keep],
key=[${partitionKey.toArray.map(fieldNames.get).mkString(", ")}],
order=[$orderString], outputInsertOnly=[$insertOnly])"
}
override def translateToExecNode(): ExecNode[_] = {
val generateUpdateBefore = ChangelogPlanUtils.generateUpdateBefore(this)
if (RankUtil.canConvertToDeduplicate(this)) {
val keepLastRow = RankUtil.keepLastDeduplicateRow(orderKey)
+ val tableConfig = unwrapTableConfig(this)
+ val outputInsertOnly =
StreamExecDeduplicate.canBeInsertOnly(tableConfig, keepLastRow)
Review Comment:
Hmm, I guess you are right. But can you confirm if I understand the code
correctly?
`ChangelogPlanUtils.isInsertOnly(this)` checks for the trait that is
actually set in `FlinkChangelogModeInferenceProgram`.
`FlinkChangelogModeInferenceProgram` works on physical plan, and therefore
those traits are being set before `StreamPhysicalRank#translateToExecNode` is
being call? So there is no need to recalculate from scratch (and defacto
duplicate the logic of checking) if this rank can output inserts only, we can
just use what is already stored/persisted in the trait?
--
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: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
