From 303fd1e245507bddb0c05110806a18a3f973fb1b Mon Sep 17 00:00:00 2001
From: yantian <lining.jln@alibaba-inc.com>
Date: Fri, 20 Dec 2024 10:13:50 +0800
Subject: [PATCH] update TableNoPermissionException to NoPermissionException
 for supporting database

---
 .../org/apache/paimon/catalog/Catalog.java    | 45 ++++++++++---------
 .../org/apache/paimon/rest/RESTCatalog.java   | 16 +++++--
 2 files changed, 35 insertions(+), 26 deletions(-)

diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java b/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java
index 28add9e6650c..3a18d2be8c8e 100644
--- a/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java
+++ b/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java
@@ -364,6 +364,29 @@ default void repairTable(Identifier identifier) throws TableNotExistException {
         throw new UnsupportedOperationException();
     }
 
+    /**
+     * Exception for trying to operate on a resource that doesn't have permission. Define as a
+     * runtime exception: 1. Other engine has no this type exception. 2. It wouldn't bring api break
+     * change.
+     */
+    class NoPermissionException extends RuntimeException {
+        private static final String MSG = "No permission for %s %s.";
+
+        public static NoPermissionException createDatabaseNoPermissionException(
+                String databaseName, Throwable cause) {
+            return new NoPermissionException("database", databaseName, cause);
+        }
+
+        public static NoPermissionException createTableNoPermissionException(
+                Identifier identifier, Throwable cause) {
+            return new NoPermissionException("table", identifier.getFullName(), cause);
+        }
+
+        public NoPermissionException(String resourceType, String resourceName, Throwable cause) {
+            super(String.format(MSG, resourceType, resourceName), cause);
+        }
+    }
+
     /** Exception for trying to drop on a database that is not empty. */
     class DatabaseNotEmptyException extends Exception {
         private static final String MSG = "Database %s is not empty.";
@@ -475,28 +498,6 @@ public Identifier identifier() {
         }
     }
 
-    /**
-     * Exception for trying to operate on a table that doesn't have permission. Define as a runtime
-     * exception: 1. Other engine has no this type exception. 2. It wouldn't bring api break change.
-     */
-    class TableNoPermissionException extends RuntimeException {
-        private static final String MSG = "No permission for Table %s.";
-        private final Identifier identifier;
-
-        public TableNoPermissionException(Identifier identifier) {
-            this(identifier, null);
-        }
-
-        public TableNoPermissionException(Identifier identifier, Throwable cause) {
-            super(String.format(MSG, identifier.getFullName()), cause);
-            this.identifier = identifier;
-        }
-
-        public Identifier identifier() {
-            return identifier;
-        }
-    }
-
     /** Exception for trying to operate on a partition that doesn't exist. */
     class PartitionNotExistException extends Exception {
 
diff --git a/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java b/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java
index f099a431b16b..33adc1b0a09e 100644
--- a/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java
+++ b/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java
@@ -197,6 +197,8 @@ public void createDatabase(String name, boolean ignoreIfExists, Map<String, Stri
             if (!ignoreIfExists) {
                 throw new DatabaseAlreadyExistException(name);
             }
+        } catch (ForbiddenException e) {
+            throw NoPermissionException.createDatabaseNoPermissionException(name, e);
         }
     }
 
@@ -212,6 +214,8 @@ public Database getDatabase(String name) throws DatabaseNotExistException {
                     name, response.options(), response.comment().orElseGet(() -> null));
         } catch (NoSuchResourceException e) {
             throw new DatabaseNotExistException(name);
+        } catch (ForbiddenException e) {
+            throw NoPermissionException.createDatabaseNoPermissionException(name, e);
         }
     }
 
@@ -228,6 +232,8 @@ public void dropDatabase(String name, boolean ignoreIfNotExists, boolean cascade
             if (!ignoreIfNotExists) {
                 throw new DatabaseNotExistException(name);
             }
+        } catch (ForbiddenException e) {
+            throw NoPermissionException.createDatabaseNoPermissionException(name, e);
         }
     }
 
@@ -255,6 +261,8 @@ public void alterDatabase(String name, List<PropertyChange> changes, boolean ign
             if (!ignoreIfNotExists) {
                 throw new DatabaseNotExistException(name);
             }
+        } catch (ForbiddenException e) {
+            throw NoPermissionException.createDatabaseNoPermissionException(name, e);
         }
     }
 
@@ -304,7 +312,7 @@ public void renameTable(Identifier fromTable, Identifier toTable, boolean ignore
                 throw new TableNotExistException(fromTable);
             }
         } catch (ForbiddenException e) {
-            throw new TableNoPermissionException(fromTable);
+            throw NoPermissionException.createTableNoPermissionException(fromTable, e);
         } catch (AlreadyExistsException e) {
             throw new TableAlreadyExistException(toTable);
         }
@@ -321,7 +329,7 @@ public void alterTable(
                 throw new TableNotExistException(identifier);
             }
         } catch (ForbiddenException e) {
-            throw new TableNoPermissionException(identifier);
+            throw NoPermissionException.createTableNoPermissionException(identifier, e);
         }
     }
 
@@ -337,7 +345,7 @@ public void dropTable(Identifier identifier, boolean ignoreIfNotExists)
                 throw new TableNotExistException(identifier);
             }
         } catch (ForbiddenException e) {
-            throw new TableNoPermissionException(identifier);
+            throw NoPermissionException.createTableNoPermissionException(identifier, e);
         }
     }
 
@@ -400,7 +408,7 @@ protected TableSchema getDataTableSchema(Identifier identifier) throws TableNotE
         } catch (NoSuchResourceException e) {
             throw new TableNotExistException(identifier);
         } catch (ForbiddenException e) {
-            throw new TableNoPermissionException(identifier, e);
+            throw NoPermissionException.createTableNoPermissionException(identifier, e);
         }
     }