[rest] Align view permission handling with table in RESTCatalog.

paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java

@@ -1433,6 +1433,33 @@ public String getTableId() {
         }
     }
 
+    /** Exception for trying to operate on the view that doesn't have permission. */
+    class ViewNoPermissionException extends RuntimeException {
+
+        private static final String MSG = "View %s has no permission. Caused by %s.";
+
+        private final Identifier identifier;
+
+        public ViewNoPermissionException(Identifier identifier, Throwable cause) {
+            super(
+                    String.format(
+                            MSG,
+                            identifier.getFullName(),
+                            cause != null && cause.getMessage() != null ? cause.getMessage() : ""),
+                    cause);
+            this.identifier = identifier;
+        }
+
+        @VisibleForTesting
+        public ViewNoPermissionException(Identifier identifier) {
+            this(identifier, null);
+        }
+
+        public Identifier identifier() {
+            return identifier;
+        }
+    }
+
     /** Exception for trying to alter a column that already exists. */
     class ColumnAlreadyExistException extends Exception {
 

paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java

@@ -899,6 +899,8 @@ public View getView(Identifier identifier) throws ViewNotExistException {
             return toView(identifier.getDatabaseName(), response);
         } catch (NoSuchResourceException e) {
             throw new ViewNotExistException(identifier);
+        } catch (ForbiddenException e) {
+            throw new ViewNoPermissionException(identifier, e);
         }
     }
 
@@ -911,6 +913,8 @@ public void dropView(Identifier identifier, boolean ignoreIfNotExists)
             if (!ignoreIfNotExists) {
                 throw new ViewNotExistException(identifier);
             }
+        } catch (ForbiddenException e) {
+            throw new ViewNoPermissionException(identifier, e);
         }
     }
 
@@ -934,6 +938,8 @@ public void createView(Identifier identifier, View view, boolean ignoreIfExists)
             }
         } catch (BadRequestException e) {
             throw new IllegalArgumentException(e.getMessage());
+        } catch (ForbiddenException e) {
+            throw new ViewNoPermissionException(identifier, e);
         }
     }
 
@@ -945,6 +951,8 @@ public List<String> listViews(String databaseName) throws DatabaseNotExistExcept
                     : api.listViews(databaseName);
         } catch (NoSuchResourceException e) {
             throw new DatabaseNotExistException(databaseName);
+        } catch (ForbiddenException e) {
+            throw new DatabaseNoPermissionException(databaseName, e);
         }
     }
 
@@ -959,6 +967,8 @@ public PagedList<String> listViewsPaged(
             return api.listViewsPaged(databaseName, maxResults, pageToken, viewNamePattern);
         } catch (NoSuchResourceException e) {
             throw new DatabaseNotExistException(databaseName);
+        } catch (ForbiddenException e) {
+            throw new DatabaseNoPermissionException(databaseName, e);
         }
     }
 
@@ -979,6 +989,8 @@ public PagedList<View> listViewDetailsPaged(
                     views.getNextPageToken());
         } catch (NoSuchResourceException e) {
             throw new DatabaseNotExistException(db);
+        } catch (ForbiddenException e) {
+            throw new DatabaseNoPermissionException(db, e);
         }
     }
 
@@ -1020,6 +1032,8 @@ public void renameView(Identifier fromView, Identifier toView, boolean ignoreIfN
             throw new ViewAlreadyExistException(toView);
         } catch (BadRequestException e) {
             throw new IllegalArgumentException(e.getMessage());
+        } catch (ForbiddenException e) {
+            throw new ViewNoPermissionException(fromView, e);
         }
     }
 
@@ -1040,6 +1054,8 @@ public void alterView(
             }
         } catch (BadRequestException e) {
             throw new IllegalArgumentException(e.getMessage());
+        } catch (ForbiddenException e) {
+            throw new ViewNoPermissionException(identifier, e);
         }
     }
 

paimon-core/src/test/java/org/apache/paimon/rest/MockRESTCatalogTest.java

@@ -347,6 +347,11 @@ protected void revokeTablePermission(Identifier identifier) {
         restCatalogServer.addNoPermissionTable(identifier);
     }
 
