This is an automated email from the ASF dual-hosted git repository.
RocMarshal pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new befec0111b8 [FLINK-40924][table-planner] Enforce require-on-conflict
under upsert-materialize FORCE
befec0111b8 is described below
commit befec0111b8ef3321c6472fb2370e68763eff414
Author: SEPURI-SAI-KRISHNA <[email protected]>
AuthorDate: Thu Oct 15 20:46:28 2026 +0530
[FLINK-40924][table-planner] Enforce require-on-conflict under
upsert-materialize FORCE
The ON CONFLICT requirement was validated inside the AUTO branch of
analyzeUpsertMaterializeStrategy, so setting
table.exec.sink.upsert-materialize to FORCE skipped it and the query
planned even though its upsert key differs from the sink's primary key
and no ON CONFLICT clause was given.
FORCE adds the sink upsert materializer, and the materializer is what
applies the conflict strategy, so the clause does take effect under
FORCE. The check that already runs under AUTO now runs there too, with
the same condition.
NONE keeps its current behaviour on purpose. It removes the
materializer, so nothing is left to apply a conflict strategy and the
clause would have no effect. Opting out that way is documented in
docs/content/docs/sql/reference/dml/insert.md.
Generated-by: Claude Code (Opus 5)
---
.../FlinkChangelogModeInferenceProgram.scala | 21 +++++++--
.../planner/plan/stream/sql/TableSinkTest.xml | 21 +++++++++
.../planner/plan/stream/sql/TableSinkTest.scala | 55 ++++++++++++++++++++++
3 files changed, 94 insertions(+), 3 deletions(-)
diff --git
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkChangelogModeInferenceProgram.scala
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkChangelogModeInferenceProgram.scala
index f3f0df6e709..82cc30ddf8e 100644
---
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkChangelogModeInferenceProgram.scala
+++
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkChangelogModeInferenceProgram.scala
@@ -1172,7 +1172,9 @@ class FlinkChangelogModeInferenceProgram extends
FlinkOptimizeProgram[StreamOpti
* 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.
+ * Also validates that an ON CONFLICT clause is specified when the upsert
key differs from the
+ * primary key. That applies to every setting that can add the
materializer, AUTO and FORCE,
+ * because the materializer is what applies the conflict strategy.
*/
private def analyzeUpsertMaterializeStrategy(sink: StreamPhysicalSink):
Boolean = {
val tableConfig = unwrapTableConfig(sink)
@@ -1209,13 +1211,26 @@ class FlinkChangelogModeInferenceProgram extends
FlinkOptimizeProgram[StreamOpti
validateSourcesHaveWatermarks(sink)
}
+ // An ON CONFLICT clause is only meaningful for an upsert sink with a
primary key, because
+ // it is rejected for append and retract sinks above.
+ val sinkIsUpsertWithPk = primaryKeys.nonEmpty && !sinkIsAppend &&
!sinkIsRetract
+
tableConfig.get(ExecutionConfigOptions.TABLE_EXEC_SINK_UPSERT_MATERIALIZE)
match {
- case UpsertMaterialize.FORCE => primaryKeys.nonEmpty && !sinkIsRetract
+ case UpsertMaterialize.FORCE =>
+ // FORCE adds the materializer, which is what applies the conflict
strategy, so an
+ // absent clause is rejected here just as it is under AUTO.
+ if (sinkIsUpsertWithPk) {
+ val upsertKeyDiffersFromPk = !sink.primaryKeysContainsUpsertKey
+ validateOnConflictSpecifiedIfRequired(sink, tableConfig,
upsertKeyDiffersFromPk)
+ }
+ primaryKeys.nonEmpty && !sinkIsRetract
+ // NONE removes the materializer, so nothing is left to apply a
conflict strategy and the
+ // clause is not required. This is the documented way to opt out of
conflict handling.
case UpsertMaterialize.NONE => false
case UpsertMaterialize.AUTO =>
// if the sink is not an UPSERT sink (has no PK, or is an APPEND or
RETRACT sink)
// we don't need to materialize results
- if (primaryKeys.isEmpty || sinkIsAppend || sinkIsRetract) {
+ if (!sinkIsUpsertWithPk) {
return false
}
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 493a6f1f932..019bfcc051f 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
@@ -798,6 +798,27 @@
Sink(table=[default_catalog.default_database.sink_agg_with_int_pk], fields=[EXPR
+- Exchange(distribution=[hash[c]])
+- Calc(select=[c])
+- DataStreamScan(table=[[default_catalog, default_database,
MyTable]], fields=[a, b, c])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testOnConflictIsNotRequiredWithoutMaterialize">
+ <Resource name="ast">
+ <![CDATA[
+LogicalSink(table=[default_catalog.default_database.sinkWithPkNoMaterialize],
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.sinkWithPkNoMaterialize],
fields=[EXPR$0, EXPR$1], 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>
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 346db891909..4183f28f4f1 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,12 +905,67 @@ class TableSinkTest extends TableTestBase {
util.verifyRelPlan(stmtSet, ExplainDetail.CHANGELOG_MODE)
}
+ @Test
+ def testOnConflictIsRequiredWithForcedMaterialize(): Unit = {
+ util.tableEnv.getConfig.set(
+ ExecutionConfigOptions.TABLE_EXEC_SINK_UPSERT_MATERIALIZE,
+ ExecutionConfigOptions.UpsertMaterialize.FORCE)
+ util.addTable(s"""
+ |CREATE TABLE sinkWithPkForcedMaterialize (
+ | `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. FORCE adds the materializer, so the clause takes
effect and is
+ // required here for the same reason it is under AUTO.
+ stmtSet.addInsertSql(
+ "INSERT INTO sinkWithPkForcedMaterialize SELECT MAX(a), COUNT(*) FROM
MyTable GROUP BY c")
+ assertThatThrownBy(() => util.verifyRelPlan(stmtSet,
ExplainDetail.CHANGELOG_MODE))
+ .isInstanceOf(classOf[ValidationException])
+ .hasMessageContaining(
+ "The query has an upsert key that differs from the primary key of the
sink table")
+ }
+
+ @Test
+ def testOnConflictIsNotRequiredWithoutMaterialize(): Unit = {
+ util.tableEnv.getConfig.set(
+ ExecutionConfigOptions.TABLE_EXEC_SINK_UPSERT_MATERIALIZE,
+ ExecutionConfigOptions.UpsertMaterialize.NONE)
+ util.addTable(s"""
+ |CREATE TABLE sinkWithPkNoMaterialize (
+ | `id` INT,
+ | `cnt` BIGINT,
+ | PRIMARY KEY (id) NOT ENFORCED
+ |) WITH (
+ | 'connector' = 'values',
+ | 'sink-insert-only' = 'false'
+ |)
+ |""".stripMargin)
+ val stmtSet = util.tableEnv.createStatementSet()
+ // NONE removes the materializer, the only thing that would apply a
conflict strategy, so the
+ // clause stays optional under it. Keep this behaviour, it is documented
with the option.
+ stmtSet.addInsertSql(
+ "INSERT INTO sinkWithPkNoMaterialize SELECT MAX(a), COUNT(*) FROM
MyTable GROUP BY c")
+ 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)
+ // This test covers what FORCE materializes, not ON CONFLICT enforcement.
The input is append
+ // only and carries no upsert key, so the clause would otherwise be
required, exactly as it
+ // already is under AUTO in testAppendOnlyInputWithoutOnConflict.
+ util.tableEnv.getConfig
+ .set(ExecutionConfigOptions.TABLE_EXEC_SINK_REQUIRE_ON_CONFLICT,
Boolean.box(false))
util.addTable(s"""
|CREATE TABLE forcedSink (
| `a` INT,
