JingsongLi commented on a change in pull request #17699:
URL: https://github.com/apache/flink/pull/17699#discussion_r753993288
##########
File path:
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkChangelogModeInferenceProgram.scala
##########
@@ -759,6 +754,84 @@ class FlinkChangelogModeInferenceProgram extends
FlinkOptimizeProgram[StreamOpti
Some(sink.copy(sinkTrait,
children.head).asInstanceOf[StreamPhysicalRel])
}
}
+
+ private def inferSinkRequiredTraits(sink: StreamPhysicalSink):
Seq[UpdateKindTrait] = {
+ val childModifyKindSet = getModifyKindSet(sink.getInput)
+ val onlyAfter = onlyAfterOrNone(childModifyKindSet)
+ val beforeAndAfter = beforeAfterOrNone(childModifyKindSet)
+ val sinkTrait = UpdateKindTrait.fromChangelogMode(
+ sink.tableSink.getChangelogMode(childModifyKindSet.toChangelogMode))
+
+ val sinkRequiredTraits = if (sinkTrait.equals(ONLY_UPDATE_AFTER)) {
+ // if sink's pk(s) are not exactly match input changeLogUpsertKeys
then it will fallback
+ // to beforeAndAfter mode for the correctness
+ var shouldFallback: Boolean = false
+ val sinkDefinedPks = toScala(sink.catalogTable.getResolvedSchema
Review comment:
Maybe you just want `getPrimaryKeyIndexes`?
##########
File path:
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkChangelogModeInferenceProgram.scala
##########
@@ -759,6 +754,84 @@ class FlinkChangelogModeInferenceProgram extends
FlinkOptimizeProgram[StreamOpti
Some(sink.copy(sinkTrait,
children.head).asInstanceOf[StreamPhysicalRel])
}
}
+
+ private def inferSinkRequiredTraits(sink: StreamPhysicalSink):
Seq[UpdateKindTrait] = {
+ val childModifyKindSet = getModifyKindSet(sink.getInput)
+ val onlyAfter = onlyAfterOrNone(childModifyKindSet)
+ val beforeAndAfter = beforeAfterOrNone(childModifyKindSet)
+ val sinkTrait = UpdateKindTrait.fromChangelogMode(
+ sink.tableSink.getChangelogMode(childModifyKindSet.toChangelogMode))
+
+ val sinkRequiredTraits = if (sinkTrait.equals(ONLY_UPDATE_AFTER)) {
+ // if sink's pk(s) are not exactly match input changeLogUpsertKeys
then it will fallback
+ // to beforeAndAfter mode for the correctness
+ var shouldFallback: Boolean = false
+ val sinkDefinedPks = toScala(sink.catalogTable.getResolvedSchema
+
.getPrimaryKey).map(_.getColumns).map(toScala[String]).getOrElse(Seq())
+ if (sinkDefinedPks.nonEmpty) {
+ val sinkColumns = sink.catalogTable.getResolvedSchema.getColumnNames
+ val sinkPks =
ImmutableBitSet.of(sinkDefinedPks.map(sinkColumns.indexOf): _*)
+ 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.size() == 0
+ || !changeLogUpsertKeys.exists {0 == _.compareTo(sinkPks)}) {
Review comment:
`compareTo` -> `equals`?
##########
File path:
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkChangelogModeInferenceProgram.scala
##########
@@ -759,6 +754,84 @@ class FlinkChangelogModeInferenceProgram extends
FlinkOptimizeProgram[StreamOpti
Some(sink.copy(sinkTrait,
children.head).asInstanceOf[StreamPhysicalRel])
}
}
+
+ private def inferSinkRequiredTraits(sink: StreamPhysicalSink):
Seq[UpdateKindTrait] = {
+ val childModifyKindSet = getModifyKindSet(sink.getInput)
+ val onlyAfter = onlyAfterOrNone(childModifyKindSet)
+ val beforeAndAfter = beforeAfterOrNone(childModifyKindSet)
+ val sinkTrait = UpdateKindTrait.fromChangelogMode(
+ sink.tableSink.getChangelogMode(childModifyKindSet.toChangelogMode))
+
+ val sinkRequiredTraits = if (sinkTrait.equals(ONLY_UPDATE_AFTER)) {
+ // if sink's pk(s) are not exactly match input changeLogUpsertKeys
then it will fallback
+ // to beforeAndAfter mode for the correctness
+ var shouldFallback: Boolean = false
+ val sinkDefinedPks = toScala(sink.catalogTable.getResolvedSchema
+
.getPrimaryKey).map(_.getColumns).map(toScala[String]).getOrElse(Seq())
+ if (sinkDefinedPks.nonEmpty) {
+ val sinkColumns = sink.catalogTable.getResolvedSchema.getColumnNames
+ val sinkPks =
ImmutableBitSet.of(sinkDefinedPks.map(sinkColumns.indexOf): _*)
+ 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.size() == 0
+ || !changeLogUpsertKeys.exists {0 == _.compareTo(sinkPks)}) {
+ shouldFallback = true
+ }
+ }
+ if (shouldFallback) {
+ Seq(beforeAndAfter)
+ } else {
+ Seq(onlyAfter, beforeAndAfter)
+ }
+ } else if (sinkTrait.equals(BEFORE_AND_AFTER)){
+ Seq(beforeAndAfter)
+ } else {
+ Seq(UpdateKindTrait.NONE)
+ }
+ sinkRequiredTraits
+ }
+
+ private def analyzeUpsertMaterializeStrategy(sink: StreamPhysicalSink):
Boolean = {
Review comment:
Can we merge this method to `inferSinkRequiredTraits`? For example,
`inferSinkRequiredTraits` returns `(Seq[UpdateKindTrait], Boolean)`
##########
File path:
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkChangelogModeInferenceProgram.scala
##########
@@ -759,6 +754,84 @@ class FlinkChangelogModeInferenceProgram extends
FlinkOptimizeProgram[StreamOpti
Some(sink.copy(sinkTrait,
children.head).asInstanceOf[StreamPhysicalRel])
}
}
+
+ private def inferSinkRequiredTraits(sink: StreamPhysicalSink):
Seq[UpdateKindTrait] = {
+ val childModifyKindSet = getModifyKindSet(sink.getInput)
+ val onlyAfter = onlyAfterOrNone(childModifyKindSet)
+ val beforeAndAfter = beforeAfterOrNone(childModifyKindSet)
+ val sinkTrait = UpdateKindTrait.fromChangelogMode(
+ sink.tableSink.getChangelogMode(childModifyKindSet.toChangelogMode))
+
+ val sinkRequiredTraits = if (sinkTrait.equals(ONLY_UPDATE_AFTER)) {
+ // if sink's pk(s) are not exactly match input changeLogUpsertKeys
then it will fallback
+ // to beforeAndAfter mode for the correctness
+ var shouldFallback: Boolean = false
Review comment:
requireBeforeAndAfter?
##########
File path:
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkChangelogModeInferenceProgram.scala
##########
@@ -759,6 +754,84 @@ class FlinkChangelogModeInferenceProgram extends
FlinkOptimizeProgram[StreamOpti
Some(sink.copy(sinkTrait,
children.head).asInstanceOf[StreamPhysicalRel])
}
}
+
+ private def inferSinkRequiredTraits(sink: StreamPhysicalSink):
Seq[UpdateKindTrait] = {
+ val childModifyKindSet = getModifyKindSet(sink.getInput)
+ val onlyAfter = onlyAfterOrNone(childModifyKindSet)
+ val beforeAndAfter = beforeAfterOrNone(childModifyKindSet)
+ val sinkTrait = UpdateKindTrait.fromChangelogMode(
+ sink.tableSink.getChangelogMode(childModifyKindSet.toChangelogMode))
+
+ val sinkRequiredTraits = if (sinkTrait.equals(ONLY_UPDATE_AFTER)) {
+ // if sink's pk(s) are not exactly match input changeLogUpsertKeys
then it will fallback
+ // to beforeAndAfter mode for the correctness
+ var shouldFallback: Boolean = false
+ val sinkDefinedPks = toScala(sink.catalogTable.getResolvedSchema
+
.getPrimaryKey).map(_.getColumns).map(toScala[String]).getOrElse(Seq())
+ if (sinkDefinedPks.nonEmpty) {
+ val sinkColumns = sink.catalogTable.getResolvedSchema.getColumnNames
+ val sinkPks =
ImmutableBitSet.of(sinkDefinedPks.map(sinkColumns.indexOf): _*)
+ 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.size() == 0
Review comment:
remove `changeLogUpsertKeys.size() == 0`?
##########
File path:
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkChangelogModeInferenceProgram.scala
##########
@@ -759,6 +754,84 @@ class FlinkChangelogModeInferenceProgram extends
FlinkOptimizeProgram[StreamOpti
Some(sink.copy(sinkTrait,
children.head).asInstanceOf[StreamPhysicalRel])
}
}
+
+ private def inferSinkRequiredTraits(sink: StreamPhysicalSink):
Seq[UpdateKindTrait] = {
+ val childModifyKindSet = getModifyKindSet(sink.getInput)
+ val onlyAfter = onlyAfterOrNone(childModifyKindSet)
+ val beforeAndAfter = beforeAfterOrNone(childModifyKindSet)
+ val sinkTrait = UpdateKindTrait.fromChangelogMode(
+ sink.tableSink.getChangelogMode(childModifyKindSet.toChangelogMode))
+
+ val sinkRequiredTraits = if (sinkTrait.equals(ONLY_UPDATE_AFTER)) {
+ // if sink's pk(s) are not exactly match input changeLogUpsertKeys
then it will fallback
+ // to beforeAndAfter mode for the correctness
+ var shouldFallback: Boolean = false
+ val sinkDefinedPks = toScala(sink.catalogTable.getResolvedSchema
+
.getPrimaryKey).map(_.getColumns).map(toScala[String]).getOrElse(Seq())
+ if (sinkDefinedPks.nonEmpty) {
+ val sinkColumns = sink.catalogTable.getResolvedSchema.getColumnNames
+ val sinkPks =
ImmutableBitSet.of(sinkDefinedPks.map(sinkColumns.indexOf): _*)
+ 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.size() == 0
+ || !changeLogUpsertKeys.exists {0 == _.compareTo(sinkPks)}) {
+ shouldFallback = true
+ }
+ }
+ if (shouldFallback) {
+ Seq(beforeAndAfter)
+ } else {
+ Seq(onlyAfter, beforeAndAfter)
+ }
+ } else if (sinkTrait.equals(BEFORE_AND_AFTER)){
+ Seq(beforeAndAfter)
+ } else {
+ Seq(UpdateKindTrait.NONE)
+ }
+ sinkRequiredTraits
+ }
+
+ private def analyzeUpsertMaterializeStrategy(sink: StreamPhysicalSink):
Boolean = {
+ val tableConfig =
sink.getCluster.getPlanner.getContext.unwrap(classOf[FlinkContext])
+ .getTableConfig
+ val inputChangelogMode = ChangelogPlanUtils.getChangelogMode(
+ sink.getInput.asInstanceOf[StreamPhysicalRel]).get
+ val catalogTable = sink.catalogTable
+ val primaryKeys = toScala(catalogTable.getResolvedSchema
Review comment:
Ditto
--
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
