This is an automated email from the ASF dual-hosted git repository.
yuxia pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 77214f138cf [FLINK-31774][doc] Add document for delete and update
statement (#22398)
77214f138cf is described below
commit 77214f138cf759a3ee5466c9b2379e717227a0ae
Author: Min <[email protected]>
AuthorDate: Thu Jun 8 09:49:12 2023 +0800
[FLINK-31774][doc] Add document for delete and update statement (#22398)
---
docs/content.zh/docs/dev/table/sql/delete.md | 208 +++++++++++++++++++++++
docs/content.zh/docs/dev/table/sql/overview.md | 2 +
docs/content.zh/docs/dev/table/sql/update.md | 221 +++++++++++++++++++++++++
docs/content/docs/dev/table/sql/delete.md | 207 +++++++++++++++++++++++
docs/content/docs/dev/table/sql/overview.md | 2 +
docs/content/docs/dev/table/sql/update.md | 220 ++++++++++++++++++++++++
6 files changed, 860 insertions(+)
diff --git a/docs/content.zh/docs/dev/table/sql/delete.md
b/docs/content.zh/docs/dev/table/sql/delete.md
new file mode 100644
index 00000000000..e945bfd4ce7
--- /dev/null
+++ b/docs/content.zh/docs/dev/table/sql/delete.md
@@ -0,0 +1,208 @@
+---
+title: "DELETE 语句"
+weight: 18
+type: docs
+aliases:
+ - /zh/dev/table/sql/delete.html
+---
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# DELETE 语句
+
+`DELETE` 语句可以用于根据条件来删除表中的数据
+
+<span class="label label-danger">注意</span> 目前, `DELETE` 语句仅支持批模式, 并且要求目标表实现了
{{< gh_link
file="flink-table/flink-table-common/src/main/java/org/apache/flink/table/connector/sink/abilities/SupportsRowLevelDelete.java"
name="SupportsRowLevelDelete" >}} 接口。
+如果在一个没有实现该接口的表上执行 `DELETE`，则会抛异常。目前 Flink 内置的连接器还没有实现该接口。
+
+## 执行删除语句
+
+{{< tabs "delete statement" >}}
+
+{{< tab "Java" >}}
+
+DELETE 语句，可以使用 `TableEnvironment` 中的 `executeSql()` 方法执行。`executeSql()` 方法执行
DELETE 语句时会立即提交一个 Flink 作业，并且返回一个 TableResult 对象，通过该对象可以获取 JobClient 方便的操作提交的作业。
+
+以下的例子展示了如何在 `TableEnvironment` 中执行一条 DELETE 语句。
+
+{{< /tab >}}
+{{< tab "Scala" >}}
+
+DELETE 语句，可以使用 `TableEnvironment` 中的 `executeSql()` 方法执行。`executeSql()` 方法执行
DELETE 语句时会立即提交一个 Flink 作业，并且返回一个 TableResult 对象，通过该对象可以获取 JobClient 方便的操作提交的作业。
+
+以下的例子展示了如何在 `TableEnvironment` 中执行一条 DELETE 语句。
+
+{{< /tab >}}
+{{< tab "Python" >}}
+
+DELETE 语句，可以使用 `TableEnvironment` 中的 `execute_sql()` 方法执行。`execute_sql()` 方法执行
DELETE 语句时会立即提交一个 Flink 作业，并且返回一个 TableResult 对象，通过该对象可以获取 JobClient 方便的操作提交的作业。
+
+以下的例子展示了如何在 `TableEnvironment` 中执行一条 DELETE 语句。
+
+{{< /tab >}}
+{{< tab "SQL CLI" >}}
+
+可以在 [SQL CLI]({{< ref "docs/dev/table/sqlClient" >}}) 中执行 DELETE 语句
+
+以下的例子展示了如何在 SQL CLI 中执行一条 DELETE 语句。
+
+{{< /tab >}}
+{{< /tabs >}}
+
+{{< tabs "9bcbe6f1-9a62-41ff-c7be-568ec5908a29" >}}
+
+{{< tab "Java" >}}
+```java
+EnvironmentSettings settings =
EnvironmentSettings.newInstance().inBatchMode().build();
+TableEnvironment tEnv = TableEnvironment.create(settings);
+// 注册一个 "Orders" 表
+tEnv.executeSql("CREATE TABLE Orders (`user` STRING, product STRING, amount
INT) WITH (...)");
+// 插入原始数据
+tEnv.executeSql("INSERT INTO Orders VALUES ('Lili', 'Apple', 1), ('Jessica',
'Banana', 2), ('Mr.White', 'Chicken', 3)").await();
+tEnv.executeSql("SELECT * FROM Orders").print();
+//
+--------------------------------+--------------------------------+-------------+
+// | user | product |
amount |
+//
+--------------------------------+--------------------------------+-------------+
+// | Lili | Apple |
1 |
+// | Jessica | Banana |
2 |
+// | Mr.White | Chicken |
3 |
+//
+--------------------------------+--------------------------------+-------------+
+// 3 rows in set
+// 根据where条件删除
+tEnv.executeSql("DELETE FROM Orders WHERE `user` = 'Lili'").await();
+tEnv.executeSql("SELECT * FROM Orders").print();
+//
+--------------------------------+--------------------------------+-------------+
+// | user | product |
amount |
+//
+--------------------------------+--------------------------------+-------------+
+// | Jessica | Banana |
2 |
+// | Mr.White | Chicken |
3 |
+//
+--------------------------------+--------------------------------+-------------+
+// 2 rows in set
+// 全表删除
+tEnv.executeSql("DELETE FROM Orders").await();
+tEnv.executeSql("SELECT * FROM Orders").print();
+// Empty set
+```
+{{< /tab >}}
+{{< tab "Scala" >}}
+```scala
+val env = StreamExecutionEnvironment.getExecutionEnvironment()
+val settings = EnvironmentSettings.newInstance().inBatchMode().build()
+val tEnv = StreamTableEnvironment.create(env, settings)
+
+// 注册一个 "Orders" 表
+tEnv.executeSql("CREATE TABLE Orders (`user` STRING, product STRING, amount
INT) WITH (...)");
+// 插入原始数据
+tEnv.executeSql("INSERT INTO Orders VALUES ('Lili', 'Apple', 1), ('Jessica',
'Banana', 2), ('Mr.White', 'Chicken', 3)").await();
+tEnv.executeSql("SELECT * FROM Orders").print();
+//
+--------------------------------+--------------------------------+-------------+
+// | user | product |
amount |
+//
+--------------------------------+--------------------------------+-------------+
+// | Lili | Apple |
1 |
+// | Jessica | Banana |
2 |
+// | Mr.White | Chicken |
3 |
+//
+--------------------------------+--------------------------------+-------------+
+// 3 rows in set
+// 根据where条件删除
+tEnv.executeSql("DELETE FROM Orders WHERE `user` = 'Lili'").await();
+tEnv.executeSql("SELECT * FROM Orders").print();
+//
+--------------------------------+--------------------------------+-------------+
+// | user | product |
amount |
+//
+--------------------------------+--------------------------------+-------------+
+// | Jessica | Banana |
2 |
+// | Mr.White | Chicken |
3 |
+//
+--------------------------------+--------------------------------+-------------+
+// 2 rows in set
+// 全表删除
+tEnv.executeSql("DELETE FROM Orders").await();
+tEnv.executeSql("SELECT * FROM Orders").print();
+// Empty set
+```
+{{< /tab >}}
+{{< tab "Python" >}}
+```python
+env_settings = EnvironmentSettings.in_batch_mode()
+table_env = TableEnvironment.create(env_settings)
+
+# 注册一个 "Orders" 表
+table_env.executeSql("CREATE TABLE Orders (`user` STRING, product STRING,
amount INT) WITH (...)");
+# 插入原始数据
+table_env.executeSql("INSERT INTO Orders VALUES ('Lili', 'Apple', 1),
('Jessica', 'Banana', 2), ('Mr.White', 'Chicken', 3)").wait();
+table_env.executeSql("SELECT * FROM Orders").print();
+#
+--------------------------------+--------------------------------+-------------+
+# | user | product |
amount |
+#
+--------------------------------+--------------------------------+-------------+
+# | Lili | Apple |
1 |
+# | Jessica | Banana |
2 |
+# | Mr.White | Chicken |
3 |
+#
+--------------------------------+--------------------------------+-------------+
+# 3 rows in set
+# 根据where条件删除
+table_env.executeSql("DELETE FROM Orders WHERE `user` = 'Lili'").wait();
+table_env.executeSql("SELECT * FROM Orders").print();
+#
+--------------------------------+--------------------------------+-------------+
+# | user | product |
amount |
+#
+--------------------------------+--------------------------------+-------------+
+# | Jessica | Banana |
2 |
+# | Mr.White | Chicken |
3 |
+#
+--------------------------------+--------------------------------+-------------+
+# 2 rows in set
+# 全表删除
+table_env.executeSql("DELETE FROM Orders").wait();
+table_env.executeSql("SELECT * FROM Orders").print();
+# Empty set
+```
+{{< /tab >}}
+{{< tab "SQL CLI" >}}
+```sql
+Flink SQL> SET 'execution.runtime-mode' = 'batch';
+[INFO] Session property has been set.
+
+Flink SQL> CREATE TABLE Orders (`user` STRING, product STRING, amount INT)
with (...);
+[INFO] Execute statement succeed.
+
+Flink SQL> INSERT INTO Orders VALUES ('Lili', 'Apple', 1), ('Jessica',
'Banana', 1), ('Mr.White', 'Chicken', 3);
+[INFO] Submitting SQL update statement to the cluster...
+[INFO] SQL update statement has been successfully submitted to the cluster:
+Job ID: bd2c46a7b2769d5c559abd73ecde82e9
+
+Flink SQL> SELECT * FROM Orders;
+ user product amount
+ Lili Apple 1
+ Jessica Banana 2
+Mr.White Chicken 3
+
+Flink SQL> DELETE FROM Orders WHERE `user` = 'Lili';
+
+ user product amount
+ Jessica Banana 2
+Mr.White Chicken 3
+
+```
+{{< /tab >}}
+{{< /tabs >}}
+
+{{< top >}}
+
+## 语法
+
+```sql
+DELETE FROM [catalog_name.][db_name.]table_name [ WHERE condition ]
+```
+
diff --git a/docs/content.zh/docs/dev/table/sql/overview.md
b/docs/content.zh/docs/dev/table/sql/overview.md
index d4732f9be8d..a416f7dc295 100644
--- a/docs/content.zh/docs/dev/table/sql/overview.md
+++ b/docs/content.zh/docs/dev/table/sql/overview.md
@@ -36,6 +36,8 @@ under the License.
- [ALTER TABLE, DATABASE, FUNCTION]({{< ref "docs/dev/table/sql/alter" >}})
- [ANALYZE TABLE]({{< ref "docs/dev/table/sql/analyze" >}})
- [INSERT]({{< ref "docs/dev/table/sql/insert" >}})
+- [UPDATE]({{< ref "docs/dev/table/sql/update" >}})
+- [DELETE]({{< ref "docs/dev/table/sql/delete" >}})
- [SQL HINTS]({{< ref "docs/dev/table/sql/queries/hints" >}})
- [DESCRIBE]({{< ref "docs/dev/table/sql/describe" >}})
- [EXPLAIN]({{< ref "docs/dev/table/sql/explain" >}})
diff --git a/docs/content.zh/docs/dev/table/sql/update.md
b/docs/content.zh/docs/dev/table/sql/update.md
new file mode 100644
index 00000000000..4c472a7e5c9
--- /dev/null
+++ b/docs/content.zh/docs/dev/table/sql/update.md
@@ -0,0 +1,221 @@
+---
+title: "UPDATE 语句"
+weight: 17
+type: docs
+aliases:
+- /zh/dev/table/sql/update.html
+---
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# UPDATE 语句
+
+`UPDATE` 语句可以用于根据条件更新表的数据。
+
+<span class="label label-danger">注意</span> 目前, `UPDATE` 语句仅支持批模式, 并且要求目标表实现了
{{< gh_link
file="flink-table/flink-table-common/src/main/java/org/apache/flink/table/connector/sink/abilities/SupportsRowLevelUpdate.java"
name="SupportsRowLevelUpdate" >}} 接口。
+如果在一个没有实现该接口的表上执行 `UPDATE`，则会抛异常。目前 Flink 内置的连接器还没有实现该接口。
+
+## 执行更新语句
+
+{{< tabs "update statement" >}}
+{{< tab "Java" >}}
+
+UPDATE 语句，可以使用 `TableEnvironment` 中的 `executeSql()` 方法执行。`executeSql()` 方法执行
UPDATE 语句时会立即提交一个 Flink 作业，并且返回一个 TableResult 对象，通过该对象可以获取 JobClient 方便的操作提交的作业。
+
+以下的例子展示了如何在 `TableEnvironment` 中执行一条 UPDATE 语句。
+
+{{< /tab >}}
+{{< tab "Scala" >}}
+
+UPDATE 语句，可以使用 `TableEnvironment` 中的 `executeSql()` 方法执行。`executeSql()` 方法执行
UPDATE 语句时会立即提交一个 Flink 作业，并且返回一个 TableResult 对象，通过该对象可以获取 JobClient 方便的操作提交的作业。
+
+以下的例子展示了如何在 `TableEnvironment` 中执行一条 UPDATE 语句。
+
+{{< /tab >}}
+{{< tab "Python" >}}
+
+UPDATE 语句，可以使用 `TableEnvironment` 中的 `execute_sql()` 方法执行。`execute_sql()` 方法执行
UPDATE 语句时会立即提交一个 Flink 作业，并且返回一个 TableResult 对象，通过该对象可以获取 JobClient 方便的操作提交的作业。
+
+以下的例子展示了如何在 `TableEnvironment` 中执行一条 UPDATE 语句。
+
+{{< /tab >}}
+{{< tab "SQL CLI" >}}
+
+可以在 [SQL CLI]({{< ref "docs/dev/table/sqlClient" >}}) 中执行 UPDATE 语句
+
+以下的例子展示了如何在 SQL CLI 中执行一条 UPDATE 语句。
+
+{{< /tab >}}
+{{< /tabs >}}
+
+{{< tabs "dc09cb72-eebd-eb88-c29b-43bc3973f9f4" >}}
+
+{{< tab "Java" >}}
+```java
+EnvironmentSettings settings =
EnvironmentSettings.newInstance().inBatchMode().build();
+TableEnvironment tEnv = TableEnvironment.create(settings);
+// 注册一个 "Orders" 表
+tEnv.executeSql("CREATE TABLE Orders (`user` STRING, product STRING, amount
INT) WITH (...)");
+// 插入原始数据
+tEnv.executeSql("insert into Orders values ('Lili', 'Apple', 1), ('Jessica',
'Banana', 1)").await();
+tEnv.executeSql("SELECT * FROM Orders").print();
+//
+--------------------------------+--------------------------------+-------------+
+// | user | product |
amount |
+//
+--------------------------------+--------------------------------+-------------+
+// | Lili | Apple |
1 |
+// | Jessica | Banana |
1 |
+//
+--------------------------------+--------------------------------+-------------+
+// 2 rows in set
+// 更新所有的amount字段
+tEnv.executeSql("UPDATE Orders SET `amount` = `amount` * 2").await();
+tEnv.executeSql("SELECT * FROM Orders").print();
+//
+--------------------------------+--------------------------------+-------------+
+// | user | product |
amount |
+//
+--------------------------------+--------------------------------+-------------+
+// | Lili | Apple |
2 |
+// | Jessica | Banana |
2 |
+//
+--------------------------------+--------------------------------+-------------+
+// 2 rows in set
+// 根据where条件更新
+tEnv.executeSql("UPDATE Orders SET `product` = 'Orange' WHERE `user` =
'Lili'").await();
+tEnv.executeSql("SELECT * FROM Orders").print();
+//
+--------------------------------+--------------------------------+-------------+
+// | user | product |
amount |
+//
+--------------------------------+--------------------------------+-------------+
+// | Lili | Orange |
2 |
+// | Jessica | Banana |
2 |
+//
+--------------------------------+--------------------------------+-------------+
+// 2 rows in set
+```
+{{< /tab >}}
+{{< tab "Scala" >}}
+```scala
+val env = StreamExecutionEnvironment.getExecutionEnvironment()
+val settings = EnvironmentSettings.newInstance().inBatchMode().build()
+val tEnv = StreamTableEnvironment.create(env, settings)
+
+// 注册一个 "Orders" 表
+tEnv.executeSql("CREATE TABLE Orders (`user` STRING, product STRING, amount
INT) WITH (...)");
+// 插入原始数据
+tEnv.executeSql("insert into Orders values ('Lili', 'Apple', 1), ('Jessica',
'Banana', 1)").await();
+tEnv.executeSql("SELECT * FROM Orders").print();
+//
+--------------------------------+--------------------------------+-------------+
+// | user | product |
amount |
+//
+--------------------------------+--------------------------------+-------------+
+// | Lili | Apple |
1 |
+// | Jessica | Banana |
1 |
+//
+--------------------------------+--------------------------------+-------------+
+// 2 rows in set
+// 更新所有的amount字段
+tEnv.executeSql("UPDATE Orders SET `amount` = `amount` * 2").await();
+tEnv.executeSql("SELECT * FROM Orders").print();
+//
+--------------------------------+--------------------------------+-------------+
+// | user | product |
amount |
+//
+--------------------------------+--------------------------------+-------------+
+// | Lili | Apple |
2 |
+// | Jessica | Banana |
2 |
+//
+--------------------------------+--------------------------------+-------------+
+// 2 rows in set
+// 根据where条件更新
+tEnv.executeSql("UPDATE Orders SET `product` = 'Orange' WHERE `user` =
'Lili'").await();
+tEnv.executeSql("SELECT * FROM Orders").print();
+//
+--------------------------------+--------------------------------+-------------+
+// | user | product |
amount |
+//
+--------------------------------+--------------------------------+-------------+
+// | Lili | Orange |
2 |
+// | Jessica | Banana |
2 |
+//
+--------------------------------+--------------------------------+-------------+
+// 2 rows in set
+```
+{{< /tab >}}
+{{< tab "Python" >}}
+```python
+env_settings = EnvironmentSettings.in_batch_mode()
+table_env = TableEnvironment.create(env_settings)
+
+# 注册一个 "Orders" 表
+table_env.executeSql("CREATE TABLE Orders (`user` STRING, product STRING,
amount INT) WITH (...)");
+# 插入原始数据
+table_env.executeSql("insert into Orders values ('Lili', 'Apple', 1),
('Jessica', 'Banana', 1)").wait();
+table_env.executeSql("SELECT * FROM Orders").print();
+#
+--------------------------------+--------------------------------+-------------+
+# | user | product |
amount |
+#
+--------------------------------+--------------------------------+-------------+
+# | Lili | Apple |
1 |
+# | Jessica | Banana |
1 |
+#
+--------------------------------+--------------------------------+-------------+
+# 2 rows in set
+# 更新所有的amount字段
+table_env.executeSql("UPDATE Orders SET `amount` = `amount` * 2").wait();
+table_env.executeSql("SELECT * FROM Orders").print();
+#
+--------------------------------+--------------------------------+-------------+
+# | user | product |
amount |
+#
+--------------------------------+--------------------------------+-------------+
+# | Lili | Apple |
2 |
+# | Jessica | Banana |
2 |
+#
+--------------------------------+--------------------------------+-------------+
+# 2 rows in set
+# 根据where条件更新
+table_env.executeSql("UPDATE Orders SET `product` = 'Orange' WHERE `user` =
'Lili'").wait();
+table_env.executeSql("SELECT * FROM Orders").print();
+#
+--------------------------------+--------------------------------+-------------+
+# | user | product |
amount |
+#
+--------------------------------+--------------------------------+-------------+
+# | Lili | Orange |
2 |
+# | Jessica | Banana |
2 |
+#
+--------------------------------+--------------------------------+-------------+
+# 2 rows in set
+```
+{{< /tab >}}
+{{< tab "SQL CLI" >}}
+```sql
+Flink SQL> SET 'execution.runtime-mode' = 'batch';
+[INFO] Session property has been set.
+
+Flink SQL> CREATE TABLE Orders (`user` STRING, product STRING, amount INT)
with (...);
+[INFO] Execute statement succeed.
+
+Flink SQL> INSERT INTO Orders VALUES ('Lili', 'Apple', 1), ('Jessica',
'Banana', 1);
+[INFO] Submitting SQL update statement to the cluster...
+[INFO] SQL update statement has been successfully submitted to the cluster:
+Job ID: bd2c46a7b2769d5c559abd73ecde82e9
+
+Flink SQL> SELECT * FROM Orders;
+ user product amount
+ Lili Apple 1
+Jessica Banana 1
+
+Flink SQL> UPDATE Orders SET amount = 2;
+
+ user product amount
+ Lili Apple 2
+Jessica Banana 2
+
+```
+{{< /tab >}}
+{{< /tabs >}}
+
+{{< top >}}
+
+## 语法
+
+```sql
+UPDATE [catalog_name.][db_name.]table_name SET column_name1 = expression1 [,
column_name2 = expression2, ...][ WHERE condition ]
+```
+
diff --git a/docs/content/docs/dev/table/sql/delete.md
b/docs/content/docs/dev/table/sql/delete.md
new file mode 100644
index 00000000000..cb757466859
--- /dev/null
+++ b/docs/content/docs/dev/table/sql/delete.md
@@ -0,0 +1,207 @@
+---
+title: "DELETE Statements"
+weight: 18
+type: docs
+aliases:
+- /dev/table/sql/delete.html
+---
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# DELETE Statements
+
+`DELETE` statement is used to perform row-level deletion on the target table
according to the filter if provided.
+
+<span class="label label-danger">Attention</span> Currently, `DELETE`
statement only supports in batch mode, and it requires the target table
connector implements the {{< gh_link
file="flink-table/flink-table-common/src/main/java/org/apache/flink/table/connector/sink/abilities/SupportsRowLevelDelete.java"
name="SupportsRowLevelDelete" >}}
+interface to support the row-level delete. An exception will be thrown if
trying to `DELETE` the table which has not implements the related interface.
Currently, there is no existing connector maintained by flink has supported
DELETE yet.
+
+## Run a DELETE statement
+
+{{< tabs "delete statement" >}}
+
+{{< tab "Java" >}}
+
+DELETE statements can be executed with the `executeSql()` method of the
`TableEnvironment`. The `executeSql()` will submit a Flink job immediately, and
return a `TableResult` instance which associates the submitted job.
+
+The following examples show how to run a single DELETE statement in
`TableEnvironment`.
+
+{{< /tab >}}
+{{< tab "Scala" >}}
+
+DELETE statements can be executed with the `executeSql()` method of the
`TableEnvironment`. The `executeSql()` will submit a Flink job immediately, and
return a `TableResult` instance which associates the submitted job.
+
+The following examples show how to run a single DELETE statement in
`TableEnvironment`.
+{{< /tab >}}
+{{< tab "Python" >}}
+
+DELETE statements can be executed with the `execute_sql()` method of the
`TableEnvironment`. The `executeSql()` will submit a Flink job immediately, and
return a `TableResult` instance which associates the submitted job.
+
+The following examples show how to run a single DELETE statement in
`TableEnvironment`.
+
+{{< /tab >}}
+{{< tab "SQL CLI" >}}
+
+DELETE statements can be executed in [SQL CLI]({{< ref
"docs/dev/table/sqlClient" >}}).
+
+The following examples show how to run a DELETE statement in SQL CLI.
+
+{{< /tab >}}
+{{< /tabs >}}
+
+{{< tabs "9bcbe6f1-9a62-41ff-c7be-568ec5908a29" >}}
+
+{{< tab "Java" >}}
+```java
+EnvironmentSettings settings =
EnvironmentSettings.newInstance().inBatchMode().build();
+TableEnvironment tEnv = TableEnvironment.create(settings);
+// register a table named "Orders"
+tEnv.executeSql("CREATE TABLE Orders (`user` STRING, product STRING, amount
INT) WITH (...)");
+// insert values
+tEnv.executeSql("INSERT INTO Orders VALUES ('Lili', 'Apple', 1), ('Jessica',
'Banana', 2), ('Mr.White', 'Chicken', 3)").await();
+tEnv.executeSql("SELECT * FROM Orders").print();
+//
+--------------------------------+--------------------------------+-------------+
+// | user | product |
amount |
+//
+--------------------------------+--------------------------------+-------------+
+// | Lili | Apple |
1 |
+// | Jessica | Banana |
2 |
+// | Mr.White | Chicken |
3 |
+//
+--------------------------------+--------------------------------+-------------+
+// 3 rows in set
+// delete by filter
+tEnv.executeSql("DELETE FROM Orders WHERE `user` = 'Lili'").await();
+tEnv.executeSql("SELECT * FROM Orders").print();
+//
+--------------------------------+--------------------------------+-------------+
+// | user | product |
amount |
+//
+--------------------------------+--------------------------------+-------------+
+// | Jessica | Banana |
2 |
+// | Mr.White | Chicken |
3 |
+//
+--------------------------------+--------------------------------+-------------+
+// 2 rows in set
+// delete entire table
+tEnv.executeSql("DELETE FROM Orders").await();
+tEnv.executeSql("SELECT * FROM Orders").print();
+// Empty set
+```
+{{< /tab >}}
+{{< tab "Scala" >}}
+```scala
+val env = StreamExecutionEnvironment.getExecutionEnvironment()
+val settings = EnvironmentSettings.newInstance().inBatchMode().build()
+val tEnv = StreamTableEnvironment.create(env, settings)
+
+// register a table named "Orders"
+tEnv.executeSql("CREATE TABLE Orders (`user` STRING, product STRING, amount
INT) WITH (...)");
+// insert values
+tEnv.executeSql("INSERT INTO Orders VALUES ('Lili', 'Apple', 1), ('Jessica',
'Banana', 2), ('Mr.White', 'Chicken', 3)").await();
+tEnv.executeSql("SELECT * FROM Orders").print();
+//
+--------------------------------+--------------------------------+-------------+
+// | user | product |
amount |
+//
+--------------------------------+--------------------------------+-------------+
+// | Lili | Apple |
1 |
+// | Jessica | Banana |
2 |
+// | Mr.White | Chicken |
3 |
+//
+--------------------------------+--------------------------------+-------------+
+// 3 rows in set
+// delete by filter
+tEnv.executeSql("DELETE FROM Orders WHERE `user` = 'Lili'").await();
+tEnv.executeSql("SELECT * FROM Orders").print();
+//
+--------------------------------+--------------------------------+-------------+
+// | user | product |
amount |
+//
+--------------------------------+--------------------------------+-------------+
+// | Jessica | Banana |
2 |
+// | Mr.White | Chicken |
3 |
+//
+--------------------------------+--------------------------------+-------------+
+// 2 rows in set
+// delete entire table
+tEnv.executeSql("DELETE FROM Orders").await();
+tEnv.executeSql("SELECT * FROM Orders").print();
+// Empty set
+```
+{{< /tab >}}
+{{< tab "Python" >}}
+```python
+env_settings = EnvironmentSettings.in_batch_mode()
+table_env = TableEnvironment.create(env_settings)
+
+# register a table named "Orders"
+table_env.executeSql("CREATE TABLE Orders (`user` STRING, product STRING,
amount INT) WITH (...)");
+# insert values
+table_env.executeSql("INSERT INTO Orders VALUES ('Lili', 'Apple', 1),
('Jessica', 'Banana', 2), ('Mr.White', 'Chicken', 3)").wait();
+table_env.executeSql("SELECT * FROM Orders").print();
+#
+--------------------------------+--------------------------------+-------------+
+# | user | product |
amount |
+#
+--------------------------------+--------------------------------+-------------+
+# | Lili | Apple |
1 |
+# | Jessica | Banana |
2 |
+# | Mr.White | Chicken |
3 |
+#
+--------------------------------+--------------------------------+-------------+
+# 3 rows in set
+# delete by filter
+table_env.executeSql("DELETE FROM Orders WHERE `user` = 'Lili'").wait();
+table_env.executeSql("SELECT * FROM Orders").print();
+#
+--------------------------------+--------------------------------+-------------+
+# | user | product |
amount |
+#
+--------------------------------+--------------------------------+-------------+
+# | Jessica | Banana |
2 |
+# | Mr.White | Chicken |
3 |
+#
+--------------------------------+--------------------------------+-------------+
+# 2 rows in set
+# delete entire table
+table_env.executeSql("DELETE FROM Orders").wait();
+table_env.executeSql("SELECT * FROM Orders").print();
+# Empty set
+```
+{{< /tab >}}
+{{< tab "SQL CLI" >}}
+```sql
+Flink SQL> SET 'execution.runtime-mode' = 'batch';
+[INFO] Session property has been set.
+
+Flink SQL> CREATE TABLE Orders (`user` STRING, product STRING, amount INT)
with (...);
+[INFO] Execute statement succeed.
+
+Flink SQL> INSERT INTO Orders VALUES ('Lili', 'Apple', 1), ('Jessica',
'Banana', 1), ('Mr.White', 'Chicken', 3);
+[INFO] Submitting SQL update statement to the cluster...
+[INFO] SQL update statement has been successfully submitted to the cluster:
+Job ID: bd2c46a7b2769d5c559abd73ecde82e9
+
+Flink SQL> SELECT * FROM Orders;
+ user product amount
+ Lili Apple 1
+ Jessica Banana 2
+Mr.White Chicken 3
+
+Flink SQL> DELETE FROM Orders WHERE `user` = 'Lili';
+
+ user product amount
+ Jessica Banana 2
+Mr.White Chicken 3
+
+```
+{{< /tab >}}
+{{< /tabs >}}
+
+{{< top >}}
+
+## DELETE ROWS
+
+```sql
+DELETE FROM [catalog_name.][db_name.]table_name [ WHERE condition ]
+```
+
diff --git a/docs/content/docs/dev/table/sql/overview.md
b/docs/content/docs/dev/table/sql/overview.md
index 97fe751c16e..8f2ba17721e 100644
--- a/docs/content/docs/dev/table/sql/overview.md
+++ b/docs/content/docs/dev/table/sql/overview.md
@@ -36,6 +36,8 @@ This page lists all the supported statements supported in
Flink SQL for now:
- [ALTER TABLE, DATABASE, FUNCTION]({{< ref "docs/dev/table/sql/alter" >}})
- [ANALYZE TABLE]({{< ref "docs/dev/table/sql/analyze" >}})
- [INSERT]({{< ref "docs/dev/table/sql/insert" >}})
+- [UPDATE]({{< ref "docs/dev/table/sql/update" >}})
+- [DELETE]({{< ref "docs/dev/table/sql/delete" >}})
- [DESCRIBE]({{< ref "docs/dev/table/sql/describe" >}})
- [EXPLAIN]({{< ref "docs/dev/table/sql/explain" >}})
- [USE]({{< ref "docs/dev/table/sql/use" >}})
diff --git a/docs/content/docs/dev/table/sql/update.md
b/docs/content/docs/dev/table/sql/update.md
new file mode 100644
index 00000000000..64364db77f8
--- /dev/null
+++ b/docs/content/docs/dev/table/sql/update.md
@@ -0,0 +1,220 @@
+---
+title: "UPDATE Statements"
+weight: 17
+type: docs
+aliases:
+- /dev/table/sql/update.html
+---
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# UPDATE Statements
+
+`UPDATE` statement is used to perform row-level updating on the target table
according to the filter if provided.
+
+<span class="label label-danger">Attention</span> Currently, `UPDATE`
statement only supports in batch mode, and it requires the target table
connector implements the {{< gh_link
file="flink-table/flink-table-common/src/main/java/org/apache/flink/table/connector/sink/abilities/SupportsRowLevelUpdate.java"
name="SupportsRowLevelUpdate" >}}
+interface to support the row-level update. An exception will be thrown if
trying to `UPDATE` the table which has not implements the related interface.
Currently, there is no existing connector maintained by flink has supported
UPDATE yet.
+
+## Run a UPDATE statement
+
+{{< tabs "update statement" >}}
+{{< tab "Java" >}}
+
+UPDATE statements can be executed with the `executeSql()` method of the
`TableEnvironment`. The `executeSql()` will submit a Flink job immediately, and
return a `TableResult` instance which associates the submitted job.
+
+The following examples show how to run a single UPDATE statement in
`TableEnvironment`.
+
+{{< /tab >}}
+{{< tab "Scala" >}}
+
+UPDATE statements can be executed with the `executeSql()` method of the
`TableEnvironment`. The `executeSql()` will submit a Flink job immediately, and
return a `TableResult` instance which associates the submitted job.
+
+The following examples show how to run a single UPDATE statement in
`TableEnvironment`.
+{{< /tab >}}
+{{< tab "Python" >}}
+
+UPDATE statements can be executed with the `execute_sql()` method of the
`TableEnvironment`. The `executeSql()` will submit a Flink job immediately, and
return a `TableResult` instance which associates the submitted job.
+
+The following examples show how to run a single UPDATE statement in
`TableEnvironment`.
+
+{{< /tab >}}
+{{< tab "SQL CLI" >}}
+
+UPDATE statements can be executed in [SQL CLI]({{< ref
"docs/dev/table/sqlClient" >}}).
+
+The following examples show how to run a UPDATE statement in SQL CLI.
+
+{{< /tab >}}
+{{< /tabs >}}
+
+{{< tabs "dc09cb72-eebd-eb88-c29b-43bc3973f9f4" >}}
+
+{{< tab "Java" >}}
+```java
+EnvironmentSettings settings =
EnvironmentSettings.newInstance().inBatchMode().build();
+TableEnvironment tEnv = TableEnvironment.create(settings);
+// register a table named "Orders"
+tEnv.executeSql("CREATE TABLE Orders (`user` STRING, product STRING, amount
INT) WITH (...)");
+// insert values
+tEnv.executeSql("insert into Orders values ('Lili', 'Apple', 1), ('Jessica',
'Banana', 1)").await();
+tEnv.executeSql("SELECT * FROM Orders").print();
+//
+--------------------------------+--------------------------------+-------------+
+// | user | product |
amount |
+//
+--------------------------------+--------------------------------+-------------+
+// | Lili | Apple |
1 |
+// | Jessica | Banana |
1 |
+//
+--------------------------------+--------------------------------+-------------+
+// 2 rows in set
+// update all the amount
+tEnv.executeSql("UPDATE Orders SET `amount` = `amount` * 2").await();
+tEnv.executeSql("SELECT * FROM Orders").print();
+//
+--------------------------------+--------------------------------+-------------+
+// | user | product |
amount |
+//
+--------------------------------+--------------------------------+-------------+
+// | Lili | Apple |
2 |
+// | Jessica | Banana |
2 |
+//
+--------------------------------+--------------------------------+-------------+
+// 2 rows in set
+// update by filter
+tEnv.executeSql("UPDATE Orders SET `product` = 'Orange' WHERE `user` =
'Lili'").await();
+tEnv.executeSql("SELECT * FROM Orders").print();
+//
+--------------------------------+--------------------------------+-------------+
+// | user | product |
amount |
+//
+--------------------------------+--------------------------------+-------------+
+// | Lili | Orange |
2 |
+// | Jessica | Banana |
2 |
+//
+--------------------------------+--------------------------------+-------------+
+// 2 rows in set
+```
+{{< /tab >}}
+{{< tab "Scala" >}}
+```scala
+val env = StreamExecutionEnvironment.getExecutionEnvironment()
+val settings = EnvironmentSettings.newInstance().inBatchMode().build()
+val tEnv = StreamTableEnvironment.create(env, settings)
+
+// register a table named "Orders"
+tEnv.executeSql("CREATE TABLE Orders (`user` STRING, product STRING, amount
INT) WITH (...)");
+// insert values
+tEnv.executeSql("insert into Orders values ('Lili', 'Apple', 1), ('Jessica',
'Banana', 1)").await();
+tEnv.executeSql("SELECT * FROM Orders").print();
+//
+--------------------------------+--------------------------------+-------------+
+// | user | product |
amount |
+//
+--------------------------------+--------------------------------+-------------+
+// | Lili | Apple |
1 |
+// | Jessica | Banana |
1 |
+//
+--------------------------------+--------------------------------+-------------+
+// 2 rows in set
+// update all the amount
+tEnv.executeSql("UPDATE Orders SET `amount` = `amount` * 2").await();
+tEnv.executeSql("SELECT * FROM Orders").print();
+//
+--------------------------------+--------------------------------+-------------+
+// | user | product |
amount |
+//
+--------------------------------+--------------------------------+-------------+
+// | Lili | Apple |
2 |
+// | Jessica | Banana |
2 |
+//
+--------------------------------+--------------------------------+-------------+
+// 2 rows in set
+// update by filter
+tEnv.executeSql("UPDATE Orders SET `product` = 'Orange' WHERE `user` =
'Lili'").await();
+tEnv.executeSql("SELECT * FROM Orders").print();
+//
+--------------------------------+--------------------------------+-------------+
+// | user | product |
amount |
+//
+--------------------------------+--------------------------------+-------------+
+// | Lili | Orange |
2 |
+// | Jessica | Banana |
2 |
+//
+--------------------------------+--------------------------------+-------------+
+// 2 rows in set
+```
+{{< /tab >}}
+{{< tab "Python" >}}
+```python
+env_settings = EnvironmentSettings.in_batch_mode()
+table_env = TableEnvironment.create(env_settings)
+
+# register a table named "Orders"
+table_env.executeSql("CREATE TABLE Orders (`user` STRING, product STRING,
amount INT) WITH (...)");
+# insert values
+table_env.executeSql("insert into Orders values ('Lili', 'Apple', 1),
('Jessica', 'Banana', 1)").wait();
+table_env.executeSql("SELECT * FROM Orders").print();
+#
+--------------------------------+--------------------------------+-------------+
+# | user | product |
amount |
+#
+--------------------------------+--------------------------------+-------------+
+# | Lili | Apple |
1 |
+# | Jessica | Banana |
1 |
+#
+--------------------------------+--------------------------------+-------------+
+# 2 rows in set
+# update all the amount
+table_env.executeSql("UPDATE Orders SET `amount` = `amount` * 2").wait();
+table_env.executeSql("SELECT * FROM Orders").print();
+#
+--------------------------------+--------------------------------+-------------+
+# | user | product |
amount |
+#
+--------------------------------+--------------------------------+-------------+
+# | Lili | Apple |
2 |
+# | Jessica | Banana |
2 |
+#
+--------------------------------+--------------------------------+-------------+
+# 2 rows in set
+# update by filter
+table_env.executeSql("UPDATE Orders SET `product` = 'Orange' WHERE `user` =
'Lili'").wait();
+table_env.executeSql("SELECT * FROM Orders").print();
+#
+--------------------------------+--------------------------------+-------------+
+# | user | product |
amount |
+#
+--------------------------------+--------------------------------+-------------+
+# | Lili | Orange |
2 |
+# | Jessica | Banana |
2 |
+#
+--------------------------------+--------------------------------+-------------+
+# 2 rows in set
+```
+{{< /tab >}}
+{{< tab "SQL CLI" >}}
+```sql
+Flink SQL> SET 'execution.runtime-mode' = 'batch';
+[INFO] Session property has been set.
+
+Flink SQL> CREATE TABLE Orders (`user` STRING, product STRING, amount INT)
with (...);
+[INFO] Execute statement succeed.
+
+Flink SQL> INSERT INTO Orders VALUES ('Lili', 'Apple', 1), ('Jessica',
'Banana', 1);
+[INFO] Submitting SQL update statement to the cluster...
+[INFO] SQL update statement has been successfully submitted to the cluster:
+Job ID: bd2c46a7b2769d5c559abd73ecde82e9
+
+Flink SQL> SELECT * FROM Orders;
+ user product amount
+ Lili Apple 1
+Jessica Banana 1
+
+Flink SQL> UPDATE Orders SET amount = 2;
+
+ user product amount
+ Lili Apple 2
+Jessica Banana 2
+
+```
+{{< /tab >}}
+{{< /tabs >}}
+
+{{< top >}}
+
+## UPDATE ROWS
+
+```sql
+UPDATE [catalog_name.][db_name.]table_name SET column_name1 = expression1 [,
column_name2 = expression2, ...][ WHERE condition ]
+```
+
