You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "Tartarus0zm (via GitHub)" <gi...@apache.org> on 2023/06/21 11:58:34 UTC
[GitHub] [flink] Tartarus0zm opened a new pull request, #22839: [FLINK-32349][table] Support atomic for CREATE TABLE AS SELECT(CTAS) statement
Tartarus0zm opened a new pull request, #22839:
URL: https://github.com/apache/flink/pull/22839
## What is the purpose of the change
Support atomic for CREATE TABLE AS SELECT(CTAS) statement
## Brief change log
Introduce StagedTable interface that support atomic operations.
## Verifying this change
StagedTableITCase#testStagedTableWithAtomicCtas
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (no)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (yes)
- The serializers: (no)
- The runtime per-record code paths (performance sensitive): (no)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (no)
- The S3 file system connector: (no)
## Documentation
- Does this pull request introduce a new feature? (yes)
- If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] luoyuxia commented on a diff in pull request #22839: [FLINK-32349][table] Support atomic for CREATE TABLE AS SELECT(CTAS) statement
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #22839:
URL: https://github.com/apache/flink/pull/22839#discussion_r1246593727
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestSupportsStagingTableFactory.java:
##########
@@ -0,0 +1,191 @@
+/*
+ * 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.factories;
+
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.DataStreamSink;
+import org.apache.flink.streaming.api.functions.sink.DiscardingSink;
+import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
+import org.apache.flink.table.catalog.StagedTable;
+import org.apache.flink.table.connector.ChangelogMode;
+import org.apache.flink.table.connector.ProviderContext;
+import org.apache.flink.table.connector.sink.DataStreamSinkProvider;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.connector.sink.abilities.SupportsStaging;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.factories.DynamicTableSinkFactory;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.util.FileUtils;
+
+import java.io.File;
+import java.util.Collections;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Set;
+
+/** A factory to create table to support staging for test purpose. */
+public class TestSupportsStagingTableFactory implements DynamicTableSinkFactory {
+
+ public static final String IDENTIFIER = "test-staging";
+
+ public static final List<String> JOB_STATUS_CHANGE_PROCESS = new LinkedList<>();
+
+ private static final ConfigOption<String> DATA_DIR =
+ ConfigOptions.key("data-dir")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("The data id used to write the rows.");
+
+ @Override
+ public DynamicTableSink createDynamicTableSink(Context context) {
+ FactoryUtil.TableFactoryHelper helper = FactoryUtil.createTableFactoryHelper(this, context);
+ helper.validate();
+ String dataDir = helper.getOptions().get(DATA_DIR);
+ return new SupportsStagingTableSink(dataDir);
+ }
+
+ @Override
+ public String factoryIdentifier() {
+ return IDENTIFIER;
+ }
+
+ @Override
+ public Set<ConfigOption<?>> requiredOptions() {
+ return Collections.singleton(DATA_DIR);
+ }
+
+ @Override
+ public Set<ConfigOption<?>> optionalOptions() {
+ return Collections.emptySet();
+ }
+
+ /** A sink that supports staging. */
+ private static class SupportsStagingTableSink implements DynamicTableSink, SupportsStaging {
+
+ private String dataDir;
+ private TestStagedTable stagedTable;
+
+ public SupportsStagingTableSink(String dataDir) {
+ this(dataDir, null);
+ }
+
+ public SupportsStagingTableSink(String dataDir, TestStagedTable stagedTable) {
+ this.dataDir = dataDir;
+ this.stagedTable = stagedTable;
+ }
+
+ @Override
+ public ChangelogMode getChangelogMode(ChangelogMode requestedMode) {
+ return ChangelogMode.insertOnly();
+ }
+
+ @Override
+ public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
+ return new DataStreamSinkProvider() {
+ @Override
+ public DataStreamSink<?> consumeDataStream(
+ ProviderContext providerContext, DataStream<RowData> dataStream) {
+ if (stagedTable != null) {
+ return dataStream
+ .addSink(new StagedSinkFunction(dataDir))
+ .setParallelism(1);
+ } else {
+ // otherwise, do nothing
+ return dataStream.addSink(new DiscardingSink<>());
+ }
+ }
+ };
+ }
+
+ @Override
+ public DynamicTableSink copy() {
+ return new SupportsStagingTableSink(dataDir, stagedTable);
+ }
+
+ @Override
+ public String asSummaryString() {
+ return "SupportsStagingTableSink";
+ }
+
+ @Override
+ public StagedTable applyStaging(StagingContext context) {
+ JOB_STATUS_CHANGE_PROCESS.clear();
Review Comment:
Don't forget to check the `StagingContext#stagingPurpose`.
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/config/TableConfigOptions.java:
##########
@@ -197,6 +197,17 @@ private TableConfigOptions() {}
.withDescription(
"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")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
Review Comment:
suggestion:
```
"Specifies if the CREATE TABLE AS SELECT statement is executed atomically. "
+ "By default, the statement is non-atomic. The target table is created in client side, and it will not be dropped even though the job fails or is cancelled. "
+ "If set this option to true and DynamicTableSink implements the SupportsStaging interface, the statement is expected to be executed atomically, "
+ "the behavior of which depends on the actual DynamicTableSink."
```
Please also don't forget to update `table_config_configuration.html`
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java:
##########
@@ -819,6 +828,57 @@ public TableResultInternal executeInternal(List<ModifyOperation> operations) {
return executeInternal(transformations, sinkIdentifierNames);
}
+ private void executeCreateTableASOperation(
+ CreateTableASOperation ctasOperation, List<ModifyOperation> mapOperations) {
+ CreateTableOperation createTableOperation = ctasOperation.getCreateTableOperation();
+ ObjectIdentifier tableIdentifier = createTableOperation.getTableIdentifier();
+ Catalog catalog = catalogManager.getCatalog(tableIdentifier.getCatalogName()).orElse(null);
+ ResolvedCatalogTable catalogTable =
+ catalogManager.resolveCatalogTable(createTableOperation.getCatalogTable());
+
+ if (!TableFactoryUtil.isLegacyConnectorOptions(
+ catalog,
+ tableConfig,
+ isStreamingMode,
+ tableIdentifier,
+ catalogTable,
+ createTableOperation.isTemporary())) {
+ DynamicTableSink dynamicTableSink =
+ ExecutableOperationUtils.createDynamicTableSink(
+ catalog,
+ () -> moduleManager.getFactory((Module::getTableSinkFactory)),
+ tableIdentifier,
+ catalogTable,
+ Collections.emptyMap(),
+ tableConfig,
+ resourceManager.getUserClassLoader(),
+ createTableOperation.isTemporary());
+ if (dynamicTableSink instanceof SupportsStaging
+ && tableConfig.get(TableConfigOptions.TABLE_CTAS_ATOMICITY_ENABLED)) {
+ // use atomic ctas
+ SupportsStaging.StagingPurpose stagingPurpose =
+ createTableOperation.isIgnoreIfExists()
+ ? SupportsStaging.StagingPurpose.CREATE_TABLE_AS_IF_NOT_EXISTS
+ : SupportsStaging.StagingPurpose.CREATE_TABLE_AS;
+ StagedTable stagedTable =
+ ((SupportsStaging) dynamicTableSink)
+ .applyStaging(new SinkStagingContext(stagingPurpose));
+ CtasJobStatusHook ctasJobStatusHook = new CtasJobStatusHook(stagedTable);
+ mapOperations.add(
+ ctasOperation.toStagedSinkModifyOperation(
+ createTableOperation.getTableIdentifier(),
+ catalogTable,
+ catalog,
+ dynamicTableSink));
+ jobStatusHookList.add(ctasJobStatusHook);
Review Comment:
I'm concern about making `jobStatusHookList` as a variable of class and then clear the `jobStatusHookList` in `createPipeline`. For me, it's too obscure and hard to understand.
Could it be some thing like:
```
jobStatusHookList = new ArayList();
// the following method will add jobStatusHook to `jobStatusHookList`
getOperation(ctasOperation, jobStatusHookList);
executeInternal(transformations, sinkIdentifierNames, jobStatusHookList );
```
Of course we then need a new method `executeInternal` that accepts `jobStatusHookList`.
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/StagedTable.java:
##########
@@ -0,0 +1,64 @@
+/*
+ * 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.catalog;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.connector.sink.abilities.SupportsStaging;
+
+import java.io.Serializable;
+
Review Comment:
suggesttion
```
/**
* The {@link StagedTable} is designed to implement atomic semantic using a two-phase commit
* protocol. The {@link StagedTable} is supposed to be returned via method {@link
* SupportsStaging#applyStaging} by the {@link DynamicTableSink} which implements the {@link
* SupportsStaging} interface.
*
* <p>When the Flink job for writing to a {@link DynamicTableSink} is CREATED, the {@link
* StagedTable#begin()} will be called; when the Flink job is FINISHED, the {@link
* StagedTable#commit()} will be called; when the Flink job is FAILED or CANCELED, the {@link
* StagedTable#abort()} will be called;
*
* <p>See more in {@link SupportsStaging}.
*/
```
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestSupportsStagingTableFactory.java:
##########
@@ -0,0 +1,191 @@
+/*
+ * 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.factories;
+
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.DataStreamSink;
+import org.apache.flink.streaming.api.functions.sink.DiscardingSink;
+import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
+import org.apache.flink.table.catalog.StagedTable;
+import org.apache.flink.table.connector.ChangelogMode;
+import org.apache.flink.table.connector.ProviderContext;
+import org.apache.flink.table.connector.sink.DataStreamSinkProvider;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.connector.sink.abilities.SupportsStaging;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.factories.DynamicTableSinkFactory;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.util.FileUtils;
+
+import java.io.File;
+import java.util.Collections;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Set;
+
+/** A factory to create table to support staging for test purpose. */
+public class TestSupportsStagingTableFactory implements DynamicTableSinkFactory {
+
+ public static final String IDENTIFIER = "test-staging";
+
+ public static final List<String> JOB_STATUS_CHANGE_PROCESS = new LinkedList<>();
+
+ private static final ConfigOption<String> DATA_DIR =
+ ConfigOptions.key("data-dir")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("The data id used to write the rows.");
+
+ @Override
+ public DynamicTableSink createDynamicTableSink(Context context) {
+ FactoryUtil.TableFactoryHelper helper = FactoryUtil.createTableFactoryHelper(this, context);
+ helper.validate();
+ String dataDir = helper.getOptions().get(DATA_DIR);
+ return new SupportsStagingTableSink(dataDir);
+ }
+
+ @Override
+ public String factoryIdentifier() {
+ return IDENTIFIER;
+ }
+
+ @Override
+ public Set<ConfigOption<?>> requiredOptions() {
+ return Collections.singleton(DATA_DIR);
+ }
+
+ @Override
+ public Set<ConfigOption<?>> optionalOptions() {
+ return Collections.emptySet();
+ }
+
+ /** A sink that supports staging. */
+ private static class SupportsStagingTableSink implements DynamicTableSink, SupportsStaging {
+
+ private String dataDir;
Review Comment:
nit:
```
private finnal String dataDir;
```
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/StagedTableITCase.java:
##########
@@ -0,0 +1,74 @@
+/*
+ * 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.batch.sql;
+
+import org.apache.flink.table.api.config.TableConfigOptions;
+import org.apache.flink.table.planner.factories.TestSupportsStagingTableFactory;
+import org.apache.flink.table.planner.factories.utils.TestCollectionTableFactory;
+import org.apache.flink.table.planner.runtime.utils.BatchTestBase;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.FileUtils;
+
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.File;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests staged table in batch mode. */
+public class StagedTableITCase extends BatchTestBase {
+
+ @Rule public final TemporaryFolder tempFolder = new TemporaryFolder();
+
+ @Before
+ @Override
+ public void before() throws Exception {
+ super.before();
+ List<Row> sourceData = Arrays.asList(Row.of(1, "ZM"));
+
+ TestCollectionTableFactory.reset();
+ TestCollectionTableFactory.initData(sourceData);
+
+ String sourceDDL = "create table t1(a int, b varchar) with ('connector' = 'COLLECTION')";
+ tEnv().executeSql(sourceDDL);
+ }
+
+ @Test
Review Comment:
use `org.junit.jupiter.api.Test`
##########
flink-python/src/main/java/org/apache/flink/table/executor/python/ChainingOptimizingExecutor.java:
##########
@@ -63,6 +66,24 @@ public Pipeline createPipeline(
return executor.createPipeline(chainedTransformations, configuration, defaultJobName);
Review Comment:
Please don't foget this comment; I mean the method
```
createPipeline(
List<Transformation<?>> transformations,
ReadableConfig configuration,
String defaultJobName)
```
can just call method
```
return createPipeline(transformations, configuration, defaultJobName, Collections.emptyList());
```
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java:
##########
@@ -784,11 +795,9 @@ public CompiledPlan compilePlan(List<ModifyOperation> operations) {
public TableResultInternal executeInternal(List<ModifyOperation> operations) {
List<ModifyOperation> mapOperations = new ArrayList<>();
for (ModifyOperation modify : operations) {
- // execute CREATE TABLE first for CTAS statements
if (modify instanceof CreateTableASOperation) {
CreateTableASOperation ctasOperation = (CreateTableASOperation) modify;
- executeInternal(ctasOperation.getCreateTableOperation());
- mapOperations.add(ctasOperation.toSinkModifyOperation(catalogManager));
+ executeCreateTableASOperation(ctasOperation, mapOperations);
Review Comment:
The method name make me confused. It doesn't really do the execution. Rename to some thing like
`getOperation(CreateTableASOperation ctasOperation)`? Then add the got operation to `mapOperations`?
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/StagedSinkModifyOperation.java:
##########
@@ -0,0 +1,87 @@
+/*
+ * 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.operations;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.catalog.ContextResolvedTable;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+
+import javax.annotation.Nullable;
+
+import java.util.Map;
+
+/**
+ * DML operation that tells to write to a sink.
Review Comment:
```suggestion
* DML operation that tells to write to a sink which implements {@link SupportsStaging}. Curerntly. this operation is only for CTAS(CREATE TABLE AS SELECT) statement.
```
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/execution/CtasJobStatusHook.java:
##########
@@ -0,0 +1,56 @@
+/*
+ * 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.execution;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.core.execution.JobStatusHook;
+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 {@link StagedTable} on job status changing.
Review Comment:
```suggestion
* call the corresponding interfaces of the inner {@link StagedTable} on job status changing.
```
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/StagedTableITCase.java:
##########
@@ -0,0 +1,74 @@
+/*
+ * 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.batch.sql;
+
+import org.apache.flink.table.api.config.TableConfigOptions;
+import org.apache.flink.table.planner.factories.TestSupportsStagingTableFactory;
+import org.apache.flink.table.planner.factories.utils.TestCollectionTableFactory;
+import org.apache.flink.table.planner.runtime.utils.BatchTestBase;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.FileUtils;
+
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.File;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests staged table in batch mode. */
+public class StagedTableITCase extends BatchTestBase {
+
+ @Rule public final TemporaryFolder tempFolder = new TemporaryFolder();
Review Comment:
The new test should follow Junit5 style:
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/StagedTableITCase.java:
##########
@@ -0,0 +1,74 @@
+/*
+ * 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.batch.sql;
+
+import org.apache.flink.table.api.config.TableConfigOptions;
+import org.apache.flink.table.planner.factories.TestSupportsStagingTableFactory;
+import org.apache.flink.table.planner.factories.utils.TestCollectionTableFactory;
+import org.apache.flink.table.planner.runtime.utils.BatchTestBase;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.FileUtils;
+
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.File;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests staged table in batch mode. */
+public class StagedTableITCase extends BatchTestBase {
+
+ @Rule public final TemporaryFolder tempFolder = new TemporaryFolder();
+
+ @Before
+ @Override
+ public void before() throws Exception {
+ super.before();
+ List<Row> sourceData = Arrays.asList(Row.of(1, "ZM"));
+
+ TestCollectionTableFactory.reset();
+ TestCollectionTableFactory.initData(sourceData);
+
+ String sourceDDL = "create table t1(a int, b varchar) with ('connector' = 'COLLECTION')";
+ tEnv().executeSql(sourceDDL);
+ }
+
+ @Test
+ public void testStagedTableWithAtomicCtas() throws Exception {
Review Comment:
I think the test can still be improved in the following aspects:
- IIUC, if `TABLE_CTAS_ATOMICITY_ENABLED` is true, the table won't be created by framewrok it self. So, please make sure the table hasn't been created in your test
- please test the case if `TABLE_CTAS_ATOMICITY_ENABLED` is false which should simiar to non-atomic ctas
- please test `abort` method, you can control wheter fail while writing data with a config , and try to make it fail so that abort method should be called
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/StagedSinkModifyOperation.java:
##########
@@ -0,0 +1,87 @@
+/*
+ * 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.operations;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.catalog.ContextResolvedTable;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+
+import javax.annotation.Nullable;
+
+import java.util.Map;
+
+/**
+ * DML operation that tells to write to a sink.
+ *
+ * <p>The sink is described by {@link #getContextResolvedTable()}, and in general is used for every
+ * sink which implementation is defined with {@link DynamicTableSink}.
+ *
+ * <p>StagedSinkModifyOperation is an extension of SinkModifyOperation to the atomic CTAS scenario.
Review Comment:
suggestion:
```
* <p>StagedSinkModifyOperation is an extension of SinkModifyOperation in the atomic CTAS scenario.
* Whiling checking whether the corresponding sink support atomic CTAS or not, we will need to get DynamicTableSink firstly and check whether it implements {@link SupportsStaging} and then call the method {@link SupportsStaging#applyStaging}. We maintain the DynamicTableSink in this operation so that we can reuse this DynamicTableSink instead of creating a new DynamicTableSink again which is error-prone.
```
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/StagedTableITCase.java:
##########
@@ -0,0 +1,74 @@
+/*
+ * 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.batch.sql;
+
+import org.apache.flink.table.api.config.TableConfigOptions;
+import org.apache.flink.table.planner.factories.TestSupportsStagingTableFactory;
+import org.apache.flink.table.planner.factories.utils.TestCollectionTableFactory;
+import org.apache.flink.table.planner.runtime.utils.BatchTestBase;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.FileUtils;
+
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.File;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests staged table in batch mode. */
+public class StagedTableITCase extends BatchTestBase {
+
+ @Rule public final TemporaryFolder tempFolder = new TemporaryFolder();
+
+ @Before
+ @Override
+ public void before() throws Exception {
+ super.before();
+ List<Row> sourceData = Arrays.asList(Row.of(1, "ZM"));
+
+ TestCollectionTableFactory.reset();
+ TestCollectionTableFactory.initData(sourceData);
+
+ String sourceDDL = "create table t1(a int, b varchar) with ('connector' = 'COLLECTION')";
+ tEnv().executeSql(sourceDDL);
+ }
+
+ @Test
+ public void testStagedTableWithAtomicCtas() throws Exception {
Review Comment:
```suggestion
void testStagedTableWithAtomicCtas() throws Exception {
```
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/StagedTableITCase.java:
##########
@@ -0,0 +1,74 @@
+/*
+ * 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.batch.sql;
+
+import org.apache.flink.table.api.config.TableConfigOptions;
+import org.apache.flink.table.planner.factories.TestSupportsStagingTableFactory;
+import org.apache.flink.table.planner.factories.utils.TestCollectionTableFactory;
+import org.apache.flink.table.planner.runtime.utils.BatchTestBase;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.FileUtils;
+
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.File;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests staged table in batch mode. */
+public class StagedTableITCase extends BatchTestBase {
+
+ @Rule public final TemporaryFolder tempFolder = new TemporaryFolder();
+
+ @Before
Review Comment:
```suggestion
@BeforeEach
```
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/connector/sink/abilities/SupportsStaging.java:
##########
@@ -0,0 +1,73 @@
+/*
+ * 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.connector.sink.abilities;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.core.execution.JobStatusHook;
+import org.apache.flink.table.catalog.StagedTable;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+
+/**
+ * Enables different staged operations to ensure atomicity in a {@link DynamicTableSink}.
+ *
+ * <p>By default, if this interface is not implemented, indicating that atomic operations are not
+ * supported, then a non-atomic implementation is used.
+ */
+@PublicEvolving
+public interface SupportsStaging {
+
+ /**
+ * Provides a {@link StagedTable} that provided transaction abstraction. StagedTable will be
+ * combined with {@link JobStatusHook} to achieve atomicity support in the Flink framework. Call
+ * the relevant API of StagedTable when the Job state is switched.
+ *
+ * <p>This method will be called at the compile stage.
+ *
+ * @param context Tell DynamicTableSink, the operation type of this StagedTable, expandable
+ * @return {@link StagedTable} that can be serialized and provides atomic operations
+ */
+ StagedTable applyStaging(StagingContext context);
+
+ /**
+ * The context is intended to tell DynamicTableSink the type of this operation. In this way,
+ * DynamicTableSink can return the corresponding implementation of StagedTable according to the
+ * specific operation. More types of operations can be extended in the future.
+ */
+ @PublicEvolving
+ interface StagingContext {
+ StagingPurpose getStagingPurpose();
+ }
+
+ /**
+ * The type of StagedTable final visibility that was expects for staging purpose.
Review Comment:
please don't ignore my comment in here.
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/StagedTableITCase.java:
##########
@@ -0,0 +1,74 @@
+/*
+ * 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.batch.sql;
+
+import org.apache.flink.table.api.config.TableConfigOptions;
+import org.apache.flink.table.planner.factories.TestSupportsStagingTableFactory;
+import org.apache.flink.table.planner.factories.utils.TestCollectionTableFactory;
+import org.apache.flink.table.planner.runtime.utils.BatchTestBase;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.FileUtils;
+
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.File;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests staged table in batch mode. */
+public class StagedTableITCase extends BatchTestBase {
+
+ @Rule public final TemporaryFolder tempFolder = new TemporaryFolder();
Review Comment:
@TempDir Path temporaryFolder;
you can find example in the codebase.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] Tartarus0zm commented on a diff in pull request #22839: [FLINK-32349][table] Support atomic for CREATE TABLE AS SELECT(CTAS) statement
Posted by "Tartarus0zm (via GitHub)" <gi...@apache.org>.
Tartarus0zm commented on code in PR #22839:
URL: https://github.com/apache/flink/pull/22839#discussion_r1255181663
##########
docs/layouts/shortcodes/generated/table_config_configuration.html:
##########
@@ -20,6 +20,12 @@
<td>String</td>
<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 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 in client side, and it will not be dropped even though the job fails or is cancelled. If set this option to true and 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:
IDEA boost and the previous comma can not be removed, the rest has been fixed
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] luoyuxia merged pull request #22839: [FLINK-32349][table] Support atomic for CREATE TABLE AS SELECT(CTAS) statement
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia merged PR #22839:
URL: https://github.com/apache/flink/pull/22839
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] Tartarus0zm commented on pull request #22839: [FLINK-32349][table] Support atomic for CREATE TABLE AS SELECT(CTAS) statement
Posted by "Tartarus0zm (via GitHub)" <gi...@apache.org>.
Tartarus0zm commented on PR #22839:
URL: https://github.com/apache/flink/pull/22839#issuecomment-1631766550
> Please don't forget to create a jira to add a doc for the atomic CTAS statement and put the jira and [FLINK-32349](https://issues.apache.org/jira/browse/FLINK-32349) under a umbrella jira.
@luoyuxia thanks for your remind
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] luoyuxia commented on a diff in pull request #22839: [FLINK-32349][table] Support atomic for CREATE TABLE AS SELECT(CTAS) statement
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #22839:
URL: https://github.com/apache/flink/pull/22839#discussion_r1259063263
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/StagedSinkModifyOperation.java:
##########
@@ -0,0 +1,88 @@
+/*
+ * 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.operations;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.catalog.ContextResolvedTable;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.connector.sink.abilities.SupportsStaging;
+
+import javax.annotation.Nullable;
+
+import java.util.Map;
+
+/**
+ * DML operation that tells to write to a sink which implements {@link SupportsStaging}. Currently.
+ * this operation is only for CTAS(CREATE TABLE AS SELECT) statement.
+ *
+ * <p>StagedSinkModifyOperation is an extension of SinkModifyOperation in the atomic CTAS scenario.
+ * Whiling checking whether the corresponding sink support atomic CTAS or not, we will need to get
+ * DynamicTableSink firstly and check whether it implements {@link SupportsStaging} and then call
+ * the method {@link SupportsStaging#applyStaging}. We maintain the DynamicTableSink in this
+ * operation so that we can reuse this DynamicTableSink instead of creating a new DynamicTableSink
+ * during translate the operation again which is error-prone.
Review Comment:
nit
```suggestion
* during translating the operation again which is error-prone.
```
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] luoyuxia commented on a diff in pull request #22839: [FLINK-32349][table] Support atomic for CREATE TABLE AS SELECT(CTAS) statement
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #22839:
URL: https://github.com/apache/flink/pull/22839#discussion_r1241151298
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java:
##########
@@ -774,11 +783,56 @@ public CompiledPlan compilePlan(List<ModifyOperation> operations) {
public TableResultInternal executeInternal(List<ModifyOperation> operations) {
List<ModifyOperation> mapOperations = new ArrayList<>();
for (ModifyOperation modify : operations) {
- // execute CREATE TABLE first for CTAS statements
if (modify instanceof CreateTableASOperation) {
CreateTableASOperation ctasOperation = (CreateTableASOperation) modify;
- executeInternal(ctasOperation.getCreateTableOperation());
- mapOperations.add(ctasOperation.toSinkModifyOperation(catalogManager));
+ CreateTableOperation createTableOperation = ctasOperation.getCreateTableOperation();
Review Comment:
I'm wondering can these code for ctas be extracted to a method for now it looks to be tedious to me.
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/execution/CtasJobStatusHook.java:
##########
@@ -0,0 +1,56 @@
+/*
+ * 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.execution;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.core.execution.JobStatusHook;
+import org.apache.flink.table.catalog.StagedTable;
+
+/**
Review Comment:
nit:
```
/**
* This hook is used to implement atomic semantics for CTAS(CREATE TABLE AS SELECT) statement. It'll
* call the corresponding interfaces of {@link StagedTable} on job status changing.
*/
```
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java:
##########
@@ -774,11 +783,56 @@ public CompiledPlan compilePlan(List<ModifyOperation> operations) {
public TableResultInternal executeInternal(List<ModifyOperation> operations) {
List<ModifyOperation> mapOperations = new ArrayList<>();
for (ModifyOperation modify : operations) {
- // execute CREATE TABLE first for CTAS statements
if (modify instanceof CreateTableASOperation) {
CreateTableASOperation ctasOperation = (CreateTableASOperation) modify;
- executeInternal(ctasOperation.getCreateTableOperation());
- mapOperations.add(ctasOperation.toSinkModifyOperation(catalogManager));
+ CreateTableOperation createTableOperation = ctasOperation.getCreateTableOperation();
+ ObjectIdentifier tableIdentifier = createTableOperation.getTableIdentifier();
+ Catalog catalog = catalogManager.getCatalog(tableIdentifier.getCatalogName()).get();
+ ResolvedCatalogTable catalogTable =
+ catalogManager.resolveCatalogTable(createTableOperation.getCatalogTable());
+
+ Optional<DynamicTableSink> dynamicTableSinkOptional =
+ TableFactoryUtil.getDynamicTableSink(
+ catalogTable,
+ tableIdentifier,
+ createTableOperation.isTemporary(),
+ catalog,
+ moduleManager,
+ tableConfig,
+ isStreamingMode,
+ resourceManager.getUserClassLoader());
+ if (dynamicTableSinkOptional.isPresent()
+ && dynamicTableSinkOptional.get() instanceof SupportsStaging
+ && tableConfig.get(TableConfigOptions.TABLE_CTAS_ATOMICITY_ENABLED)) {
+ DynamicTableSink dynamicTableSink = dynamicTableSinkOptional.get();
+ StagedTable stagedTable =
+ ((SupportsStaging) dynamicTableSink)
+ .applyStaging(
+ new SupportsStaging.StagingContext() {
Review Comment:
I'm not a fans for lambda function. I think we can create a calss name `SinkStagingContext` just like `SinkRuntimeProviderContext`.
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/delegation/Executor.java:
##########
@@ -61,6 +62,21 @@ Pipeline createPipeline(
ReadableConfig tableConfiguration,
@Nullable String defaultJobName);
+ /**
+ * Translates the given transformations to a {@link Pipeline}.
Review Comment:
nit:
```suggestion
* Translates the given transformations with a list of {@link JobStatusHook}s to a {@link Pipeline} .
```
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/connector/sink/abilities/SupportsStaging.java:
##########
@@ -0,0 +1,73 @@
+/*
+ * 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.connector.sink.abilities;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.core.execution.JobStatusHook;
+import org.apache.flink.table.catalog.StagedTable;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+
+/**
Review Comment:
Please elaborate the interface in the java doc as it's the core interface such as what the interface is for and when this interface will be consider. And when the interface is considered, which method will be called.
A good doc make it easy to understand the mechanism and convenient for connector to integrate
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/connector/sink/abilities/SupportsStaging.java:
##########
@@ -0,0 +1,73 @@
+/*
+ * 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.connector.sink.abilities;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.core.execution.JobStatusHook;
+import org.apache.flink.table.catalog.StagedTable;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+
+/**
+ * Enables different staged operations to ensure atomicity in a {@link DynamicTableSink}.
+ *
+ * <p>By default, if this interface is not implemented, indicating that atomic operations are not
+ * supported, then a non-atomic implementation is used.
+ */
+@PublicEvolving
+public interface SupportsStaging {
+
+ /**
+ * Provides a {@link StagedTable} that provided transaction abstraction. StagedTable will be
+ * combined with {@link JobStatusHook} to achieve atomicity support in the Flink framework. Call
+ * the relevant API of StagedTable when the Job state is switched.
+ *
+ * <p>This method will be called at the compile stage.
+ *
+ * @param context Tell DynamicTableSink, the operation type of this StagedTable, expandable
+ * @return {@link StagedTable} that can be serialized and provides atomic operations
+ */
+ StagedTable applyStaging(StagingContext context);
+
+ /**
Review Comment:
nit:
```
/**
* The context is intended to tell DynamicTableSink the type of this operation. Currently, it'll
* provide what kind of operation the staging sink is for. In this way, the DynamicTableSink can
* return the corresponding implementation of StagedTable according to the specific operation.
* More types of operations can be extended in the future.
*/
```
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/StagedTable.java:
##########
@@ -0,0 +1,65 @@
+/*
+ * 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.catalog;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.core.execution.JobStatusHook;
+
+import java.io.Serializable;
+
+/**
+ * A {@link StagedTable} for atomic semantics using a two-phase commit protocol, combined with
+ * {@link JobStatusHook} for atomic CTAS. {@link StagedTable} will be a member variable of
+ * CtasJobStatusHook and can be serialized;
+ *
+ * <p>CtasJobStatusHook#onCreated will call the begin method of StagedTable;
+ * CtasJobStatusHook#onFinished will call the commit method of StagedTable;
+ * CtasJobStatusHook#onFailed and CtasJobStatusHook#onCanceled will call the abort method of
+ * StagedTable;
+ */
+@PublicEvolving
+public interface StagedTable extends Serializable {
+
+ /**
+ * This method will be called when the job is started. Similar to what it means to open a
+ * transaction in a relational database; In Flink's atomic CTAS scenario, it is used to do some
+ * initialization work; For example, initializing the client of the underlying service, the tmp
+ * path of the underlying storage, or even call the start transaction API of the underlying
+ * service, etc.
+ */
+ void begin();
+
+ /**
+ * This method will be called when the job is succeeds. Similar to what it means to commit the
+ * transaction in a relational database; In Flink's atomic CTAS scenario, it is used to do some
+ * data visibility related work; For example, moving the underlying data to the target
+ * directory, writing buffer data to the underlying storage service, or even call the commit
+ * transaction API of the underlying service, etc.
+ */
+ void commit();
+
+ /**
+ * This method will be called when the job is failed or canceled. Similar to what it means to
Review Comment:
Suggestion:
```
/**
* This method will be called when the job is failed or canceled. In Flink's atomic CTAS
* scenario, it is expected to do some cleaning work for a writing; For example, delete the data
* in tmp directory, delete the temporary data in the underlying storage service, or even call
* the rollback transaction API of the underlying service, etc.
*/
```
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/StagedTableITCase.java:
##########
@@ -0,0 +1,70 @@
+/*
+ * 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.batch.sql;
+
+import org.apache.flink.table.api.config.TableConfigOptions;
+import org.apache.flink.table.planner.factories.utils.TestCollectionTableFactory;
+import org.apache.flink.table.planner.runtime.utils.BatchTestBase;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.FileUtils;
+
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.File;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests staged table in a table environment. */
+public class StagedTableITCase extends BatchTestBase {
+
+ @Rule public final TemporaryFolder tempFolder = new TemporaryFolder();
+
+ @Before
+ @Override
+ public void before() throws Exception {
+ super.before();
+ List<Row> sourceData = Arrays.asList(Row.of(1, "ZM"));
+
+ TestCollectionTableFactory.reset();
+ TestCollectionTableFactory.initData(sourceData);
+
+ String sourceDDL = "create table t1(a int, b varchar) with ('connector' = 'COLLECTION')";
+ tEnv().executeSql(sourceDDL);
+ }
+
+ @Test
+ public void testStagedTableWithAtomicCtas() throws Exception {
Review Comment:
How can you make sure the data is invisible before commiting in this test?
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/StagedTable.java:
##########
@@ -0,0 +1,65 @@
+/*
+ * 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.catalog;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.core.execution.JobStatusHook;
+
+import java.io.Serializable;
+
+/**
+ * A {@link StagedTable} for atomic semantics using a two-phase commit protocol, combined with
+ * {@link JobStatusHook} for atomic CTAS. {@link StagedTable} will be a member variable of
Review Comment:
Also, please elaborate the interface in the java doc as it's the core interface such as what the interface is for and who should return it And when the interface is considered, which method will be called in which condition.
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/execution/CtasJobStatusHook.java:
##########
@@ -0,0 +1,56 @@
+/*
+ * 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.execution;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.core.execution.JobStatusHook;
+import org.apache.flink.table.catalog.StagedTable;
+
+/**
Review Comment:
Another minor comment on the java doc of `JobStatusHook`.
```Hooks provided by users on job status changing```,
but the hook now shouldn't be provided by users since it has marked as internal.
Change it to `Hooks on job status changing`?
That means we don't intend users to use it, althogh users can use it anyway.
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/delegation/Executor.java:
##########
@@ -61,6 +62,21 @@ Pipeline createPipeline(
ReadableConfig tableConfiguration,
@Nullable String defaultJobName);
+ /**
+ * Translates the given transformations to a {@link Pipeline}.
+ *
+ * @param transformations list of transformations
+ * @param tableConfiguration table-specific configuration options
+ * @param defaultJobName default job name if not specified via {@link PipelineOptions#NAME}
+ * @param jobStatusHookList list of JobStatusHooks
Review Comment:
nit:
```suggestion
* @param jobStatusHookList list of {@link JobStatusHook}s
```
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/factories/TableFactoryUtil.java:
##########
@@ -164,4 +171,54 @@ public static boolean isLegacyConnectorOptions(
}
}
}
+
+ /**
+ * Create a DynamicTableSink for ResolvedCatalogTable using table factory associated with the
+ * catalog.
+ */
+ public static Optional<DynamicTableSink> getDynamicTableSink(
Review Comment:
`TableFactoryUtil` is not suitable to put this method. I also need this util method whiling supporting to truncate table in my [pr](https://github.com/apache/flink/pull/22696/files#diff-b4a2c05dbc05190ed95ee36ccae3084356dea84a1c316bf89254e6217471d9d7R47).
Would you mind rebase after my pr is merged?
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/StagedSinkModifyOperation.java:
##########
@@ -0,0 +1,87 @@
+/*
+ * 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.operations;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.catalog.ContextResolvedTable;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+
+import javax.annotation.Nullable;
+
+import java.util.Map;
+
+/**
Review Comment:
nit:
```
/**
* DML operation that tells to write to a sink which implements {@link SupportsStaging}.
*
* <p>The sink is described by {@link #getContextResolvedTable()}, and in general is used for every
* sink which implementation is defined with {@link DynamicTableSink}.
*
* <p>StagedSinkModifyOperation is an extension of SinkModifyOperation in the atomic CTAS scenario.
* Whiling checking whether the corresponding sink support atomic CTAS or not, we will need to get
* DynamicTableSink firstly and check whether it implements {@link SupportsStaging} and then call
* the method {@link SupportsStaging#applyStaging}. We maintain the DynamicTableSink in this
* operation so that we can reuse this DynamicTableSink instead of creating a new DynamicTableSink
* which is error-prone.
*/
```
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/delegation/DefaultExecutor.java:
##########
@@ -85,6 +86,21 @@ public Pipeline createPipeline(
return streamGraph;
}
+ @Override
+ public Pipeline createPipeline(
+ List<Transformation<?>> transformations,
+ ReadableConfig tableConfiguration,
+ @Nullable String defaultJobName,
+ List<JobStatusHook> jobStatusHookList) {
+ StreamGraph streamGraph =
+ (StreamGraph) createPipeline(transformations, tableConfiguration, defaultJobName);
+ for (JobStatusHook hook : jobStatusHookList) {
+ streamGraph.registerJobStatusHook(hook);
+ }
+ jobStatusHookList.clear();
Review Comment:
Why clear the `jobStatusHookList`?
##########
flink-python/src/main/java/org/apache/flink/table/executor/python/ChainingOptimizingExecutor.java:
##########
@@ -63,6 +66,24 @@ public Pipeline createPipeline(
return executor.createPipeline(chainedTransformations, configuration, defaultJobName);
Review Comment:
this method can be simplied to
```return createPipeline(transformations, configuration, defaultJobName, Collections.emptyList());```
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/connector/sink/abilities/SupportsStaging.java:
##########
@@ -0,0 +1,73 @@
+/*
+ * 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.connector.sink.abilities;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.core.execution.JobStatusHook;
+import org.apache.flink.table.catalog.StagedTable;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+
+/**
+ * Enables different staged operations to ensure atomicity in a {@link DynamicTableSink}.
+ *
+ * <p>By default, if this interface is not implemented, indicating that atomic operations are not
+ * supported, then a non-atomic implementation is used.
+ */
+@PublicEvolving
+public interface SupportsStaging {
+
+ /**
+ * Provides a {@link StagedTable} that provided transaction abstraction. StagedTable will be
Review Comment:
Suggestion:
```
/**
* Provides a {@link StagingContext} for the sink modification and return a {@link StagedTable}.
* The {@link StagedTable} provides transaction abstraction to support atomicity for CTAS. Flink
* will call the relevant API of StagedTable when the Job status switches,
*
* <p>Note: This method will be called at the compile stage.
*
* @param context The context for the sink modification
* @return {@link StagedTable} that will be leveraged by Flink framework to provide atomicity
* semantics.
*/
```
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/connector/sink/abilities/SupportsStaging.java:
##########
@@ -0,0 +1,73 @@
+/*
+ * 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.connector.sink.abilities;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.core.execution.JobStatusHook;
+import org.apache.flink.table.catalog.StagedTable;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+
+/**
+ * Enables different staged operations to ensure atomicity in a {@link DynamicTableSink}.
+ *
+ * <p>By default, if this interface is not implemented, indicating that atomic operations are not
+ * supported, then a non-atomic implementation is used.
+ */
+@PublicEvolving
+public interface SupportsStaging {
+
+ /**
+ * Provides a {@link StagedTable} that provided transaction abstraction. StagedTable will be
+ * combined with {@link JobStatusHook} to achieve atomicity support in the Flink framework. Call
+ * the relevant API of StagedTable when the Job state is switched.
+ *
+ * <p>This method will be called at the compile stage.
+ *
+ * @param context Tell DynamicTableSink, the operation type of this StagedTable, expandable
+ * @return {@link StagedTable} that can be serialized and provides atomic operations
+ */
+ StagedTable applyStaging(StagingContext context);
+
+ /**
+ * The context is intended to tell DynamicTableSink the type of this operation. In this way,
+ * DynamicTableSink can return the corresponding implementation of StagedTable according to the
+ * specific operation. More types of operations can be extended in the future.
+ */
+ @PublicEvolving
+ interface StagingContext {
+ StagingPurpose getStagingPurpose();
+ }
+
+ /**
+ * The type of StagedTable final visibility that was expects for staging purpose.
Review Comment:
suggestion:
```
/**
* The type of operation the staging sink is for.
*
* <p>Currently, two types of operation are supported:
*
* <ul>
* <li>CREATE_TABLE_AS - for the operation of CREATE TABLE AS SELECT statement
* <li>CREATE_TABLE_AS_IF_NOT_EXISTS - for the operation of CREATE TABLE AS SELECT IF NOT
* EXISTS statement.
* </ul>
*/
```
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/StagedTable.java:
##########
@@ -0,0 +1,65 @@
+/*
+ * 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.catalog;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.core.execution.JobStatusHook;
+
+import java.io.Serializable;
+
+/**
+ * A {@link StagedTable} for atomic semantics using a two-phase commit protocol, combined with
+ * {@link JobStatusHook} for atomic CTAS. {@link StagedTable} will be a member variable of
+ * CtasJobStatusHook and can be serialized;
+ *
+ * <p>CtasJobStatusHook#onCreated will call the begin method of StagedTable;
+ * CtasJobStatusHook#onFinished will call the commit method of StagedTable;
+ * CtasJobStatusHook#onFailed and CtasJobStatusHook#onCanceled will call the abort method of
+ * StagedTable;
+ */
+@PublicEvolving
+public interface StagedTable extends Serializable {
+
+ /**
+ * This method will be called when the job is started. Similar to what it means to open a
+ * transaction in a relational database; In Flink's atomic CTAS scenario, it is used to do some
+ * initialization work; For example, initializing the client of the underlying service, the tmp
+ * path of the underlying storage, or even call the start transaction API of the underlying
+ * service, etc.
+ */
+ void begin();
+
+ /**
Review Comment:
Suggestion:
```
/**
* This method will be called when the job is succeeds. In Flink's atomic CTAS scenario, it is
* expected to do some commit work. For example, moving the underlying data to the target
* directory to make it visible, writing buffer data to the underlying storage service, or even
* call the commit transaction API of the underlying service, etc.
*/
```
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/StagedTable.java:
##########
@@ -0,0 +1,65 @@
+/*
+ * 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.catalog;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.core.execution.JobStatusHook;
+
+import java.io.Serializable;
+
+/**
+ * A {@link StagedTable} for atomic semantics using a two-phase commit protocol, combined with
+ * {@link JobStatusHook} for atomic CTAS. {@link StagedTable} will be a member variable of
+ * CtasJobStatusHook and can be serialized;
+ *
+ * <p>CtasJobStatusHook#onCreated will call the begin method of StagedTable;
+ * CtasJobStatusHook#onFinished will call the commit method of StagedTable;
+ * CtasJobStatusHook#onFailed and CtasJobStatusHook#onCanceled will call the abort method of
+ * StagedTable;
+ */
+@PublicEvolving
+public interface StagedTable extends Serializable {
+
+ /**
Review Comment:
Suggestion:
```
/**
* This method will be called when the job is started. In Flink's atomic CTAS scenario, it is
* expected to do initialization work; For example, initializing the client of the underlying
* service, the tmp path of the underlying storage, or even call the start transaction API of
* the underlying service, etc.
*/
```
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/StagedTable.java:
##########
@@ -0,0 +1,65 @@
+/*
+ * 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.catalog;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.core.execution.JobStatusHook;
+
+import java.io.Serializable;
+
+/**
+ * A {@link StagedTable} for atomic semantics using a two-phase commit protocol, combined with
+ * {@link JobStatusHook} for atomic CTAS. {@link StagedTable} will be a member variable of
Review Comment:
Please don't refer `JobStatusHook` in here since it's not a public api. Talk about `JobStatusHook` will make devs confused.
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/StagedTableITCase.java:
##########
@@ -0,0 +1,70 @@
+/*
+ * 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.batch.sql;
+
+import org.apache.flink.table.api.config.TableConfigOptions;
+import org.apache.flink.table.planner.factories.utils.TestCollectionTableFactory;
+import org.apache.flink.table.planner.runtime.utils.BatchTestBase;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.FileUtils;
+
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.File;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests staged table in a table environment. */
+public class StagedTableITCase extends BatchTestBase {
Review Comment:
Please test it in both batch and stream mode.
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestSupportsStagingTableFactory.java:
##########
@@ -0,0 +1,171 @@
+/*
+ * 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.factories;
+
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.DataStreamSink;
+import org.apache.flink.streaming.api.functions.sink.DiscardingSink;
+import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
+import org.apache.flink.table.catalog.StagedTable;
+import org.apache.flink.table.connector.ChangelogMode;
+import org.apache.flink.table.connector.ProviderContext;
+import org.apache.flink.table.connector.sink.DataStreamSinkProvider;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.connector.sink.abilities.SupportsStaging;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.factories.DynamicTableSinkFactory;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.util.FileUtils;
+
+import java.io.File;
+import java.util.Collections;
+import java.util.Set;
+
+/** A factory to create table to support staging for test purpose. */
+public class TestSupportsStagingTableFactory implements DynamicTableSinkFactory {
+
+ public static final String IDENTIFIER = "test-staging";
+
+ private static final ConfigOption<String> DATA_DIR =
+ ConfigOptions.key("data-dir")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("The data id used to write the rows.");
+
+ @Override
+ public DynamicTableSink createDynamicTableSink(Context context) {
+ FactoryUtil.TableFactoryHelper helper = FactoryUtil.createTableFactoryHelper(this, context);
+ helper.validate();
+ String dataDir = helper.getOptions().get(DATA_DIR);
+ return new SupportsStagingTableSink(dataDir);
+ }
+
+ @Override
+ public String factoryIdentifier() {
+ return IDENTIFIER;
+ }
+
+ @Override
+ public Set<ConfigOption<?>> requiredOptions() {
+ return Collections.singleton(DATA_DIR);
+ }
+
+ @Override
+ public Set<ConfigOption<?>> optionalOptions() {
+ return Collections.emptySet();
+ }
+
+ /** A sink that supports staging. */
+ private static class SupportsStagingTableSink implements DynamicTableSink, SupportsStaging {
+
+ private String dataDir;
+ private TestStagedTable stagedTable;
+
+ public SupportsStagingTableSink(String dataDir) {
+ this(dataDir, null);
+ }
+
+ public SupportsStagingTableSink(String dataDir, TestStagedTable stagedTable) {
+ this.dataDir = dataDir;
+ this.stagedTable = stagedTable;
+ }
+
+ @Override
+ public ChangelogMode getChangelogMode(ChangelogMode requestedMode) {
+ return ChangelogMode.insertOnly();
+ }
+
+ @Override
+ public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
+ return new DataStreamSinkProvider() {
+ @Override
+ public DataStreamSink<?> consumeDataStream(
+ ProviderContext providerContext, DataStream<RowData> dataStream) {
+ if (stagedTable != null) {
+ return dataStream
+ .addSink(new StagedSinkFunction(dataDir))
+ .setParallelism(1);
+ } else {
+ // otherwise, do nothing
+ return dataStream.addSink(new DiscardingSink<>());
+ }
+ }
+ };
+ }
+
+ @Override
+ public DynamicTableSink copy() {
+ return new SupportsStagingTableSink(dataDir, stagedTable);
+ }
+
+ @Override
+ public String asSummaryString() {
+ return "SupportsStagingTableSink";
+ }
+
+ @Override
+ public StagedTable applyStaging(StagingContext context) {
+ stagedTable = new TestStagedTable();
+ return stagedTable;
+ }
+ }
+
+ /** The sink for delete existing data. */
+ private static class StagedSinkFunction extends RichSinkFunction<RowData> {
+
+ private String dataDir;
+
+ public StagedSinkFunction(String dataDir) {
+ this.dataDir = dataDir;
+ }
+
+ @Override
+ public void open(Configuration parameters) throws Exception {
+ super.open(parameters);
+ File parentDir = new File(dataDir);
+ if (parentDir.exists()) {
+ parentDir.delete();
+ }
+ parentDir.mkdirs();
+ new File(dataDir, "data").createNewFile();
+ }
+
+ @Override
+ public void invoke(RowData value, Context context) throws Exception {
+ FileUtils.writeFileUtf8(
+ new File(dataDir, "data"), value.getInt(0) + "," + value.getString(1));
+ }
+ }
+
+ /** A StagedTable for test. */
+ private static class TestStagedTable implements StagedTable {
+
+ @Override
+ public void begin() {}
Review Comment:
If we do nothing in these method, we can never know these method will be really called when job begin, finish, and abort.
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/StagedTable.java:
##########
@@ -0,0 +1,65 @@
+/*
+ * 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.catalog;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.core.execution.JobStatusHook;
+
+import java.io.Serializable;
+
+/**
+ * A {@link StagedTable} for atomic semantics using a two-phase commit protocol, combined with
+ * {@link JobStatusHook} for atomic CTAS. {@link StagedTable} will be a member variable of
+ * CtasJobStatusHook and can be serialized;
+ *
+ * <p>CtasJobStatusHook#onCreated will call the begin method of StagedTable;
+ * CtasJobStatusHook#onFinished will call the commit method of StagedTable;
+ * CtasJobStatusHook#onFailed and CtasJobStatusHook#onCanceled will call the abort method of
+ * StagedTable;
+ */
+@PublicEvolving
+public interface StagedTable extends Serializable {
+
+ /**
Review Comment:
No need to refer `relational database`.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] Tartarus0zm commented on a diff in pull request #22839: [FLINK-32349][table] Support atomic for CREATE TABLE AS SELECT(CTAS) statement
Posted by "Tartarus0zm (via GitHub)" <gi...@apache.org>.
Tartarus0zm commented on code in PR #22839:
URL: https://github.com/apache/flink/pull/22839#discussion_r1255181663
##########
docs/layouts/shortcodes/generated/table_config_configuration.html:
##########
@@ -20,6 +20,12 @@
<td>String</td>
<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 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 in client side, and it will not be dropped even though the job fails or is cancelled. If set this option to true and 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:
IDEA boost and the previous comma can not be removed, the rest has been fixed
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] luoyuxia commented on a diff in pull request #22839: [FLINK-32349][table] Support atomic for CREATE TABLE AS SELECT(CTAS) statement
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #22839:
URL: https://github.com/apache/flink/pull/22839#discussion_r1258138669
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java:
##########
@@ -849,6 +858,56 @@ private ModifyOperation getOperation(ReplaceTableAsOperation rtasOperation) {
return rtasOperation.toSinkModifyOperation(catalogManager);
}
+ private ModifyOperation getOperation(
Review Comment:
nit:
I'm not sure, but May be `getModifyOperation()` may clear?
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/connector/sink/abilities/SupportsStaging.java:
##########
@@ -0,0 +1,79 @@
+/*
+ * 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.connector.sink.abilities;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.table.catalog.StagedTable;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+
+/**
+ * Interface for {@link DynamicTableSink}s that support atomic semantic for CTAS(CREATE TABLE AS
+ * SELECT) statement using a two-phase commit protocol. The table sink is responsible for returning
+ * a {@link StagedTable} to tell the Flink how to implement the atomicity semantics.
+ *
+ * <p>If the user turns on {@link TableConfigOptions#TABLE_CTAS_ATOMICITY_ENABLED}, and the {@link
+ * DynamicTableSink} implements {@link SupportsStaging}, the planner will call method {@link
+ * #applyStaging(StagingContext)} to get the {@link StagedTable} returned by the sink, then {@link
+ * StagedTable} will be used by Flink to implement a two-phase commit with the actual implementation
+ * of {@link StagedTable}.
Review Comment:
```suggestion
* of the {@link StagedTable}.
```
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/StagedSinkModifyOperation.java:
##########
@@ -0,0 +1,88 @@
+/*
+ * 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.operations;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.catalog.ContextResolvedTable;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.connector.sink.abilities.SupportsStaging;
+
+import javax.annotation.Nullable;
+
+import java.util.Map;
+
+/**
+ * DML operation that tells to write to a sink which implements {@link SupportsStaging}. Currently.
+ * this operation is only for CTAS(CREATE TABLE AS SELECT) statement.
+ *
+ * <p>StagedSinkModifyOperation is an extension of SinkModifyOperation in the atomic CTAS scenario.
+ * Whiling checking whether the corresponding sink support atomic CTAS or not, we will need to get
+ * DynamicTableSink firstly and check whether it implements {@link SupportsStaging} and then call
+ * the method {@link SupportsStaging#applyStaging}. We maintain the DynamicTableSink in this
+ * operation so that we can reuse this DynamicTableSink instead of creating a new DynamicTableSink
Review Comment:
```suggestion
* operation so that we can reuse this DynamicTableSink instead of creating a new DynamicTableSink during translate the operation
```
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/AtomicCtasITCase.java:
##########
@@ -0,0 +1,136 @@
+/*
+ * 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.batch.sql;
+
+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.table.planner.runtime.utils.BatchTestBase;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.FileUtils;
+
+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.Arrays;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+
+/** Tests atomic ctas in batch mode. */
+public class AtomicCtasITCase extends BatchTestBase {
Review Comment:
Also, Just found the code in stream/batch AtomicCtasITCase is almost same.
Would it make sense that we extract the code into a common abstract class like `AtomicCtasITCaseBase` just like `JoinReorderITCaseBase`? And then we will only need to rewrite `getTableEnvironment` to provided a env with stream/batch.
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/AtomicCtasITCase.java:
##########
@@ -0,0 +1,136 @@
+/*
+ * 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.stream.sql;
+
+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.table.planner.runtime.utils.StreamingTestBase;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.FileUtils;
+
+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.Arrays;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+
+/** Tests atomic ctas in stream mode. */
+public class AtomicCtasITCase extends StreamingTestBase {
+
+ @BeforeEach
+ void setup() throws Exception {
+ super.before();
+ List<Row> sourceData = Arrays.asList(Row.of(1, "ZM"));
Review Comment:
```suggestion
List<Row> sourceData = Collections.singletonList(Row.of(1, "ZM"));
```
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/AtomicCtasITCase.java:
##########
@@ -0,0 +1,136 @@
+/*
+ * 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.batch.sql;
+
+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.table.planner.runtime.utils.BatchTestBase;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.FileUtils;
+
+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.Arrays;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+
+/** Tests atomic ctas in batch mode. */
+public class AtomicCtasITCase extends BatchTestBase {
+
+ @BeforeEach
+ void setup() throws Exception {
+ super.before();
+ List<Row> sourceData = Arrays.asList(Row.of(1, "ZM"));
Review Comment:
nit:
List<Row> sourceData = Collections.singletonList(Row.of(1, "ZM"));
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/connector/sink/abilities/SupportsStaging.java:
##########
@@ -0,0 +1,79 @@
+/*
+ * 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.connector.sink.abilities;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.table.catalog.StagedTable;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+
+/**
+ * Interface for {@link DynamicTableSink}s that support atomic semantic for CTAS(CREATE TABLE AS
+ * SELECT) statement using a two-phase commit protocol. The table sink is responsible for returning
+ * a {@link StagedTable} to tell the Flink how to implement the atomicity semantics.
+ *
+ * <p>If the user turns on {@link TableConfigOptions#TABLE_CTAS_ATOMICITY_ENABLED}, and the {@link
+ * DynamicTableSink} implements {@link SupportsStaging}, the planner will call method {@link
+ * #applyStaging(StagingContext)} to get the {@link StagedTable} returned by the sink, then {@link
Review Comment:
```suggestion
* #applyStaging(StagingContext)} to get the {@link StagedTable} returned by the sink, then the {@link
```
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] luoyuxia commented on pull request #22839: [FLINK-32349][table] Support atomic for CREATE TABLE AS SELECT(CTAS) statement
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on PR #22839:
URL: https://github.com/apache/flink/pull/22839#issuecomment-1631729536
Please don't forget to create a jira to add a doc for the atomic CTAS statement and put the jira and FLINK-32349 under a umbrella jira.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #22839: [FLINK-32349][table] Support atomic for CREATE TABLE AS SELECT(CTAS) statement
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #22839:
URL: https://github.com/apache/flink/pull/22839#issuecomment-1600710495
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "9604ce4793f3f9dcbf6d65fbbb9cf64bd690e6ce",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "9604ce4793f3f9dcbf6d65fbbb9cf64bd690e6ce",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 9604ce4793f3f9dcbf6d65fbbb9cf64bd690e6ce UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] Tartarus0zm commented on a diff in pull request #22839: [FLINK-32349][table] Support atomic for CREATE TABLE AS SELECT(CTAS) statement
Posted by "Tartarus0zm (via GitHub)" <gi...@apache.org>.
Tartarus0zm commented on code in PR #22839:
URL: https://github.com/apache/flink/pull/22839#discussion_r1243535740
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/factories/TableFactoryUtil.java:
##########
@@ -164,4 +171,54 @@ public static boolean isLegacyConnectorOptions(
}
}
}
+
+ /**
+ * Create a DynamicTableSink for ResolvedCatalogTable using table factory associated with the
+ * catalog.
+ */
+ public static Optional<DynamicTableSink> getDynamicTableSink(
Review Comment:
ok
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] luoyuxia commented on pull request #22839: [FLINK-32349][table] Support atomic for CREATE TABLE AS SELECT(CTAS) statement
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on PR #22839:
URL: https://github.com/apache/flink/pull/22839#issuecomment-1628875052
Btw, please also remember to rebase master when you fix my comments.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] Tartarus0zm commented on a diff in pull request #22839: [FLINK-32349][table] Support atomic for CREATE TABLE AS SELECT(CTAS) statement
Posted by "Tartarus0zm (via GitHub)" <gi...@apache.org>.
Tartarus0zm commented on code in PR #22839:
URL: https://github.com/apache/flink/pull/22839#discussion_r1255185193
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java:
##########
@@ -816,7 +825,56 @@ public TableResultInternal executeInternal(List<ModifyOperation> operations) {
List<Transformation<?>> transformations = translate(mapOperations);
List<String> sinkIdentifierNames = extractSinkIdentifierNames(mapOperations);
- return executeInternal(transformations, sinkIdentifierNames);
+ return executeInternal(transformations, sinkIdentifierNames, jobStatusHookList);
+ }
+
+ private ModifyOperation getOperation(
+ CreateTableASOperation ctasOperation, List<JobStatusHook> jobStatusHookList) {
+ CreateTableOperation createTableOperation = ctasOperation.getCreateTableOperation();
+ ObjectIdentifier tableIdentifier = createTableOperation.getTableIdentifier();
+ Catalog catalog = catalogManager.getCatalog(tableIdentifier.getCatalogName()).orElse(null);
+ ResolvedCatalogTable catalogTable =
+ catalogManager.resolveCatalogTable(createTableOperation.getCatalogTable());
+
+ if (!TableFactoryUtil.isLegacyConnectorOptions(
+ catalog,
+ tableConfig,
+ isStreamingMode,
+ tableIdentifier,
+ catalogTable,
+ createTableOperation.isTemporary())) {
+ DynamicTableSink dynamicTableSink =
+ ExecutableOperationUtils.createDynamicTableSink(
+ catalog,
+ () -> moduleManager.getFactory((Module::getTableSinkFactory)),
+ tableIdentifier,
+ catalogTable,
+ Collections.emptyMap(),
+ tableConfig,
+ resourceManager.getUserClassLoader(),
+ createTableOperation.isTemporary());
+ if (dynamicTableSink instanceof SupportsStaging
Review Comment:
good catch
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] Tartarus0zm commented on pull request #22839: [FLINK-32349][table] Support atomic for CREATE TABLE AS SELECT(CTAS) statement
Posted by "Tartarus0zm (via GitHub)" <gi...@apache.org>.
Tartarus0zm commented on PR #22839:
URL: https://github.com/apache/flink/pull/22839#issuecomment-1630171905
@luoyuxia hi, the test is pass, PTAL
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] Tartarus0zm commented on pull request #22839: [FLINK-32349][table] Support atomic for CREATE TABLE AS SELECT(CTAS) statement
Posted by "Tartarus0zm (via GitHub)" <gi...@apache.org>.
Tartarus0zm commented on PR #22839:
URL: https://github.com/apache/flink/pull/22839#issuecomment-1612642680
@luoyuxia hello, I have fix your comment, PTAL
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] luoyuxia commented on a diff in pull request #22839: [FLINK-32349][table] Support atomic for CREATE TABLE AS SELECT(CTAS) statement
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #22839:
URL: https://github.com/apache/flink/pull/22839#discussion_r1250008294
##########
docs/layouts/shortcodes/generated/table_config_configuration.html:
##########
@@ -20,6 +20,12 @@
<td>String</td>
<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 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 in client side, and it will not be dropped even though the job fails or is cancelled. If set this option to true and 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:
nit
```suggestion
<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>
```
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java:
##########
@@ -816,7 +825,56 @@ public TableResultInternal executeInternal(List<ModifyOperation> operations) {
List<Transformation<?>> transformations = translate(mapOperations);
List<String> sinkIdentifierNames = extractSinkIdentifierNames(mapOperations);
- return executeInternal(transformations, sinkIdentifierNames);
+ return executeInternal(transformations, sinkIdentifierNames, jobStatusHookList);
+ }
+
+ private ModifyOperation getOperation(
+ CreateTableASOperation ctasOperation, List<JobStatusHook> jobStatusHookList) {
+ CreateTableOperation createTableOperation = ctasOperation.getCreateTableOperation();
+ ObjectIdentifier tableIdentifier = createTableOperation.getTableIdentifier();
+ Catalog catalog = catalogManager.getCatalog(tableIdentifier.getCatalogName()).orElse(null);
+ ResolvedCatalogTable catalogTable =
+ catalogManager.resolveCatalogTable(createTableOperation.getCatalogTable());
+
+ if (!TableFactoryUtil.isLegacyConnectorOptions(
+ catalog,
+ tableConfig,
+ isStreamingMode,
+ tableIdentifier,
+ catalogTable,
+ createTableOperation.isTemporary())) {
+ DynamicTableSink dynamicTableSink =
+ ExecutableOperationUtils.createDynamicTableSink(
+ catalog,
+ () -> moduleManager.getFactory((Module::getTableSinkFactory)),
+ tableIdentifier,
+ catalogTable,
+ Collections.emptyMap(),
+ tableConfig,
+ resourceManager.getUserClassLoader(),
+ createTableOperation.isTemporary());
+ if (dynamicTableSink instanceof SupportsStaging
Review Comment:
You can check `tableConfig.get(TableConfigOptions.TABLE_CTAS_ATOMICITY_ENABLED)` at first so that we can avoid creating dynamicTableSink in the earlier stage.
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/StagedTable.java:
##########
@@ -0,0 +1,66 @@
+/*
+ * 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.catalog;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.connector.sink.abilities.SupportsStaging;
+
+import java.io.Serializable;
+
+/**
+ * The {@link StagedTable} is designed to implement atomic semantic using a two-phase commit
Review Comment:
```suggestion
* The {@link StagedTable} is designed to implement Flink's atomic semantic for CTAS(CREATE TABLE AS SELECT) statement using a two-phase commit
```
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/StagedTable.java:
##########
@@ -0,0 +1,66 @@
+/*
+ * 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.catalog;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.connector.sink.abilities.SupportsStaging;
+
+import java.io.Serializable;
+
+/**
+ * The {@link StagedTable} is designed to implement atomic semantic using a two-phase commit
+ * protocol. The {@link StagedTable} is supposed to be returned via method {@link
+ * SupportsStaging#applyStaging} by the {@link DynamicTableSink} which implements the {@link
+ * SupportsStaging} interface.
+ *
+ * <p>When the Flink job for writing to a {@link DynamicTableSink} is CREATED, the {@link
+ * StagedTable#begin()} will be called; when the Flink job is FINISHED, the {@link
+ * StagedTable#commit()} will be called; when the Flink job is FAILED or CANCELED, the {@link
+ * StagedTable#abort()} will be called;
+ *
+ * <p>See more in {@link SupportsStaging}.
+ */
+@PublicEvolving
+public interface StagedTable extends Serializable {
+
+ /**
+ * This method will be called when the job is started. In Flink's atomic CTAS scenario, it is
+ * expected to do initialization work; For example, initializing the client of the underlying
+ * service, the tmp path of the underlying storage, or even call the start transaction API of
+ * the underlying service, etc.
+ */
+ void begin();
+
+ /**
+ * This method will be called when the job is succeeds. In Flink's atomic CTAS scenario, it is
+ * expected to do some commit work. For example, moving the underlying data to the target
+ * directory to make it visible, writing buffer data to the underlying storage service, or even
+ * call the commit transaction API of the underlying service, etc.
+ */
+ void commit();
+
+ /**
+ * This method will be called when the job is failed or canceled. In Flink's atomic CTAS
Review Comment:
```suggestion
* This method will be called when the job is failed or is canceled. In Flink's atomic CTAS
```
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/StagedTableITCase.java:
##########
@@ -0,0 +1,148 @@
+/*
+ * 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.batch.sql;
+
+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.table.planner.runtime.utils.BatchTestBase;
+import org.apache.flink.testutils.junit.utils.TempDirUtils;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.FileUtils;
+
+import org.junit.jupiter.api.AfterEach;
+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.nio.file.Path;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+
+/** Tests staged table in batch mode. */
+public class StagedTableITCase extends BatchTestBase {
+
+ @TempDir Path temporaryFolder;
+
+ private File tmpDataFolder;
+
+ @BeforeEach
+ void setup() throws Exception {
+ super.before();
+ List<Row> sourceData = Arrays.asList(Row.of(1, "ZM"));
+
+ TestCollectionTableFactory.reset();
+ TestCollectionTableFactory.initData(sourceData);
+
+ String sourceDDL = "create table t1(a int, b varchar) with ('connector' = 'COLLECTION')";
+ tEnv().executeSql(sourceDDL);
+ tmpDataFolder = TempDirUtils.newFolder(temporaryFolder);
+ }
+
+ @AfterEach
+ void close() {
+ if (tmpDataFolder != null) {
+ tmpDataFolder.delete();
Review Comment:
Why delete it. Can't you provide a temPath for each method?
like:
testStagedTableWithAtomicCtas(@TempDir Path temporaryFolder)
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/connector/sink/abilities/SupportsStaging.java:
##########
@@ -0,0 +1,80 @@
+/*
+ * 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.connector.sink.abilities;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.core.execution.JobStatusHook;
+import org.apache.flink.table.catalog.StagedTable;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+
+/**
+ * Interface for {@link DynamicTableSink}s that support CTAS(CREATE TABLE AS SELECT) statement
+ * atomic semantic using a two-phase commit protocol. The table sink is responsible for telling
+ * planner how to implement atomicity semantics via {@link StagedTable}.
+ *
+ * <p>If the user turns on {@link TableConfigOptions#TABLE_CTAS_ATOMICITY_ENABLED}, and {@link
Review Comment:
suggestion:
If the user turns on {@link TableConfigOptions#TABLE_CTAS_ATOMICITY_ENABLED}, and the {@link DynamicTableSink} implements {@link SupportsStaging}, the planner will call method {@link #applyStaging(StagingContext)} to get the {@link StagedTable} returned by the sink, then {@link StagedTable} will be used by Flink to implement a two-phase commit with the actual implementation of {@link StagedTable}.
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/StagedTableITCase.java:
##########
@@ -0,0 +1,148 @@
+/*
+ * 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.batch.sql;
+
+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.table.planner.runtime.utils.BatchTestBase;
+import org.apache.flink.testutils.junit.utils.TempDirUtils;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.FileUtils;
+
+import org.junit.jupiter.api.AfterEach;
+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.nio.file.Path;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+
+/** Tests staged table in batch mode. */
+public class StagedTableITCase extends BatchTestBase {
+
+ @TempDir Path temporaryFolder;
+
+ private File tmpDataFolder;
+
+ @BeforeEach
+ void setup() throws Exception {
+ super.before();
+ List<Row> sourceData = Arrays.asList(Row.of(1, "ZM"));
+
+ TestCollectionTableFactory.reset();
+ TestCollectionTableFactory.initData(sourceData);
+
+ String sourceDDL = "create table t1(a int, b varchar) with ('connector' = 'COLLECTION')";
+ tEnv().executeSql(sourceDDL);
+ tmpDataFolder = TempDirUtils.newFolder(temporaryFolder);
+ }
+
+ @AfterEach
+ void close() {
+ if (tmpDataFolder != null) {
+ tmpDataFolder.delete();
+ }
+ }
+
+ @Test
+ void testStagedTableWithAtomicCtas() throws Exception {
Review Comment:
```suggestion
void testAtomicCtas() throws Exception {
```
It's not for test staged table, it's for test atomic ctas.
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/execution/CtasJobStatusHook.java:
##########
@@ -0,0 +1,56 @@
+/*
+ * 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.execution;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.core.execution.JobStatusHook;
+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 changing.
Review Comment:
```suggestion
* call the corresponding interfaces of the inner {@link StagedTable} on job status changes.
```
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/config/TableConfigOptions.java:
##########
@@ -197,6 +197,17 @@ private TableConfigOptions() {}
.withDescription(
"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")
Review Comment:
dito: don't forget to also update in here.
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/StagedTable.java:
##########
@@ -0,0 +1,66 @@
+/*
+ * 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.catalog;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.connector.sink.abilities.SupportsStaging;
+
+import java.io.Serializable;
+
+/**
+ * The {@link StagedTable} is designed to implement atomic semantic using a two-phase commit
+ * protocol. The {@link StagedTable} is supposed to be returned via method {@link
+ * SupportsStaging#applyStaging} by the {@link DynamicTableSink} which implements the {@link
+ * SupportsStaging} interface.
+ *
+ * <p>When the Flink job for writing to a {@link DynamicTableSink} is CREATED, the {@link
+ * StagedTable#begin()} will be called; when the Flink job is FINISHED, the {@link
+ * StagedTable#commit()} will be called; when the Flink job is FAILED or CANCELED, the {@link
+ * StagedTable#abort()} will be called;
+ *
+ * <p>See more in {@link SupportsStaging}.
+ */
+@PublicEvolving
+public interface StagedTable extends Serializable {
+
+ /**
+ * This method will be called when the job is started. In Flink's atomic CTAS scenario, it is
+ * expected to do initialization work; For example, initializing the client of the underlying
+ * service, the tmp path of the underlying storage, or even call the start transaction API of
+ * the underlying service, etc.
+ */
+ void begin();
+
+ /**
+ * This method will be called when the job is succeeds. In Flink's atomic CTAS scenario, it is
Review Comment:
```suggestion
* This method will be called when the job succeeds. In Flink's atomic CTAS scenario, it is
```
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/connector/sink/abilities/SupportsStaging.java:
##########
@@ -0,0 +1,80 @@
+/*
+ * 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.connector.sink.abilities;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.core.execution.JobStatusHook;
+import org.apache.flink.table.catalog.StagedTable;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+
+/**
+ * Interface for {@link DynamicTableSink}s that support CTAS(CREATE TABLE AS SELECT) statement
Review Comment:
suggestion:
Interface for {@link DynamicTableSink}s that support atomic semantic for CTAS(CREATE TABLE AS SELECT) statement using a two-phase commit protocol. The table sink is responsible for returning
a {@link StagedTable} to tell the Flink how to implement the atomicity semantics.
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/StagedTable.java:
##########
@@ -0,0 +1,66 @@
+/*
+ * 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.catalog;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.connector.sink.abilities.SupportsStaging;
+
+import java.io.Serializable;
+
+/**
+ * The {@link StagedTable} is designed to implement atomic semantic using a two-phase commit
+ * protocol. The {@link StagedTable} is supposed to be returned via method {@link
+ * SupportsStaging#applyStaging} by the {@link DynamicTableSink} which implements the {@link
+ * SupportsStaging} interface.
+ *
+ * <p>When the Flink job for writing to a {@link DynamicTableSink} is CREATED, the {@link
+ * StagedTable#begin()} will be called; when the Flink job is FINISHED, the {@link
+ * StagedTable#commit()} will be called; when the Flink job is FAILED or CANCELED, the {@link
+ * StagedTable#abort()} will be called;
+ *
+ * <p>See more in {@link SupportsStaging}.
+ */
+@PublicEvolving
+public interface StagedTable extends Serializable {
+
+ /**
+ * This method will be called when the job is started. In Flink's atomic CTAS scenario, it is
+ * expected to do initialization work; For example, initializing the client of the underlying
+ * service, the tmp path of the underlying storage, or even call the start transaction API of
+ * the underlying service, etc.
+ */
+ void begin();
+
+ /**
+ * This method will be called when the job is succeeds. In Flink's atomic CTAS scenario, it is
+ * expected to do some commit work. For example, moving the underlying data to the target
+ * directory to make it visible, writing buffer data to the underlying storage service, or even
+ * call the commit transaction API of the underlying service, etc.
+ */
+ void commit();
+
+ /**
+ * This method will be called when the job is failed or canceled. In Flink's atomic CTAS
+ * scenario, it is expected to do some cleaning work for a writing; For example, delete the data
+ * in tmp directory, delete the temporary data in the underlying storage service, or even call
Review Comment:
```suggestion
* in the tmp directory, delete the temporary data in the underlying storage service, or even call
```
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/StagedTable.java:
##########
@@ -0,0 +1,66 @@
+/*
+ * 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.catalog;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.connector.sink.abilities.SupportsStaging;
+
+import java.io.Serializable;
+
+/**
+ * The {@link StagedTable} is designed to implement atomic semantic using a two-phase commit
+ * protocol. The {@link StagedTable} is supposed to be returned via method {@link
+ * SupportsStaging#applyStaging} by the {@link DynamicTableSink} which implements the {@link
+ * SupportsStaging} interface.
+ *
+ * <p>When the Flink job for writing to a {@link DynamicTableSink} is CREATED, the {@link
Review Comment:
```suggestion
* <p>When the Flink job for writing to a {@link DynamicTableSink} with atomic semantic supporting is CREATED, the {@link
```
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/StagedTable.java:
##########
@@ -0,0 +1,66 @@
+/*
+ * 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.catalog;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.connector.sink.abilities.SupportsStaging;
+
+import java.io.Serializable;
+
+/**
+ * The {@link StagedTable} is designed to implement atomic semantic using a two-phase commit
+ * protocol. The {@link StagedTable} is supposed to be returned via method {@link
+ * SupportsStaging#applyStaging} by the {@link DynamicTableSink} which implements the {@link
+ * SupportsStaging} interface.
+ *
+ * <p>When the Flink job for writing to a {@link DynamicTableSink} is CREATED, the {@link
+ * StagedTable#begin()} will be called; when the Flink job is FINISHED, the {@link
+ * StagedTable#commit()} will be called; when the Flink job is FAILED or CANCELED, the {@link
+ * StagedTable#abort()} will be called;
+ *
+ * <p>See more in {@link SupportsStaging}.
+ */
+@PublicEvolving
+public interface StagedTable extends Serializable {
+
+ /**
+ * This method will be called when the job is started. In Flink's atomic CTAS scenario, it is
+ * expected to do initialization work; For example, initializing the client of the underlying
+ * service, the tmp path of the underlying storage, or even call the start transaction API of
+ * the underlying service, etc.
+ */
+ void begin();
+
+ /**
+ * This method will be called when the job is succeeds. In Flink's atomic CTAS scenario, it is
+ * expected to do some commit work. For example, moving the underlying data to the target
+ * directory to make it visible, writing buffer data to the underlying storage service, or even
+ * call the commit transaction API of the underlying service, etc.
+ */
+ void commit();
+
+ /**
+ * This method will be called when the job is failed or canceled. In Flink's atomic CTAS
+ * scenario, it is expected to do some cleaning work for a writing; For example, delete the data
Review Comment:
```suggestion
* scenario, it is expected to do some cleaning work for writing; For example, delete the data
```
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/StagedTableITCase.java:
##########
@@ -0,0 +1,149 @@
+/*
+ * 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.stream.sql;
+
+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.table.planner.runtime.utils.StreamingTestBase;
+import org.apache.flink.testutils.junit.utils.TempDirUtils;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.FileUtils;
+
+import org.junit.jupiter.api.AfterEach;
+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.nio.file.Path;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+
+/** Tests staged table in stream mode. */
+public class StagedTableITCase extends StreamingTestBase {
Review Comment:
dito
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/StagedTableITCase.java:
##########
@@ -0,0 +1,148 @@
+/*
+ * 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.batch.sql;
+
+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.table.planner.runtime.utils.BatchTestBase;
+import org.apache.flink.testutils.junit.utils.TempDirUtils;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.FileUtils;
+
+import org.junit.jupiter.api.AfterEach;
+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.nio.file.Path;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+
+/** Tests staged table in batch mode. */
+public class StagedTableITCase extends BatchTestBase {
Review Comment:
```suggestion
public class AtomicCtasITCase extends BatchTestBase {
```
?
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/StagedTableITCase.java:
##########
@@ -0,0 +1,148 @@
+/*
+ * 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.batch.sql;
+
+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.table.planner.runtime.utils.BatchTestBase;
+import org.apache.flink.testutils.junit.utils.TempDirUtils;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.FileUtils;
+
+import org.junit.jupiter.api.AfterEach;
+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.nio.file.Path;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+
+/** Tests staged table in batch mode. */
+public class StagedTableITCase extends BatchTestBase {
+
+ @TempDir Path temporaryFolder;
+
+ private File tmpDataFolder;
+
+ @BeforeEach
+ void setup() throws Exception {
+ super.before();
+ List<Row> sourceData = Arrays.asList(Row.of(1, "ZM"));
+
+ TestCollectionTableFactory.reset();
+ TestCollectionTableFactory.initData(sourceData);
+
+ String sourceDDL = "create table t1(a int, b varchar) with ('connector' = 'COLLECTION')";
+ tEnv().executeSql(sourceDDL);
+ tmpDataFolder = TempDirUtils.newFolder(temporaryFolder);
+ }
+
+ @AfterEach
+ void close() {
+ if (tmpDataFolder != null) {
+ tmpDataFolder.delete();
+ }
+ }
+
+ @Test
+ void testStagedTableWithAtomicCtas() throws Exception {
+ commonTestStagedTableWithAtomicCtas("ctas_batch_table", false);
+ }
+
+ @Test
+ void testStagedTableWithAtomicCtasIfNotExists() throws Exception {
+ commonTestStagedTableWithAtomicCtas("ctas_if_not_exists_batch_table", true);
+ }
+
+ void commonTestStagedTableWithAtomicCtas(String tableName, boolean ifNotExists)
+ throws Exception {
+ tEnv().getConfig().set(TableConfigOptions.TABLE_CTAS_ATOMICITY_ENABLED, true);
+ String dataDir = tmpDataFolder.getAbsolutePath();
+ String sqlFragment = ifNotExists ? " if not exists " + tableName : tableName;
+ tEnv().executeSql(
+ "create table "
+ + sqlFragment
+ + " with ('connector' = 'test-staging', 'data-dir' = '"
+ + dataDir
+ + "') as select * from t1")
+ .await();
+ assertThat(tEnv().listTables()).doesNotContain(tableName);
+ File file = new File(dataDir, "data");
+ assertThat(file).exists();
+ assertThat(file).isFile();
+ assertThat(FileUtils.readFileUtf8(file)).isEqualTo("1,ZM");
+ 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 (ifNotExists) {
+ assertThat(TestSupportsStagingTableFactory.STAGING_PURPOSE_LIST)
+ .contains(SupportsStaging.StagingPurpose.CREATE_TABLE_AS_IF_NOT_EXISTS);
+ } else {
+ assertThat(TestSupportsStagingTableFactory.STAGING_PURPOSE_LIST)
+ .contains(SupportsStaging.StagingPurpose.CREATE_TABLE_AS);
+ }
+ }
+
+ @Test
+ void testFailStagedTableWithAtomicCtas() {
+ tEnv().getConfig().set(TableConfigOptions.TABLE_CTAS_ATOMICITY_ENABLED, true);
+ String dataDir = tmpDataFolder.getAbsolutePath();
+ assertThatCode(
+ () ->
+ tEnv().executeSql(
+ "create table ctas_batch_table_fail with ('connector' = 'test-staging', 'data-dir' = '"
+ + dataDir
+ + "', 'sink-fail' = '"
+ + true
+ + "') as select * from t1")
+ .await())
+ .hasRootCauseMessage("Test StagedTable abort method.");
+
+ assertThat(TestSupportsStagingTableFactory.JOB_STATUS_CHANGE_PROCESS).hasSize(2);
+ assertThat(TestSupportsStagingTableFactory.JOB_STATUS_CHANGE_PROCESS)
+ .contains("begin", "abort");
+ }
+
+ @Test
+ void testStagedTableWithoutAtomicCtas() throws Exception {
+ tEnv().getConfig().set(TableConfigOptions.TABLE_CTAS_ATOMICITY_ENABLED, false);
+ String dataDir = tmpDataFolder.getAbsolutePath();
+ tEnv().executeSql(
+ "create table ctas_batch_table with ('connector' = 'test-staging', 'data-dir' = '"
+ + dataDir
+ + "') as select * from t1")
+ .await();
+ assertThat(tEnv().listTables()).contains("ctas_batch_table");
+ // Not using StagedTable, so need to read the hidden file
+ File file = new File(dataDir, "_data");
+ assertThat(file).exists();
Review Comment:
Please extract
```
assertThat(file).exists();
assertThat(file).isFile();
assertThat(FileUtils.readFileUtf8(file)).isEqualTo("1,ZM");
```
to a common method
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] luoyuxia commented on a diff in pull request #22839: [FLINK-32349][table] Support atomic for CREATE TABLE AS SELECT(CTAS) statement
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #22839:
URL: https://github.com/apache/flink/pull/22839#discussion_r1242157031
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestSupportsStagingTableFactory.java:
##########
@@ -0,0 +1,171 @@
+/*
+ * 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.factories;
+
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.DataStreamSink;
+import org.apache.flink.streaming.api.functions.sink.DiscardingSink;
+import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
+import org.apache.flink.table.catalog.StagedTable;
+import org.apache.flink.table.connector.ChangelogMode;
+import org.apache.flink.table.connector.ProviderContext;
+import org.apache.flink.table.connector.sink.DataStreamSinkProvider;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.connector.sink.abilities.SupportsStaging;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.factories.DynamicTableSinkFactory;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.util.FileUtils;
+
+import java.io.File;
+import java.util.Collections;
+import java.util.Set;
+
+/** A factory to create table to support staging for test purpose. */
+public class TestSupportsStagingTableFactory implements DynamicTableSinkFactory {
+
+ public static final String IDENTIFIER = "test-staging";
+
+ private static final ConfigOption<String> DATA_DIR =
+ ConfigOptions.key("data-dir")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("The data id used to write the rows.");
+
+ @Override
+ public DynamicTableSink createDynamicTableSink(Context context) {
+ FactoryUtil.TableFactoryHelper helper = FactoryUtil.createTableFactoryHelper(this, context);
+ helper.validate();
+ String dataDir = helper.getOptions().get(DATA_DIR);
+ return new SupportsStagingTableSink(dataDir);
+ }
+
+ @Override
+ public String factoryIdentifier() {
+ return IDENTIFIER;
+ }
+
+ @Override
+ public Set<ConfigOption<?>> requiredOptions() {
+ return Collections.singleton(DATA_DIR);
+ }
+
+ @Override
+ public Set<ConfigOption<?>> optionalOptions() {
+ return Collections.emptySet();
+ }
+
+ /** A sink that supports staging. */
+ private static class SupportsStagingTableSink implements DynamicTableSink, SupportsStaging {
+
+ private String dataDir;
+ private TestStagedTable stagedTable;
+
+ public SupportsStagingTableSink(String dataDir) {
+ this(dataDir, null);
+ }
+
+ public SupportsStagingTableSink(String dataDir, TestStagedTable stagedTable) {
+ this.dataDir = dataDir;
+ this.stagedTable = stagedTable;
+ }
+
+ @Override
+ public ChangelogMode getChangelogMode(ChangelogMode requestedMode) {
+ return ChangelogMode.insertOnly();
+ }
+
+ @Override
+ public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
+ return new DataStreamSinkProvider() {
+ @Override
+ public DataStreamSink<?> consumeDataStream(
+ ProviderContext providerContext, DataStream<RowData> dataStream) {
+ if (stagedTable != null) {
+ return dataStream
+ .addSink(new StagedSinkFunction(dataDir))
+ .setParallelism(1);
+ } else {
+ // otherwise, do nothing
+ return dataStream.addSink(new DiscardingSink<>());
+ }
+ }
+ };
+ }
+
+ @Override
+ public DynamicTableSink copy() {
+ return new SupportsStagingTableSink(dataDir, stagedTable);
+ }
+
+ @Override
+ public String asSummaryString() {
+ return "SupportsStagingTableSink";
+ }
+
+ @Override
+ public StagedTable applyStaging(StagingContext context) {
+ stagedTable = new TestStagedTable();
+ return stagedTable;
+ }
+ }
+
+ /** The sink for delete existing data. */
+ private static class StagedSinkFunction extends RichSinkFunction<RowData> {
+
+ private String dataDir;
+
+ public StagedSinkFunction(String dataDir) {
+ this.dataDir = dataDir;
+ }
+
+ @Override
+ public void open(Configuration parameters) throws Exception {
+ super.open(parameters);
+ File parentDir = new File(dataDir);
+ if (parentDir.exists()) {
+ parentDir.delete();
+ }
+ parentDir.mkdirs();
+ new File(dataDir, "data").createNewFile();
+ }
+
+ @Override
+ public void invoke(RowData value, Context context) throws Exception {
+ FileUtils.writeFileUtf8(
+ new File(dataDir, "data"), value.getInt(0) + "," + value.getString(1));
+ }
+ }
+
+ /** A StagedTable for test. */
+ private static class TestStagedTable implements StagedTable {
+
+ @Override
+ public void begin() {}
Review Comment:
I think we should think of a way to test it.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] Tartarus0zm commented on pull request #22839: [FLINK-32349][table] Support atomic for CREATE TABLE AS SELECT(CTAS) statement
Posted by "Tartarus0zm (via GitHub)" <gi...@apache.org>.
Tartarus0zm commented on PR #22839:
URL: https://github.com/apache/flink/pull/22839#issuecomment-1624814266
@luoyuxia hello, I have fix your comments, PTAL
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org