From a06db4147ad51e1d13515b1e1278867581d05216 Mon Sep 17 00:00:00 2001
From: Gustavo de Morais <gdemorais@confluent.io>
Date: Wed, 5 Aug 2026 18:24:29 +0200
Subject: [PATCH] [FLINK-40334][table] Insert-only input should stay
 unmaterialized under FORCE

FLINK-38928 dropped the guard that skipped the sink upsert materializer for
insert-only input, because DO ERROR and DO NOTHING need the operator to compare
inserts that share a primary key. Without it, an insert-only query into a sink
with a primary key gets a SinkUpsertMaterializer and a keyed shuffle whenever
upsert materialization is forced, even though there are no changes to reconcile.

Restore the guard and pass the conflict strategy into the sink transformation so
that only DO ERROR and DO NOTHING keep materializing insert-only input.

The decision stays at translation time on purpose. Plans compiled before the
clause existed persist requireUpsertMaterialize = true for this case, so
suppressing the operator in the planner instead would make it appear on restore.
---
 .../plan/nodes/exec/batch/BatchExecSink.java  |   1 +
 .../nodes/exec/common/CommonExecSink.java     |  24 ++-
 .../nodes/exec/stream/StreamExecSink.java     |  14 +-
 .../planner/plan/stream/sql/TableSinkTest.xml | 173 ++++++++++++++++++
 .../plan/stream/sql/TableSinkTest.scala       |  45 +++++
 5 files changed, 244 insertions(+), 13 deletions(-)

diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecSink.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecSink.java
index 8a1dd04bcfe87..cb267e24e80d1 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecSink.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecSink.java
@@ -120,6 +120,7 @@ protected Transformation<Object> translateToPlanInternal(
                 tableSink,
                 -1,
                 false,
+                null,
                 null);
     }
 
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecSink.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecSink.java
index 31c0bc4074feb..3738840ab2761 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecSink.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecSink.java
@@ -36,6 +36,8 @@
 import org.apache.flink.streaming.api.transformations.PartitionTransformation;
 import org.apache.flink.streaming.api.transformations.TransformationWithLineage;
 import org.apache.flink.streaming.runtime.partitioner.KeyGroupStreamPartitioner;
+import org.apache.flink.table.api.InsertConflictStrategy;
+import org.apache.flink.table.api.InsertConflictStrategy.ConflictBehavior;
 import org.apache.flink.table.api.TableException;
 import org.apache.flink.table.api.config.ExecutionConfigOptions;
 import org.apache.flink.table.catalog.ResolvedSchema;
@@ -84,6 +86,8 @@
 
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
 
+import javax.annotation.Nullable;
+
 import java.util.Arrays;
 import java.util.List;
 import java.util.Objects;
