[FLINK-40324][table] Append-only should stay unmaterialized without an ON CONFLICT clause

flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkChangelogModeInferenceProgram.scala

@@ -18,7 +18,7 @@
 package org.apache.flink.table.planner.plan.optimize.program
 
 import org.apache.flink.legacy.table.sinks.{AppendStreamTableSink, RetractStreamTableSink, StreamTableSink, UpsertStreamTableSink}
-import org.apache.flink.table.api.{TableException, ValidationException}
+import org.apache.flink.table.api.{TableConfig, TableException, ValidationException}
 import org.apache.flink.table.api.InsertConflictStrategy.ConflictBehavior
 import org.apache.flink.table.api.config.ExecutionConfigOptions
 import org.apache.flink.table.api.config.ExecutionConfigOptions.UpsertMaterialize
@@ -1148,7 +1148,8 @@ class FlinkChangelogModeInferenceProgram extends FlinkOptimizeProgram[StreamOpti
      * Analyze whether to enable upsertMaterialize or not. In these case will return true:
      *   1. when `TABLE_EXEC_SINK_UPSERT_MATERIALIZE` set to FORCE and sink's primary key nonempty.
      *      2. when `TABLE_EXEC_SINK_UPSERT_MATERIALIZE` set to AUTO and sink's primary key doesn't
-     *      contain upsertKeys of the input update stream.
+     *      contain upsertKeys of the input update stream, unless the input is insert only and the
+     *      effective conflict strategy is DEDUPLICATE.
      *
      * Also validates that ON CONFLICT clause is specified when upsert key differs from primary key.
      */
@@ -1197,41 +1198,50 @@ class FlinkChangelogModeInferenceProgram extends FlinkOptimizeProgram[StreamOpti
             return false
           }
 
-          // For a DEDUPLICATE strategy and INSERT only input, we simply let the inserts be handled
-          // as UPSERT_AFTER and overwrite previous value
-          if (inputIsAppend && sink.isDeduplicateConflictStrategy) {
-            return false
-          }
-
           // if input has updates and primary key != upsert key  we should enable upsertMaterialize.
           //
           // An optimize is: do not enable upsertMaterialize when sink pk(s) contains input
           // changeLogUpsertKeys
           val upsertKeyDiffersFromPk = !sink.primaryKeysContainsUpsertKey
+          validateOnConflictSpecifiedIfRequired(sink, tableConfig, upsertKeyDiffersFromPk)
 
-          // Validate that ON CONFLICT is specified when upsert key differs from primary key
-          val requireOnConflict =
-            tableConfig.get(ExecutionConfigOptions.TABLE_EXEC_SINK_REQUIRE_ON_CONFLICT)
-          if (requireOnConflict && upsertKeyDiffersFromPk && sink.conflictStrategy == null) {
-            val pkNames = sink.getPrimaryKeyNames
-            val upsertKeyNames = sink.getUpsertKeyNames
-            throw new ValidationException(
-              "The query has an upsert key that differs from the primary key of the sink table " +
-                s"'${sink.contextResolvedTable.getIdentifier.asSummaryString}'. " +
-                s"Primary key: $pkNames, upsert key: $upsertKeyNames. " +
-                "This can lead to non-deterministic results when multiple records with different " +
-                "upsert keys map to the same primary key. " +
-                "Please specify an ON CONFLICT clause to define how conflicts should be handled: " +
-                "ON CONFLICT DO DEDUPLICATE (update to the latest record, state intensive, since we" +
-                " need to keep the entire history), or " +
-                "ON CONFLICT DO ERROR (fail on conflict), or " +
-                "ON CONFLICT DO NOTHING (keep first record).")
+          // Once enforcement above has passed, an absent clause leaves DEDUPLICATE as the strategy.
+          val deduplicatesOnConflict =
+            sink.conflictStrategy == null || sink.isDeduplicateConflictStrategy
+
+          // For a DEDUPLICATE strategy and INSERT only input, we simply let the inserts be handled
+          // as UPDATE_AFTER and overwrite previous value
+          if (deduplicatesOnConflict && inputIsAppend) {
+            return false
           }
 
           upsertKeyDiffersFromPk
       }
     }
 
