diff --git a/flink-python/pyflink/table/tests/test_catalog_completeness.py b/flink-python/pyflink/table/tests/test_catalog_completeness.py index 5b14f867d3e23..1cb532112a284 100644 --- a/flink-python/pyflink/table/tests/test_catalog_completeness.py +++ b/flink-python/pyflink/table/tests/test_catalog_completeness.py @@ -52,6 +52,7 @@ def excluded_methods(cls): 'listConnections', 'createConnection', 'alterConnection', + 'renameConnection', 'convertTableToMaterializedTable'} diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/CatalogManager.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/CatalogManager.java index ca31e17ebf4e4..3667dbe0c31ad 100644 --- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/CatalogManager.java +++ b/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 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. * diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/GenericInMemoryCatalog.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/GenericInMemoryCatalog.java index c353dbe2965e6..347c9e642a75f 100644 --- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/GenericInMemoryCatalog.java +++ b/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 { diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/AlterConnectionRenameOperation.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/AlterConnectionRenameOperation.java new file mode 100644 index 0000000000000..93690e72da0c0 --- /dev/null +++ b/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; + } +} diff --git a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/catalog/CatalogManagerTest.java b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/catalog/CatalogManagerTest.java index 907d03dfcf8d1..4c29d1de04a22 100644 --- a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/catalog/CatalogManagerTest.java +++ b/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 options = + new HashMap() { + { + 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() diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/Catalog.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/Catalog.java index 2f3b478a62a4f..182c0fb6b7b70 100644 --- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/Catalog.java +++ b/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. * diff --git a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/catalog/CatalogTest.java b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/catalog/CatalogTest.java index f93e9f0575b77..b2cce120e7b66 100644 --- a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/catalog/CatalogTest.java +++ b/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()) { diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlAlterConnectionRenameConverter.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlAlterConnectionRenameConverter.java new file mode 100644 index 0000000000000..183e69fc985a5 --- /dev/null +++ b/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 { + + @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()); + } +} 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 ce8f41df99229..d9c5da83fd0bc 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 SqlAlterConnectionRenameConverter()); } 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 a410c95ec0b65..2a03bef687755 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 @@ -23,6 +23,7 @@ import org.apache.flink.table.catalog.ObjectIdentifier; import org.apache.flink.table.catalog.SensitiveConnection; import org.apache.flink.table.operations.Operation; +import org.apache.flink.table.operations.ddl.AlterConnectionRenameOperation; import org.apache.flink.table.operations.ddl.CreateConnectionOperation; import org.junit.jupiter.api.Test; @@ -121,4 +122,32 @@ void testCreateConnectionWithEmptyOptionsRejected() { .isInstanceOf(SqlValidateException.class) .hasMessageContaining("Connection property list can not be empty."); } + + @Test + void testAlterConnectionRename() { + Operation operation = parse("ALTER CONNECTION my_conn RENAME TO new_conn"); + assertThat(operation).isInstanceOf(AlterConnectionRenameOperation.class); + AlterConnectionRenameOperation op = (AlterConnectionRenameOperation) operation; + + assertThat(op.getConnectionIdentifier()) + .isEqualTo(ObjectIdentifier.of("builtin", "default", "my_conn")); + assertThat(op.getNewConnectionName()).isEqualTo("new_conn"); + assertThat(op.ignoreIfNotExists()).isFalse(); + } + + @Test + void testAlterConnectionRenameIfExists() { + Operation operation = parse("ALTER CONNECTION IF EXISTS my_conn RENAME TO new_conn"); + AlterConnectionRenameOperation op = (AlterConnectionRenameOperation) operation; + assertThat(op.ignoreIfNotExists()).isTrue(); + } + + @Test + void testAlterConnectionRenameWithFullyQualifiedName() { + Operation operation = parse("ALTER CONNECTION cat1.db1.my_conn RENAME TO new_conn"); + AlterConnectionRenameOperation op = (AlterConnectionRenameOperation) operation; + assertThat(op.getConnectionIdentifier()) + .isEqualTo(ObjectIdentifier.of("cat1", "db1", "my_conn")); + assertThat(op.getNewConnectionName()).isEqualTo("new_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 a9d99cbfe5d32..ffd7966bf7509 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,36 @@ void testCreatePermanentConnectionRejectedWithoutSecretStore() { .hasMessageContaining("WritableSecretStore must be configured"); } + @Test + void testAlterConnectionRenameTemporaryConnection() { + tEnv().executeSql("CREATE TEMPORARY CONNECTION my_conn WITH ('k' = 'v')"); + + tEnv().executeSql("ALTER CONNECTION my_conn RENAME TO new_conn"); + + assertThat(catalogManager().getConnection(connectionIdentifier("my_conn"))).isEmpty(); + assertThat(catalogManager().getConnection(connectionIdentifier("new_conn"))) + .hasValueSatisfying( + connection -> + assertThat(connection.getOptions()).containsOnly(entry("k", "v"))); + } + + @Test + void testAlterConnectionRenameIfExists() { + tEnv().executeSql("ALTER CONNECTION IF EXISTS missing_conn RENAME TO new_conn"); + + assertThat(catalogManager().getConnection(connectionIdentifier("new_conn"))).isEmpty(); + } + + @Test + void testAlterConnectionRenameToExistingConnectionRejected() { + tEnv().executeSql("CREATE TEMPORARY CONNECTION my_conn WITH ('k' = 'v')"); + tEnv().executeSql("CREATE TEMPORARY CONNECTION new_conn WITH ('k' = 'v')"); + + assertThatThrownBy(() -> tEnv().executeSql("ALTER CONNECTION my_conn RENAME TO new_conn")) + .isInstanceOf(ValidationException.class) + .hasMessageContaining("already exists"); + } + private CatalogManager catalogManager() { return ((TableEnvironmentInternal) tEnv()).getCatalogManager(); }