+    @Override
+    protected void revokeViewPermission(Identifier identifier) {
+        restCatalogServer.addNoPermissionView(identifier);
+    }
+
     @Override
     protected void authTableColumns(Identifier identifier, List<String> columns) {
         restCatalogServer.addTableColumnAuth(identifier, columns);

paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogServer.java

@@ -195,6 +195,7 @@ public class RESTCatalogServer {
     private final Map<String, TableSnapshot> tableWithSnapshotId2SnapshotStore = new HashMap<>();
     private final List<String> noPermissionDatabases = new ArrayList<>();
     private final List<String> noPermissionTables = new ArrayList<>();
+    private final List<String> noPermissionViews = new ArrayList<>();
     private final Map<String, Function> functionStore = new HashMap<>();
     private final Map<String, List<String>> columnAuthHandler = new HashMap<>();
     private final Map<String, List<Predicate>> rowFilterAuthHandler = new HashMap<>();
@@ -276,6 +277,10 @@ public void addNoPermissionTable(Identifier identifier) {
         noPermissionTables.add(identifier.getFullName());
     }
 
+    public void addNoPermissionView(Identifier identifier) {
+        noPermissionViews.add(identifier.getFullName());
+    }
+
     public void addTableColumnAuth(Identifier identifier, List<String> select) {
         columnAuthHandler.put(identifier.getFullName(), select);
     }
@@ -631,6 +636,14 @@ && isTableByIdRequest(request.getPath())) {
                                     e.getMessage(),
                                     403);
                     return mockResponse(response, 403);
+                } catch (Catalog.ViewNoPermissionException e) {
+                    response =
+                            new ErrorResponse(
+                                    ErrorResponse.RESOURCE_TYPE_VIEW,
+                                    e.identifier().getTableName(),
+                                    e.getMessage(),
+                                    403);
+                    return mockResponse(response, 403);
                 } catch (Catalog.DatabaseAlreadyExistException e) {
                     response =
                             new ErrorResponse(
@@ -2327,6 +2340,9 @@ private List<Identifier> listViews(Map<String, String> parameters) {
     private MockResponse viewHandle(String method, Identifier identifier, String requestData)
             throws Exception {
         RESTResponse response;
+        if (noPermissionViews.contains(identifier.getFullName())) {
+            throw new Catalog.ViewNoPermissionException(identifier);
+        }
         if (viewStore.containsKey(identifier.getFullName())) {
             switch (method) {
                 case "GET":
@@ -2431,6 +2447,9 @@ private MockResponse renameViewHandle(String data) throws Exception {
         RenameTableRequest requestBody = RESTApi.fromJson(data, RenameTableRequest.class);
         Identifier fromView = requestBody.getSource();
         Identifier toView = requestBody.getDestination();
+        if (noPermissionViews.contains(fromView.getFullName())) {
+            throw new Catalog.ViewNoPermissionException(fromView);
+        }
         if (!viewStore.containsKey(fromView.getFullName())) {
             throw new Catalog.ViewNotExistException(fromView);
         }

paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java

@@ -278,6 +278,34 @@ void testDatabaseApiWhenNoPermission() {
                                 false));
     }
 
+    @Test
+    void testApiWhenViewNoPermission() throws Exception {
+        Identifier identifier = Identifier.create("test_view_db", "no_permission_view");
+        catalog.createDatabase(identifier.getDatabaseName(), false);
+        View view = createView(identifier);
+        catalog.createView(identifier, view, false);
+        revokeViewPermission(identifier);
+        assertThrows(Catalog.ViewNoPermissionException.class, () -> catalog.getView(identifier));
+        assertThrows(
+                Catalog.ViewNoPermissionException.class, () -> catalog.dropView(identifier, false));
+        assertThrows(
+                Catalog.ViewNoPermissionException.class,
+                () ->
+                        catalog.renameView(
+                                identifier,
+                                Identifier.create("test_view_db", "no_permission_view2"),
+                                false));
+        assertThrows(
+                Catalog.ViewNoPermissionException.class,
+                () ->
+                        catalog.alterView(
+                                identifier,
+                                ImmutableList.of(
+                                        ViewChange.addDialect(
+                                                "flink_1", "SELECT * FROM FLINK_TABLE_1")),
+                                false));
+    }
+
     @Test
     void testApiWhenDatabaseNoExistAndNotIgnore() {
         String database = "test_no_exist_db";
@@ -3932,6 +3960,8 @@ protected abstract Catalog newRestCatalogWithDataToken(Map<String, String> extra
 
     protected abstract void revokeTablePermission(Identifier identifier);
 
+    protected abstract void revokeViewPermission(Identifier identifier);
+
     protected abstract void authTableColumns(Identifier identifier, List<String> columns);
 
     protected abstract void revokeDatabasePermission(String database);