You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by yu...@apache.org on 2023/06/29 03:03:51 UTC
[flink] branch master updated (7466df0c3c5 -> 300902016bd)
This is an automated email from the ASF dual-hosted git repository.
yuxia pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
from 7466df0c3c5 [FLINK-32283][network] Implement the SortBufferAccumulator for the tiered storage
new 45ce1926274 [FLINK-32234][table] Convert SQLTruncateTable to TruncateTableOperation
new f8df2d608af [FLINK-32234][table] Support to execute truncate table operation in TableEnvironmentImpl
new 300902016bd [FLINK-32234][sql-client] Add test for truncate table in sql-client This closes #22696
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../flink/table/client/cli/CliClientITCase.java | 5 +-
.../test/resources/sql/{delete.q => truncate.q} | 32 ++----
.../internal/ExecutableOperationContextImpl.java | 11 +-
.../table/api/internal/TableEnvironmentImpl.java | 3 +-
.../table/operations/ExecutableOperation.java | 2 +
.../table/operations/TruncateTableOperation.java | 119 +++++++++++++++++++++
.../operations/utils/ExecutableOperationUtils.java | 75 +++++++++++++
.../planner/operations/DeletePushDownUtils.java | 26 ++---
.../operations/converters/SqlNodeConverters.java | 1 +
...nverter.java => SqlTruncateTableConverter.java} | 26 ++---
.../table/planner/calcite/FlinkPlannerImpl.scala | 3 +-
.../factories/TestUpdateDeleteTableFactory.java | 8 +-
.../operations/SqlDmlToOperationConverterTest.java | 13 +++
.../runtime/batch/sql/TruncateTableITCase.java | 105 ++++++++++++++++++
...teTableITCase.java => TruncateTableITCase.java} | 20 ++--
15 files changed, 370 insertions(+), 79 deletions(-)
copy flink-table/flink-sql-client/src/test/resources/sql/{delete.q => truncate.q} (69%)
create mode 100644 flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/TruncateTableOperation.java
create mode 100644 flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/utils/ExecutableOperationUtils.java
copy flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/{SqlAlterViewRenameConverter.java => SqlTruncateTableConverter.java} (52%)
create mode 100644 flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/TruncateTableITCase.java
copy flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/{DeleteTableITCase.java => TruncateTableITCase.java} (62%)
[flink] 02/03: [FLINK-32234][table] Support to execute truncate table operation in TableEnvironmentImpl
Posted by yu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yuxia pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit f8df2d608aff7cfa4a592532cfe280139a2960ad
Author: luoyuxia <lu...@alumni.sjtu.edu.cn>
AuthorDate: Thu Jun 1 18:04:58 2023 +0800
[FLINK-32234][table] Support to execute truncate table operation in TableEnvironmentImpl
---
.../internal/ExecutableOperationContextImpl.java | 11 ++-
.../table/api/internal/TableEnvironmentImpl.java | 3 +-
.../table/operations/ExecutableOperation.java | 2 +
.../table/operations/TruncateTableOperation.java | 79 +++++++++++++++-
.../operations/utils/ExecutableOperationUtils.java | 75 +++++++++++++++
.../planner/operations/DeletePushDownUtils.java | 26 ++---
.../factories/TestUpdateDeleteTableFactory.java | 8 +-
.../runtime/batch/sql/TruncateTableITCase.java | 105 +++++++++++++++++++++
.../runtime/stream/sql/TruncateTableITCase.java | 39 ++++++++
9 files changed, 324 insertions(+), 24 deletions(-)
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/ExecutableOperationContextImpl.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/ExecutableOperationContextImpl.java
index 17503b4b060..6c45e8de574 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/ExecutableOperationContextImpl.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/ExecutableOperationContextImpl.java
@@ -34,17 +34,21 @@ public class ExecutableOperationContextImpl implements ExecutableOperation.Conte
private final ResourceManager resourceManager;
private final TableConfig tableConfig;
+ private final boolean isStreamingMode;
+
public ExecutableOperationContextImpl(
CatalogManager catalogManager,
FunctionCatalog functionCatalog,
ModuleManager moduleManager,
ResourceManager resourceManager,
- TableConfig tableConfig) {
+ TableConfig tableConfig,
+ boolean isStreamingMode) {
this.catalogManager = catalogManager;
this.functionCatalog = functionCatalog;
this.moduleManager = moduleManager;
this.resourceManager = resourceManager;
this.tableConfig = tableConfig;
+ this.isStreamingMode = isStreamingMode;
}
@Override
@@ -71,4 +75,9 @@ public class ExecutableOperationContextImpl implements ExecutableOperation.Conte
public TableConfig getTableConfig() {
return tableConfig;
}
+
+ @Override
+ public boolean isStreamingMode() {
+ return isStreamingMode;
+ }
}
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java
index 75198df1c8a..4c7fda752fc 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java
@@ -208,7 +208,8 @@ public class TableEnvironmentImpl implements TableEnvironmentInternal {
functionCatalog,
moduleManager,
resourceManager,
- tableConfig);
+ tableConfig,
+ isStreamingMode);
}
public static TableEnvironmentImpl create(Configuration configuration) {
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ExecutableOperation.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ExecutableOperation.java
index 5109268a5a9..21c4e054bf8 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ExecutableOperation.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ExecutableOperation.java
@@ -63,5 +63,7 @@ public interface ExecutableOperation extends Operation {
ResourceManager getResourceManager();
TableConfig getTableConfig();
+
+ boolean isStreamingMode();
}
}
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/TruncateTableOperation.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/TruncateTableOperation.java
index c0d698ee0fe..64715b49708 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/TruncateTableOperation.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/TruncateTableOperation.java
@@ -19,10 +19,23 @@
package org.apache.flink.table.operations;
import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.api.internal.TableResultInternal;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogManager;
+import org.apache.flink.table.catalog.ContextResolvedTable;
import org.apache.flink.table.catalog.ObjectIdentifier;
+import org.apache.flink.table.catalog.ResolvedCatalogTable;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.connector.sink.abilities.SupportsTruncate;
+import org.apache.flink.table.factories.TableFactoryUtil;
+import org.apache.flink.table.module.Module;
+import org.apache.flink.table.operations.utils.ExecutableOperationUtils;
+
+import java.util.Collections;
/** Operation to describe an {@code TRUNCATE TABLE} statement. */
-public class TruncateTableOperation implements Operation {
+public class TruncateTableOperation implements ExecutableOperation {
private final ObjectIdentifier tableIdentifier;
@@ -30,6 +43,70 @@ public class TruncateTableOperation implements Operation {
this.tableIdentifier = tableIdentifier;
}
+ @Override
+ public TableResultInternal execute(Context ctx) {
+ if (ctx.isStreamingMode()) {
+ throw new TableException(
+ "TRUNCATE TABLE statement is not supported in streaming mode.");
+ }
+
+ CatalogManager catalogManager = ctx.getCatalogManager();
+ ContextResolvedTable contextResolvedTable = catalogManager.getTableOrError(tableIdentifier);
+ CatalogBaseTable catalogBaseTable = contextResolvedTable.getTable();
+ if (catalogBaseTable.getTableKind() == CatalogBaseTable.TableKind.VIEW) {
+ throw new TableException("TRUNCATE TABLE statement is not supported for view.");
+ }
+
+ ResolvedCatalogTable resolvedTable = contextResolvedTable.getResolvedTable();
+ ObjectIdentifier objectIdentifier = contextResolvedTable.getIdentifier();
+ if (contextResolvedTable.isAnonymous()) {
+ throw new TableException(
+ "TRUNCATE TABLE statement is not supported for anonymous table.");
+ }
+
+ if (TableFactoryUtil.isLegacyConnectorOptions(
+ catalogManager.getCatalog(objectIdentifier.getCatalogName()).orElse(null),
+ ctx.getTableConfig(),
+ ctx.isStreamingMode(),
+ tableIdentifier,
+ resolvedTable,
+ contextResolvedTable.isTemporary())) {
+ throw new TableException(
+ String.format(
+ "Can't perform truncate table operation of the table %s"
+ + " because the corresponding table sink is the legacy "
+ + "TableSink."
+ + " Please implement %s for it.",
+ tableIdentifier, DynamicTableSink.class.getName()));
+ }
+
+ DynamicTableSink tableSink =
+ ExecutableOperationUtils.createDynamicTableSink(
+ contextResolvedTable.getCatalog().orElse(null),
+ () -> ctx.getModuleManager().getFactory((Module::getTableSinkFactory)),
+ objectIdentifier,
+ resolvedTable,
+ Collections.emptyMap(),
+ ctx.getTableConfig(),
+ ctx.getResourceManager().getUserClassLoader(),
+ contextResolvedTable.isTemporary());
+ if (tableSink instanceof SupportsTruncate) {
+ SupportsTruncate supportsTruncateTableSink = (SupportsTruncate) tableSink;
+ try {
+ supportsTruncateTableSink.executeTruncation();
+ return TableResultInternal.TABLE_RESULT_OK;
+ } catch (Exception e) {
+ throw new TableException("Fail to execute TRUNCATE TABLE statement.", e);
+ }
+ } else {
+ throw new TableException(
+ String.format(
+ "TRUNCATE TABLE statement is not supported for the table %s since the table hasn't "
+ + "implemented the interface %s.",
+ tableIdentifier.asSummaryString(), SupportsTruncate.class.getName()));
+ }
+ }
+
@VisibleForTesting
public ObjectIdentifier getTableIdentifier() {
return tableIdentifier;
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/utils/ExecutableOperationUtils.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/utils/ExecutableOperationUtils.java
new file mode 100644
index 00000000000..24908c627b4
--- /dev/null
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/utils/ExecutableOperationUtils.java
@@ -0,0 +1,75 @@
+/*
+ * 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.utils;
+
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.ObjectIdentifier;
+import org.apache.flink.table.catalog.ResolvedCatalogTable;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.factories.DynamicTableSinkFactory;
+import org.apache.flink.table.factories.Factory;
+import org.apache.flink.table.factories.FactoryUtil;
+
+import javax.annotation.Nullable;
+
+import java.util.Map;
+import java.util.Optional;
+import java.util.function.Supplier;
+
+/** Utils for the executable operation. */
+public class ExecutableOperationUtils {
+
+ /**
+ * Creates a {@link DynamicTableSink} from a {@link CatalogTable}.
+ *
+ * <p>It'll try to create table sink from to {@param catalog}, then try to create from {@param
+ * sinkFactorySupplier} passed secondly. Otherwise, an attempt is made to discover a matching
+ * factory using Java SPI (see {@link Factory} for details).
+ */
+ public static DynamicTableSink createDynamicTableSink(
+ @Nullable Catalog catalog,
+ Supplier<Optional<DynamicTableSinkFactory>> sinkFactorySupplier,
+ ObjectIdentifier objectIdentifier,
+ ResolvedCatalogTable catalogTable,
+ Map<String, String> enrichmentOptions,
+ ReadableConfig configuration,
+ ClassLoader classLoader,
+ boolean isTemporary) {
+ DynamicTableSinkFactory dynamicTableSinkFactory = null;
+ if (catalog != null
+ && catalog.getFactory().isPresent()
+ && catalog.getFactory().get() instanceof DynamicTableSinkFactory) {
+ // try get from catalog
+ dynamicTableSinkFactory = (DynamicTableSinkFactory) catalog.getFactory().get();
+ }
+ if (dynamicTableSinkFactory == null) {
+ dynamicTableSinkFactory = sinkFactorySupplier.get().orElse(null);
+ }
+ return FactoryUtil.createDynamicTableSink(
+ dynamicTableSinkFactory,
+ objectIdentifier,
+ catalogTable,
+ enrichmentOptions,
+ configuration,
+ classLoader,
+ isTemporary);
+ }
+}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/DeletePushDownUtils.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/DeletePushDownUtils.java
index c9682ed5078..d5270cb89b7 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/DeletePushDownUtils.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/DeletePushDownUtils.java
@@ -30,10 +30,9 @@ import org.apache.flink.table.connector.sink.DynamicTableSink;
import org.apache.flink.table.expressions.Expression;
import org.apache.flink.table.expressions.ResolvedExpression;
import org.apache.flink.table.expressions.resolver.ExpressionResolver;
-import org.apache.flink.table.factories.DynamicTableSinkFactory;
-import org.apache.flink.table.factories.FactoryUtil;
import org.apache.flink.table.factories.TableFactoryUtil;
import org.apache.flink.table.module.Module;
+import org.apache.flink.table.operations.utils.ExecutableOperationUtils;
import org.apache.flink.table.planner.calcite.FlinkContext;
import org.apache.flink.table.planner.plan.rules.logical.SimplifyFilterConditionRule;
import org.apache.flink.table.planner.plan.utils.FlinkRexUtil;
@@ -94,26 +93,13 @@ public class DeletePushDownUtils {
objectIdentifier,
resolvedTable,
isTemporary)) {
- DynamicTableSinkFactory dynamicTableSinkFactory = null;
- if (optionalCatalog.isPresent()
- && optionalCatalog.get().getFactory().isPresent()
- && optionalCatalog.get().getFactory().get()
- instanceof DynamicTableSinkFactory) {
- // try get from catalog
- dynamicTableSinkFactory =
- (DynamicTableSinkFactory) optionalCatalog.get().getFactory().get();
- }
-
- if (dynamicTableSinkFactory == null) {
- Optional<DynamicTableSinkFactory> factoryFromModule =
- context.getModuleManager().getFactory((Module::getTableSinkFactory));
- // then try get from module
- dynamicTableSinkFactory = factoryFromModule.orElse(null);
- }
// create table dynamic table sink
DynamicTableSink tableSink =
- FactoryUtil.createDynamicTableSink(
- dynamicTableSinkFactory,
+ ExecutableOperationUtils.createDynamicTableSink(
+ optionalCatalog.orElse(null),
+ () ->
+ context.getModuleManager()
+ .getFactory((Module::getTableSinkFactory)),
objectIdentifier,
resolvedTable,
Collections.emptyMap(),
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestUpdateDeleteTableFactory.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestUpdateDeleteTableFactory.java
index 2e9c9141d54..05ee1dc044e 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestUpdateDeleteTableFactory.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestUpdateDeleteTableFactory.java
@@ -42,6 +42,7 @@ import org.apache.flink.table.connector.sink.DynamicTableSink;
import org.apache.flink.table.connector.sink.abilities.SupportsDeletePushDown;
import org.apache.flink.table.connector.sink.abilities.SupportsRowLevelDelete;
import org.apache.flink.table.connector.sink.abilities.SupportsRowLevelUpdate;
+import org.apache.flink.table.connector.sink.abilities.SupportsTruncate;
import org.apache.flink.table.connector.source.DynamicTableSource;
import org.apache.flink.table.connector.source.ScanTableSource;
import org.apache.flink.table.connector.source.SourceFunctionProvider;
@@ -666,7 +667,7 @@ public class TestUpdateDeleteTableFactory
/** A sink that supports delete push down and row-level update. */
public static class SupportsDeletePushDownSink extends SupportsRowLevelUpdateSink
- implements SupportsDeletePushDown {
+ implements SupportsDeletePushDown, SupportsTruncate {
private final String dataId;
private final boolean onlyAcceptEqualPredicate;
@@ -742,6 +743,11 @@ public class TestUpdateDeleteTableFactory
}
return Optional.empty();
}
+
+ @Override
+ public void executeTruncation() {
+ registeredRowData.put(dataId, Collections.emptyList());
+ }
}
private static int[] getRequiredColumnIndexes(
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/TruncateTableITCase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/TruncateTableITCase.java
new file mode 100644
index 00000000000..5e09fd2b521
--- /dev/null
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/TruncateTableITCase.java
@@ -0,0 +1,105 @@
+/*
+ * 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.TableException;
+import org.apache.flink.table.api.TableResult;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.planner.factories.TestUpdateDeleteTableFactory;
+import org.apache.flink.table.planner.runtime.utils.BatchTestBase;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.CollectionUtil;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** The IT case for Truncate table statement in batch mode. */
+public class TruncateTableITCase extends BatchTestBase {
+ private static final int ROW_NUM = 2;
+
+ @Test
+ void testTruncateTable() {
+ String dataId = registerData();
+ tEnv().executeSql(
+ String.format(
+ "CREATE TABLE t (a int) WITH"
+ + " ('connector' = 'test-update-delete',"
+ + " 'data-id' = '%s',"
+ + " 'only-accept-equal-predicate' = 'true'"
+ + ")",
+ dataId));
+ List<Row> rows = toRows(tEnv().executeSql("SELECT * FROM t"));
+ assertThat(rows.toString()).isEqualTo("[+I[0], +I[1]]");
+ tEnv().executeSql("TRUNCATE TABLE t");
+ rows = toRows(tEnv().executeSql("SELECT * FROM t"));
+ assertThat(rows).isEmpty();
+ }
+
+ @Test
+ void testTruncateTableWithoutImplementation() {
+ tEnv().executeSql("CREATE TABLE t (a int) WITH" + " ('connector' = 'values')");
+ assertThatThrownBy(() -> tEnv().executeSql("TRUNCATE TABLE t"))
+ .isInstanceOf(TableException.class)
+ .hasMessage(
+ "TRUNCATE TABLE statement is not supported for the table default_catalog.default_database.t"
+ + " since the table hasn't implemented the interface"
+ + " org.apache.flink.table.connector.sink.abilities.SupportsTruncate.");
+ }
+
+ @Test
+ void testTruncateLegacyTable() {
+ tEnv().executeSql(
+ "CREATE TABLE t (a int, b string, c double) WITH"
+ + " ('connector' = 'COLLECTION')");
+ assertThatThrownBy(() -> tEnv().executeSql("TRUNCATE TABLE t"))
+ .isInstanceOf(TableException.class)
+ .hasMessage(
+ String.format(
+ "Can't perform truncate table operation of the table %s "
+ + "because the corresponding table sink is the legacy "
+ + "TableSink."
+ + " Please implement %s for it.",
+ "`default_catalog`.`default_database`.`t`",
+ DynamicTableSink.class.getName()));
+ }
+
+ private String registerData() {
+ List<RowData> values = createValue();
+ return TestUpdateDeleteTableFactory.registerRowData(values);
+ }
+
+ private List<RowData> createValue() {
+ List<RowData> values = new ArrayList<>();
+ for (int i = 0; i < ROW_NUM; i++) {
+ values.add(GenericRowData.of(i));
+ }
+ return values;
+ }
+
+ private List<Row> toRows(TableResult result) {
+ return CollectionUtil.iteratorToList(result.collect());
+ }
+}
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/TruncateTableITCase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/TruncateTableITCase.java
new file mode 100644
index 00000000000..70ac9267059
--- /dev/null
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/TruncateTableITCase.java
@@ -0,0 +1,39 @@
+/*
+ * 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.TableException;
+import org.apache.flink.table.planner.runtime.utils.StreamingTestBase;
+
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** The IT case for Truncate table statement in streaming mode. */
+public class TruncateTableITCase extends StreamingTestBase {
+
+ @Test
+ void testTruncateTable() {
+ // should throw exception since truncate table is not support in streaming mode
+ tEnv().executeSql("CREATE TABLE t (a int) WITH ('connector' = 'test-update-delete')");
+ assertThatThrownBy(() -> tEnv().executeSql("TRUNCATE TABLE t"))
+ .isInstanceOf(TableException.class)
+ .hasMessage("TRUNCATE TABLE statement is not supported in streaming mode.");
+ }
+}
[flink] 01/03: [FLINK-32234][table] Convert SQLTruncateTable to TruncateTableOperation
Posted by yu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yuxia pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 45ce1926274d66e98bf0141f1977cb220124e75e
Author: luoyuxia <lu...@alumni.sjtu.edu.cn>
AuthorDate: Thu Jun 1 18:03:59 2023 +0800
[FLINK-32234][table] Convert SQLTruncateTable to TruncateTableOperation
---
.../table/operations/TruncateTableOperation.java | 42 ++++++++++++++++++++++
.../operations/converters/SqlNodeConverters.java | 1 +
.../converters/SqlTruncateTableConverter.java | 39 ++++++++++++++++++++
.../table/planner/calcite/FlinkPlannerImpl.scala | 3 +-
.../operations/SqlDmlToOperationConverterTest.java | 13 +++++++
5 files changed, 97 insertions(+), 1 deletion(-)
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/TruncateTableOperation.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/TruncateTableOperation.java
new file mode 100644
index 00000000000..c0d698ee0fe
--- /dev/null
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/TruncateTableOperation.java
@@ -0,0 +1,42 @@
+/*
+ * 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.VisibleForTesting;
+import org.apache.flink.table.catalog.ObjectIdentifier;
+
+/** Operation to describe an {@code TRUNCATE TABLE} statement. */
+public class TruncateTableOperation implements Operation {
+
+ private final ObjectIdentifier tableIdentifier;
+
+ public TruncateTableOperation(ObjectIdentifier tableIdentifier) {
+ this.tableIdentifier = tableIdentifier;
+ }
+
+ @VisibleForTesting
+ public ObjectIdentifier getTableIdentifier() {
+ return tableIdentifier;
+ }
+
+ @Override
+ public String asSummaryString() {
+ return "TRUNCATE TABLE " + tableIdentifier.toString();
+ }
+}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlNodeConverters.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlNodeConverters.java
index 16eb030a0bd..0eb68e1465f 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlNodeConverters.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlNodeConverters.java
@@ -47,6 +47,7 @@ public class SqlNodeConverters {
register(new SqlAlterTableDropPartitionConverter());
register(new SqlQueryConverter());
register(new SqlShowPartitionsConverter());
+ register(new SqlTruncateTableConverter());
}
/**
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlTruncateTableConverter.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlTruncateTableConverter.java
new file mode 100644
index 00000000000..2c096957529
--- /dev/null
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlTruncateTableConverter.java
@@ -0,0 +1,39 @@
+/*
+ * 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.operations.converters;
+
+import org.apache.flink.sql.parser.dml.SqlTruncateTable;
+import org.apache.flink.table.catalog.CatalogManager;
+import org.apache.flink.table.catalog.ObjectIdentifier;
+import org.apache.flink.table.catalog.UnresolvedIdentifier;
+import org.apache.flink.table.operations.Operation;
+import org.apache.flink.table.operations.TruncateTableOperation;
+
+/** A converter for {@link SqlTruncateTable}. */
+public class SqlTruncateTableConverter implements SqlNodeConverter<SqlTruncateTable> {
+
+ @Override
+ public Operation convertSqlNode(SqlTruncateTable sqlTruncateTable, ConvertContext context) {
+ UnresolvedIdentifier unresolvedIdentifier =
+ UnresolvedIdentifier.of(sqlTruncateTable.fullTableName());
+ CatalogManager catalogManager = context.getCatalogManager();
+ ObjectIdentifier tableIdentifier = catalogManager.qualifyIdentifier(unresolvedIdentifier);
+ return new TruncateTableOperation(tableIdentifier);
+ }
+}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/calcite/FlinkPlannerImpl.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/calcite/FlinkPlannerImpl.scala
index 9a1e5996f69..b6a364800cf 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/calcite/FlinkPlannerImpl.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/calcite/FlinkPlannerImpl.scala
@@ -19,7 +19,7 @@ package org.apache.flink.table.planner.calcite
import org.apache.flink.sql.parser.ExtendedSqlNode
import org.apache.flink.sql.parser.ddl.{SqlCompilePlan, SqlReset, SqlSet, SqlUseModules}
-import org.apache.flink.sql.parser.dml.{RichSqlInsert, SqlBeginStatementSet, SqlCompileAndExecutePlan, SqlEndStatementSet, SqlExecute, SqlExecutePlan, SqlStatementSet}
+import org.apache.flink.sql.parser.dml.{RichSqlInsert, SqlBeginStatementSet, SqlCompileAndExecutePlan, SqlEndStatementSet, SqlExecute, SqlExecutePlan, SqlStatementSet, SqlTruncateTable}
import org.apache.flink.sql.parser.dql._
import org.apache.flink.table.api.{TableException, ValidationException}
import org.apache.flink.table.planner.hint.JoinStrategy
@@ -149,6 +149,7 @@ class FlinkPlannerImpl(
|| sqlNode.isInstanceOf[SqlSet]
|| sqlNode.isInstanceOf[SqlReset]
|| sqlNode.isInstanceOf[SqlExecutePlan]
+ || sqlNode.isInstanceOf[SqlTruncateTable]
) {
return sqlNode
}
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlDmlToOperationConverterTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlDmlToOperationConverterTest.java
index e11f0bb4459..23b4c2cb98a 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlDmlToOperationConverterTest.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlDmlToOperationConverterTest.java
@@ -34,6 +34,7 @@ import org.apache.flink.table.operations.Operation;
import org.apache.flink.table.operations.QueryOperation;
import org.apache.flink.table.operations.SinkModifyOperation;
import org.apache.flink.table.operations.StatementSetOperation;
+import org.apache.flink.table.operations.TruncateTableOperation;
import org.apache.flink.table.planner.calcite.FlinkPlannerImpl;
import org.apache.flink.table.planner.factories.TestUpdateDeleteTableFactory;
import org.apache.flink.table.planner.parse.CalciteParser;
@@ -308,6 +309,18 @@ public class SqlDmlToOperationConverterTest extends SqlNodeToOperationConversion
checkUpdateOperation(operation);
}
+ @Test
+ public void testTruncateTable() {
+ String sql = "TRUNCATE TABLE t1";
+ FlinkPlannerImpl planner = getPlannerBySqlDialect(SqlDialect.DEFAULT);
+ final CalciteParser parser = getParserBySqlDialect(SqlDialect.DEFAULT);
+ Operation operation = parse(sql, planner, parser);
+ assertThat(operation).isInstanceOf(TruncateTableOperation.class);
+ TruncateTableOperation truncateTableOperation = (TruncateTableOperation) operation;
+ assertThat(truncateTableOperation.getTableIdentifier())
+ .isEqualTo(ObjectIdentifier.of("builtin", "default", "t1"));
+ }
+
private void checkExplainSql(String sql) {
FlinkPlannerImpl planner = getPlannerBySqlDialect(SqlDialect.DEFAULT);
CalciteParser parser = getParserBySqlDialect(SqlDialect.DEFAULT);
[flink] 03/03: [FLINK-32234][sql-client] Add test for truncate table in sql-client This closes #22696
Posted by yu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yuxia pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 300902016bd41961f28dacdd9b5fd028b8edb925
Author: luoyuxia <lu...@alumni.sjtu.edu.cn>
AuthorDate: Thu Jun 1 18:05:24 2023 +0800
[FLINK-32234][sql-client] Add test for truncate table in sql-client
This closes #22696
---
.../flink/table/client/cli/CliClientITCase.java | 5 +-
.../src/test/resources/sql/truncate.q | 62 ++++++++++++++++++++++
2 files changed, 65 insertions(+), 2 deletions(-)
diff --git a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/CliClientITCase.java b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/CliClientITCase.java
index 533c8c9fe94..0caf1043a33 100644
--- a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/CliClientITCase.java
+++ b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/CliClientITCase.java
@@ -163,7 +163,8 @@ class CliClientITCase {
replaceVars.put("$VAR_PIPELINE_JARS_URL", udfDependency.toString());
replaceVars.put("$VAR_REST_PORT", configuration.get(PORT).toString());
replaceVars.put("$VAR_JOBMANAGER_RPC_ADDRESS", configuration.get(ADDRESS));
- replaceVars.put("$VAR_DELETE_TABLE_DATA_ID", prepareDataForDeleteStatement());
+ replaceVars.put("$VAR_DELETE_TABLE_DATA_ID", prepareData());
+ replaceVars.put("$VAR_TRUNCATE_TABLE_DATA_ID", prepareData());
}
@BeforeEach
@@ -347,7 +348,7 @@ class CliClientITCase {
}
}
- private static String prepareDataForDeleteStatement() {
+ private static String prepareData() {
List<RowData> values = new ArrayList<>();
for (int i = 0; i < 5; i++) {
values.add(GenericRowData.of(i, StringData.fromString("b_" + i), i * 2.0));
diff --git a/flink-table/flink-sql-client/src/test/resources/sql/truncate.q b/flink-table/flink-sql-client/src/test/resources/sql/truncate.q
new file mode 100644
index 00000000000..0fda300a192
--- /dev/null
+++ b/flink-table/flink-sql-client/src/test/resources/sql/truncate.q
@@ -0,0 +1,62 @@
+# truncate.q - test truncate table statement
+#
+# 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.
+
+# first set batch mode
+SET 'execution.runtime-mode' = 'batch';
+[INFO] Execute statement succeed.
+!info
+
+SET 'sql-client.execution.result-mode' = 'tableau';
+[INFO] Execute statement succeed.
+!info
+
+SET 'table.dml-sync' = 'true';
+[INFO] Execute statement succeed.
+!info
+
+# create a table first
+CREATE TABLE t (a int PRIMARY KEY NOT ENFORCED, b string, c double)
+WITH (
+ 'connector' = 'test-update-delete',
+ 'data-id' = '$VAR_TRUNCATE_TABLE_DATA_ID'
+);
+[INFO] Execute statement succeed.
+!info
+
+# query the table first
+SELECT * FROM t;
++---+-----+-----+
+| a | b | c |
++---+-----+-----+
+| 0 | b_0 | 0.0 |
+| 1 | b_1 | 2.0 |
+| 2 | b_2 | 4.0 |
+| 3 | b_3 | 6.0 |
+| 4 | b_4 | 8.0 |
++---+-----+-----+
+5 rows in set
+!ok
+
+# truncate the table
+TRUNCATE TABLE t;
+[INFO] Execute statement succeed.
+!info
+
+# query the table again
+SELECT * FROM t;
+Empty set
+!ok