Status

Current state: "Under Discussion"

...

...

...

...

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

Motivation

With the efforts in FLIP-24 and FLIP-91, Flink SQL client supports submitting SQL jobs but lacks further support for their lifecycles afterward which is crucial for streaming use cases. That means Flink SQL client users have to turn to other clients (e.g. CLI) or APIs (e.g. REST API) to manage the jobs, like triggering savepoints or canceling queries, which makes the user experience of SQL client incomplete. 

...

This statement stops a non-terminated SQL, which is similar to `flink stop` in CLI. As stop command has a `--drain` option, we should introduce a table config like `sql-client.stop-with-drain` to support the same functionality.and `flink cancel` in CLI. 

STOP JOB '<job_id>' [WITH SAVEPOINT] [WITH DRAIN]

The result would the savepoint path.

+-----------------------------------------------------------------|
|                     savepoint_path                              |
+-----------------------------------------------------------------|
| hdfs://mycluster/flink-savepoints/savepoint-cca7bc-bb1e257f0dab |
+-----------------------------------------------------------------|

Related Table Options

In addition, thereThere're two related table options to control the fine-grained behavior:

1.

...

WITH SAVEPOINT

If specified, the stop statement stops a SQL job with a savepoint, which is similar to `flink stop` in CLI.

OtherwiseIf it's true (true by default), the stop statement stops a SQL job without a savepoint, which is similar to ungracefully, just like `flink cancel` in In CLI. Since an ungrateful drop doesn’t trigger a savepoint, the result would be a simple OK, like the one returned by DDL.

2.

...

WITH DRAIN

If specified

...

If it's true (false by default), the stop statement stops a SQL job and increases the watermark to MAX_WATERMARK to trigger all the timers, which is similar to `flink stop .. --drain` in CLI.

...

CREATE SAVEPOINT '[savepoint_path]' FOR JOB'<job_id>'

The result would the savepoint path.

...

Flink SQL> INSERT INTO tbl_a SELECT * FROM tbl_b;
[INFO] Submitting SQL update statement to the cluster...
[INFO] SQL update statement has been successfully submitted to the cluster:
Job ID: 6b1af540c0c0bb3fcfcad50ac037c862

Flink SQL> SHOW QUERIESJOBS;
+----------------------------------+--------------------+---------+---------------------+---------------------+-------------+----------------------+
|           job_id                 |       job_name     | status  |    start_time       |      end_time       |   duration  |       web_url        |
+----------------------------------+--------------------+---------|---------------------|---------------------|-------------|----------------------|
| 6b1af540c0c0bb3fcfcad50ac037c862 | INSERT INTO tbl_a..| RUNNING | 2022-05-01 10:20:33 | 2022-05-01 10:20:53 |  0h 0m 20s  | http://127.0.0.1:8081|
+----------------------------------+--------------------+---------+---------------------+---------------------+-------------+----------------------+

Flink SQL > CREATE SAVEPOINT FOR JOB '6b1af540c0c0bb3fcfcad50ac037c862';
+------------------------------------------------------------------|
|                            savepoint_path                        |
+------------------------------------------------------------------|
| hdfs://mycluster/flink-savepoints/savepoint-cca7bc-bb1e257f0dab  |
+------------------------------------------------------------------|

Flink SQL > STOP JOB '6b1af540c0c0bb3fcfcad50ac037c862';
[INFO] The specified job is stopped.

Flink SQL > DROP SAVEPOINT 'hdfs://mycluster/flink-savepoints/savepoint-cca7bc-bb1e257f0dab';
[INFO] The specified savepoint is dropped.

...

This FLIP introduces new SQL keywords, which may cause troubles for the existing SQLs. Users need to escape the new keywords if they use them as SQL identifiers.

The new keywords are:

• • JOB (new)
  • QUERY JOBS (new)
  • QUERIES STOP (new)
  • RELEASE DRAIN (new)
  • SAVEPOINT (already reserved)
  • SAVEPOINTS (already reserved)

Rejected Alternatives

...