luoyuxia commented on code in PR #22995:
URL: https://github.com/apache/flink/pull/22995#discussion_r1263660087
##########
docs/layouts/shortcodes/generated/table_config_configuration.html:
##########
@@ -21,10 +21,10 @@
<td>The name of the default database in the initial catalog to be
created when instantiating TableEnvironment.</td>
</tr>
<tr>
- <td><h5>table.ctas.atomicity-enabled</h5><br> <span class="label
label-primary">Batch</span> <span class="label
label-primary">Streaming</span></td>
+ <td><h5>table.rtas-ctas.atomicity-enabled</h5><br> <span
class="label label-primary">Batch</span> <span class="label
label-primary">Streaming</span></td>
<td style="word-wrap: break-word;">false</td>
<td>Boolean</td>
- <td>Specifies if the CREATE TABLE AS SELECT statement is executed
atomically. By default, the statement is non-atomic. The target table is
created on the client side, and it will not be dropped even though the job
fails or is canceled. If set this option to true and the underlying
DynamicTableSink implements the SupportsStaging interface, the statement is
expected to be executed atomically, the behavior of which depends on the actual
DynamicTableSink.</td>
+ <td>Specifies if the CREATE TABLE/REPLACE TABLE/CREATE OR REPLACE
AS SELECT statement is executed atomically. By default, the statement is
non-atomic. The target table is created/replaced on the client side, and it
will not be rollback even though the job fails or is canceled. If set this
option to true and the underlying DynamicTableSink implements the
SupportsStaging interface, the statement is expected to be executed atomically,
the behavior of which depends on the actual DynamicTableSink.</td>
Review Comment:
```suggestion
<td>Specifies if the CREATE TABLE/REPLACE TABLE/CREATE OR
REPLACE AS SELECT statement is executed atomically. By default, the statement
is non-atomic. The target table is created/replaced on the client side, and it
will not be rollbacked even though the job fails or is canceled. If set this
option to true and the underlying DynamicTableSink implements the
SupportsStaging interface, the statement is expected to be executed atomically,
the behavior of which depends on the actual DynamicTableSink.</td>
```
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java:
##########
@@ -833,10 +832,44 @@ public TableResultInternal
executeInternal(List<ModifyOperation> operations) {
return executeInternal(transformations, sinkIdentifierNames,
jobStatusHookList);
}
- private ModifyOperation getOperation(ReplaceTableAsOperation
rtasOperation) {
- // rtas drop table first, then create
+ private ModifyOperation getModifyOperation(
+ ReplaceTableAsOperation rtasOperation, List<JobStatusHook>
jobStatusHookList) {
CreateTableOperation createTableOperation =
rtasOperation.getCreateTableOperation();
Review Comment:
Get warning from my IDE
`Duplicated code fragment (7 lines long) `
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java:
##########
@@ -833,10 +832,44 @@ public TableResultInternal
executeInternal(List<ModifyOperation> operations) {
return executeInternal(transformations, sinkIdentifierNames,
jobStatusHookList);
}
- private ModifyOperation getOperation(ReplaceTableAsOperation
rtasOperation) {
- // rtas drop table first, then create
+ private ModifyOperation getModifyOperation(
+ ReplaceTableAsOperation rtasOperation, List<JobStatusHook>
jobStatusHookList) {
CreateTableOperation createTableOperation =
rtasOperation.getCreateTableOperation();
ObjectIdentifier tableIdentifier =
createTableOperation.getTableIdentifier();
+ Catalog catalog =
catalogManager.getCatalog(tableIdentifier.getCatalogName()).orElse(null);
+ ResolvedCatalogTable catalogTable =
+
catalogManager.resolveCatalogTable(createTableOperation.getCatalogTable());
+ Optional<DynamicTableSink> stagingDynamicTableSink =
+ getSupportsStagingDynamicTableSink(createTableOperation,
catalog, catalogTable);
+ if (stagingDynamicTableSink.isPresent()) {
+ // use atomic rtas
+ DynamicTableSink dynamicTableSink = stagingDynamicTableSink.get();
+ if (!rtasOperation.isCreateOrReplace()) {
+ Optional<ContextResolvedTable> driverTable =
Review Comment:
Why name `driverTable`? It really confused me.
Use `oldTable`?
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java:
##########
@@ -833,10 +832,44 @@ public TableResultInternal
executeInternal(List<ModifyOperation> operations) {
return executeInternal(transformations, sinkIdentifierNames,
jobStatusHookList);
}
- private ModifyOperation getOperation(ReplaceTableAsOperation
rtasOperation) {
- // rtas drop table first, then create
+ private ModifyOperation getModifyOperation(
+ ReplaceTableAsOperation rtasOperation, List<JobStatusHook>
jobStatusHookList) {
CreateTableOperation createTableOperation =
rtasOperation.getCreateTableOperation();
ObjectIdentifier tableIdentifier =
createTableOperation.getTableIdentifier();
+ Catalog catalog =
catalogManager.getCatalog(tableIdentifier.getCatalogName()).orElse(null);
+ ResolvedCatalogTable catalogTable =
+
catalogManager.resolveCatalogTable(createTableOperation.getCatalogTable());
+ Optional<DynamicTableSink> stagingDynamicTableSink =
+ getSupportsStagingDynamicTableSink(createTableOperation,
catalog, catalogTable);
+ if (stagingDynamicTableSink.isPresent()) {
+ // use atomic rtas
+ DynamicTableSink dynamicTableSink = stagingDynamicTableSink.get();
+ if (!rtasOperation.isCreateOrReplace()) {
Review Comment:
I'm wondering what's the behavior for staging replace in spark.
Is it also will check the table exists or not? And if not, will it also
throw an exception?
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/utils/AtomicRtasITCaseBase.java:
##########
@@ -0,0 +1,223 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.table.planner.runtime.utils;
+
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.api.config.TableConfigOptions;
+import org.apache.flink.table.connector.sink.abilities.SupportsStaging;
+import
org.apache.flink.table.planner.factories.TestSupportsStagingTableFactory;
+import
org.apache.flink.table.planner.factories.utils.TestCollectionTableFactory;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.FileUtils;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.Collections;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** The base case of atomic rtas ITCase. */
+public abstract class AtomicRtasITCaseBase extends TestLogger {
+
+ protected TableEnvironment tEnv;
+
+ protected abstract TableEnvironment getTableEnvironment();
+
+ @BeforeEach
+ void setup() {
+ tEnv = getTableEnvironment();
+ List<Row> sourceData = Collections.singletonList(Row.of(1, "ZM"));
+
+ TestCollectionTableFactory.reset();
+ TestCollectionTableFactory.initData(sourceData);
+
+ String sourceDDL = "create table t1(a int, b varchar) with
('connector' = 'COLLECTION')";
+ tEnv.executeSql(sourceDDL);
+ // clean data
+ TestSupportsStagingTableFactory.JOB_STATUS_CHANGE_PROCESS.clear();
+ TestSupportsStagingTableFactory.STAGING_PURPOSE_LIST.clear();
+ }
+
+ @Test
+ void testAtomicReplaceTableAs(@TempDir Path temporaryFolder) throws
Exception {
+ commonTestForAtomicReplaceTableAs(
+ "atomic_replace_table", false, true, temporaryFolder.toFile());
+ }
+
+ @Test
+ void testAtomicReplaceTableAsWithDriverTableNotExists(@TempDir Path
temporaryFolder)
+ throws Exception {
+ commonTestForAtomicReplaceTableAs(
+ "atomic_replace_table_not_exists", false, false,
temporaryFolder.toFile());
+ }
+
+ @Test
+ void testAtomicCreateOrReplaceTableAs(@TempDir Path temporaryFolder)
throws Exception {
+ commonTestForAtomicReplaceTableAs(
+ "atomic_create_or_replace_table", true, true,
temporaryFolder.toFile());
+ }
+
+ @Test
+ void testAtomicCreateOrReplaceTableAsWithDriverTableNotExists(@TempDir
Path temporaryFolder)
+ throws Exception {
+ commonTestForAtomicReplaceTableAs(
+ "atomic_create_or_replace_table_not_exists", true, false,
temporaryFolder.toFile());
+ }
+
+ private void commonTestForAtomicReplaceTableAs(
+ String tableName,
+ boolean isCreateOrReplace,
+ boolean createDriverTable,
+ File tmpDataFolder)
+ throws Exception {
+ if (createDriverTable) {
+ tEnv.executeSql("create table " + tableName + " (a int) with
('connector' = 'PRINT')");
+ }
+
+
tEnv.getConfig().set(TableConfigOptions.TABLE_RTAS_CTAS_ATOMICITY_ENABLED,
true);
+ String dataDir = tmpDataFolder.getAbsolutePath();
+ String sqlFragment =
Review Comment:
nit:
I think the code lines
` String sqlFragment =
isCreateOrReplace
? " create or replace table " + tableName
: " replace table " + tableName;`
can be extracted to a method.
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java:
##########
@@ -833,10 +832,44 @@ public TableResultInternal
executeInternal(List<ModifyOperation> operations) {
return executeInternal(transformations, sinkIdentifierNames,
jobStatusHookList);
}
- private ModifyOperation getOperation(ReplaceTableAsOperation
rtasOperation) {
- // rtas drop table first, then create
+ private ModifyOperation getModifyOperation(
+ ReplaceTableAsOperation rtasOperation, List<JobStatusHook>
jobStatusHookList) {
CreateTableOperation createTableOperation =
rtasOperation.getCreateTableOperation();
ObjectIdentifier tableIdentifier =
createTableOperation.getTableIdentifier();
+ Catalog catalog =
catalogManager.getCatalog(tableIdentifier.getCatalogName()).orElse(null);
+ ResolvedCatalogTable catalogTable =
+
catalogManager.resolveCatalogTable(createTableOperation.getCatalogTable());
+ Optional<DynamicTableSink> stagingDynamicTableSink =
+ getSupportsStagingDynamicTableSink(createTableOperation,
catalog, catalogTable);
+ if (stagingDynamicTableSink.isPresent()) {
+ // use atomic rtas
+ DynamicTableSink dynamicTableSink = stagingDynamicTableSink.get();
+ if (!rtasOperation.isCreateOrReplace()) {
+ Optional<ContextResolvedTable> driverTable =
+ catalogManager.getTable(tableIdentifier);
+ if (!driverTable.isPresent()) {
+ throw new TableException(
+ String.format(
+ "The table %s to be replaced doesn't
exist. "
+ + "You can try to use CREATE TABLE
AS statement or "
+ + "CREATE OR REPLACE TABLE AS
statement.",
+ tableIdentifier));
+ }
+ }
+ SupportsStaging.StagingPurpose stagingPurpose =
+ rtasOperation.isCreateOrReplace()
+ ?
SupportsStaging.StagingPurpose.CREATE_OR_REPLACE_TABLE_AS
+ : SupportsStaging.StagingPurpose.REPLACE_TABLE_AS;
+
+ StagedTable stagedTable =
+ ((SupportsStaging) dynamicTableSink)
+ .applyStaging(new
SinkStagingContext(stagingPurpose));
+ AtomicJobStatusHook atomicJobStatusHook = new
AtomicJobStatusHook(stagedTable);
+ jobStatusHookList.add(atomicJobStatusHook);
+ return rtasOperation.toStagedSinkModifyOperation(
+ tableIdentifier, catalogTable, catalog, dynamicTableSink);
+ }
+ // non-atomic rtas drop table first, then create
try {
catalogManager.dropTable(tableIdentifier,
rtasOperation.isCreateOrReplace());
} catch (ValidationException e) {
Review Comment:
Just notice we will use error message to check whether the table to be
dropped exists or not. For me, it's hacky and when the error message changes,
the check may well fail.
Can we first get the table and then drop?
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/execution/AtomicJobStatusHook.java:
##########
@@ -23,14 +23,15 @@
import org.apache.flink.table.catalog.StagedTable;
/**
- * This hook is used to implement atomic semantics for CTAS(CREATE TABLE AS
SELECT) statement. It'll
- * call the corresponding interfaces of the inner {@link StagedTable} on job
status changes.
+ * This hook is used to implement atomic semantics for CTAS(CREATE TABLE AS
SELECT) or RTAS([CREATE
+ * OR] REPLACE TABLE AS SELECT) statement. It'll call the corresponding
interfaces of the inner
+ * {@link StagedTable} on job status changes.
*/
-public class CtasJobStatusHook implements JobStatusHook {
+public class AtomicJobStatusHook implements JobStatusHook {
Review Comment:
StagingSinkJobStatusHook?
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/config/TableConfigOptions.java:
##########
@@ -198,13 +198,13 @@ private TableConfigOptions() {}
"Local directory that is used by planner for
storing downloaded resources.");
@Documentation.TableOption(execMode =
Documentation.ExecMode.BATCH_STREAMING)
- public static final ConfigOption<Boolean> TABLE_CTAS_ATOMICITY_ENABLED =
- key("table.ctas.atomicity-enabled")
+ public static final ConfigOption<Boolean>
TABLE_RTAS_CTAS_ATOMICITY_ENABLED =
+ key("table.rtas-ctas.atomicity-enabled")
.booleanType()
.defaultValue(false)
.withDescription(
- "Specifies if the CREATE TABLE AS SELECT statement
is executed atomically. By default, the statement is non-atomic. "
- + "The target table is created on the
client side, and it will not be dropped even though the job fails or is
canceled. "
+ "Specifies if the CREATE TABLE/REPLACE
TABLE/CREATE OR REPLACE AS SELECT statement is executed atomically. By default,
the statement is non-atomic. "
+ + "The target table is created/replaced on
the client side, and it will not be rollback even though the job fails or is
canceled. "
Review Comment:
dito
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/utils/AtomicRtasITCaseBase.java:
##########
@@ -0,0 +1,223 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.table.planner.runtime.utils;
+
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.api.config.TableConfigOptions;
+import org.apache.flink.table.connector.sink.abilities.SupportsStaging;
+import
org.apache.flink.table.planner.factories.TestSupportsStagingTableFactory;
+import
org.apache.flink.table.planner.factories.utils.TestCollectionTableFactory;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.FileUtils;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.Collections;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** The base case of atomic rtas ITCase. */
+public abstract class AtomicRtasITCaseBase extends TestLogger {
+
+ protected TableEnvironment tEnv;
+
+ protected abstract TableEnvironment getTableEnvironment();
+
+ @BeforeEach
+ void setup() {
+ tEnv = getTableEnvironment();
+ List<Row> sourceData = Collections.singletonList(Row.of(1, "ZM"));
+
+ TestCollectionTableFactory.reset();
+ TestCollectionTableFactory.initData(sourceData);
+
+ String sourceDDL = "create table t1(a int, b varchar) with
('connector' = 'COLLECTION')";
+ tEnv.executeSql(sourceDDL);
+ // clean data
+ TestSupportsStagingTableFactory.JOB_STATUS_CHANGE_PROCESS.clear();
+ TestSupportsStagingTableFactory.STAGING_PURPOSE_LIST.clear();
+ }
+
+ @Test
+ void testAtomicReplaceTableAs(@TempDir Path temporaryFolder) throws
Exception {
+ commonTestForAtomicReplaceTableAs(
+ "atomic_replace_table", false, true, temporaryFolder.toFile());
+ }
+
+ @Test
+ void testAtomicReplaceTableAsWithDriverTableNotExists(@TempDir Path
temporaryFolder)
Review Comment:
I'm really confused about the naming.
Why is `DriverTable`? I mean, why name driver?
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/utils/AtomicRtasITCaseBase.java:
##########
@@ -0,0 +1,223 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.table.planner.runtime.utils;
+
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.api.config.TableConfigOptions;
+import org.apache.flink.table.connector.sink.abilities.SupportsStaging;
+import
org.apache.flink.table.planner.factories.TestSupportsStagingTableFactory;
+import
org.apache.flink.table.planner.factories.utils.TestCollectionTableFactory;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.FileUtils;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.Collections;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** The base case of atomic rtas ITCase. */
+public abstract class AtomicRtasITCaseBase extends TestLogger {
+
+ protected TableEnvironment tEnv;
+
+ protected abstract TableEnvironment getTableEnvironment();
+
+ @BeforeEach
+ void setup() {
+ tEnv = getTableEnvironment();
+ List<Row> sourceData = Collections.singletonList(Row.of(1, "ZM"));
+
+ TestCollectionTableFactory.reset();
+ TestCollectionTableFactory.initData(sourceData);
+
+ String sourceDDL = "create table t1(a int, b varchar) with
('connector' = 'COLLECTION')";
+ tEnv.executeSql(sourceDDL);
+ // clean data
+ TestSupportsStagingTableFactory.JOB_STATUS_CHANGE_PROCESS.clear();
+ TestSupportsStagingTableFactory.STAGING_PURPOSE_LIST.clear();
+ }
+
+ @Test
+ void testAtomicReplaceTableAs(@TempDir Path temporaryFolder) throws
Exception {
+ commonTestForAtomicReplaceTableAs(
+ "atomic_replace_table", false, true, temporaryFolder.toFile());
+ }
+
+ @Test
+ void testAtomicReplaceTableAsWithDriverTableNotExists(@TempDir Path
temporaryFolder)
Review Comment:
If no strong reason for `DriverTable`, I would like to name it as
`isCreateReplacedTable`
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/utils/AtomicRtasITCaseBase.java:
##########
@@ -0,0 +1,223 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.table.planner.runtime.utils;
+
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.api.config.TableConfigOptions;
+import org.apache.flink.table.connector.sink.abilities.SupportsStaging;
+import
org.apache.flink.table.planner.factories.TestSupportsStagingTableFactory;
+import
org.apache.flink.table.planner.factories.utils.TestCollectionTableFactory;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.FileUtils;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.Collections;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** The base case of atomic rtas ITCase. */
+public abstract class AtomicRtasITCaseBase extends TestLogger {
+
+ protected TableEnvironment tEnv;
+
+ protected abstract TableEnvironment getTableEnvironment();
+
+ @BeforeEach
+ void setup() {
+ tEnv = getTableEnvironment();
+ List<Row> sourceData = Collections.singletonList(Row.of(1, "ZM"));
+
+ TestCollectionTableFactory.reset();
+ TestCollectionTableFactory.initData(sourceData);
+
+ String sourceDDL = "create table t1(a int, b varchar) with
('connector' = 'COLLECTION')";
+ tEnv.executeSql(sourceDDL);
+ // clean data
Review Comment:
nit:
for clean, I would like to have a @AfterEach method.
`AtomicRtasITCaseBase` may also need to update.
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/utils/AtomicRtasITCaseBase.java:
##########
@@ -0,0 +1,223 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.table.planner.runtime.utils;
+
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.api.config.TableConfigOptions;
+import org.apache.flink.table.connector.sink.abilities.SupportsStaging;
+import
org.apache.flink.table.planner.factories.TestSupportsStagingTableFactory;
+import
org.apache.flink.table.planner.factories.utils.TestCollectionTableFactory;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.FileUtils;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.Collections;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** The base case of atomic rtas ITCase. */
+public abstract class AtomicRtasITCaseBase extends TestLogger {
+
+ protected TableEnvironment tEnv;
+
+ protected abstract TableEnvironment getTableEnvironment();
+
+ @BeforeEach
+ void setup() {
+ tEnv = getTableEnvironment();
+ List<Row> sourceData = Collections.singletonList(Row.of(1, "ZM"));
+
+ TestCollectionTableFactory.reset();
+ TestCollectionTableFactory.initData(sourceData);
+
+ String sourceDDL = "create table t1(a int, b varchar) with
('connector' = 'COLLECTION')";
+ tEnv.executeSql(sourceDDL);
+ // clean data
+ TestSupportsStagingTableFactory.JOB_STATUS_CHANGE_PROCESS.clear();
+ TestSupportsStagingTableFactory.STAGING_PURPOSE_LIST.clear();
+ }
+
+ @Test
+ void testAtomicReplaceTableAs(@TempDir Path temporaryFolder) throws
Exception {
+ commonTestForAtomicReplaceTableAs(
+ "atomic_replace_table", false, true, temporaryFolder.toFile());
+ }
+
+ @Test
+ void testAtomicReplaceTableAsWithDriverTableNotExists(@TempDir Path
temporaryFolder)
+ throws Exception {
+ commonTestForAtomicReplaceTableAs(
+ "atomic_replace_table_not_exists", false, false,
temporaryFolder.toFile());
+ }
+
+ @Test
+ void testAtomicCreateOrReplaceTableAs(@TempDir Path temporaryFolder)
throws Exception {
+ commonTestForAtomicReplaceTableAs(
+ "atomic_create_or_replace_table", true, true,
temporaryFolder.toFile());
+ }
+
+ @Test
+ void testAtomicCreateOrReplaceTableAsWithDriverTableNotExists(@TempDir
Path temporaryFolder)
+ throws Exception {
+ commonTestForAtomicReplaceTableAs(
+ "atomic_create_or_replace_table_not_exists", true, false,
temporaryFolder.toFile());
+ }
+
+ private void commonTestForAtomicReplaceTableAs(
+ String tableName,
+ boolean isCreateOrReplace,
+ boolean createDriverTable,
+ File tmpDataFolder)
+ throws Exception {
+ if (createDriverTable) {
+ tEnv.executeSql("create table " + tableName + " (a int) with
('connector' = 'PRINT')");
+ }
+
+
tEnv.getConfig().set(TableConfigOptions.TABLE_RTAS_CTAS_ATOMICITY_ENABLED,
true);
+ String dataDir = tmpDataFolder.getAbsolutePath();
+ String sqlFragment =
+ isCreateOrReplace
+ ? " create or replace table " + tableName
+ : " replace table " + tableName;
+ String sql =
+ sqlFragment
+ + " with ('connector' = 'test-staging', 'data-dir' = '"
+ + dataDir
+ + "') as select * from t1";
+ if (!isCreateOrReplace && !createDriverTable) {
+ assertThatThrownBy(() -> tEnv.executeSql(sql))
+ .isInstanceOf(TableException.class)
+ .hasMessage(
+ "The table `default_catalog`.`default_database`.`"
+ + tableName
+ + "` to be replaced doesn't exist."
+ + " You can try to use CREATE TABLE AS
statement or CREATE OR REPLACE TABLE AS statement.");
+ } else {
+ tEnv.executeSql(sql).await();
+ if (createDriverTable) {
+ assertThat(tEnv.listTables()).contains(tableName);
+ } else {
+ assertThat(tEnv.listTables()).doesNotContain(tableName);
+ }
+ verifyDataFile(dataDir, "data");
+
assertThat(TestSupportsStagingTableFactory.JOB_STATUS_CHANGE_PROCESS).hasSize(2);
+
assertThat(TestSupportsStagingTableFactory.JOB_STATUS_CHANGE_PROCESS)
+ .contains("begin", "commit");
+
assertThat(TestSupportsStagingTableFactory.STAGING_PURPOSE_LIST).hasSize(1);
+ if (isCreateOrReplace) {
+
assertThat(TestSupportsStagingTableFactory.STAGING_PURPOSE_LIST)
+
.contains(SupportsStaging.StagingPurpose.CREATE_OR_REPLACE_TABLE_AS);
+ } else {
+
assertThat(TestSupportsStagingTableFactory.STAGING_PURPOSE_LIST)
+
.contains(SupportsStaging.StagingPurpose.REPLACE_TABLE_AS);
+ }
+ }
+ }
+
+ @Test
+ void testAtomicReplaceTableAsWithException(@TempDir Path temporaryFolder)
throws Exception {
+ commonTestForAtomicReplaceTableAsWithException(
+ "atomic_replace_table_fail", false, temporaryFolder.toFile());
+ }
+
+ @Test
+ void testAtomicCreateOrReplaceTableAsWithException(@TempDir Path
temporaryFolder)
+ throws Exception {
+ commonTestForAtomicReplaceTableAsWithException(
+ "atomic_create_or_replace_table_fail", false,
temporaryFolder.toFile());
+ }
+
+ private void commonTestForAtomicReplaceTableAsWithException(
+ String tableName, boolean isCreateOrReplace, File tmpDataFolder) {
Review Comment:
get warning from my IDE:
`Actual value of parameter 'isCreateOrReplace' is always 'false' `
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
