[
https://issues.apache.org/jira/browse/FLINK-29781?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
godfrey he closed FLINK-29781.
------------------------------
Resolution: Fixed
> ChangelogNormalize uses wrong keys after transformation by
> WatermarkAssignerChangelogNormalizeTransposeRule
> ------------------------------------------------------------------------------------------------------------
>
> Key: FLINK-29781
> URL: https://issues.apache.org/jira/browse/FLINK-29781
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
> Affects Versions: 1.16.0, 1.15.3
> Reporter: lincoln lee
> Assignee: lincoln lee
> Priority: Major
> Labels: pull-request-available
> Fix For: 1.17.0, 1.16.1
>
>
> currently WatermarkAssignerChangelogNormalizeTransposeRule didn't remap the
> uniquekey indexes for its new input after plan rewrite, this may produce
> wrong result.
> A simple case:
> {code}
> @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)
> val sql =
> """
> |SELECT t1.a, t1.b, t2.f
> |FROM t1 INNER JOIN t2 FOR SYSTEM_TIME AS OF t1.ingestion_time
> | ON t1.a = t2.a WHERE t2.f = true
> |""".stripMargin
> util.verifyRelPlan(sql, ExplainDetail.CHANGELOG_MODE)
> }
> {code}
> the generated plan is incorrect for now:
> {color:red}ChangelogNormalize(key=[ingestion_time]){color} uses wrong key
> 'ingestion_time' (should be 'a')
> optimize result:
> {code}
> Calc(select=[a, b, f])
> +- 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])
> :- Exchange(distribution=[hash[a]])
> : +- WatermarkAssigner(rowtime=[ingestion_time],
> watermark=[ingestion_time])
> : +- Calc(select=[CAST(ingestion_time AS TIMESTAMP(3) *ROWTIME*) AS
> ingestion_time, a, b])
> : +- TableSourceScan(table=[[default_catalog, default_database,
> t1]], fields=[a, b, ingestion_time])
> +- Exchange(distribution=[hash[a]])
> +- Calc(select=[ingestion_time, a, f], where=[f])
> +- ChangelogNormalize(key=[ingestion_time])
> +- Exchange(distribution=[hash[a]])
> +- WatermarkAssigner(rowtime=[ingestion_time],
> watermark=[ingestion_time])
> +- Calc(select=[CAST(ingestion_time AS TIMESTAMP(3)
> *ROWTIME*) AS ingestion_time, a, f])
> +- TableSourceScan(table=[[default_catalog,
> default_database, t2, project=[a, f], metadata=[ts]]], fields=[a, f,
> ingestion_time])
> {code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
