[FLINK-30302] WatermarkAssignerChangelogNormalizeTransposeRuleTest failed

• Type: Bug
• Status: Closed
• Priority: Major
• Resolution: Cannot Reproduce
• Affects Version/s: 1.17.0
• Component/s: Table SQL / Planner
• Labels: stale-major test-stability
• Assignee: Unassigned
• Reporter: Matthias Pohl
• Created: 05/Dec/22 11:19
• Updated: 14/Aug/23 13:31
• Resolved: 14/Aug/23 13:31

Description

https://dev.azure.com/apache-flink/apache-flink/_build/results?buildId=43715&view=logs&j=0c940707-2659-5648-cbe6-a1ad63045f0a&t=075c2716-8010-5565-fe08-3c4bb45824a4&l=11792

Dec 05 04:26:28 [ERROR] org.apache.flink.table.planner.plan.rules.physical.stream.WatermarkAssignerChangelogNormalizeTransposeRuleTest.testPushdownCalcNotAffectChangelogNormalizeKey  Time elapsed: 0.601 s  <<< FAILURE!
Dec 05 04:26:28 org.junit.ComparisonFailure: 
Dec 05 04:26:28 optimized rel plan expected:<...c(select=[a, b, f], [changelogMode=[I])
Dec 05 04:26:28 +- TemporalJoin(joinType=[InnerJoin], where=[AND(=(a, a0), __TEMPORAL_JOIN_CONDITION(ingestion_time, ingestion_time0, __TEMPORAL_JOIN_CONDITION_PRIMARY_KEY(a0), __TEMPORAL_JOIN_LEFT_KEY(a), __TEMPORAL_JOIN_RIGHT_KEY(a0)))], select=[ingestion_time, a, b, ingestion_time0, a0, f], changelogMode=[I])
Dec 05 04:26:28    :- Exchange(distribution=[hash[a]], changelogMode=[I])
Dec 05 04:26:28    :  +- WatermarkAssigner(rowtime=[ingestion_time], watermark=[ingestion_time], changelogMode=[I])
Dec 05 04:26:28    :     +- Calc(select=[CAST(ingestion_time AS TIMESTAMP(3) *ROWTIME*) AS ingestion_time, a, b], changelogMode=[I])
Dec 05 04:26:28    :        +- TableSourceScan(table=[[default_catalog, default_database, t1]], fields=[a, b, ingestion_time], changelogMode=[I])
Dec 05 04:26:28    +- Exchange(distribution=[hash[a]], changelogMode=[I,UB,UA,D])
Dec 05 04:26:28       +- Calc(select=[ingestion_time, a, f], where=[f], changelogMode=[I,UB,UA,D])
Dec 05 04:26:28          +- ChangelogNormalize(key=[a], changelogMode=[I,UB,UA,D])
Dec 05 04:26:28             +- Exchange(distribution=[hash[a]], changelogMode=[I,UA,D])
Dec 05 04:26:28                +- WatermarkAssigner(rowtime=[ingestion_time], watermark=[ingestion_time], changelogMode=[I,UA,D])
Dec 05 04:26:28                   +- Calc(select=[CAST(ingestion_time AS TIMESTAMP(3) *ROWTIME*) AS ingestion_time, a, f], changelogMode=[I,UA,D])
Dec 05 04:26:28    ]                  +-...> but was:<...c(select=[a, b, f], [where=[f], changelogMode=[I])
Dec 05 04:26:28 +- TemporalJoin(joinType=[InnerJoin], where=[AND(=(a, a0), __TEMPORAL_JOIN_CONDITION(ingestion_time, ingestion_time0, __TEMPORAL_JOIN_CONDITION_PRIMARY_KEY(a0), __TEMPORAL_JOIN_LEFT_KEY(a), __TEMPORAL_JOIN_RIGHT_KEY(a0)))], select=[ingestion_time, a, b, ingestion_time0, a0, f], changelogMode=[I])
Dec 05 04:26:28    :- Exchange(distribution=[hash[a]], changelogMode=[I])
Dec 05 04:26:28    :  +- WatermarkAssigner(rowtime=[ingestion_time], watermark=[ingestion_time], changelogMode=[I])
Dec 05 04:26:28    :     +- Calc(select=[CAST(ingestion_time AS TIMESTAMP(3) *ROWTIME*) AS ingestion_time, a, b], changelogMode=[I])
Dec 05 04:26:28    :        +- TableSourceScan(table=[[default_catalog, default_database, t1]], fields=[a, b, ingestion_time], changelogMode=[I])
Dec 05 04:26:28    +- Exchange(distribution=[hash[a]], changelogMode=[I,UA,D])
Dec 05 04:26:28       +- ChangelogNormalize(key=[a], changelogMode=[I,UA,D])
Dec 05 04:26:28          +- Exchange(distribution=[hash[a]], changelogMode=[I,UA,D])
Dec 05 04:26:28             +- WatermarkAssigner(rowtime=[ingestion_time], watermark=[ingestion_time], changelogMode=[I,UA,D])
Dec 05 04:26:28                +- Calc(select=[CAST(ingestion_time AS TIMESTAMP(3) *ROWTIME*) AS ingestion_time, a, f], changelogMode=[I,UA,D])
Dec 05 04:26:28 ]                  +-...>