This is an automated email from the ASF dual-hosted git repository.
gustavodemorais 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 5d91e07c8f2 [FLINK-40324][table] Append-only should stay
unmaterialized without an ON CONFLICT clause
5d91e07c8f2 is described below
commit 5d91e07c8f2cfc3b4d0e779748e08311ea306bd3
Author: Gustavo de Morais <[email protected]>
AuthorDate: Wed Aug 5 10:54:36 2026 +0200
[FLINK-40324][table] Append-only should stay unmaterialized without an ON
CONFLICT clause
This closes #28918
---
.../FlinkChangelogModeInferenceProgram.scala | 60 +++++++++++++---------
.../planner/plan/stream/sql/TableSinkTest.xml | 37 +++++++++++++
.../planner/plan/stream/sql/TableSinkTest.scala | 41 +++++++++++++++
3 files changed, 113 insertions(+), 25 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 1c91e960a8a..5d578e2b076 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
@@ -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)
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 1ae3035de31..90d4c284865 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
@@ -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>
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 31f0d6ca6fa..d3e46cfc7bd 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
@@ -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).
