Motivation

Currently there are many probelms in sql client. For example, users can't register the table source/sink proposed by FLIP-95. Sql client is a warpper 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 convient, reuseable way to configure the client settings. 

Proposed Changes

Configure the client 

Use the command to configure the client

Currently sql-client uses YAML file to configure the client, which has its own grammer rather than the commands used in the client. It causes overhead for users because users have to study and it's very tricky for users to debug line by line. Considering the Table Api has developed sophisticated grammer,  it's better to use the same commands in the client for users to configure their setting 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 setttings to the default. For example, users submit a job that restored from the specified savepoint. But the specified savepont 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.

Flink SQL> RESET execution.savepoint.path;

Add -i parameter to specify the initlization file

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.

-- use specified settings
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',
  'name' = 'TestCatalog'
);

USE CATALOG TestCatalog;

-- 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 initalization files.

Use the sql client as the tool to submit job

Add -f parameter to specify the execution file

Currenlty, users can only submit job 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.

SET pipeline.name='SqlJob';

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'
);

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.

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

Support Multiple DML

Currently we have discussed the grammer about multiline statement. It's time to support it in the sql parser. In the sql-client, users should submit statement as follows.

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

[Info] Submit the statement into the statement set.

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

[Info] Submit the statement into the statement set.

Flink SQL> END;

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

Compatibility, Deprecation, and Migration Plan

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

The YAML initialization file and -u parameter will be deprecated. It is recommended that users use the `-i` to specify the initialization file and `-f` to specify the execution file. 

Summary

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

Supported Option in sql client

1000000

Supported Command in sql client

Supported command line parameter