[flink] Adjust renameView after renameTable failed

herefree · Nov 1, 2024 · 0ed3ff5 · 0ed3ff5
1 parent f9b61d4
commit 0ed3ff5
Show file tree

Hide file tree

Showing 2 changed files with 20 additions and 13 deletions.
diff --git a/paimon-core/src/test/java/org/apache/paimon/catalog/CatalogTestBase.java b/paimon-core/src/test/java/org/apache/paimon/catalog/CatalogTestBase.java
@@ -901,11 +901,20 @@ public void testView() throws Exception {
         assertThatThrownBy(() -> catalog.createView(identifier, view, false))
                 .isInstanceOf(Catalog.ViewAlreadyExistException.class);
 
-        catalog.dropView(identifier, false);
-        assertThat(catalog.viewExists(identifier)).isFalse();
+        Identifier newIdentifier = new Identifier("view_db", "new_view");
+        catalog.renameView(new Identifier("view_db", "unknown"), newIdentifier, true);
+        assertThatThrownBy(
+                        () ->
+                                catalog.renameView(
+                                        new Identifier("view_db", "unknown"), newIdentifier, false))
+                .isInstanceOf(Catalog.ViewNotExistException.class);
+        catalog.renameView(identifier, newIdentifier, false);
+
+        catalog.dropView(newIdentifier, false);
+        assertThat(catalog.viewExists(newIdentifier)).isFalse();
 
-        catalog.dropView(identifier, true);
-        assertThatThrownBy(() -> catalog.dropView(identifier, false))
+        catalog.dropView(newIdentifier, true);
+        assertThatThrownBy(() -> catalog.dropView(newIdentifier, false))
                 .isInstanceOf(Catalog.ViewNotExistException.class);
     }
 }
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalog.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalog.java
@@ -1083,19 +1083,17 @@ public final void renameTable(
             ObjectPath tablePath, String newTableName, boolean ignoreIfNotExists)
             throws CatalogException, TableNotExistException, TableAlreadyExistException {
         ObjectPath toTable = new ObjectPath(tablePath.getDatabaseName(), newTableName);
-        if (catalog.viewExists(toIdentifier(tablePath))) {
+        try {
+            catalog.renameTable(toIdentifier(tablePath), toIdentifier(toTable), ignoreIfNotExists);
+        } catch (Catalog.TableNotExistException e) {
             try {
                 catalog.renameView(
                         toIdentifier(tablePath), toIdentifier(toTable), ignoreIfNotExists);
-                return;
-            } catch (Catalog.ViewNotExistException | Catalog.ViewAlreadyExistException e) {
-                throw new RuntimeException("Unexpected exception.", e);
+            } catch (Catalog.ViewNotExistException ex) {
+                throw new TableNotExistException(getName(), tablePath);
+            } catch (Catalog.ViewAlreadyExistException ex) {
+                throw new TableAlreadyExistException(getName(), toTable);
             }
-        }
-        try {
-            catalog.renameTable(toIdentifier(tablePath), toIdentifier(toTable), ignoreIfNotExists);
-        } catch (Catalog.TableNotExistException e) {
-            throw new TableNotExistException(getName(), tablePath);
         } catch (Catalog.TableAlreadyExistException e) {
             throw new TableAlreadyExistException(getName(), toTable);
         }