[FLINK-40497][table] Allow materialized tables to default to ON CONFLICT DO ERROR

docs/content.zh/docs/sql/reference/dml/insert.md

@@ -314,6 +314,11 @@ This check is controlled by the configuration option `table.exec.sink.require-on
 
 Alternatively, if you do not need consistency guarantees for conflicting keys, you can disable the sink upsert materializer entirely by setting `table.exec.sink.upsert-materialize` to `NONE`. This removes the materializer operator from the pipeline, so no buffering, compaction, or conflict resolution is performed. Records are passed directly to the sink as they arrive.
 
+### Materialized tables
+
+A materialized table has no syntax for an explicit `ON CONFLICT` clause, so it always falls back to the implicit `DO DEDUPLICATE` strategy when its primary key differs from its query's upsert key. Setting `table.exec.sink.materialized-table-forces-on-conflict-error` (default: `false`) makes it default to `DO ERROR` instead, avoiding `DO DEDUPLICATE`'s higher state cost. Since `DO ERROR` requires a watermark on every source, this only applies when all sources already have one - otherwise the table keeps the `DO DEDUPLICATE` default so its creation never fails because of this option. This option adds `ON CONFLICT DO ERROR` independently from `table.exec.sink.require-on-conflict`. Optimally, when working with Materialized Tables, disable `table.exec.sink.require-on-conflict` to avoid confusion. If you enable both options, all your source tables will have to define a watermark strategy.
+
+
 ### Syntax
 
 ```sql

docs/content/docs/sql/reference/dml/insert.md

@@ -324,6 +324,12 @@ This check is controlled by the configuration option `table.exec.sink.require-on
 
 Alternatively, if you do not need consistency guarantees for conflicting keys, you can disable the sink upsert materializer entirely by setting `table.exec.sink.upsert-materialize` to `NONE`. This removes the materializer operator from the pipeline, so no buffering, compaction, or conflict resolution is performed. Records are passed directly to the sink as they arrive.
 
+### Materialized tables
+
+A materialized table has no syntax for an explicit `ON CONFLICT` clause, so it always falls back to the implicit `DO DEDUPLICATE` strategy when its primary key differs from its query's upsert key. Setting `table.exec.sink.materialized-table-forces-on-conflict-error` (default: `false`) makes it default to `DO ERROR` instead, avoiding `DO DEDUPLICATE`'s higher state cost. Since `DO ERROR` requires a watermark on every source, this only applies when all sources already have one - otherwise the table keeps the `DO DEDUPLICATE` default so its creation never fails because of this option. This option adds `ON CONFLICT DO ERROR` independently from `table.exec.sink.require-on-conflict`. Optimally, when working with Materialized Tables, disable `table.exec.sink.require-on-conflict` to avoid confusion. If you enable both options, all your source tables will have to define a watermark strategy.
+
+
+
 ### Syntax
 
 ```sql

docs/layouts/shortcodes/generated/execution_config_configuration.html

@@ -260,6 +260,12 @@
             <td><p>Enum</p></td>
             <td>In order to minimize the distributed disorder problem when writing data into table with primary keys that many users suffers. FLINK will auto add a keyed shuffle by default when the sink parallelism differs from upstream operator and sink parallelism is not 1. This works only when the upstream ensures the multi-records' order on the primary key, if not, the added shuffle can not solve the problem (In this situation, a more proper way is to consider the deduplicate operation for the source firstly or use an upsert source with primary key definition which truly reflect the records evolution).<br />By default, the keyed shuffle will be added when the sink's parallelism differs from upstream operator. You can set to no shuffle(NONE) or force shuffle(FORCE).<br /><br />Possible values:<ul><li>"NONE"</li><li>"AUTO"</li><li>"FORCE"</li></ul></td>
         </tr>
+        <tr>
+            <td><h5>table.exec.sink.materialized-table-forces-on-conflict-error</h5><br> <span class="label label-primary">Streaming</span></td>
+            <td style="word-wrap: break-word;">false</td>
+            <td>Boolean</td>
+            <td>When enabled, a materialized table whose declared primary key differs from its query's upsert key defaults to ON CONFLICT DO ERROR instead of the state-heavy deduplicating materializer.<br /><br />Independent of table.exec.sink.require-on-conflict, which controls whether an explicit ON CONFLICT clause is required elsewhere.</td>
+        </tr>
         <tr>
             <td><h5>table.exec.sink.nested-constraint-enforcer</h5><br> <span class="label label-primary">Batch</span> <span class="label label-primary">Streaming</span></td>
             <td style="word-wrap: break-word;">IGNORE</td>

flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/config/ExecutionConfigOptions.java

@@ -277,6 +277,27 @@ public class ExecutionConfigOptions {
                                                     + "results in certain streaming scenarios.")
                                     .build());
 
+    @Documentation.TableOption(execMode = Documentation.ExecMode.STREAMING)
+    public static final ConfigOption<Boolean>
+            TABLE_EXEC_SINK_MATERIALIZED_TABLE_FORCES_ON_CONFLICT_ERROR =
+                    key("table.exec.sink.materialized-table-forces-on-conflict-error")
+                            .booleanType()
+                            .defaultValue(false)
+                            .withDescription(
+                                    Description.builder()
+                                            .text(
+                                                    "When enabled, a materialized table whose declared primary key "
+                                                            + "differs from its query's upsert key defaults to ON CONFLICT "
+                                                            + "DO ERROR instead of the state-heavy deduplicating "
+                                                            + "materializer.")
+                                            .linebreak()
+                                            .linebreak()
+                                            .text(
+                                                    "Independent of table.exec.sink.require-on-conflict, which "
+                                                            + "controls whether an explicit ON CONFLICT clause is "
+                                                            + "required elsewhere.")
+                                            .build());
+
     // ------------------------------------------------------------------------
     //  Sort Options
     // ------------------------------------------------------------------------

flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/connectors/DynamicSinkUtils.java

@@ -83,6 +83,7 @@
 import org.apache.calcite.rel.RelNode;
 import org.apache.calcite.rel.core.Project;
 import org.apache.calcite.rel.core.TableModify;
+import org.apache.calcite.rel.core.TableScan;
 import org.apache.calcite.rel.hint.RelHint;
 import org.apache.calcite.rel.logical.LogicalFilter;
 import org.apache.calcite.rel.logical.LogicalTableModify;
@@ -263,6 +264,10 @@ public static RelNode convertCreateTableAsToRel(
      * AS) into a {@link RelNode} that writes into the table, adding helper projections if
      * necessary. The caller resolves the target table to its {@link ResolvedCatalogTable} form (the
      * new definition for an alter, the created definition for a create).
+     *
+     * <p>The table defaults to {@code ON CONFLICT DO ERROR} when {@link
+     * ExecutionConfigOptions#TABLE_EXEC_SINK_MATERIALIZED_TABLE_FORCES_ON_CONFLICT_ERROR} is set
+     * and every source has a watermark; otherwise it keeps the deduplicating default.
      */
     public static RelNode convertMaterializedTableAsToRel(
             FlinkRelBuilder relBuilder,
@@ -272,10 +277,19 @@ public static RelNode convertMaterializedTableAsToRel(
             ResolvedCatalogTable resolvedTable,
             Map<String, String> staticPartitions,
             boolean isOverwrite,
-            DynamicTableSink sink) {
+            DynamicTableSink sink,
+            ReadableConfig configuration) {
         final ContextResolvedTable contextResolvedTable =
                 ContextResolvedTable.permanent(identifier, catalog, resolvedTable);
 
+        final InsertConflictStrategy conflictStrategy =
+                configuration.get(
+                                        ExecutionConfigOptions
+                                                .TABLE_EXEC_SINK_MATERIALIZED_TABLE_FORCES_ON_CONFLICT_ERROR)
+                                && allSourcesHaveWatermarks(input)
+                        ? InsertConflictStrategy.error()
+                        : null;
+
         return convertSinkToRel(
                 relBuilder,
                 input,
@@ -285,7 +299,28 @@ public static RelNode convertMaterializedTableAsToRel(
                 null,
                 isOverwrite,
                 sink,
-                null);
+                conflictStrategy);
+    }
+
+    /** Whether every table scanned by {@code rel} declares a watermark. */
+    private static boolean allSourcesHaveWatermarks(RelNode rel) {
+        if (rel instanceof TableScan) {
+            final TableSourceTable table =
+                    ((TableScan) rel).getTable().unwrap(TableSourceTable.class);
+            // An unresolvable scan can't be proven to lack a watermark, so it isn't treated as a
+            // reason to fall back - matching the same convention used for the physical-tree check.
+            return table == null
+                    || !table.contextResolvedTable()
+                            .getResolvedSchema()
+                            .getWatermarkSpecs()
+                            .isEmpty();
+        }
+        for (RelNode input : rel.getInputs()) {
+            if (!allSourcesHaveWatermarks(input)) {
+                return false;
+            }
+        }
+        return true;
     }
 
     private static RelNode convertSinkToRel(

flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/delegation/PlannerBase.scala

@@ -615,7 +615,8 @@ abstract class PlannerBase(
       resolvedTable,
       staticPartitions,
       false,
-      tableSink)
+      tableSink,
+      getTableConfig)
   }
 
   protected def createSerdeContext: SerdeContext = {

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

@@ -1249,17 +1249,34 @@ class FlinkChangelogModeInferenceProgram extends FlinkOptimizeProgram[StreamOpti
       if (requireOnConflict && upsertKeyDiffersFromPk && sink.conflictStrategy == null) {
         val pkNames = sink.getPrimaryKeyNames
         val upsertKeyNames = sink.getUpsertKeyNames
+        val identifier = sink.contextResolvedTable.getIdentifier.asSummaryString
+        val mtOnConflictDoError = tableConfig.get(
+          ExecutionConfigOptions.TABLE_EXEC_SINK_MATERIALIZED_TABLE_FORCES_ON_CONFLICT_ERROR)
+        val materializedTableHint =
+          if (mtOnConflictDoError) {
+            // When both requireOnConflict and mtOnConflictDoError are enabled, every source
+            // must declare a watermark for ON CONFLICT DO ERROR to apply.
+            val sourcesWithoutWatermarks = new java.util.ArrayList[String]()
+            collectSourcesWithoutWatermarks(sink.getInput, sourcesWithoutWatermarks)
+            " If this is a materialized table: table.exec.sink.materialized-table-forces-on-" +
+              "conflict-error is enabled, but a watermark is missing on: " +
+              s"${sourcesWithoutWatermarks.toArray.mkString(", ")}. Add one to every source to " +
+              "satisfy both options."
+          } else {
+            ""
+          }
         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"The query has an upsert key that differs from the primary key of the sink table " +
+            s"'$identifier'. " +
             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).")
+            "ON CONFLICT DO NOTHING (keep first record)." +
+            materializedTableHint)
       }
     }