fix(catalog): grant view when grant table

e2e_test/batch/catalog/has_privilege.slt.part

@@ -32,6 +32,9 @@ GRANT ALL PRIVILEGES ON foo TO test_user GRANTED BY root;
 statement ok
 GRANT INSERT ON bar TO test_user WITH GRANT OPTION GRANTED BY root;
 
+statement ok
+GRANT INSERT ON foo_view TO test_user WITH GRANT OPTION GRANTED BY root;
+
 statement ok
 GRANT SELECT ON ALL TABLES IN SCHEMA public TO test_user WITH GRANT OPTION GRANTED BY root;
 
@@ -44,6 +47,9 @@ GRANT SELECT ON ALL SOURCES IN SCHEMA public TO test_user WITH GRANT OPTION GRAN
 statement ok
 GRANT CREATE ON SCHEMA test_schema TO test_user;
 
+query error table not found: bar_err
+GRANT INSERT ON bar_err TO test_user WITH GRANT OPTION GRANTED BY root;
+
 query error Invalid parameter user: User test_user_err not found
 SELECT has_table_privilege('test_user_err', 'foo', 'SELECT');
 
@@ -86,21 +92,25 @@ SELECT has_table_privilege('test_user', 'foo', 'INSERT WITH GRANT OPTION');
 ----
 f
 
-# FIXME(Kexiang): Currently, RW's grant privilege on all table doesn't apply to VIEWS.
 query I
 SELECT has_table_privilege('test_user', 'foo_view', 'SELECT');
 ----
-f
+t
 
 query I
 SELECT has_table_privilege('test_user', 'foo_view'::regclass, 'INSERT');
 ----
+t
+
+query I
+SELECT has_table_privilege('test_user', 'foo_view'::regclass, 'UPDATE');
+----
 f
 
 query I
 SELECT has_any_column_privilege('test_user', 'foo_view'::regclass, 'INSERT');
 ----
-f
+t
 
 query I
 SELECT has_table_privilege('test_user', 'foo_mv', 'SELECT');
@@ -188,6 +198,32 @@ SELECT has_schema_privilege('test_user', 'test_schema', 'CREATE');
 ----
 t
 
+statement ok
+REVOKE SELECT ON ALL TABLES IN SCHEMA public FROM test_user GRANTED BY root;
+
+query I
+SELECT has_table_privilege('test_user', 'bar'::regclass, 'SELECT');
+----
+f
+
+query I
+SELECT has_table_privilege('test_user', 'foo_view', 'SELECT');
+----
+f
+
+query I
+SELECT has_table_privilege('test_user', 'foo_view', 'INSERT');
+----
+t
+
+statement ok
+REVOKE INSERT ON foo_view FROM test_user GRANTED BY root;
+
+query I
+SELECT has_table_privilege('test_user', 'foo_view', 'INSERT');
+----
+f
+
 statement ok
 DROP SOURCE foo_source;
 

proto/user.proto

@@ -66,7 +66,7 @@ message GrantPrivilege {
 
     uint32 all_tables_schema_id = 11;
     uint32 all_sources_schema_id = 12;
-    uint32 all_dml_tables_schema_id = 13;
+    uint32 all_dml_relations_schema_id = 13;
     uint32 subscription_id = 14;
   }
   repeated ActionWithGrantOption action_with_opts = 7;

src/common/src/acl/mod.rs

@@ -100,11 +100,11 @@ pub static ALL_AVAILABLE_DATABASE_MODES: LazyLock<AclModeSet> =
     LazyLock::new(|| make_bitflags!(AclMode::{Create | Connect}).into());
 pub static ALL_AVAILABLE_SCHEMA_MODES: LazyLock<AclModeSet> =
     LazyLock::new(|| make_bitflags!(AclMode::{Create | Usage}).into());
+// Including TABLES and VIEWS
 pub static ALL_AVAILABLE_TABLE_MODES: LazyLock<AclModeSet> =
     LazyLock::new(|| make_bitflags!(AclMode::{Select | Insert | Update | Delete}).into());
 pub static ALL_AVAILABLE_SOURCE_MODES: LazyLock<AclModeSet> = LazyLock::new(AclModeSet::readonly);
 pub static ALL_AVAILABLE_MVIEW_MODES: LazyLock<AclModeSet> = LazyLock::new(AclModeSet::readonly);
