Status

Current state: "Under Discussion"

Discussion thread: http://apache-flink-mailing-list-archive.1008284.n3.nabble.com/DISCUSS-FLIP-163-SQL-Client-Improvements-td48354.html

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 commands 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.

Currently the TableEnvironment uses the TableResult#collect() to fetch the results. The client uses the JM as the man in the middle to communicate with the socket sink and JM knows the address and port of the client. For more details, please refer to the references[1][2]. After apply this changes to the sql-client, users don't need to set the parameters of the socket sink any more.

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 (separated by comma).

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

The sql file looks as follows.

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

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

USE CATALOG HiveCatalog;

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

CREATE IF NOT EXISTS 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 TEMPORARY 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 sync mode
SET table.dml-sync=true;

-- 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=100;

-- restore from the specific savepoint path
SET execution.savepoint.path=/tmp/flink-savepoints/savepoint-cca7bc-bb1e257f0dab;

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;

Sometimes, users may add multiple INSERT INTO statements in the execution files and expect to execute queries one by one in batch mode. Therefore, we would add a new option `table.dml-sync`. The default value for this option is false, which means the client will execute the next job when client submits the job. When setting this option true,  the client will execute the next job until job finishes. One more words, this option also works for the `TableEnvironment#executeSql`, which is as same as invoking `TableResult#await()` automatically.

If the sql file contains SELECT statement, it should take up the terminal until job finishes because terminal is the "sink" in the job. However, the behavior is undefined, such as how to cancel the job if the source is unbounded. In this version, the sql file doesn't allow SELECT statements.

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.

Simplify the command to start up the client

If users want to start up the sql client in embedded mode, they have to input  the command as follows.

> ./sql-client.sh embedded

It seems a little cumbersome for users because we still haven't support gateway right now and users use embedded client in most cases. Therefore, we propose a simplified command without parameter `embedded` to start up the client in the embedded mode. That means embedded is the default mode. 

> ./sql-client.sh

Use the sql client as the interaction tool

Support more interactive commands.

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

[Info] Add the jar into current session.

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

[Info] Delete the jar from current session.

Flink SQL> SHOW JARS;
/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 ESTIMATED_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 when an error happens, because currently only exception message is shown on the terminal which is not useful for debugging problems. 

Support Multiple DML

Currently we have discussed the grammar about multi-line statements[3] 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. 

Implementation Details

In our blueprint, we will maintain only one parser. It will first use regex to match simple commands, e.g. SET Commands. If the regex doesn't find any avaliable commands, the statement will be delegated to the underlying Calcite parser. 

For better understanding, the chart below demonstrates which commands belong to the sql-clinet and which command belong to the table environment.

When use -f to submit job, we will create a DumpTerminal(InputStream, OutputStream). The input stream is a FileStream created from the specified file and the output stram is std output. In this way, we can unify the execution. 

Summary

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

TableEnvironment

Added Option in table environment

Add `TableEnvironment.create(Configuration)`

In sql client and table environment, we can create table environment from the specified options in the configuration.

Supported Option in sql client

1000000

Supported Command in sql client

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

Supported command line parameter

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

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, we may totally remove the support of Yaml file to configure the client properties.

References

[1]https://issues.apache.org/jira/browse/FLINK-14807 

[2]https://docs.google.com/document/d/13Ata18-e89_hAdfukzEJYreOg2FBZO_Y0RohLDAme6Y/edit#

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