Matthias Pohl created FLINK-30302: ------------------------------------- Summary: WatermarkAssignerChangelogNormalizeTransposeRuleTest failed Key: FLINK-30302 URL: https://issues.apache.org/jira/browse/FLINK-30302 Project: Flink Issue Type: Bug Components: Table SQL / Planner Affects Versions: 1.17.0 Reporter: Matthias Pohl
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 {code} 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 ] +-...> {code} -- This message was sent by Atlassian Jira (v8.20.10#820010)