Timo Walther created FLINK-23663:
------------------------------------
Summary: Reduce state size in ChangelogNormalize through filter
push down
Key: FLINK-23663
URL: https://issues.apache.org/jira/browse/FLINK-23663
Project: Flink
Issue Type: Improvement
Components: Table SQL / Planner
Reporter: Timo Walther
{{ChangelogNormalize}} is an expensive stateful operation as it stores data for
each key.
Filters are generally not pushed through a ChangelogNormalize node which means
that users have no possibility to at least limit the key space. Pushing filters
like {{a < 10}} into a source like {{upsert-kafka}} that is emitting {{+I[key1,
a=9]}} and {{-D[key1, a=10]}}, is problematic as the deletion will be filtered
and leads to wrong results. But limiting the filter push down to key space
should be safe.
Furthermore, it seems the current implementation is also wrong as it pushes
filters through {{ChangelogNormalize}} but only if the source implements filter
push down.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