+    private def validateOnConflictSpecifiedIfRequired(
+        sink: StreamPhysicalSink,
+        tableConfig: TableConfig,
+        upsertKeyDiffersFromPk: Boolean): Unit = {
+      val requireOnConflict =
+        tableConfig.get(ExecutionConfigOptions.TABLE_EXEC_SINK_REQUIRE_ON_CONFLICT)
+      if (requireOnConflict && upsertKeyDiffersFromPk && sink.conflictStrategy == null) {
+        val pkNames = sink.getPrimaryKeyNames
+        val upsertKeyNames = sink.getUpsertKeyNames
+        throw new ValidationException(
+          "The query has an upsert key that differs from the primary key of the sink table " +
+            s"'${sink.contextResolvedTable.getIdentifier.asSummaryString}'. " +
+            s"Primary key: $pkNames, upsert key: $upsertKeyNames. " +
+            "This can lead to non-deterministic results when multiple records with different " +
+            "upsert keys map to the same primary key. " +
+            "Please specify an ON CONFLICT clause to define how conflicts should be handled: " +
+            "ON CONFLICT DO DEDUPLICATE (update to the latest record, state intensive, since we" +
+            " need to keep the entire history), or " +
+            "ON CONFLICT DO ERROR (fail on conflict), or " +
+            "ON CONFLICT DO NOTHING (keep first record).")
+      }
+    }
+
     private def validateSourcesHaveWatermarks(sink: StreamPhysicalSink): Unit = {
       val sourcesWithoutWatermarks = new java.util.ArrayList[String]()
       collectSourcesWithoutWatermarks(sink.getInput, sourcesWithoutWatermarks)

flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/TableSinkTest.xml

@@ -16,6 +16,22 @@ See the License for the specific language governing permissions and
 limitations under the License.
 -->
 <Root>
+  <TestCase name="testAppendOnlyInputWithoutOnConflict">
+    <Resource name="ast">
+      <![CDATA[
+LogicalSink(table=[default_catalog.default_database.sinkWithPk], fields=[a, b])
++- LogicalProject(a=[$0], b=[$1])
+   +- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
+]]>
+    </Resource>
+    <Resource name="optimized rel plan">
+      <![CDATA[
+Sink(table=[default_catalog.default_database.sinkWithPk], fields=[a, b], changelogMode=[NONE])
++- Calc(select=[a, b], changelogMode=[I])
+   +- DataStreamScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c], changelogMode=[I])
+]]>
+    </Resource>
+  </TestCase>
   <TestCase name="testAppendSink">
     <Resource name="ast">
       <![CDATA[
@@ -852,6 +868,27 @@ Sink(table=[default_catalog.default_database.SinkRankChangeLog], fields=[person,
             +- GroupAggregate(groupBy=[person], select=[person, SUM(votes) AS sum_votes])
                +- Exchange(distribution=[hash[person]])
                   +- TableSourceScan(table=[[default_catalog, default_database, src]], fields=[person, votes])
+]]>
+    </Resource>
+  </TestCase>
+  <TestCase name="testUpdatingInputWithoutOnConflict">
+    <Resource name="ast">
+      <![CDATA[
+LogicalSink(table=[default_catalog.default_database.updatingSinkWithPk], fields=[EXPR$0, EXPR$1])
++- LogicalProject(EXPR$0=[$1], EXPR$1=[$2])
+   +- LogicalAggregate(group=[{0}], EXPR$0=[MAX($1)], EXPR$1=[COUNT()])
+      +- LogicalProject(c=[$2], a=[$0])
+         +- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
+]]>
+    </Resource>
+    <Resource name="optimized rel plan">
+      <![CDATA[
+Sink(table=[default_catalog.default_database.updatingSinkWithPk], fields=[EXPR$0, EXPR$1], upsertMaterialize=[true], changelogMode=[NONE])
++- Calc(select=[EXPR$0, EXPR$1], changelogMode=[I,UB,UA])
+   +- GroupAggregate(groupBy=[c], select=[c, MAX(a) AS EXPR$0, COUNT(*) AS EXPR$1], changelogMode=[I,UB,UA])
+      +- Exchange(distribution=[hash[c]], changelogMode=[I])
+         +- Calc(select=[c, a], changelogMode=[I])
+            +- DataStreamScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c], changelogMode=[I])
 ]]>
     </Resource>
   </TestCase>

flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/TableSinkTest.scala

@@ -864,6 +864,47 @@ class TableSinkTest extends TableTestBase {
     util.verifyRelPlan(stmtSet, ExplainDetail.CHANGELOG_MODE)
   }
 
+  @Test
+  def testAppendOnlyInputWithoutOnConflict(): Unit = {
+    util.tableEnv.getConfig
+      .set(ExecutionConfigOptions.TABLE_EXEC_SINK_REQUIRE_ON_CONFLICT, Boolean.box(false))
+    util.addTable(s"""
+                     |CREATE TABLE sinkWithPk (
+                     |  `a` INT,
+                     |  `b` BIGINT,
+                     |  PRIMARY KEY (a) NOT ENFORCED
+                     |) WITH (
+                     |  'connector' = 'values',
+                     |  'sink-insert-only' = 'false'
+                     |)
+                     |""".stripMargin)
+    val stmtSet = util.tableEnv.createStatementSet()
+    stmtSet.addInsertSql("INSERT INTO sinkWithPk SELECT a, b FROM MyTable")
+    util.verifyRelPlan(stmtSet, ExplainDetail.CHANGELOG_MODE)
+  }
+
+  @Test
+  def testUpdatingInputWithoutOnConflict(): Unit = {
+    util.tableEnv.getConfig
+      .set(ExecutionConfigOptions.TABLE_EXEC_SINK_REQUIRE_ON_CONFLICT, Boolean.box(false))
+    util.addTable(s"""
+                     |CREATE TABLE updatingSinkWithPk (
+                     |  `id` INT,
+                     |  `cnt` BIGINT,
+                     |  PRIMARY KEY (id) NOT ENFORCED
+                     |) WITH (
+                     |  'connector' = 'values',
+                     |  'sink-insert-only' = 'false'
+                     |)
+                     |""".stripMargin)
+    val stmtSet = util.tableEnv.createStatementSet()
+    // The upsert key is the grouping key c, which is not written to the sink, so it can never
+    // match the primary key.
+    stmtSet.addInsertSql(
+      "INSERT INTO updatingSinkWithPk SELECT MAX(a), COUNT(*) FROM MyTable GROUP BY c")
+    util.verifyRelPlan(stmtSet, ExplainDetail.CHANGELOG_MODE)
+  }
+
   @Test
   def testInjectiveCastPreservesUpsertKey(): Unit = {
     // Aggregation produces upsert stream with key (a).