[FLINK-12557] Improve the DDL property keys to be more concise and readble

• Type: Improvement
• Status: Reopened
• Priority: Not a Priority
• Resolution: Unresolved
• Component/s: Table SQL / API
• Labels: auto-deprioritized-major auto-deprioritized-minor
• Assignee: Unassigned
• Reporter: Danny Chen
• Created: 20/May/19 10:20
• Updated: 25/Nov/21 10:43

Description

The with option in table DDL defines the properties needed for specific connector to create TableSource/Sink. The properties structure for SqlClient config YAML is defined in Improvements to the Unified SQL Connector API, in this design, the properties can be categorized into 4 parts:

 

1. Top level properties: name, type(source, sink), update-mode ...
2. Connector specific properties: connector.type, connector.path ...
3. Format properties: format.type, format.fields.1.name ...
4. Table schema properties: (can be omitted for DDL)

 

This properties structure is reasonable for YAML, but they are not that concise enough for developers.  So there also defines a tool class named DescriptorProperties to reconstruct the data structure(like TableSchema) from the flat k-v strings.

 

So in order to reduce complexity and keep the KV consistency for DDL with properties and TableFactory properties, i proposed to simplify the DDL with properties keys as following (corresponding to above 4 categories):

 

1. Top level properties: keep same as that in the YAML e.g. connector, update-mode
2. Connector specific properties: start with prefix named the connector type e.g. for kafka connector, the properties are defined as kafka.k1 = v1, kafka.k2 = v2
3. Format properties: format.type simplified to format and the others with prefix of the format name e.g. format = 'json', json.line-delimiter = "\n"
4. Table schema properties: omitted.

Here is a demo of creat table DDL:

CREATE TABLE Kafka10SourceTable (
  intField INTEGER,
  stringField VARCHAR(128) COMMENT 'User IP address',
  longField BIGINT,
  rowTimeField TIMESTAMP,
  WATERMARK wm01 FOR  'longField' AS BOUNDED WITH DELAY '60' SECOND
)
COMMENT 'Kafka Source Table of topic user_ip_address'
WITH (
  connector='kafka',
  kafka.property-version='1',
  kafka.version='0.10',
  kafka.topic='test-kafka-topic',
  kafka.startup-mode = 'latest-offset'
  kafka.specific-offset = 'offset'
  format='json'
  json.property-version = '1'
  json.version='1'
  json.derive-schema='true'
)

Issue Links

• Blocked: FLINK-10232 Add a SQL DDL
• is related to: FLINK-14644 Improve the usability of connector properties
• is related to: FLINK-12567 Rework DescriptorProperties to adapt unified DDL with clause and Descriptor key value pairs