Skip to content

[FLINK-40301][table] Merge Calc nodes before ChangelogNormalize projection pushdown #28882

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Au-Miner wants to merge 1 commit into
base: master
Choose a base branch
from
+69 −25
Open

[FLINK-40301][table] Merge Calc nodes before ChangelogNormalize projection pushdown #28882

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions ...lanner/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkStreamRuleSets.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -540,6 +540,7 @@ object FlinkStreamRuleSets {
val CHANGELOG_NORMALIZE_TRANSPOSE_RULES: RuleSet = RuleSets.ofList(
WatermarkAssignerChangelogNormalizeTransposeRule.WITH_CALC,
WatermarkAssignerChangelogNormalizeTransposeRule.WITHOUT_CALC,
FlinkCalcMergeRule.STREAM_PHYSICAL_INSTANCE,
// reduce state size in ChangelogNormalize
PushCalcPastChangelogNormalizeRule.INSTANCE
)
Expand Down
32 changes: 32 additions & 0 deletions ...anner/plan/rules/physical/stream/WatermarkAssignerChangelogNormalizeTransposeRuleTest.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -122,6 +122,38 @@ Calc(select=[a, b, f], where=[f], changelogMode=[I])
+- WatermarkAssigner(rowtime=[ingestion_time], watermark=[ingestion_time], changelogMode=[I,UA,D])
+- Calc(select=[CAST(ingestion_time AS TIMESTAMP(3) *ROWTIME*) AS ingestion_time, a, f], changelogMode=[I,UA,D])
+- TableSourceScan(table=[[default_catalog, default_database, t2, project=[a, f], metadata=[ts]]], fields=[a, f, ingestion_time], changelogMode=[I,UA,D])
]]>
</Resource>
</TestCase>
<TestCase name="testPushdownCalcNotAffectChangelogNormalizeKey2">
<Resource name="sql">
<![CDATA[
SELECT f, count(*)
FROM (select ingestion_time, f, a from t2) t2
group by f
]]>
</Resource>
<Resource name="ast">
<![CDATA[
LogicalAggregate(group=[{0}], EXPR$1=[COUNT()])
+- LogicalProject(f=[$1])
+- LogicalProject(ingestion_time=[$1], f=[$3], a=[$2])
+- LogicalWatermarkAssigner(rowtime=[ingestion_time], watermark=[$1])
+- LogicalProject(k=[$0], ingestion_time=[CAST($3):TIMESTAMP(3) *ROWTIME*], a=[$1], f=[$2])
+- LogicalTableScan(table=[[default_catalog, default_database, t2, metadata=[ts]]])
]]>
</Resource>
<Resource name="optimized rel plan">
<![CDATA[
GroupAggregate(groupBy=[f], select=[f, COUNT_RETRACT(*) AS EXPR$1], changelogMode=[I,UA,D])
+- Exchange(distribution=[hash[f]], changelogMode=[I,UB,UA,D])
+- Calc(select=[f], changelogMode=[I,UB,UA,D])
+- ChangelogNormalize(key=[a], changelogMode=[I,UB,UA,D])
+- Exchange(distribution=[hash[a]], changelogMode=[I,UA,D])
+- Calc(select=[f, a], changelogMode=[I,UA,D])
+- WatermarkAssigner(rowtime=[ingestion_time], watermark=[ingestion_time], changelogMode=[I,UA,D])
+- Calc(select=[f, CAST(ingestion_time AS TIMESTAMP(3) *ROWTIME*) AS ingestion_time, a], changelogMode=[I,UA,D])
+- TableSourceScan(table=[[default_catalog, default_database, t2, project=[f, a], metadata=[ts]]], fields=[f, a, ingestion_time], changelogMode=[I,UA,D])
]]>
</Resource>
</TestCase>
Expand Down
61 changes: 36 additions & 25 deletions ...ner/plan/rules/physical/stream/WatermarkAssignerChangelogNormalizeTransposeRuleTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -75,6 +75,31 @@ class WatermarkAssignerChangelogNormalizeTransposeRuleTest extends TableTestBase
| 'enable-watermark-push-down' = 'true'
|)
|""".stripMargin)
util.addTable("""
|CREATE TABLE t1 (
| ingestion_time TIMESTAMP(3) METADATA FROM 'ts',
| a VARCHAR NOT NULL,
| b VARCHAR NOT NULL,
| WATERMARK FOR ingestion_time AS ingestion_time
|) WITH (
| 'connector' = 'values',
| 'readable-metadata' = 'ts:TIMESTAMP(3)'
|)
""".stripMargin)
util.addTable("""
|CREATE TABLE t2 (
| k VARBINARY,
| ingestion_time TIMESTAMP(3) METADATA FROM 'ts',
| a VARCHAR NOT NULL,
| f BOOLEAN NOT NULL,
| WATERMARK FOR `ingestion_time` AS `ingestion_time`,
| PRIMARY KEY (`a`) NOT ENFORCED
|) WITH (
| 'connector' = 'values',
| 'readable-metadata' = 'ts:TIMESTAMP(3)',
| 'changelog-mode' = 'I,UA,D'
|)
""".stripMargin)
}

// ----------------------------------------------------------------------------------------
Expand Down Expand Up @@ -169,31 +194,6 @@ class WatermarkAssignerChangelogNormalizeTransposeRuleTest extends TableTestBase

@Test
def testPushdownCalcNotAffectChangelogNormalizeKey(): Unit = {
util.addTable("""
|CREATE TABLE t1 (
| ingestion_time TIMESTAMP(3) METADATA FROM 'ts',
| a VARCHAR NOT NULL,
| b VARCHAR NOT NULL,
| WATERMARK FOR ingestion_time AS ingestion_time
|) WITH (
| 'connector' = 'values',
| 'readable-metadata' = 'ts:TIMESTAMP(3)'
|)
""".stripMargin)
util.addTable("""
|CREATE TABLE t2 (
| k VARBINARY,
| ingestion_time TIMESTAMP(3) METADATA FROM 'ts',
| a VARCHAR NOT NULL,
| f BOOLEAN NOT NULL,
| WATERMARK FOR `ingestion_time` AS `ingestion_time`,
| PRIMARY KEY (`a`) NOT ENFORCED
|) WITH (
| 'connector' = 'values',
| 'readable-metadata' = 'ts:TIMESTAMP(3)',
| 'changelog-mode' = 'I,UA,D'
|)
""".stripMargin)
// After FLINK-28988 applied, the filter will not be pushed down into left input of join and get
// a more optimal plan (upsert mode without ChangelogNormalize).
val sql =
Expand All @@ -204,4 +204,15 @@ class WatermarkAssignerChangelogNormalizeTransposeRuleTest extends TableTestBase
|""".stripMargin
util.verifyRelPlan(sql, ExplainDetail.CHANGELOG_MODE)
}

@Test
def testPushdownCalcNotAffectChangelogNormalizeKey2(): Unit = {
val sql =
"""
|SELECT f, count(*)
|FROM (select ingestion_time, f, a from t2) t2
|group by f
|""".stripMargin
util.verifyRelPlan(sql, ExplainDetail.CHANGELOG_MODE)
}
}