diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/DropConnectionOperation.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/DropConnectionOperation.java new file mode 100644 index 00000000000000..b0f6d2aea2866f --- /dev/null +++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/DropConnectionOperation.java @@ -0,0 +1,89 @@ +/* + * 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.ddl; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.table.api.internal.TableResultImpl; +import org.apache.flink.table.api.internal.TableResultInternal; +import org.apache.flink.table.catalog.ObjectIdentifier; +import org.apache.flink.table.operations.Operation; +import org.apache.flink.table.operations.OperationUtils; + +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.Map; + +/** Operation to describe a DROP CONNECTION statement. */ +@Internal +public class DropConnectionOperation implements DropOperation { + + private final ObjectIdentifier connectionIdentifier; + private final boolean ifExists; + private final boolean isTemporary; + private final boolean isSystemConnection; + + public DropConnectionOperation( + ObjectIdentifier connectionIdentifier, + boolean ifExists, + boolean isTemporary, + boolean isSystemConnection) { + this.connectionIdentifier = connectionIdentifier; + this.ifExists = ifExists; + this.isTemporary = isTemporary; + this.isSystemConnection = isSystemConnection; + } + + public ObjectIdentifier getConnectionIdentifier() { + return connectionIdentifier; + } + + public boolean isIfExists() { + return ifExists; + } + + public boolean isTemporary() { + return isTemporary; + } + + public boolean isSystemConnection() { + return isSystemConnection; + } + + @Override + public String asSummaryString() { + Map params = new LinkedHashMap<>(); + params.put("identifier", connectionIdentifier); + params.put("ifExists", ifExists); + params.put("isTemporary", isTemporary); + params.put("isSystemConnection", isSystemConnection); + + return OperationUtils.formatWithChildren( + "DROP CONNECTION", params, Collections.emptyList(), Operation::asSummaryString); + } + + @Override + public TableResultInternal execute(Context ctx) { + if (isTemporary) { + ctx.getCatalogManager().dropTemporaryConnection(connectionIdentifier, ifExists); + } else { + ctx.getCatalogManager().dropConnection(connectionIdentifier, ifExists); + } + return TableResultImpl.TABLE_RESULT_OK; + } +} diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlDropConnectionConverter.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlDropConnectionConverter.java new file mode 100644 index 00000000000000..4bab847f198463 --- /dev/null +++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlDropConnectionConverter.java @@ -0,0 +1,43 @@ +/* + * 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.ddl.connection.SqlDropConnection; +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.ddl.DropConnectionOperation; + +/** A converter for {@link SqlDropConnection}. */ +public class SqlDropConnectionConverter implements SqlNodeConverter { + + @Override + public Operation convertSqlNode(SqlDropConnection sqlDropConnection, ConvertContext context) { + UnresolvedIdentifier unresolvedIdentifier = + UnresolvedIdentifier.of(sqlDropConnection.getFullName()); + ObjectIdentifier identifier = + context.getCatalogManager().qualifyIdentifier(unresolvedIdentifier); + + return new DropConnectionOperation( + identifier, + sqlDropConnection.getIfExists(), + sqlDropConnection.isTemporary(), + sqlDropConnection.isSystemConnection()); + } +} 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 ce8f41df992295..b7eeedaaf07899 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 @@ -141,6 +141,7 @@ private static void registerCatalogConverters() { private static void registerConnectionConverters() { register(new SqlCreateConnectionConverter()); + register(new SqlDropConnectionConverter()); } private static void registerMaterializedTableConverters() { diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlConnectionOperationConverterTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlConnectionOperationConverterTest.java index a410c95ec0b659..3d09391c314daf 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlConnectionOperationConverterTest.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlConnectionOperationConverterTest.java @@ -24,6 +24,7 @@ import org.apache.flink.table.catalog.SensitiveConnection; import org.apache.flink.table.operations.Operation; import org.apache.flink.table.operations.ddl.CreateConnectionOperation; +import org.apache.flink.table.operations.ddl.DropConnectionOperation; import org.junit.jupiter.api.Test; @@ -121,4 +122,51 @@ void testCreateConnectionWithEmptyOptionsRejected() { .isInstanceOf(SqlValidateException.class) .hasMessageContaining("Connection property list can not be empty."); } + + @Test + void testDropConnection() { + Operation operation = parse("DROP CONNECTION my_conn"); + assertThat(operation).isInstanceOf(DropConnectionOperation.class); + DropConnectionOperation op = (DropConnectionOperation) operation; + + assertThat(op.getConnectionIdentifier()) + .isEqualTo(ObjectIdentifier.of("builtin", "default", "my_conn")); + assertThat(op.isIfExists()).isFalse(); + assertThat(op.isTemporary()).isFalse(); + assertThat(op.isSystemConnection()).isFalse(); + } + + @Test + void testDropConnectionIfExists() { + Operation operation = parse("DROP CONNECTION IF EXISTS my_conn"); + DropConnectionOperation op = (DropConnectionOperation) operation; + assertThat(op.isIfExists()).isTrue(); + assertThat(op.isTemporary()).isFalse(); + assertThat(op.isSystemConnection()).isFalse(); + } + + @Test + void testDropTemporaryConnection() { + Operation operation = parse("DROP TEMPORARY CONNECTION my_conn"); + DropConnectionOperation op = (DropConnectionOperation) operation; + assertThat(op.isIfExists()).isFalse(); + assertThat(op.isTemporary()).isTrue(); + assertThat(op.isSystemConnection()).isFalse(); + } + + @Test + void testDropTemporarySystemConnection() { + Operation operation = parse("DROP TEMPORARY SYSTEM CONNECTION my_conn"); + DropConnectionOperation op = (DropConnectionOperation) operation; + assertThat(op.isTemporary()).isTrue(); + assertThat(op.isSystemConnection()).isTrue(); + } + + @Test + void testDropConnectionWithFullyQualifiedName() { + Operation operation = parse("DROP CONNECTION cat1.db1.my_conn"); + DropConnectionOperation op = (DropConnectionOperation) operation; + assertThat(op.getConnectionIdentifier()) + .isEqualTo(ObjectIdentifier.of("cat1", "db1", "my_conn")); + } } diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/CreateConnectionITCase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/CreateConnectionITCase.java index a9d99cbfe5d328..3ec95bfbb8bbe7 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/CreateConnectionITCase.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/CreateConnectionITCase.java @@ -73,6 +73,29 @@ void testCreatePermanentConnectionRejectedWithoutSecretStore() { .hasMessageContaining("WritableSecretStore must be configured"); } + @Test + void testDropTemporaryConnection() { + tEnv().executeSql("CREATE TEMPORARY CONNECTION my_conn WITH ('k' = 'v')"); + + tEnv().executeSql("DROP TEMPORARY CONNECTION my_conn"); + + assertThat(catalogManager().getConnection(connectionIdentifier("my_conn"))).isEmpty(); + } + + @Test + void testDropTemporaryConnectionIfExists() { + tEnv().executeSql("DROP TEMPORARY CONNECTION IF EXISTS my_conn"); + + assertThat(catalogManager().getConnection(connectionIdentifier("my_conn"))).isEmpty(); + } + + @Test + void testDropMissingTemporaryConnectionRejected() { + assertThatThrownBy(() -> tEnv().executeSql("DROP TEMPORARY CONNECTION my_conn")) + .isInstanceOf(ValidationException.class) + .hasMessageContaining("Temporary connection with identifier"); + } + private CatalogManager catalogManager() { return ((TableEnvironmentInternal) tEnv()).getCatalogManager(); }