-pub static ALL_AVAILABLE_VIEW_MODES: LazyLock<AclModeSet> = LazyLock::new(AclModeSet::readonly);
 pub static ALL_AVAILABLE_SINK_MODES: LazyLock<AclModeSet> = LazyLock::new(AclModeSet::empty);
 pub static ALL_AVAILABLE_SUBSCRIPTION_MODES: LazyLock<AclModeSet> =
     LazyLock::new(AclModeSet::empty);

src/frontend/src/handler/handle_privilege.rs

@@ -21,6 +21,7 @@ use super::RwPgResponse;
 use crate::binder::Binder;
 use crate::catalog::root_catalog::SchemaPath;
 use crate::catalog::table_catalog::TableType;
+use crate::catalog::CatalogError;
 use crate::error::{ErrorCode, Result};
 use crate::handler::HandlerArgs;
 use crate::session::SessionImpl;
@@ -93,17 +94,31 @@ fn make_prost_privilege(
                     Binder::resolve_schema_qualified_name(db_name, name)?;
                 let schema_path = SchemaPath::new(schema_name.as_deref(), &search_path, user_name);
 
-                let (table, _) = reader.get_table_by_name(db_name, schema_path, &table_name)?;
-                match table.table_type() {
-                    TableType::Table => {}
-                    _ => {
-                        return Err(ErrorCode::InvalidInputSyntax(format!(
-                            "{table_name} is not a table",
-                        ))
-                        .into());
+                match reader.get_table_by_name(db_name, schema_path, &table_name) {
+                    Ok((table, _)) => {
+                        match table.table_type() {
+                            TableType::Table => {
+                                grant_objs.push(PbObject::TableId(table.id().table_id));
+                                continue;
+                            }
+                            _ => {
+                                return Err(ErrorCode::InvalidInputSyntax(format!(
+                                    "{table_name} is not a table",
+                                ))
+                                .into());
+                            }
+                        };
+                    }
+                    Err(CatalogError::NotFound("table", _)) => {
+                        let (view, _) = reader
+                            .get_view_by_name(db_name, schema_path, &table_name)
+                            .map_err(|_| CatalogError::NotFound("table", table_name))?;
+                        grant_objs.push(PbObject::ViewId(view.id));
+                    }
+                    Err(e) => {
+                        return Err(e.into());
                     }
                 }
-                grant_objs.push(PbObject::TableId(table.id().table_id));
             }
         }
         GrantObjects::Sources(sources) => {
@@ -138,7 +153,7 @@ fn make_prost_privilege(
             for schema in schemas {
                 let schema_name = Binder::resolve_schema_name(schema)?;
                 let schema = reader.get_schema_by_name(session.database(), &schema_name)?;
-                grant_objs.push(PbObject::AllDmlTablesSchemaId(schema.id()));
+                grant_objs.push(PbObject::AllDmlRelationsSchemaId(schema.id()));
             }
         }
         o => {

src/frontend/src/user/user_privilege.rs

@@ -94,7 +94,7 @@ pub fn available_prost_privilege(object: PbObject, for_dml_table: bool) -> PbGra
                 &acl::ALL_AVAILABLE_MVIEW_MODES
             }
         }
-        PbObject::ViewId(_) => &acl::ALL_AVAILABLE_VIEW_MODES,
+        PbObject::ViewId(_) => &acl::ALL_AVAILABLE_TABLE_MODES,
         PbObject::SinkId(_) => &acl::ALL_AVAILABLE_SINK_MODES,
         PbObject::SubscriptionId(_) => &acl::ALL_AVAILABLE_SUBSCRIPTION_MODES,
         PbObject::FunctionId(_) => &acl::ALL_AVAILABLE_FUNCTION_MODES,

src/meta/service/src/user_service.rs

@@ -76,7 +76,7 @@ impl UserServiceImpl {
                     }
                     expanded_privileges.push(privilege);
                 }
