[FLINK-38262][table] alter connection rename operation

flink-python/pyflink/table/tests/test_catalog_completeness.py

@@ -52,6 +52,7 @@ def excluded_methods(cls):
             'listConnections',
             'createConnection',
             'alterConnection',
+            'renameConnection',
             'convertTableToMaterializedTable'}
 
 

flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/CatalogManager.java

@@ -2111,6 +2111,62 @@ public void alterConnection(
         }
     }
 
+    /**
+     * Rename a connection in the given fully qualified path.
+     *
+     * @param objectIdentifier The fully qualified path of the connection to rename.
+     * @param newConnectionName The new connection name.
+     * @param ignoreIfNotExists If false exception will be thrown if the connection to be renamed
+     *     does not exist.
+     */
+    public void renameConnection(
+            ObjectIdentifier objectIdentifier,
+            String newConnectionName,
+            boolean ignoreIfNotExists) {
+        checkArgument(!StringUtils.isNullOrWhitespaceOnly(newConnectionName));
+        ObjectIdentifier newIdentifier =
+                ObjectIdentifier.of(
+                        objectIdentifier.getCatalogName(),
+                        objectIdentifier.getDatabaseName(),
+                        newConnectionName);
+        CatalogConnection temporaryConnection = temporaryConnections.get(objectIdentifier);
+        if (temporaryConnection != null) {
+            if (getConnection(newIdentifier).isPresent()) {
+                throw new ValidationException(
+                        String.format(
+                                "Connection with identifier '%s' already exists.",
+                                newIdentifier.asSummaryString()));
+            }
+            temporaryConnections.remove(objectIdentifier);
+            temporaryConnections.put(newIdentifier, temporaryConnection);
+            return;
+        }
+
+        Optional<CatalogConnection> existingOpt = getConnection(objectIdentifier);
+        if (!existingOpt.isPresent()) {
+            if (ignoreIfNotExists) {
+                return;
+            }
+            throw new ValidationException(
+                    String.format(
+                            "Connection with identifier '%s' does not exist.",
+                            objectIdentifier.asSummaryString()));
+        }
+        if (getConnection(newIdentifier).isPresent()) {
+            throw new ValidationException(
+                    String.format(
+                            "Connection with identifier '%s' already exists.",
+                            newIdentifier.asSummaryString()));
+        }
+
+        execute(
+                (catalog, path) ->
+                        catalog.renameConnection(path, newConnectionName, ignoreIfNotExists),
+                objectIdentifier,
+                ignoreIfNotExists,
+                "RenameConnection");
+    }
+
     /**
      * Drop a permanent connection from the given fully qualified path.
      *

flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/GenericInMemoryCatalog.java

@@ -511,6 +511,27 @@ public void alterConnection(
         connections.put(connectionPath, newConnection.copy());
     }
 
+    @Override
+    public void renameConnection(
+            ObjectPath connectionPath, String newConnectionName, boolean ignoreIfNotExists)
+            throws ConnectionNotExistException, ConnectionAlreadyExistException {
+        checkNotNull(connectionPath);
+        checkArgument(!StringUtils.isNullOrWhitespaceOnly(newConnectionName));
+
+        if (connectionExists(connectionPath)) {
+            ObjectPath newPath =
+                    new ObjectPath(connectionPath.getDatabaseName(), newConnectionName);
+
+            if (connectionExists(newPath)) {
+                throw new ConnectionAlreadyExistException(getName(), newPath);
+            } else {
+                connections.put(newPath, connections.remove(connectionPath));
+            }
+        } else if (!ignoreIfNotExists) {
+            throw new ConnectionNotExistException(getName(), connectionPath);
+        }
+    }
+
     @Override
     public void dropConnection(ObjectPath connectionPath, boolean ignoreIfNotExists)
             throws ConnectionNotExistException {

flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/AlterConnectionRenameOperation.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.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;
+
+/** Operation for ALTER CONNECTION .. RENAME TO .. statements. */
+@Internal
+public class AlterConnectionRenameOperation implements AlterOperation {
+
+    private final ObjectIdentifier connectionIdentifier;
+    private final String newConnectionName;
+    private final boolean ignoreIfNotExists;
+
+    public AlterConnectionRenameOperation(
+            ObjectIdentifier connectionIdentifier,
+            String newConnectionName,
+            boolean ignoreIfNotExists) {
+        this.connectionIdentifier = connectionIdentifier;
+        this.newConnectionName = newConnectionName;
+        this.ignoreIfNotExists = ignoreIfNotExists;
+    }
+
+    public ObjectIdentifier getConnectionIdentifier() {
+        return connectionIdentifier;
+    }
+
+    public String getNewConnectionName() {
+        return newConnectionName;
+    }
+
+    public boolean ignoreIfNotExists() {
+        return ignoreIfNotExists;
+    }
+
+    @Override
+    public String asSummaryString() {
+        return String.format(
+                "ALTER CONNECTION %s%s RENAME TO %s",
+                ignoreIfNotExists ? "IF EXISTS " : "",
+                connectionIdentifier.asSummaryString(),
+                newConnectionName);
+    }
+
+    @Override
+    public TableResultInternal execute(Context ctx) {
+        ctx.getCatalogManager()
+                .renameConnection(connectionIdentifier, newConnectionName, ignoreIfNotExists);
+        return TableResultImpl.TABLE_RESULT_OK;
+    }
+}

flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/catalog/CatalogManagerTest.java

@@ -544,6 +544,51 @@ public void testCreateConnectionWithoutTypeFallsBackToDefaultFactory() throws Ex
                 .containsEntry("bootstrap.servers", "localhost:9092");
     }
 
+    @Test
+    public void testRenameTemporaryConnection() {
+        CatalogManager catalogManager = createCatalogManager(null);
+        ObjectIdentifier source =
+                ObjectIdentifier.of(
+                        catalogManager.getCurrentCatalog(),
+                        catalogManager.getCurrentDatabase(),
+                        "conn1");
+        ObjectIdentifier target =
+                ObjectIdentifier.of(
+                        catalogManager.getCurrentCatalog(),
+                        catalogManager.getCurrentDatabase(),
+                        "conn2");
+        Map<String, String> options =
+                new HashMap<String, String>() {
+                    {
+                        put("type", "default");
+                        put("bootstrap.servers", "localhost:9092");
+                    }
+                };
+        catalogManager.createTemporaryConnection(
+                SensitiveConnection.of(options, null), source, false);
+
+        catalogManager.renameConnection(source, "conn2", false);
+
+        assertThat(catalogManager.getConnection(source)).isEmpty();
+        assertThat(catalogManager.getConnection(target))
+                .hasValueSatisfying(
+                        connection ->
+                                assertThat(connection.getOptions())
+                                        .containsEntry("bootstrap.servers", "localhost:9092"));
+    }
+
+    @Test
+    public void testRenameMissingTemporaryConnectionIfExists() {
+        CatalogManager catalogManager = createCatalogManager(null);
+        ObjectIdentifier source =
+                ObjectIdentifier.of(
+                        catalogManager.getCurrentCatalog(),
+                        catalogManager.getCurrentDatabase(),
+                        "conn1");
+
+        catalogManager.renameConnection(source, "conn2", true);
+    }
+
     private CatalogManager createCatalogManager(@Nullable CatalogModificationListener listener) {
         CatalogManager.Builder builder =
                 CatalogManager.newBuilder()

flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/Catalog.java

@@ -1041,6 +1041,26 @@ default void alterConnection(
                         this.getClass()));
     }
 
