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)