-            } else if let Some(Object::AllDmlTablesSchemaId(schema_id)) = &privilege.object {
+            } else if let Some(Object::AllDmlRelationsSchemaId(schema_id)) = &privilege.object {
                 let tables = match &self.metadata_manager {
                     MetadataManager::V1(mgr) => {
                         mgr.catalog_manager.list_dml_table_ids(*schema_id).await
@@ -89,6 +89,16 @@ impl UserServiceImpl {
                         .map(|id| id as _)
                         .collect(),
                 };
+                let views = match &self.metadata_manager {
+                    MetadataManager::V1(mgr) => mgr.catalog_manager.list_view_ids(*schema_id).await,
+                    MetadataManager::V2(mgr) => mgr
+                        .catalog_controller
+                        .list_view_ids(*schema_id as _)
+                        .await?
+                        .into_iter()
+                        .map(|id| id as _)
+                        .collect(),
+                };
                 for table_id in tables {
                     let mut privilege = privilege.clone();
                     privilege.object = Some(Object::TableId(table_id));
@@ -100,6 +110,17 @@ impl UserServiceImpl {
                     }
                     expanded_privileges.push(privilege);
                 }
+                for view_id in views {
+                    let mut privilege = privilege.clone();
+                    privilege.object = Some(Object::ViewId(view_id));
+                    if let Some(true) = with_grant_option {
+                        privilege
+                            .action_with_opts
+                            .iter_mut()
+                            .for_each(|p| p.with_grant_option = true);
+                    }
+                    expanded_privileges.push(privilege);
+                }
             } else if let Some(Object::AllSourcesSchemaId(schema_id)) = &privilege.object {
                 let sources = match &self.metadata_manager {
                     MetadataManager::V1(mgr) => {

src/meta/src/controller/catalog.rs

@@ -31,7 +31,7 @@ use risingwave_meta_model_v2::{
     ActorUpstreamActors, ColumnCatalogArray, ConnectionId, CreateType, DatabaseId, FragmentId,
     FunctionId, I32Array, IndexId, JobStatus, ObjectId, PrivateLinkService, Property, SchemaId,
     SinkId, SourceId, StreamNode, StreamSourceInfo, StreamingParallelism, SubscriptionId, TableId,
-    UserId,
+    UserId, ViewId,
 };
 use risingwave_pb::catalog::subscription::SubscriptionState;
 use risingwave_pb::catalog::table::PbTableType;
@@ -2428,6 +2428,19 @@ impl CatalogController {
         Ok(table_ids)
     }
 
+    pub async fn list_view_ids(&self, schema_id: SchemaId) -> MetaResult<Vec<ViewId>> {
+        let inner = self.inner.read().await;
+        let view_ids: Vec<ViewId> = View::find()
+            .select_only()
+            .column(view::Column::ViewId)
+            .join(JoinType::InnerJoin, view::Relation::Object.def())
+            .filter(object::Column::SchemaId.eq(schema_id))
+            .into_tuple()
+            .all(&inner.db)
+            .await?;
+        Ok(view_ids)
+    }
+
     pub async fn list_tables_by_type(&self, table_type: TableType) -> MetaResult<Vec<PbTable>> {
         let inner = self.inner.read().await;
         let table_objs = Table::find()

src/meta/src/manager/catalog/database.rs

@@ -363,6 +363,14 @@ impl DatabaseManager {
             .collect_vec()
     }
 
+    pub fn list_view_ids(&self, schema_id: SchemaId) -> Vec<ViewId> {
+        self.views
+            .values()
+            .filter(|view| view.schema_id == schema_id)
+            .map(|view| view.id)
+            .collect_vec()
+    }
+
     pub fn list_sources(&self) -> Vec<Source> {
         self.sources.values().cloned().collect_vec()
     }

src/meta/src/manager/catalog/mod.rs

@@ -3608,6 +3608,10 @@ impl CatalogManager {
             .list_dml_table_ids(schema_id)
     }
 
+    pub async fn list_view_ids(&self, schema_id: SchemaId) -> Vec<TableId> {
+        self.core.lock().await.database.list_view_ids(schema_id)
+    }
+
     pub async fn list_sources(&self) -> Vec<Source> {
         self.core.lock().await.database.list_sources()
     }