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/09/21 01:33:53 UTC
[flink] branch master updated (f2cb1d24728 -> d5e3edb98f1)
This is an automated email from the ASF dual-hosted git repository.
yuxia pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
from f2cb1d24728 [FLINK-32848][tests][JUnit5 migration] Migrate flink-runtime/rpc tests to JUnit5 (#23301)
new 0e97ae3549f [FLINK-33050][table] Atomicity is not supported prompting the user to disable
new d5e3edb98f1 [FLINK-33050][docs] Add notice of data duplicates for RTAS in docs This closes #23389
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
docs/content.zh/docs/dev/table/sql/create.md | 1 +
docs/content/docs/dev/table/sql/create.md | 1 +
.../table/api/internal/TableEnvironmentImpl.java | 36 ++++++++++++++--------
3 files changed, 26 insertions(+), 12 deletions(-)
[flink] 02/02: [FLINK-33050][docs] Add notice of data duplicates for RTAS in docs This closes #23389
Posted by yu...@apache.org.
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
commit d5e3edb98f1f485fdafd8ce1fd88a930af7cf829
Author: Tartarus0zm <zh...@163.com>
AuthorDate: Wed Sep 20 11:55:54 2023 +0800
[FLINK-33050][docs] Add notice of data duplicates for RTAS in docs
This closes #23389
---
docs/content.zh/docs/dev/table/sql/create.md | 1 +
docs/content/docs/dev/table/sql/create.md | 1 +
2 files changed, 2 insertions(+)
diff --git a/docs/content.zh/docs/dev/table/sql/create.md b/docs/content.zh/docs/dev/table/sql/create.md
index 7024b457585..f6c8674b6c8 100644
--- a/docs/content.zh/docs/dev/table/sql/create.md
+++ b/docs/content.zh/docs/dev/table/sql/create.md
@@ -615,6 +615,7 @@ INSERT INTO my_rtas_table SELECT id, name, age FROM source_table WHERE mod(id, 1
* 暂不支持主键约束。
**注意：** 默认情况下，RTAS 是非原子性的，这意味着如果在向表中插入数据时发生错误，该表不会被自动删除或还原成原来的表。
+**注意：** RTAS 会先删除表，然后创建表并写入数据。但如果表是在基于内存的 Catalog 里，删除表只会将其从 Catalog 里移除，并不会移除物理表中的数据。因此，执行RTAS语句之前的数据仍然存在。
### 原子性
diff --git a/docs/content/docs/dev/table/sql/create.md b/docs/content/docs/dev/table/sql/create.md
index 8dc17cd525d..3d94b111c53 100644
--- a/docs/content/docs/dev/table/sql/create.md
+++ b/docs/content/docs/dev/table/sql/create.md
@@ -616,6 +616,7 @@ INSERT INTO my_rtas_table SELECT id, name, age FROM source_table WHERE mod(id, 1
* 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.
+**Note:** RTAS will drop the table first, then create the table and insert the data. But if the table is in the in-memory catalog, dropping table will only remove it from the catalog without removing the data in the physical table. So, the data before executing RTAS statement will still exist.
### Atomicity
[flink] 01/02: [FLINK-33050][table] Atomicity is not supported prompting the user to disable
Posted by yu...@apache.org.
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
commit 0e97ae3549f77578b36ee9d6da97d98fc6e76fa4
Author: Tartarus0zm <zh...@163.com>
AuthorDate: Wed Sep 20 11:55:39 2023 +0800
[FLINK-33050][table] Atomicity is not supported prompting the user to disable
---
.../table/api/internal/TableEnvironmentImpl.java | 36 ++++++++++++++--------
1 file changed, 24 insertions(+), 12 deletions(-)
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java
index 1b9648f6cb6..d25df607fd1 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java
@@ -143,6 +143,7 @@ import static org.apache.flink.table.api.config.TableConfigOptions.TABLE_DML_SYN
*/
@Internal
public class TableEnvironmentImpl implements TableEnvironmentInternal {
+
// Flag that tells if the TableSource/TableSink used in this environment is stream table
// source/sink,
// and this should always be true. This avoids too many hard code.
@@ -951,18 +952,29 @@ public class TableEnvironmentImpl implements TableEnvironmentInternal {
createTableOperation.getTableIdentifier(),
catalogTable,
createTableOperation.isTemporary())) {
- DynamicTableSink dynamicTableSink =
- ExecutableOperationUtils.createDynamicTableSink(
- catalog,
- () -> moduleManager.getFactory((Module::getTableSinkFactory)),
- createTableOperation.getTableIdentifier(),
- catalogTable,
- Collections.emptyMap(),
- tableConfig,
- resourceManager.getUserClassLoader(),
- createTableOperation.isTemporary());
- if (dynamicTableSink instanceof SupportsStaging) {
- return Optional.of(dynamicTableSink);
+ try {
+ DynamicTableSink dynamicTableSink =
+ ExecutableOperationUtils.createDynamicTableSink(
+ catalog,
+ () -> moduleManager.getFactory((Module::getTableSinkFactory)),
+ createTableOperation.getTableIdentifier(),
+ catalogTable,
+ Collections.emptyMap(),
+ tableConfig,
+ resourceManager.getUserClassLoader(),
+ createTableOperation.isTemporary());
+ if (dynamicTableSink instanceof SupportsStaging) {
+ return Optional.of(dynamicTableSink);
+ }
+ } catch (Exception e) {
+ throw new TableException(
+ String.format(
+ "Fail to create DynamicTableSink for the table %s, "
+ + "maybe the table does not support atomicity of CTAS/RTAS, "
+ + "please set %s to false and try again.",
+ createTableOperation.getTableIdentifier(),
+ TableConfigOptions.TABLE_RTAS_CTAS_ATOMICITY_ENABLED.key()),
+ e);
}
}
}