@@ -146,7 +150,8 @@ protected Transformation<Object> createSinkTransformation(
             DynamicTableSink tableSink,
             int rowtimeFieldIndex,
             boolean upsertMaterialize,
-            int[] inputUpsertKey) {
+            int[] inputUpsertKey,
+            @Nullable InsertConflictStrategy conflictStrategy) {
         final ResolvedSchema schema = tableSinkSpec.getContextResolvedTable().getResolvedSchema();
         final SinkRuntimeProvider runtimeProvider =
                 tableSink.getSinkRuntimeProvider(
@@ -187,6 +192,11 @@ protected Transformation<Object> createSinkTransformation(
         Optional<LineageVertex> lineageVertexOpt =
                 TableLineageUtils.extractLineageDataset(outputObject);
 
+        // only add materialization if input has changes, unless the conflict strategy has to
+        // compare every insert against the row stored under the same primary key
+        final boolean needMaterialization =
+                upsertMaterialize && (!inputInsertOnly || detectsDuplicateKeys(conflictStrategy));
+
         Transformation<RowData> sinkTransform =
                 applyConstraintValidations(inputTransform, config, persistedRowType);
 
@@ -199,10 +209,10 @@ protected Transformation<Object> createSinkTransformation(
                             primaryKeys,
                             sinkParallelism,
                             inputParallelism,
-                            upsertMaterialize);
+                            needMaterialization);
         }
 
-        if (upsertMaterialize) {
+        if (needMaterialization) {
             sinkTransform =
                     applyUpsertMaterialize(
                             sinkTransform,
@@ -246,6 +256,14 @@ protected Transformation<Object> createSinkTransformation(
         return transformation;
     }
 
+    /** Whether the conflict strategy has to detect rows that share a primary key. */
+    protected static boolean detectsDuplicateKeys(
+            @Nullable InsertConflictStrategy conflictStrategy) {
+        return conflictStrategy != null
+                && (conflictStrategy.getBehavior() == ConflictBehavior.ERROR
+                        || conflictStrategy.getBehavior() == ConflictBehavior.NOTHING);
+    }
+
     /**
      * Apply an operator to filter or report error to process not-null values for not-null fields.
      */
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecSink.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecSink.java
index 22dce136a0e24..acc7fa10b2268 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecSink.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecSink.java
@@ -26,7 +26,6 @@
 import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
 import org.apache.flink.streaming.api.transformations.OneInputTransformation;
 import org.apache.flink.table.api.InsertConflictStrategy;
-import org.apache.flink.table.api.InsertConflictStrategy.ConflictBehavior;
 import org.apache.flink.table.api.TableException;
 import org.apache.flink.table.api.config.ExecutionConfigOptions;
 import org.apache.flink.table.api.config.ExecutionConfigOptions.RowtimeInserter;
@@ -291,7 +290,8 @@ protected Transformation<Object> translateToPlanInternal(
                 tableSink,
                 rowtimeFieldIndex,
                 upsertMaterialize,
-                inputUpsertKey);
+                inputUpsertKey,
+                conflictStrategy);
     }
 
     @Override
@@ -359,7 +359,7 @@ protected Transformation<RowData> applyUpsertMaterialize(
         // This assigns the current watermark as the timestamp to each record,
         // which is required for the WatermarkCompactingSinkMaterializer to work correctly
         Transformation<RowData> transformForMaterializer = inputTransform;
-        if (isErrorOrNothingConflictStrategy()) {
+        if (detectsDuplicateKeys(conflictStrategy)) {
             // Use input parallelism to preserve watermark semantics
             transformForMaterializer =
                     ExecNodeUtil.createOneInputTransformation(
@@ -410,7 +410,7 @@ private OneInputStreamOperator<RowData, RowData> createSumOperator(
             GeneratedHashFunction rowHashFunction) {
 
         // Check if we should use the watermark-compacting materializer for ERROR/NOTHING strategies
-        if (isErrorOrNothingConflictStrategy()) {
+        if (detectsDuplicateKeys(conflictStrategy)) {
             RowType keyType = RowTypeUtils.projectRowType(physicalRowType, primaryKeys);
 
             return WatermarkCompactingSinkMaterializer.create(
@@ -450,12 +450,6 @@ private OneInputStreamOperator<RowData, RowData> createSumOperator(
                                 config));
     }
 
-    private boolean isErrorOrNothingConflictStrategy() {
-        return conflictStrategy != null
-                && (conflictStrategy.getBehavior() == ConflictBehavior.ERROR
-                        || conflictStrategy.getBehavior() == ConflictBehavior.NOTHING);
-    }
-
     private static SequencedMultiSetStateConfig createStateConfig(
             SinkUpsertMaterializeStrategy strategy,
             TimeDomain ttlTimeDomain,
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/TableSinkTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/TableSinkTest.xml
index 90d4c28486542..486e4d045ec2c 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/TableSinkTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/TableSinkTest.xml
@@ -504,6 +504,179 @@ Sink(table=[default_catalog.default_database.sink], fields=[a, b])
 ]]>
     </Resource>
   </TestCase>
+  <TestCase name="testForcedMaterializeWithAppendOnlyInput">
+    <Resource name="explain">
+      <![CDATA[== Abstract Syntax Tree ==
+LogicalSink(table=[default_catalog.default_database.forcedSink], fields=[a, b])
++- LogicalProject(a=[$0], b=[$1])
+   +- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
+
+== Optimized Physical Plan ==
+Sink(table=[default_catalog.default_database.forcedSink], fields=[a, b], upsertMaterialize=[true])
++- Calc(select=[a, b])
+   +- DataStreamScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c])
+
+== Optimized Execution Plan ==
+Sink(table=[default_catalog.default_database.forcedSink], fields=[a, b], upsertMaterialize=[true])
++- Calc(select=[a, b])
+   +- DataStreamScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c])
+
+== Physical Execution Plan ==
+{
+  "nodes" : [ {
+    "id" : ,
+    "type" : "Source: Collection Source",
+    "pact" : "Data Source",
+    "contents" : "Source: Collection Source",
+    "parallelism" : 1
+  }, {
+    "id" : ,
+    "type" : "SourceConversion[]",
+    "pact" : "Operator",
+    "contents" : "[]:SourceConversion(table=[default_catalog.default_database.MyTable], fields=[a, b, c])",
+    "parallelism" : 1,
+    "predecessors" : [ {
+      "id" : ,
+      "ship_strategy" : "FORWARD",
+      "side" : "second"
+    } ]
+  }, {
+    "id" : ,
+    "type" : "Calc[]",
+    "pact" : "Operator",
+    "contents" : "[]:Calc(select=[a, b])",
+    "parallelism" : 1,
+    "predecessors" : [ {
+      "id" : ,
+      "ship_strategy" : "FORWARD",
+      "side" : "second"
+    } ]
+  }, {
+    "id" : ,
+    "type" : "ConstraintEnforcer[]",
+    "pact" : "Operator",
+    "contents" : "[]:ConstraintEnforcer[NotNullEnforcer(fields=[a])]",
+    "parallelism" : 1,
+    "predecessors" : [ {
+      "id" : ,
+      "ship_strategy" : "FORWARD",
+      "side" : "second"
+    } ]
+  }, {
+    "id" : ,
+    "type" : "Sink: forcedSink[]",
+    "pact" : "Data Sink",
+    "contents" : "[]:Sink(table=[default_catalog.default_database.forcedSink], fields=[a, b], upsertMaterialize=[true])",
+    "parallelism" : 1,
+    "predecessors" : [ {
+      "id" : ,
+      "ship_strategy" : "FORWARD",
+      "side" : "second"
+    } ]
+  } ]
+}]]>
+    </Resource>
+  </TestCase>
+  <TestCase name="testForcedMaterializeWithUpdatingInput">
+    <Resource name="explain">
+      <![CDATA[== Abstract Syntax Tree ==
+LogicalSink(table=[default_catalog.default_database.forcedSinkWithCount], fields=[c, EXPR$1])
++- LogicalAggregate(group=[{0}], EXPR$1=[COUNT()])
+   +- LogicalProject(c=[$2])
+      +- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
+
+== Optimized Physical Plan ==
+Sink(table=[default_catalog.default_database.forcedSinkWithCount], fields=[c, EXPR$1], upsertMaterialize=[true])
++- GroupAggregate(groupBy=[c], select=[c, COUNT(*) AS EXPR$1])
+   +- Exchange(distribution=[hash[c]])
+      +- Calc(select=[c])
+         +- DataStreamScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c])
+
+== Optimized Execution Plan ==
+Sink(table=[default_catalog.default_database.forcedSinkWithCount], fields=[c, EXPR$1], upsertMaterialize=[true])
++- GroupAggregate(groupBy=[c], select=[c, COUNT(*) AS EXPR$1])
+   +- Exchange(distribution=[hash[c]])
+      +- Calc(select=[c])
+         +- DataStreamScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c])
+
+== Physical Execution Plan ==
+{
+  "nodes" : [ {
+    "id" : ,
+    "type" : "Source: Collection Source",
+    "pact" : "Data Source",
+    "contents" : "Source: Collection Source",
+    "parallelism" : 1
+  }, {
+    "id" : ,
+    "type" : "SourceConversion[]",
+    "pact" : "Operator",
+    "contents" : "[]:SourceConversion(table=[default_catalog.default_database.MyTable], fields=[a, b, c])",
+    "parallelism" : 1,
+    "predecessors" : [ {
+      "id" : ,
+      "ship_strategy" : "FORWARD",
+      "side" : "second"
+    } ]
+  }, {
+    "id" : ,
+    "type" : "Calc[]",
+    "pact" : "Operator",
+    "contents" : "[]:Calc(select=[c])",
+    "parallelism" : 1,
+    "predecessors" : [ {
+      "id" : ,
+      "ship_strategy" : "FORWARD",
+      "side" : "second"
+    } ]
+  }, {
+    "id" : ,
+    "type" : "GroupAggregate[]",
+    "pact" : "Operator",
+    "contents" : "[]:GroupAggregate(groupBy=[c], select=[c, COUNT(*) AS EXPR$1])",
+    "parallelism" : 1,
+    "predecessors" : [ {
+      "id" : ,
+      "ship_strategy" : "HASH",
+      "side" : "second"
+    } ]
+  }, {
+    "id" : ,
+    "type" : "ConstraintEnforcer[]",
+    "pact" : "Operator",
+    "contents" : "[]:ConstraintEnforcer[NotNullEnforcer(fields=[c])]",
+    "parallelism" : 1,
+    "predecessors" : [ {
+      "id" : ,
+      "ship_strategy" : "FORWARD",
+      "side" : "second"
+    } ]
+  }, {
+    "id" : ,
+    "type" : "SinkMaterializer[]",
+    "pact" : "Operator",
+    "contents" : "[]:SinkMaterializer(pk=[c])",
+    "parallelism" : 1,
+    "predecessors" : [ {
+      "id" : ,
+      "ship_strategy" : "HASH",
+      "side" : "second"
+    } ]
+  }, {
+    "id" : ,
+    "type" : "Sink: forcedSinkWithCount[]",
+    "pact" : "Data Sink",
+    "contents" : "[]:Sink(table=[default_catalog.default_database.forcedSinkWithCount], fields=[c, EXPR$1], upsertMaterialize=[true])",
+    "parallelism" : 1,
+    "predecessors" : [ {
+      "id" : ,
+      "ship_strategy" : "FORWARD",
+      "side" : "second"
+    } ]
+  } ]
+}]]>
+    </Resource>
+  </TestCase>
   <TestCase name="testInjectiveCastPreservesUpsertKey">
     <Resource name="ast">
       <![CDATA[
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/TableSinkTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/TableSinkTest.scala
index d3e46cfc7bdc6..a2bad91048f99 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/TableSinkTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/TableSinkTest.scala
@@ -905,6 +905,51 @@ class TableSinkTest extends TableTestBase {
     util.verifyRelPlan(stmtSet, ExplainDetail.CHANGELOG_MODE)
   }
 
+  @Test
+  def testForcedMaterializeWithAppendOnlyInput(): Unit = {
+    util.getStreamEnv.setParallelism(1)
+    util.tableEnv.getConfig.set(
+      ExecutionConfigOptions.TABLE_EXEC_SINK_UPSERT_MATERIALIZE,
+      ExecutionConfigOptions.UpsertMaterialize.FORCE)
+    util.addTable(s"""
+                     |CREATE TABLE forcedSink (
+                     |  `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 forcedSink SELECT a, b FROM MyTable")
+    // There is nothing to materialize, so the plan must not contain a SinkMaterializer.
+    util.verifyExplain(stmtSet, ExplainDetail.JSON_EXECUTION_PLAN)
+  }
+
+  @Test
+  def testForcedMaterializeWithUpdatingInput(): Unit = {
+    util.getStreamEnv.setParallelism(1)
+    util.tableEnv.getConfig.set(
+      ExecutionConfigOptions.TABLE_EXEC_SINK_UPSERT_MATERIALIZE,
+      ExecutionConfigOptions.UpsertMaterialize.FORCE)
+    util.addTable(s"""
+                     |CREATE TABLE forcedSinkWithCount (
+                     |  `c` STRING,
+                     |  `cnt` BIGINT,
+                     |  PRIMARY KEY (c) NOT ENFORCED
+                     |) WITH (
+                     |  'connector' = 'values',
+                     |  'sink-insert-only' = 'false'
+                     |)
+                     |""".stripMargin)
+    val stmtSet = util.tableEnv.createStatementSet()
+    stmtSet.addInsertSql(
+      "INSERT INTO forcedSinkWithCount SELECT c, COUNT(*) FROM MyTable GROUP BY c")
+    // The upsert key already matches the primary key, so only FORCE asks for a SinkMaterializer.
+    util.verifyExplain(stmtSet, ExplainDetail.JSON_EXECUTION_PLAN)
+  }
+
   @Test
   def testInjectiveCastPreservesUpsertKey(): Unit = {
     // Aggregation produces upsert stream with key (a).