Jark Wu created FLINK-20369: ------------------------------- Summary: Improve the digest of TableSourceScan and Sink node Key: FLINK-20369 URL: https://issues.apache.org/jira/browse/FLINK-20369 Project: Flink Issue Type: Improvement Components: Table SQL / Planner Reporter: Jark Wu
Currently, 1. the digest of {{TableSourceScan}} and {{Sink}} doesn't contain the connector information which will be quite useful when debugging. 2. The table name is quite verbose when under default catalog and database, would be better to simplify it to only table name if under default catalog and database. 3. Maybe it's nicer to have changelog mode of source and sink, because it's a meta information of {{DynamicTableSource/Sink#getChangelogMode}}. {code} Sink(table=[default_catalog.default_database.sink_kafka_count_city], fields=[city_name, count_customer, sum_gender], changelogMode=[NONE]) +- Calc(select=[city_name, CAST(count_customer) AS count_customer, CAST(sum_gender) AS sum_gender], changelogMode=[I,UA,D]) +- Join(joinType=[InnerJoin], where=[=(city_id, id)], select=[city_id, count_customer, sum_gender, id, city_name], leftInputSpec=[JoinKeyContainsUniqueKey], rightInputSpec=[JoinKeyContainsUniqueKey], changelogMode=[I,UA,D]) :- Exchange(distribution=[hash[city_id]], changelogMode=[I,UA,D]) : +- GlobalGroupAggregate(groupBy=[city_id], select=[city_id, COUNT_RETRACT(count1$0) AS count_customer, SUM_RETRACT((sum$1, count$2)) AS sum_gender], changelogMode=[I,UA,D]) : +- Exchange(distribution=[hash[city_id]], changelogMode=[I]) : +- LocalGroupAggregate(groupBy=[city_id], select=[city_id, COUNT_RETRACT(*) AS count1$0, SUM_RETRACT(gender) AS (sum$1, count$2)], changelogMode=[I]) : +- Calc(select=[city_id, gender], changelogMode=[I,UB,UA,D]) : +- ChangelogNormalize(key=[customer_id], changelogMode=[I,UB,UA,D]) : +- Exchange(distribution=[hash[customer_id]], changelogMode=[UA,D]) : +- MiniBatchAssigner(interval=[3000ms], mode=[ProcTime], changelogMode=[UA,D]) : +- TableSourceScan(table=[[default_catalog, default_database, source_customer]], fields=[customer_id, city_id, age, gender, update_time], changelogMode=[UA,D]) +- Exchange(distribution=[hash[id]], changelogMode=[I,UA,D]) +- ChangelogNormalize(key=[id], changelogMode=[I,UA,D]) +- Exchange(distribution=[hash[id]], changelogMode=[UA,D]) +- MiniBatchAssigner(interval=[3000ms], mode=[ProcTime], changelogMode=[UA,D]) +- TableSourceScan(table=[[default_catalog, default_database, source_city]], fields=[id, city_name], changelogMode=[UA,D]) {code} -- This message was sent by Atlassian Jira (v8.3.4#803005)