+    /**
+     * Rename an existing connection.
+     *
+     * @param connectionPath Path of the connection to be renamed
+     * @param newConnectionName the new name of the connection
+     * @param ignoreIfNotExists Flag to specify behavior when the connection does not exist: if set
+     *     to false, throw an exception, if set to true, do nothing.
+     * @throws ConnectionNotExistException if the connection does not exist
+     * @throws ConnectionAlreadyExistException if a connection with the new name already exists
+     * @throws CatalogException in case of any runtime exception
+     */
+    default void renameConnection(
+            ObjectPath connectionPath, String newConnectionName, boolean ignoreIfNotExists)
+            throws ConnectionNotExistException, ConnectionAlreadyExistException, CatalogException {
+        throw new UnsupportedOperationException(
+                String.format(
+                        "renameConnection(ObjectPath, String, boolean) is not implemented for %s.",
+                        this.getClass()));
+    }
+
     /**
      * Drop a connection.
      *

flink-table/flink-table-common/src/test/java/org/apache/flink/table/catalog/CatalogTest.java

@@ -652,6 +652,61 @@ public void testAlterMissingConnectionIgnoreIfNotExist() throws Exception {
         catalog.alterConnection(connectionPath1, newConnection, true);
     }
 
+    @Test
+    public void testRenameConnection() throws Exception {
+        if (!supportsConnections()) {
+            return;
+        }
+        catalog.createDatabase(db1, createDb(), false);
+        CatalogConnection connection = createConnection();
+        catalog.createConnection(connectionPath1, connection, false);
+
+        catalog.renameConnection(connectionPath1, c2, false);
+
+        assertThat(catalog.connectionExists(connectionPath1)).isFalse();
+        assertThat(catalog.getConnection(connectionPath2).getOptions())
+                .isEqualTo(connection.getOptions());
+    }
+
+    @Test
+    public void testRenameConnection_ConnectionAlreadyExistException() throws Exception {
+        if (!supportsConnections()) {
+            return;
+        }
+        catalog.createDatabase(db1, createDb(), false);
+        catalog.createConnection(connectionPath1, createConnection(), false);
+        catalog.createConnection(connectionPath2, createConnection(), false);
+
+        assertThatThrownBy(() -> catalog.renameConnection(connectionPath1, c2, false))
+                .isInstanceOf(ConnectionAlreadyExistException.class)
+                .hasMessage(
+                        "Connection 'db1.c2' already exists in catalog '"
+                                + TEST_CATALOG_NAME
+                                + "'.");
+    }
+
+    @Test
+    public void testRenameMissingConnectionNotExistException() throws Exception {
+        if (!supportsConnections()) {
+            return;
+        }
+        catalog.createDatabase(db1, createDb(), false);
+
+        assertThatThrownBy(() -> catalog.renameConnection(connectionPath1, c2, false))
+                .isInstanceOf(ConnectionNotExistException.class)
+                .hasMessage("Connection 'db1.c1' does not exist in catalog 'test-catalog'.");
+    }
+
+    @Test
+    public void testRenameMissingConnectionIgnoreIfNotExist() throws Exception {
+        if (!supportsConnections()) {
+            return;
+        }
+        catalog.createDatabase(db1, createDb(), false);
+
+        catalog.renameConnection(connectionPath1, c2, true);
+    }
+
     @Test
     public void testDropMissingConnectionNotExistException() throws Exception {
         if (!supportsConnections()) {

flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlAlterConnectionRenameConverter.java

@@ -0,0 +1,44 @@
+/*
+ * 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.SqlAlterConnectionRename;
+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.AlterConnectionRenameOperation;
+
+/** A converter for {@link SqlAlterConnectionRename}. */
+public class SqlAlterConnectionRenameConverter
+        implements SqlNodeConverter<SqlAlterConnectionRename> {
+
+    @Override
+    public Operation convertSqlNode(
+            SqlAlterConnectionRename sqlAlterConnectionRename, ConvertContext context) {
+        ObjectIdentifier connectionIdentifier =
+                context.getCatalogManager()
+                        .qualifyIdentifier(
+                                UnresolvedIdentifier.of(sqlAlterConnectionRename.getFullName()));
+
+        return new AlterConnectionRenameOperation(
+                connectionIdentifier,
+                sqlAlterConnectionRename.getNewConnectionName().getSimple(),
+                sqlAlterConnectionRename.ifConnectionExists());
+    }
+}

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 SqlAlterConnectionRenameConverter());
     }
 
     private static void registerMaterializedTableConverters() {