[hotfix][table] Fix plan tests for the new PushCalcPastChangelogNorma…

…lizeRule
liuyongvs · Mar 15, 2023 · 617fcaa · 617fcaa
1 parent 1b262d5
commit 617fcaa
Show file tree

Hide file tree

Showing 2 changed files with 12 additions and 10 deletions.
diff --git a/.../org/apache/flink/table/planner/plan/rules/physical/stream/ChangelogModeInferenceTest.xml b/.../org/apache/flink/table/planner/plan/rules/physical/stream/ChangelogModeInferenceTest.xml
@@ -468,9 +468,9 @@ Sink(table=[default_catalog.default_database.all_change_sink_table], fields=[id,
       :     +- Exchange(distribution=[hash[id]], changelogMode=[I,UA,D])
       :        +- TableSourceScan(table=[[default_catalog, default_database, upsert_managed_table]], fields=[id, col1, col2], changelogMode=[I,UA,D])
       +- Exchange(distribution=[hash[id]], changelogMode=[I,UB,UA,D])
-         +- Calc(select=[id], changelogMode=[I,UB,UA,D])
-            +- ChangelogNormalize(key=[id], changelogMode=[I,UB,UA,D])
-               +- Exchange(distribution=[hash[id]], changelogMode=[I,UA,D])
+         +- ChangelogNormalize(key=[id], changelogMode=[I,UB,UA,D])
+            +- Exchange(distribution=[hash[id]], changelogMode=[I,UA,D])
+               +- Calc(select=[id], changelogMode=[I,UA,D])
                   +- TableSourceScan(table=[[default_catalog, default_database, upsert_managed_table]], fields=[id, col1, col2], changelogMode=[I,UA,D])
 ]]>
 		</Resource>

diff --git a/...anner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/TableScanTest.xml b/...anner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/TableScanTest.xml
@@ -98,10 +98,11 @@ LogicalProject(a=[$1], b=[$2], c=[$3])
 Calc(select=[a, b, c], where=[>(a, 1)], changelogMode=[I,UB,UA,D])
 +- ChangelogNormalize(key=[id], changelogMode=[I,UB,UA,D])
    +- Exchange(distribution=[hash[id]], changelogMode=[I,UA,D])
-      +- WatermarkAssigner(rowtime=[ts], watermark=[-(ts, 1000:INTERVAL SECOND)], changelogMode=[I,UA,D])
-         +- Calc(select=[id, a, +(a, 1) AS b, c, TO_TIMESTAMP(c) AS ts], changelogMode=[I,UA,D])
-            +- DropUpdateBefore(changelogMode=[I,UA,D])
-               +- TableSourceScan(table=[[default_catalog, default_database, src]], fields=[id, a, c], changelogMode=[I,UB,UA,D])
+      +- Calc(select=[id, a, b, c], changelogMode=[I,UA,D])
+         +- WatermarkAssigner(rowtime=[ts], watermark=[-(ts, 1000:INTERVAL SECOND)], changelogMode=[I,UA,D])
+            +- Calc(select=[id, a, +(a, 1) AS b, c, TO_TIMESTAMP(c) AS ts], changelogMode=[I,UA,D])
+               +- DropUpdateBefore(changelogMode=[I,UA,D])
+                  +- TableSourceScan(table=[[default_catalog, default_database, src]], fields=[id, a, c], changelogMode=[I,UB,UA,D])
 ]]>
     </Resource>
   </TestCase>
@@ -630,9 +631,10 @@ LogicalProject(a=[$1], b=[$2], c=[$3])
 Calc(select=[a, b, c], where=[>(a, 1)], changelogMode=[I,UB,UA,D])
 +- ChangelogNormalize(key=[id], changelogMode=[I,UB,UA,D])
    +- Exchange(distribution=[hash[id]], changelogMode=[UA,D])
-      +- WatermarkAssigner(rowtime=[ts], watermark=[-(ts, 1000:INTERVAL SECOND)], changelogMode=[UA,D])
-         +- Calc(select=[id, a, +(a, 1) AS b, c, TO_TIMESTAMP(c) AS ts], changelogMode=[UA,D])
-            +- TableSourceScan(table=[[default_catalog, default_database, src]], fields=[id, a, c], changelogMode=[UA,D])
+      +- Calc(select=[id, a, b, c], changelogMode=[UA,D])
+         +- WatermarkAssigner(rowtime=[ts], watermark=[-(ts, 1000:INTERVAL SECOND)], changelogMode=[UA,D])
+            +- Calc(select=[id, a, +(a, 1) AS b, c, TO_TIMESTAMP(c) AS ts], changelogMode=[UA,D])
+               +- TableSourceScan(table=[[default_catalog, default_database, src]], fields=[id, a, c], changelogMode=[UA,D])
 ]]>
     </Resource>
   </TestCase>