[
https://issues.apache.org/jira/browse/FLINK-41111?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Gustavo de Morais reassigned FLINK-41111:
-----------------------------------------
Assignee: Moritz Manner
> Changelog mode errors of FlinkChangelogModeInferenceProgram do not identify
> the part of the query that produces the changes
> ---------------------------------------------------------------------------------------------------------------------------
>
> Key: FLINK-41111
> URL: https://issues.apache.org/jira/browse/FLINK-41111
> Project: Flink
> Issue Type: Improvement
> Components: Table SQL / Planner
> Reporter: Moritz Manner
> Assignee: Moritz Manner
> Priority: Minor
>
> When a sink or an operator cannot consume the changes of its input,
> FlinkChangelogModeInferenceProgram throws a TableException whose message only
> contains the consumer and the description of the producing node:
> {noformat}
> StreamPhysicalOverAggregate doesn't support consuming update and delete
> changes which is produced by node ChangelogNormalize(key=[k])
> {noformat}
> The exception carries no structured information, so tools that want to report
> this error differently have to parse the message. The message itself is also
> hard to act on:
> * When the producing node is a ChangelogNormalize, only its key is shown, not
> the source table. With several upsert sources, it is not clear which one is
> meant.
> * For a process table function, the consumer is reported as
> StreamPhysicalProcessTableFunction, without the function name or the table
> argument.
> * The consumer is a planner class name (StreamPhysicalOverAggregate) instead
> of the SQL construct.
> * The sink is written as 'cat.db.snk', which is ambiguous when a part of the
> identifier contains a dot.
> * The plan is not included, and EXPLAIN fails with the same error.
> h3. Repro
> Sink that only accepts inserts:
> {code:sql}
> CREATE TABLE src (k INT, v INT) WITH ('connector' = 'datagen');
> CREATE TABLE snk (k INT, c BIGINT) WITH ('connector' = 'filesystem', 'path' =
> 'file:///tmp/snk', 'format' = 'csv');
> INSERT INTO snk SELECT k, COUNT(*) FROM src GROUP BY k;
> {code}
> {noformat}
> Table sink 'default_catalog.default_database.snk' doesn't support consuming
> update changes which is produced by node GroupAggregate(groupBy=[k],
> select=[k, COUNT(*) AS EXPR$1])
> {noformat}
> Operator after an upsert source, where the source table is not named:
> {code:sql}
> CREATE TABLE orders (k INT, v INT, ts TIMESTAMP(3), WATERMARK FOR ts AS ts,
> PRIMARY KEY (k) NOT ENFORCED)
> WITH ('connector' = 'upsert-kafka', 'topic' = 'orders',
> 'properties.bootstrap.servers' = 'localhost:9092',
> 'key.format' = 'json', 'value.format' = 'json');
> SELECT k, COUNT(*) OVER (PARTITION BY k ORDER BY ts ROWS BETWEEN UNBOUNDED
> PRECEDING AND CURRENT ROW) FROM orders;
> {code}
> {noformat}
> StreamPhysicalOverAggregate doesn't support consuming update and delete
> changes which is produced by node ChangelogNormalize(key=[k])
> {noformat}
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
