Status

Current state: "Under Discussion"

Discussion thread: 

Vote thread: 

JIRA: 

Released: 1.13.0

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

Motivation

Currently there are many problems in sql client. For example, users can't register the table source/sink proposed by FLIP-95. Sql client is a wrapper of the Table Environment where users can use commands or SQL. Therefore, Sql client has two purposes:

• Sql client works as the entry for users to submit their sql job;
• Sql client works as the interactive tool;

Besides, sql clients should offer users a convenient, reusable way to configure the client settings. 

Proposed Changes

Configure the client 

Use the command to configure the client

Currently sql-client uses a YAML file to configure the client, which has its own grammar rather than the commands used in the client. It causes overhead for users because users have to study both gramars and it's very tricky for users to debug the YAML problems. Considering the Table Api has developed sophisticated grammar,  it's better to use the same commands in the client for users to configure their settings including set the runtime/table settings and register Catalog/Module/Database/Table/Function. 

For better understanding, the chart below lists the relationship between YAML and the command in the client.

However, some users may need to reset some settings to the default. For example, users submit a job that is restored from the specified savepoint. But the specified savepoint path lives in the session. Every job submitted later will try to load the specified savepoint. Therefore, we propose to add a new command in the sql client.

Flink SQL> RESET <key>;

In the case mentioned above, users can use the command as follows to clear the savepoint path.

Flink SQL> RESET execution.savepoint.path;

Add -i parameter to specify the initlization files

Users can use the parameter -i --init to start up the sql client with the initialization files.

> ./sql-client.sh embedded -i init1.sql init2.sql

The sql file looks as follows.

-- set up the session properties
SET sql-client.execution.mode=batch;
SET parallism.default=100;
SET pipeline.auto-watermark-interval=500;

-- set up the catalog
CREATE CATALOG HiveCatalog WITH (
  'type' = 'hive'
);

USE CATALOG HiveCatalog;

-- register the table
CREATE TABLE pageviews (
  user_id BIGINT,
  page_id BIGINT,
  viewtime TIMESTAMP,
  proctime AS PROCTIME()
) WITH (
  'connector' = 'kafka',
  'topic' = 'pageviews',
  'properties.bootstrap.servers' = '...',
  'format' = 'avro'
);

CREATE TABLE pageviews_enriched (
  user_id BIGINT,
  page_id BIGINT,
  viewtime TIMESTAMP,
  user_region STRING,
  WATERMARK FOR viewtime AS viewtime - INTERVAL '2' SECOND
) WITH (
  'connector' = 'kafka',
  'topic' = 'pageviews_enriched',
  ...
);

Considering the -i parameter is used to set up the environment, it's forbidden to add any INSERT statements and Queries in the initialization files.

Use the sql client as the tool to submit job

Add -f parameter to specify the execution file

Currently, users can only submit a single INSERT statement by -u command.  We proposed to add a new command line parameter -f for users to specify the execution file. Users can use the following command to start up the sql client.

> ./sql-client.sh embedded -i init1.sql init2.sql -f sqljob.sql

The sql file looks as follows.

CREATE TABLE users (
  user_id BIGINT,
  user_name STRING,
  user_level STRING,
  region STRING,
  PRIMARY KEY (user_id) NOT ENFORCED
) WITH (
  'connector' = 'upsert-kafka',
  'topic' = 'users',
  'properties.bootstrap.servers' = '...',
  'key.format' = 'csv',
  'value.format' = 'avro'
);

-- set the job name
SET pipeline.name=SqlJob; 
-- set the queue that the job submit to
SET yarn.application.queue=root;
-- set the job parallism
SET parallism.default=
10;


INSERT INTO pageviews_enriched
SELECT *
FROM pageviews AS p
LEFT JOIN users FOR SYSTEM_TIME AS OF p.proctime AS u
ON p.user_id = u.user_id;

Difference between -i and -f parameter

Users can specify the -i and -f together when starting up the sql client. In this section, we list the differences between two parameters.

Use the sql client as the interaction tool

Support more interactive commands.

Flink SQL> ADD JAR /local/test.jar;

[Info] Add the jar into the list.

Flink SQL> DELETE JAR /local/test.jar;

[Info] Delete the jar from the list.

Flink SQL> LIST JAR;
/local/addedJar.jar
/local/usedJar.jar

Flink SQL>

Flink SQL> SHOW CREATE TABLE MyTable;
CREATE TABLE `MyTable` (
  `id` INT,
  `value` BIGINT
) WITH (
  'connector' = 'kafka',
  'topic' = 'users',
  'properties.bootstrap.servers' = '...',
  'key.format' = 'csv',
  'value.format' = 'avro'
)

Flink SQL> EXPLAIN COST, CHANGELOG SELECT * FROM MyTable;
...

What's more, we will add a new option 'sql-client.verbose' to allow users to print the exception stack.

Support Multiple DML

Currently we have discussed the grammar about multi-line statements[1] as follows. 

BEGIN STATEMENT SET;
INSERT INTO ... SELECT ...;
INSERT INTO ... SELECT ...;
END;

In the sql-client, users should submit statements as follows.

Flink SQL> BEGIN STATEMENT SET;

[Info] Begin the statement set.

Flink SQL> INSERT INTO emps1 SELECT * FROM emps(x, y);

[Info] Add the statement into the statement set.

Flink SQL> INSERT INTO emps2 SELECT * FROM emps(x, y);

[Info] Add the statement into the statement set.

Flink SQL> END;

[Info] Submitting SQL statement set to the cluster... 

If users press CRTL+C after input BEGIN STATEMENT SET but before input END statement, the client will clear the statments in the set and exit statement set mode. 

Summary

After this FLIP finishes, the sql client will have the options, commands and command line parameters as follows. 

Supported Option in sql client

1000000

Supported Command in sql client

We use '+', '-' to identify the added and deprecated command.

Supported command line parameter

We use '+', '-' to identify the added and deprecated parameter.

Compatibility, Deprecation, and Migration Plan

This FLIP is an improvment for sql client.  Compatibility is not affected immediately.

The YAML initialization file will be deleted in this version and command line parameters related to the YAML file are deprecated. But we will continue to support the Yaml file if users specify explicitly. In the future, it's not supported to use the Yaml file to configure the client properties.

References

[1] http://apache-flink-mailing-list-archive.1008284.n3.nabble.com/DISCUSS-SQL-Syntax-for-Table-API-StatementSet-td42515.html