Timo Walther created FLINK-20346:
------------------------------------
Summary: Explain ChangelogMode for sinks
Key: FLINK-20346
URL:
https://issues.apache.org/jira/browse/FLINK-20346 Project: Flink
Issue Type: Improvement
Components: Table SQL / Planner
Reporter: Timo Walther
When explaining an `INSERT INTO` statement, the output does not show a changelog mode. However, this might be useful for users to know which kind of updates end up in a connector such as Upsert Kafka.
For example:
{code}
String initialValues = "INSERT INTO upsert_kafka\n"
+ "VALUES\n"
+ " (1, 'name 1', TIMESTAMP '2020-03-08 13:12:11.123', 100, 41, 'payload 1'),\n"
+ " (2, 'name 2', TIMESTAMP '2020-03-09 13:12:11.123', 101, 42, 'payload 2'),\n"
+ " (3, 'name 3', TIMESTAMP '2020-03-10 13:12:11.123', 102, 43, 'payload 3'),\n"
+ " (2, 'name 2', TIMESTAMP '2020-03-11 13:12:11.123', 101, 42, 'payload')";
System.out.println(tEnv.explainSql(initialValues, ExplainDetail.CHANGELOG_MODE));
{code}
Leads to `changelogMode=[NONE]`:
{code}
== Optimized Logical Plan ==
Sink(table=[default_catalog.default_database.upsert_kafka], fields=[k_user_id, name, k_event_id, user_id, payload, timestamp], changelogMode=[NONE])
+- Calc(select=[CAST(EXPR$0) AS k_user_id, CAST(EXPR$1) AS name, CAST(EXPR$3) AS k_event_id, CAST(EXPR$4) AS user_id, CAST(EXPR$5) AS payload, CAST(EXPR$2) AS timestamp], changelogMode=[I])
+- Values(type=[RecordType(INTEGER EXPR$0, CHAR(6) EXPR$1, TIMESTAMP(3) EXPR$2, INTEGER EXPR$3, INTEGER EXPR$4, VARCHAR(9) EXPR$5)], tuples=[[{ 1, _UTF-16LE'name 1', 2020-03-08 13:12:11.123, 100, 41, _UTF-16LE'payload 1' }, { 2, _UTF-16LE'name 2', 2020-03-09 13:12:11.123, 101, 42, _UTF-16LE'payload 2' }, { 3, _UTF-16LE'name 3', 2020-03-10 13:12:11.123, 102, 43, _UTF-16LE'payload 3' }, { 2, _UTF-16LE'name 2', 2020-03-11 13:12:11.123, 101, 42, _UTF-16LE'payload' }]], changelogMode=[I])
{code}
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
Locked
