You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by yu...@apache.org on 2023/08/10 11:16:06 UTC
[flink] branch master updated: [FLINK-32581][docs] Add docs for atomic CTAS and RTAS (#23116)
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 3cd0bf0a670 [FLINK-32581][docs] Add docs for atomic CTAS and RTAS (#23116)
3cd0bf0a670 is described below
commit 3cd0bf0a67016fe9c0b9ddbf033391ddb87ea496
Author: zhangmang <zh...@163.com>
AuthorDate: Thu Aug 10 19:15:58 2023 +0800
[FLINK-32581][docs] Add docs for atomic CTAS and RTAS (#23116)
---
docs/content.zh/docs/dev/table/sourcesSinks.md | 10 +++++++---
docs/content.zh/docs/dev/table/sql/create.md | 22 ++++++++++++++++++----
docs/content/docs/dev/table/sourcesSinks.md | 5 +++++
docs/content/docs/dev/table/sql/create.md | 22 ++++++++++++++++++----
4 files changed, 48 insertions(+), 11 deletions(-)
diff --git a/docs/content.zh/docs/dev/table/sourcesSinks.md b/docs/content.zh/docs/dev/table/sourcesSinks.md
index aeac54398b6..7c75eb83ac3 100644
--- a/docs/content.zh/docs/dev/table/sourcesSinks.md
+++ b/docs/content.zh/docs/dev/table/sourcesSinks.md
@@ -268,15 +268,19 @@ Flink 会对工厂类逐个进行检查，确保其“标识符”是全局唯
</tr>
<tr>
<td>{{< gh_link file="flink-table/flink-table-common/src/main/java/org/apache/flink/table/connector/sink/abilities/SupportsDeletePushDown.java" name="SupportsDeletePushDown" >}}</td>
- <td>支持将 <code>DELETE</code> 语句中的过滤条件下推到 <code>DynamicTableSink</code>，sink 端可以直接根据过滤条件来删除数据。
+ <td>支持将 <code>DELETE</code> 语句中的过滤条件下推到 <code>DynamicTableSink</code>，sink 端可以直接根据过滤条件来删除数据。</td>
</tr>
<tr>
<td>{{< gh_link file="flink-table/flink-table-common/src/main/java/org/apache/flink/table/connector/sink/abilities/SupportsRowLevelDelete.java" name="SupportsRowLevelDelete" >}}</td>
- <td>支持 <code>DynamicTableSink</code> 根据行级别的变更来删除已有的数据。该接口的实现者需要告诉 Planner 如何产生这些行变更，并且需要消费这些行变更从而达到删除数据的目的。
+ <td>支持 <code>DynamicTableSink</code> 根据行级别的变更来删除已有的数据。该接口的实现者需要告诉 Planner 如何产生这些行变更，并且需要消费这些行变更从而达到删除数据的目的。</td>
</tr>
<tr>
<td>{{< gh_link file="flink-table/flink-table-common/src/main/java/org/apache/flink/table/connector/sink/abilities/SupportsRowLevelUpdate.java" name="SupportsRowLevelUpdate" >}}</td>
- <td>支持 <code>DynamicTableSink</code> 根据行级别的变更来更新已有的数据。该接口的实现者需要告诉 Planner 如何产生这些行变更，并且需要消费这些行变更从而达到更新数据的目的。
+ <td>支持 <code>DynamicTableSink</code> 根据行级别的变更来更新已有的数据。该接口的实现者需要告诉 Planner 如何产生这些行变更，并且需要消费这些行变更从而达到更新数据的目的。</td>
+ </tr>
+ <tr>
+ <td>{{< gh_link file="flink-table/flink-table-common/src/main/java/org/apache/flink/table/connector/sink/abilities/SupportsStaging.java" name="SupportsStaging" >}}</td>
+ <td>支持 <code>DynamicTableSink</code> 提供 CTAS(CREATE TABLE AS SELECT) 或 RTAS([CREATE OR] REPLACE TABLE AS SELECT) 的原子性语义。该接口的实现者需要返回一个提供原子性语义实现的 <code>StagedTable</code> 对象。</td>
</tr>
</tbody>
</table>
diff --git a/docs/content.zh/docs/dev/table/sql/create.md b/docs/content.zh/docs/dev/table/sql/create.md
index 1606928b78d..7024b457585 100644
--- a/docs/content.zh/docs/dev/table/sql/create.md
+++ b/docs/content.zh/docs/dev/table/sql/create.md
@@ -547,14 +547,20 @@ CREATE TABLE my_ctas_table (
INSERT INTO my_ctas_table SELECT id, name, age FROM source_table WHERE mod(id, 10) = 0;
```
-**注意** CTAS 有如下约束：
+**注意：** CTAS 有如下约束：
* 暂不支持创建临时表。
* 暂不支持指定列信息。
* 暂不支持指定 Watermark。
* 暂不支持创建分区表。
* 暂不支持主键约束。
-**注意** 目前，CTAS 创建的目标表是非原子性的，如果在向表中插入数据时发生错误，该表不会被自动删除。
+**注意：** 默认情况下，CTAS 是非原子性的，这意味着如果在向表中插入数据时发生错误，该表不会被自动删除。
+
+#### 原子性
+
+如果要启用 CTAS 的原子性，则应确保：
+* 对应的 Connector sink 已经实现了 CTAS 的原子性语义，你可能需要阅读对应 Connector 的文档看是否已经支持了原子性语义。如果开发者想要实现原子性语义，请参考文档 [SupportsStaging]({{< ref "docs/dev/table/sourcesSinks" >}}#sink-abilities)。
+* 设置配置项 [table.rtas-ctas.atomicity-enabled]({{< ref "docs/dev/table/config" >}}#table-rtas-ctas-atomicity-enabled) 为 `true`。
{{< top >}}
@@ -566,7 +572,7 @@ WITH (key1=val1, key2=val2, ...)
AS select_query
```
-**注意** RTAS 有如下语义:
+**注意：** RTAS 有如下语义:
* REPLACE TABLE AS SELECT 语句：要被替换的目标表必须存在，否则会报错。
* CREATE OR REPLACE TABLE AS SELECT 语句：要被替换的目标表如果不存在，引擎会自动创建；如果存在的话，引擎就直接替换它。
@@ -601,13 +607,21 @@ CREATE TABLE my_rtas_table (
INSERT INTO my_rtas_table SELECT id, name, age FROM source_table WHERE mod(id, 10) = 0;
```
-**注意** RTAS 有如下约束：
+**注意：** RTAS 有如下约束：
* 暂不支持替换临时表。
* 暂不支持指定列信息。
* 暂不支持指定 Watermark。
* 暂不支持创建分区表。
* 暂不支持主键约束。
+**注意：** 默认情况下，RTAS 是非原子性的，这意味着如果在向表中插入数据时发生错误，该表不会被自动删除或还原成原来的表。
+
+### 原子性
+
+如果要启用 RTAS 的原子性，则应确保：
+* 对应的 Connector sink 已经实现了 RTAS 的原子性语义，你可能需要阅读对应 Connector 的文档看是否已经支持了原子性语义。如果开发者想要实现原子性语义，请参考文档 [SupportsStaging]({{< ref "docs/dev/table/sourcesSinks" >}}#sink-abilities)。
+* 设置配置项 [table.rtas-ctas.atomicity-enabled]({{< ref "docs/dev/table/config" >}}#table-rtas-ctas-atomicity-enabled) 为 `true`。
+
{{< top >}}
## CREATE CATALOG
diff --git a/docs/content/docs/dev/table/sourcesSinks.md b/docs/content/docs/dev/table/sourcesSinks.md
index d9b780d5e9f..ca0c8f0186b 100644
--- a/docs/content/docs/dev/table/sourcesSinks.md
+++ b/docs/content/docs/dev/table/sourcesSinks.md
@@ -344,6 +344,11 @@ that a sink can still work on common data structures and perform a conversion at
and consuming them to achieve the purpose of row(s) update.
</td>
</tr>
+ <tr>
+ <td>{{< gh_link file="flink-table/flink-table-common/src/main/java/org/apache/flink/table/connector/sink/abilities/SupportsStaging.java" name="SupportsStaging" >}}</td>
+ <td>Enables to support atomic semantic for CTAS(CREATE TABLE AS SELECT) or RTAS([CREATE OR] REPLACE TABLE AS SELECT) in a <code>DynamicTableSink</code>. The table sink is responsible for returning <code>StagedTable</code> object that provides atomic semantics.
+ </td>
+ </tr>
</tbody>
</table>
diff --git a/docs/content/docs/dev/table/sql/create.md b/docs/content/docs/dev/table/sql/create.md
index 13a4d59608d..8dc17cd525d 100644
--- a/docs/content/docs/dev/table/sql/create.md
+++ b/docs/content/docs/dev/table/sql/create.md
@@ -547,14 +547,20 @@ CREATE TABLE my_ctas_table (
INSERT INTO my_ctas_table SELECT id, name, age FROM source_table WHERE mod(id, 10) = 0;
```
-**Note** CTAS has these restrictions:
+**Note:** CTAS has these restrictions:
* Does not support creating a temporary table yet.
* Does not support specifying explicit columns yet.
* Does not support specifying explicit watermark yet.
* Does not support creating partitioned table yet.
* Does not support specifying primary key constraints yet.
-**Note** The target table created by CTAS is non-atomic currently, the table won't be dropped automatically if occur errors while inserting data into the table.
+**Note:** By default, CTAS is non-atomic which means the table created won't be dropped automatically if occur errors while inserting data into the table.
+
+#### Atomicity
+
+If you want to enable atomicity for CTAS, then you should make sure:
+* The sink has implemented the atomicity semantics for CTAS. You may refer to the doc for the corresponding connector sink to know the atomicity semantics is available or not. For devs who want to implement the atomicity semantics, please refer to the doc [SupportsStaging]({{< ref "docs/dev/table/sourcesSinks" >}}#sink-abilities).
+* Set option [table.rtas-ctas.atomicity-enabled]({{< ref "docs/dev/table/config" >}}#table-rtas-ctas-atomicity-enabled) to `true`.
{{< top >}}
@@ -566,7 +572,7 @@ WITH (key1=val1, key2=val2, ...)
AS select_query
```
-**Note** RTAS has the following semantic:
+**Note:** RTAS has the following semantic:
* REPLACE TABLE AS SELECT statement: the target table to be replaced must exist, otherwise, an exception will be thrown.
* CREATE OR REPLACE TABLE AS SELECT statement: the target table to be replaced will be created if it does not exist; if it does exist, it'll be replaced.
@@ -601,7 +607,7 @@ CREATE TABLE my_rtas_table (
INSERT INTO my_rtas_table SELECT id, name, age FROM source_table WHERE mod(id, 10) = 0;
```
-**Note** RTAS has these restrictions:
+**Note:** RTAS has these restrictions:
* Does not support replacing a temporary table yet.
* Does not support specifying explicit columns yet.
@@ -609,6 +615,14 @@ INSERT INTO my_rtas_table SELECT id, name, age FROM source_table WHERE mod(id, 1
* Does not support creating partitioned table yet.
* Does not support specifying primary key constraints yet.
+**Note:** By default, RTAS is non-atomic which means the table won't be dropped or restored to its origin automatically if occur errors while inserting data into the table.
+
+### Atomicity
+
+If you want to enable atomicity for RTAS, then you should make sure:
+* The sink has implemented the atomicity semantics for RTAS. You may refer to the doc for the corresponding connector sink to know the atomicity semantics is available or not. For devs who want to implement the atomicity semantics, please refer to the doc [SupportsStaging]({{< ref "docs/dev/table/sourcesSinks" >}}#sink-abilities).
+* Set option [table.rtas-ctas.atomicity-enabled]({{< ref "docs/dev/table/config" >}}#table-rtas-ctas-atomicity-enabled) to `true`.
+
{{< top >}}
## CREATE CATALOG