[
https://issues.apache.org/jira/browse/FLINK-39369?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18070481#comment-18070481
]
Xuyang Zhong commented on FLINK-39369:
--------------------------------------
This bug has been fixed in https://issues.apache.org/jira/browse/FLINK-37475
and I have verified in Fluss repo.
{code:java}
// add this test in FlinkTableSourceITCase
@Test
void test() {
tEnv.executeSql(
"create table read_snapshot_test (a int not null primary key not
enforced, b varchar) with ('table.changelog.image' = 'WAL')");
System.err.println(
tEnv.explainSql("select * from read_snapshot_test",
ExplainDetail.CHANGELOG_MODE));
}
// in 1.20
== Optimized Execution Plan ==
ChangelogNormalize(key=[a])
+- Exchange(distribution=[hash[a]])
+- TableSourceScan(table=[[testcatalog, defaultdb, read_snapshot_test]],
fields=[a, b])
// in 2.2
== Optimized Execution Plan ==
TableSourceScan(table=[[testcatalog, defaultdb, read_snapshot_test]],
fields=[a, b]){code}
[~ipolyzos] You can take a look again if you have time.
> ChangelogNormalize Operator In WAL Image Mode with Fluss
> --------------------------------------------------------
>
> Key: FLINK-39369
> URL: https://issues.apache.org/jira/browse/FLINK-39369
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / API
> Affects Versions: 1.20.0
> Reporter: Giannis Polyzos
> Priority: Minor
> Attachments: Screenshot 2026-03-31 at 6.09.11 PM.png
>
>
> Flink 1.20 and Fluss 0.9
> When using `'table.changelog.image' = 'WAL'` Fluss skips the `UPDATE_BEFORE`
> on the WAL.
> However during streaming reads, because `UPDATE_BEFORE` is not generated,
> this makes Flink to introduce the `ChangelogNormalize` operator which is
> expensive and unwanted.
> I tried to disable the operator with
> `table.exec.sink.upsert-materialize=NONE` however seems like its not taking
> effect
> (see attached image)
>
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
