This is an automated email from the ASF dual-hosted git repository.
twalthr 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 4732a5210a8 [FLINK-40111][table-planner] Fall back to retract for a
keyless sink without an upsert key
4732a5210a8 is described below
commit 4732a5210a876a55a9367e31f3aa9f8cf0f45fd7
Author: Gustavo de Morais <[email protected]>
AuthorDate: Fri Jul 10 09:26:54 2026 +0200
[FLINK-40111][table-planner] Fall back to retract for a keyless sink
without an upsert key
This closes #28704.
---
.../FlinkChangelogModeInferenceProgram.scala | 10 +++--
.../physical/stream/ChangelogModeInferenceTest.xml | 52 ++++++++++++++++++++++
.../plan/stream/sql/DagOptimizationTest.xml | 12 ++---
.../stream/ChangelogModeInferenceTest.scala | 37 +++++++++++++++
.../plan/stream/sql/DagOptimizationTest.scala | 8 ++--
5 files changed, 107 insertions(+), 12 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 77fc7208d24..8313e4c1d96 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
@@ -1074,15 +1074,19 @@ class FlinkChangelogModeInferenceProgram extends
FlinkOptimizeProgram[StreamOpti
* <p>Notice: even if sink pk is a subset of the upsert key, the pk is NOT
considered satisfied
* when the upsert key has columns outside sink pk. This differs from
batch job's unique key
* inference.
+ *
+ * <p>A sink without a primary key is satisfied whenever the input carries
any upsert key.
*/
private def canUpsertKeysWithImmutableColsSatisfyPk(sink:
StreamPhysicalSink): Boolean = {
val sinkDefinedPks =
sink.contextResolvedTable.getResolvedSchema.getPrimaryKeyIndexes
+ val fmq =
FlinkRelMetadataQuery.reuseOrCreate(sink.getCluster.getMetadataQuery)
+ val changeLogUpsertKeys = fmq.getUpsertKeys(sink.getInput)
if (sinkDefinedPks.isEmpty) {
- return true
+ // A keyless sink cannot apply UPDATE_AFTER in place, so it can only
accept upsert when the
+ // input itself carries an upsert key; otherwise fall back to
beforeAndAfter.
+ return changeLogUpsertKeys != null && !changeLogUpsertKeys.isEmpty
}
val sinkPks = ImmutableBitSet.of(sinkDefinedPks: _*)
- val fmq =
FlinkRelMetadataQuery.reuseOrCreate(sink.getCluster.getMetadataQuery)
- val changeLogUpsertKeys = fmq.getUpsertKeys(sink.getInput)
// if upsert key is null, pk cannot be satisfied, should fall back to
beforeAndAfter
if (changeLogUpsertKeys == null) {
return false
diff --git
a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/physical/stream/ChangelogModeInferenceTest.xml
b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/physical/stream/ChangelogModeInferenceTest.xml
index 03c86424dd1..1e453bc0c9e 100644
---
a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/physical/stream/ChangelogModeInferenceTest.xml
+++
b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/physical/stream/ChangelogModeInferenceTest.xml
@@ -172,6 +172,58 @@ GroupAggregate(groupBy=[cnt], select=[cnt,
COUNT_RETRACT(cnt) AS frequency], cha
: +- LegacyTableSourceScan(table=[[default_catalog,
default_database, MyTable, source: [CollectionTableSource(word, number)]]],
fields=[word, number], changelogMode=[I])
+- Calc(select=[CAST(cnt AS BIGINT) AS cnt], changelogMode=[I])
+- LegacyTableSourceScan(table=[[default_catalog, default_database,
MyTable2, source: [CollectionTableSource(word, cnt)]]], fields=[word, cnt],
changelogMode=[I])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase
name="testKeylessUpsertSinkFallsBackToRetractWhenUpsertKeyProjectedAway">
+ <Resource name="sql">
+ <![CDATA[INSERT INTO keyless_upsert_sink_no_key SELECT rate FROM
DeduplicatedView]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalSink(table=[default_catalog.default_database.keyless_upsert_sink_no_key],
fields=[rate])
++- LogicalProject(rate=[$1])
+ +- LogicalProject(currency=[$0], rate=[$1], rowtime=[$2])
+ +- LogicalFilter(condition=[=($3, 1)])
+ +- LogicalProject(currency=[$0], rate=[$1], rowtime=[$2],
rowNum=[ROW_NUMBER() OVER (PARTITION BY $0 ORDER BY $2 DESC NULLS LAST)])
+ +- LogicalWatermarkAssigner(rowtime=[rowtime], watermark=[$2])
+ +- LogicalTableScan(table=[[default_catalog, default_database,
ratesHistory, source: [CollectionTableSource(currency, rate, rowtime)]]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+Sink(table=[default_catalog.default_database.keyless_upsert_sink_no_key],
fields=[rate], changelogMode=[NONE])
++- Calc(select=[rate], changelogMode=[I,UB,UA,D])
+ +- Rank(strategy=[AppendFastStrategy], rankType=[ROW_NUMBER],
rankRange=[rankStart=1, rankEnd=1], partitionBy=[currency], orderBy=[ROWTIME
rowtime DESC], select=[currency, rate, rowtime], changelogMode=[I,UB,UA,D])
+ +- Exchange(distribution=[hash[currency]], changelogMode=[I])
+ +- WatermarkAssigner(rowtime=[rowtime], watermark=[rowtime],
changelogMode=[I])
+ +- LegacyTableSourceScan(table=[[default_catalog,
default_database, ratesHistory, source: [CollectionTableSource(currency, rate,
rowtime)]]], fields=[currency, rate, rowtime], changelogMode=[I])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testKeylessUpsertSinkKeepsUpsertWhenInputHasUpsertKey">
+ <Resource name="sql">
+ <![CDATA[INSERT INTO keyless_upsert_sink SELECT currency, rate FROM
DeduplicatedView]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalSink(table=[default_catalog.default_database.keyless_upsert_sink],
fields=[currency, rate])
++- LogicalProject(currency=[$0], rate=[$1])
+ +- LogicalProject(currency=[$0], rate=[$1], rowtime=[$2])
+ +- LogicalFilter(condition=[=($3, 1)])
+ +- LogicalProject(currency=[$0], rate=[$1], rowtime=[$2],
rowNum=[ROW_NUMBER() OVER (PARTITION BY $0 ORDER BY $2 DESC NULLS LAST)])
+ +- LogicalWatermarkAssigner(rowtime=[rowtime], watermark=[$2])
+ +- LogicalTableScan(table=[[default_catalog, default_database,
ratesHistory, source: [CollectionTableSource(currency, rate, rowtime)]]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+Sink(table=[default_catalog.default_database.keyless_upsert_sink],
fields=[currency, rate], changelogMode=[NONE])
++- Calc(select=[currency, rate], changelogMode=[I,UA,D])
+ +- Rank(strategy=[AppendFastStrategy], rankType=[ROW_NUMBER],
rankRange=[rankStart=1, rankEnd=1], partitionBy=[currency], orderBy=[ROWTIME
rowtime DESC], select=[currency, rate, rowtime], changelogMode=[I,UA,D])
+ +- Exchange(distribution=[hash[currency]], changelogMode=[I])
+ +- WatermarkAssigner(rowtime=[rowtime], watermark=[rowtime],
changelogMode=[I])
+ +- LegacyTableSourceScan(table=[[default_catalog,
default_database, ratesHistory, source: [CollectionTableSource(currency, rate,
rowtime)]]], fields=[currency, rate, rowtime], changelogMode=[I])
]]>
</Resource>
</TestCase>
diff --git
a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/DagOptimizationTest.xml
b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/DagOptimizationTest.xml
index 9f7107c8013..67d54af4a83 100644
---
a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/DagOptimizationTest.xml
+++
b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/DagOptimizationTest.xml
@@ -994,7 +994,7 @@ Sink(table=[default_catalog.default_database.appendSink],
fields=[a, b, c, d, e,
<TestCase name="testUnionAndAggWithDifferentGroupings">
<Resource name="ast">
<![CDATA[
-LogicalSink(table=[default_catalog.default_database.upsertSink], fields=[b, c,
a_sum])
+LogicalSink(table=[default_catalog.default_database.retractSink], fields=[b,
c, a_sum])
+- LogicalUnion(all=[true])
:- LogicalAggregate(group=[{0, 1}], a_sum=[SUM($2)])
: +- LogicalProject(b=[$1], c=[$2], a=[$0])
@@ -1007,13 +1007,13 @@
LogicalSink(table=[default_catalog.default_database.upsertSink], fields=[b, c, a
</Resource>
<Resource name="optimized rel plan">
<![CDATA[
-Sink(table=[default_catalog.default_database.upsertSink], fields=[b, c,
a_sum], changelogMode=[NONE])
-+- Union(all=[true], union=[b, c, a_sum], changelogMode=[I,UA])
- :- GroupAggregate(groupBy=[b, c], select=[b, c, SUM(a) AS a_sum],
changelogMode=[I,UA])
+Sink(table=[default_catalog.default_database.retractSink], fields=[b, c,
a_sum], changelogMode=[NONE])
++- Union(all=[true], union=[b, c, a_sum], changelogMode=[I,UB,UA])
+ :- GroupAggregate(groupBy=[b, c], select=[b, c, SUM(a) AS a_sum],
changelogMode=[I,UB,UA])
: +- Exchange(distribution=[hash[b, c]], changelogMode=[I])
: +- TableSourceScan(table=[[default_catalog, default_database,
MyTable]], fields=[a, b, c], changelogMode=[I])
- +- Calc(select=[CAST(1 AS BIGINT) AS b, c, a_sum], changelogMode=[I,UA])
- +- GroupAggregate(groupBy=[c], select=[c, SUM(a) AS a_sum],
changelogMode=[I,UA])
+ +- Calc(select=[CAST(1 AS BIGINT) AS b, c, a_sum], changelogMode=[I,UB,UA])
+ +- GroupAggregate(groupBy=[c], select=[c, SUM(a) AS a_sum],
changelogMode=[I,UB,UA])
+- Exchange(distribution=[hash[c]], changelogMode=[I])
+- Calc(select=[c, a], changelogMode=[I])
+- TableSourceScan(table=[[default_catalog, default_database,
MyTable]], fields=[a, b, c], changelogMode=[I])
diff --git
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/physical/stream/ChangelogModeInferenceTest.scala
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/physical/stream/ChangelogModeInferenceTest.scala
index ce2dc3d4468..787cbf39116 100644
---
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/physical/stream/ChangelogModeInferenceTest.scala
+++
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/physical/stream/ChangelogModeInferenceTest.scala
@@ -546,4 +546,41 @@ class ChangelogModeInferenceTest extends TableTestBase {
// upsert key {id, name} is NOT a subset of sink pk {id}, so
BEFORE_AND_AFTER is required
util.verifyRelPlan(statementSet, ExplainDetail.CHANGELOG_MODE)
}
+
+ @Test
+ def testKeylessUpsertSinkKeepsUpsertWhenInputHasUpsertKey(): Unit = {
+ // Input keeps its upsert key (currency), so a keyless upsert-only sink
stays upsert.
+ util.addTable("""
+ |CREATE TABLE keyless_upsert_sink (
+ | currency STRING,
+ | rate INT
+ |) WITH (
+ | 'connector' = 'values',
+ | 'sink-insert-only' = 'false',
+ | 'sink-changelog-mode-enforced' = 'I,UA,D'
+ |)
+ |""".stripMargin)
+ util.verifyRelPlanInsert(
+ "INSERT INTO keyless_upsert_sink SELECT currency, rate FROM
DeduplicatedView",
+ ExplainDetail.CHANGELOG_MODE)
+ }
+
+ @Test
+ def testKeylessUpsertSinkFallsBackToRetractWhenUpsertKeyProjectedAway():
Unit = {
+ // The input upsert key (currency) is projected away, so a keyless
upsert-only sink cannot
+ // upsert and the planner falls back to retract; the deduplicate's
ChangelogNormalize supplies
+ // UPDATE_BEFORE.
+ util.addTable("""
+ |CREATE TABLE keyless_upsert_sink_no_key (
+ | rate INT
+ |) WITH (
+ | 'connector' = 'values',
+ | 'sink-insert-only' = 'false',
+ | 'sink-changelog-mode-enforced' = 'I,UA,D'
+ |)
+ |""".stripMargin)
+ util.verifyRelPlanInsert(
+ "INSERT INTO keyless_upsert_sink_no_key SELECT rate FROM
DeduplicatedView",
+ ExplainDetail.CHANGELOG_MODE)
+ }
}
diff --git
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/DagOptimizationTest.scala
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/DagOptimizationTest.scala
index 5430545af23..0daee92964e 100644
---
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/DagOptimizationTest.scala
+++
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/DagOptimizationTest.scala
@@ -573,14 +573,16 @@ class DagOptimizationTest extends TableTestBase {
""".stripMargin
val table = util.tableEnv.sqlQuery(sqlQuery)
+ // The union of two differently-grouped aggregates has no common upsert
key, so it feeds a
+ // retract sink.
TestSinkUtil.addValuesSink(
util.tableEnv,
- "upsertSink",
+ "retractSink",
List("b", "c", "a_sum"),
List(LONG, STRING, INT),
- ChangelogMode.upsert()
+ ChangelogMode.all()
)
- util.verifyRelPlanInsert(table, "upsertSink", ExplainDetail.CHANGELOG_MODE)
+ util.verifyRelPlanInsert(table, "retractSink",
ExplainDetail.CHANGELOG_MODE)
}
@Test
