raminqaf commented on code in PR #28793:
URL: https://github.com/apache/flink/pull/28793#discussion_r3656100254
##########
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()) {
Review Comment:
nit: inline and simplify
```suggestion
if (getConnection(objectIdentifier).isEmpty()) {
```
##########
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()) {
Review Comment:
```suggestion
if (temporaryConnection != null &&
getConnection(newIdentifier).isPresent()) {
```
##########
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()));
+ }
+
Review Comment:
Is the community aligned with adding this to the public API? I haven't seen
a Mailing Thread or FLIP on this. Maybe I have missed it.
##########
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");
+ }
+ };
Review Comment:
use `Map.of(Map.entry(...), ...)`
##########
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()) {
Review Comment:
`getConnection(newIdentifier).isPresent()` can be also extracted and re-used
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
