Moritz Manner created FLINK-41111:
-------------------------------------
Summary: 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
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)
