[FLINK-38262][table] Add DROP CONNECTION operation

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<String, Object> 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;
+    }
+}

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<SqlDropConnection> {
+
+    @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());
+    }
+}

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() {

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"));
+    }
 }

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();
     }