This is an automated email from the ASF dual-hosted git repository.
gustavodemorais pushed a commit to branch release-2.3
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-2.3 by this push:
new fa2ec19c464 [FLINK-40368][table] Deduplicate output should only be
insert-only when ordered by time attribute (#29372)
fa2ec19c464 is described below
commit fa2ec19c46454d00778146618985e37e245e951e
Author: Gustavo de Morais <[email protected]>
AuthorDate: Mon Oct 5 15:02:17 2026 +0200
[FLINK-40368][table] Deduplicate output should only be insert-only when
ordered by time attribute (#29372)
* [FLINK-40368][table] Deduplicate output should only be insert-only when
ordered by time attribute
* [FLINK-40368][table] Reuse isDeduplicationOnTimeAttribute in
FlinkRelMdModifiedMonotonicity
Generated-by: Claude Code (Claude Opus 5.5)
(cherry picked from commit 6eeffe83da9c9ac130c7591243b78c6820a8b3ca)
---
.../metadata/FlinkRelMdModifiedMonotonicity.scala | 15 +----
.../FlinkChangelogModeInferenceProgram.scala | 4 +-
.../flink/table/planner/plan/utils/RankUtil.scala | 16 +++---
.../planner/plan/stream/sql/DeduplicateTest.xml | 66 ++++++++++++++++++++++
.../planner/plan/stream/sql/DeduplicateTest.scala | 33 +++++++++++
.../runtime/stream/sql/DeduplicateITCase.scala | 57 +++++++++++++++++++
6 files changed, 169 insertions(+), 22 deletions(-)
diff --git
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdModifiedMonotonicity.scala
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdModifiedMonotonicity.scala
index 8b638bef45c..b8fd6836841 100644
---
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdModifiedMonotonicity.scala
+++
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdModifiedMonotonicity.scala
@@ -188,19 +188,8 @@ class FlinkRelMdModifiedMonotonicity private extends
MetadataHandler[ModifiedMon
def getRelModifiedMonotonicity(rel: Rank, mq: RelMetadataQuery):
RelModifiedMonotonicity = {
rel match {
- // Only ranks that can be converted to a Deduplicate operator are
handled here, mirroring the
- // pre-FLINK-34702 behavior when a dedicated StreamPhysicalDeduplicate
node carried these
- // invariants. This is the same condition as
RankUtil.canConvertToDeduplicate(FlinkLogicalRank):
- // a Top-1 ROW_NUMBER without rank number output, sorted on a single
time attribute. We inline
- // it rather than calling canConvertToDeduplicate(StreamPhysicalRank)
because the latter reads
- // the ModifyKindSetTrait, which is still undefined at the time this
metadata is computed.
- // Any other rank (multi-column or non-time-attribute ORDER BY, i.e. a
real Top-1 Rank that
- // retracts and re-emits the kept row) falls through to the generic
logic below.
- case physicalRank: StreamPhysicalRank
- if RankUtil.isDeduplication(rel) &&
- RankUtil.sortOnTimeAttributeOnly(
- physicalRank.orderKey,
- physicalRank.getInput.getRowType) =>
+ // Only Ranks that can be converted to a Deduplicate operator are
handled here
+ case physicalRank: StreamPhysicalRank if
RankUtil.isDeduplicationOnTimeAttribute(rel) =>
getPhysicalRankModifiedMonotonicity(physicalRank, mq)
case _ =>
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 75be86373b9..a3bc12d7ae6 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
@@ -256,7 +256,7 @@ class FlinkChangelogModeInferenceProgram extends
FlinkOptimizeProgram[StreamOpti
val providedTrait = ModifyKindSetTrait.INSERT_ONLY
createNewNode(rel, children, providedTrait, requiredTrait, requester)
- case rank: StreamPhysicalRank if RankUtil.isDeduplication(rank) =>
+ case rank: StreamPhysicalRank if
RankUtil.isDeduplicationOnTimeAttribute(rank) =>
val children = visitChildren(rel, ModifyKindSetTrait.ALL_CHANGES)
val tableConfig = unwrapTableConfig(rank)
@@ -278,7 +278,7 @@ class FlinkChangelogModeInferenceProgram extends
FlinkOptimizeProgram[StreamOpti
createNewNode(rel, children, providedTrait, requiredTrait, requester)
- case rank: StreamPhysicalRank if !RankUtil.isDeduplication(rank) =>
+ case rank: StreamPhysicalRank =>
// Rank supports consuming all changes
val children = visitChildren(rel, ModifyKindSetTrait.ALL_CHANGES)
createNewNode(rel, children, ModifyKindSetTrait.ALL_CHANGES,
requiredTrait, requester)
diff --git
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/RankUtil.scala
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/RankUtil.scala
index adac0e1d188..27359fbb2c6 100644
---
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/RankUtil.scala
+++
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/RankUtil.scala
@@ -383,17 +383,19 @@ object RankUtil {
!rank.outputRankNumber && rank.rankType == RankType.ROW_NUMBER &&
isTop1(rank.rankRange)
}
+ /**
+ * Whether the given rank is a deduplication ordered by a single time
attribute. Any other
+ * deduplication is executed as a regular Top-1 Rank that retracts the
previously kept row.
+ */
+ def isDeduplicationOnTimeAttribute(rank: Rank): Boolean =
+ isDeduplication(rank) && sortOnTimeAttributeOnly(rank.orderKey,
rank.getInput.getRowType)
+
/**
* Whether the given [[StreamPhysicalRank]] could be converted to
*
[[org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecDeduplicate]].
*/
- def canConvertToDeduplicate(rank: StreamPhysicalRank): Boolean = {
- lazy val inputInsertOnly = ChangelogPlanUtils.inputInsertOnly(rank)
- lazy val sortOnTimeAttributeOnly =
- RankUtil.sortOnTimeAttributeOnly(rank.orderKey, rank.getInput.getRowType)
-
- isDeduplication(rank) && inputInsertOnly && sortOnTimeAttributeOnly
- }
+ def canConvertToDeduplicate(rank: StreamPhysicalRank): Boolean =
+ isDeduplicationOnTimeAttribute(rank) &&
ChangelogPlanUtils.inputInsertOnly(rank)
/**
* Determines if the given order key indicates that the last row should be
kept for deduplication.
diff --git
a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/DeduplicateTest.xml
b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/DeduplicateTest.xml
index f5f7e4edfbf..a91a4b8ba57 100644
---
a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/DeduplicateTest.xml
+++
b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/DeduplicateTest.xml
@@ -16,6 +16,72 @@ See the License for the specific language governing
permissions and
limitations under the License.
-->
<Root>
+ <TestCase name="testFirstRowOnNonTimeAttribute">
+ <Resource name="explain">
+ <![CDATA[== Abstract Syntax Tree ==
+LogicalProject(EXPR$0=[$1], b=[$0], EXPR$2=[$2])
++- LogicalAggregate(group=[{0}], EXPR$0=[SUM($1)], EXPR$2=[SUM($2)])
+ +- LogicalProject(b=[$1], a=[$0], c=[$2])
+ +- LogicalProject(a=[$0], b=[$1], c=[$2])
+ +- LogicalFilter(condition=[=($5, 1)])
+ +- LogicalProject(a=[$0], b=[$1], c=[$2], proctime=[$3],
rowtime=[$4], rank_num=[ROW_NUMBER() OVER (PARTITION BY $0 ORDER BY $2 NULLS
FIRST)])
+ +- LogicalTableScan(table=[[default_catalog, default_database,
MyTable]])
+
+== Optimized Physical Plan ==
+Calc(select=[EXPR$0, b, EXPR$2], changelogMode=[I,UA,D])
++- GroupAggregate(groupBy=[b], select=[b, SUM_RETRACT(a) AS EXPR$0,
SUM_RETRACT(c) AS EXPR$2], changelogMode=[I,UA,D])
+ +- Exchange(distribution=[hash[b]], changelogMode=[I,UB,UA,D])
+ +- Calc(select=[b, a, c], changelogMode=[I,UB,UA,D])
+ +- Rank(strategy=[AppendFastStrategy], rankType=[ROW_NUMBER],
rankRange=[rankStart=1, rankEnd=1], partitionBy=[a], orderBy=[c ASC],
select=[a, b, c], changelogMode=[I,UB,UA,D])
+ +- Exchange(distribution=[hash[a]], changelogMode=[I])
+ +- Calc(select=[a, b, c], changelogMode=[I])
+ +- DataStreamScan(table=[[default_catalog, default_database,
MyTable]], fields=[a, b, c, proctime, rowtime], changelogMode=[I])
+
+== Optimized Execution Plan ==
+Calc(select=[EXPR$0, b, EXPR$2])
++- GroupAggregate(groupBy=[b], select=[b, SUM_RETRACT(a) AS EXPR$0,
SUM_RETRACT(c) AS EXPR$2])
+ +- Exchange(distribution=[hash[b]])
+ +- Calc(select=[b, a, c])
+ +- Rank(strategy=[AppendFastStrategy], rankType=[ROW_NUMBER],
rankRange=[rankStart=1, rankEnd=1], partitionBy=[a], orderBy=[c ASC],
select=[a, b, c])
+ +- Exchange(distribution=[hash[a]])
+ +- Calc(select=[a, b, c])
+ +- DataStreamScan(table=[[default_catalog, default_database,
MyTable]], fields=[a, b, c, proctime, rowtime])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testFirstRowOnTimeAndNonTimeAttribute">
+ <Resource name="explain">
+ <![CDATA[== Abstract Syntax Tree ==
+LogicalProject(EXPR$0=[$1], b=[$0], EXPR$2=[$2])
++- LogicalAggregate(group=[{0}], EXPR$0=[SUM($1)], EXPR$2=[SUM($2)])
+ +- LogicalProject(b=[$1], a=[$0], c=[$2])
+ +- LogicalProject(a=[$0], b=[$1], c=[$2])
+ +- LogicalFilter(condition=[=($5, 1)])
+ +- LogicalProject(a=[$0], b=[$1], c=[$2], proctime=[$3],
rowtime=[$4], rank_num=[ROW_NUMBER() OVER (PARTITION BY $0 ORDER BY $4 NULLS
FIRST, $2 NULLS FIRST)])
+ +- LogicalTableScan(table=[[default_catalog, default_database,
MyTable]])
+
+== Optimized Physical Plan ==
+Calc(select=[EXPR$0, b, EXPR$2], changelogMode=[I,UA,D])
++- GroupAggregate(groupBy=[b], select=[b, SUM_RETRACT(a) AS EXPR$0,
SUM_RETRACT(c) AS EXPR$2], changelogMode=[I,UA,D])
+ +- Exchange(distribution=[hash[b]], changelogMode=[I,UB,UA,D])
+ +- Calc(select=[b, a, c], changelogMode=[I,UB,UA,D])
+ +- Rank(strategy=[AppendFastStrategy], rankType=[ROW_NUMBER],
rankRange=[rankStart=1, rankEnd=1], partitionBy=[a], orderBy=[ROWTIME rowtime
ASC, c ASC], select=[a, b, c, rowtime], changelogMode=[I,UB,UA,D])
+ +- Exchange(distribution=[hash[a]], changelogMode=[I])
+ +- Calc(select=[a, b, c, rowtime], changelogMode=[I])
+ +- DataStreamScan(table=[[default_catalog, default_database,
MyTable]], fields=[a, b, c, proctime, rowtime], changelogMode=[I])
+
+== Optimized Execution Plan ==
+Calc(select=[EXPR$0, b, EXPR$2])
++- GroupAggregate(groupBy=[b], select=[b, SUM_RETRACT(a) AS EXPR$0,
SUM_RETRACT(c) AS EXPR$2])
+ +- Exchange(distribution=[hash[b]])
+ +- Calc(select=[b, a, c])
+ +- Rank(strategy=[AppendFastStrategy], rankType=[ROW_NUMBER],
rankRange=[rankStart=1, rankEnd=1], partitionBy=[a], orderBy=[ROWTIME rowtime
ASC, c ASC], select=[a, b, c, rowtime])
+ +- Exchange(distribution=[hash[a]])
+ +- Calc(select=[a, b, c, rowtime])
+ +- DataStreamScan(table=[[default_catalog, default_database,
MyTable]], fields=[a, b, c, proctime, rowtime])
+]]>
+ </Resource>
+ </TestCase>
<TestCase name="testInvalidChangelogInput">
<Resource name="explain">
<![CDATA[== Abstract Syntax Tree ==
diff --git
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/DeduplicateTest.scala
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/DeduplicateTest.scala
index a1b72a9ad62..b4681339550 100644
---
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/DeduplicateTest.scala
+++
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/DeduplicateTest.scala
@@ -307,4 +307,37 @@ class DeduplicateTest extends TableTestBase {
util.verifyExplain(sqlQuery, ExplainDetail.CHANGELOG_MODE)
}
+ @Test
+ def testFirstRowOnNonTimeAttribute(): Unit = {
+ val sql =
+ """
+ |SELECT sum(a), b, sum(c) FROM (
+ | SELECT a, b, c
+ | FROM (
+ | SELECT *,
+ | ROW_NUMBER() OVER (PARTITION BY a ORDER BY c ASC) as rank_num
+ | FROM MyTable)
+ | WHERE rank_num = 1)
+ |GROUP BY b
+ """.stripMargin
+
+ util.verifyExplain(sql, ExplainDetail.CHANGELOG_MODE)
+ }
+
+ @Test
+ def testFirstRowOnTimeAndNonTimeAttribute(): Unit = {
+ val sql =
+ """
+ |SELECT sum(a), b, sum(c) FROM (
+ | SELECT a, b, c
+ | FROM (
+ | SELECT *,
+ | ROW_NUMBER() OVER (PARTITION BY a ORDER BY rowtime ASC, c
ASC) as rank_num
+ | FROM MyTable)
+ | WHERE rank_num = 1)
+ |GROUP BY b
+ """.stripMargin
+
+ util.verifyExplain(sql, ExplainDetail.CHANGELOG_MODE)
+ }
}
diff --git
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/DeduplicateITCase.scala
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/DeduplicateITCase.scala
index 5065c236062..cfe7275fe00 100644
---
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/DeduplicateITCase.scala
+++
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/DeduplicateITCase.scala
@@ -305,6 +305,63 @@ class DeduplicateITCase(miniBatch: MiniBatchMode, mode:
StateBackendMode, enable
assertThat(rawResult.sorted).isEqualTo(expected.sorted)
}
+ @TestTemplate
+ def testFirstRowOnNonTimeAttributeFollowedByUnboundedAgg(): Unit = {
+ val data = List((1, 100L, "a"), (1, 10L, "b"), (2, 1L, "c"))
+ val t = StreamingEnvUtil.fromCollection(env, data).toTable(tEnv, 'a, 'b,
'c)
+ tEnv.createTemporaryView("T", t)
+
+ val sql =
+ """
+ |SELECT SUM(b) FROM (
+ | SELECT a, b
+ | FROM (
+ | SELECT *,
+ | ROW_NUMBER() OVER (PARTITION BY a ORDER BY b) as rowNum
+ | FROM T
+ | )
+ | WHERE rowNum = 1
+ |)
+ """.stripMargin
+
+ val sink = new TestingRetractSink
+ tEnv.sqlQuery(sql).toRetractStream[Row].addSink(sink).setParallelism(1)
+ env.execute()
+
+ // SUM over the first rows: 10 for key 1 (replacing 100) + 1 for key 2
+ assertThat(sink.getRetractResults).isEqualTo(List("11"))
+ }
+
+ @TestTemplate
+ def testFirstRowOnTimeAndNonTimeAttributeFollowedByUnboundedAgg(): Unit = {
+ val data = List((1, 100L, "a"), (1, 10L, "b"), (2, 1L, "c"))
+ val t = StreamingEnvUtil
+ .fromCollection(env, data)
+ .assignTimestampsAndWatermarks(new RowtimeExtractor)
+ .toTable(tEnv, 'a, 'b, 'c, 'rowtime.rowtime())
+ tEnv.createTemporaryView("T", t)
+
+ val sql =
+ """
+ |SELECT SUM(b) FROM (
+ | SELECT a, b
+ | FROM (
+ | SELECT *,
+ | ROW_NUMBER() OVER (PARTITION BY a ORDER BY rowtime, c) as rowNum
+ | FROM T
+ | )
+ | WHERE rowNum = 1
+ |)
+ """.stripMargin
+
+ val sink = new TestingRetractSink
+ tEnv.sqlQuery(sql).toRetractStream[Row].addSink(sink).setParallelism(1)
+ env.execute()
+
+ // SUM over the first rows: 10 for key 1 (replacing 100) + 1 for key 2
+ assertThat(sink.getRetractResults).isEqualTo(List("11"))
+ }
+
@TestTemplate
def testLastRowOnRowtime(): Unit = {
val t = StreamingEnvUtil
