feat(frontend): support acl column in pg_namespace

src/frontend/src/catalog/pg_catalog/mod.rs

@@ -4,7 +4,7 @@
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+// http://www.apache.org/licenses/&LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
@@ -19,7 +19,7 @@ pub mod pg_namespace;
 pub mod pg_type;
 pub mod pg_user;
 
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
 use std::sync::Arc;
 
 use async_trait::async_trait;
@@ -28,6 +28,8 @@ use risingwave_common::array::Row;
 use risingwave_common::catalog::{ColumnDesc, SysCatalogReader, TableId, DEFAULT_SUPER_USER_ID};
 use risingwave_common::error::{ErrorCode, Result};
 use risingwave_common::types::{DataType, ScalarImpl};
+use risingwave_pb::user::grant_privilege::{Action, Object};
+use risingwave_pb::user::UserInfo;
 use serde_json::json;
 
 use crate::catalog::catalog_service::CatalogReader;
@@ -43,6 +45,7 @@ use crate::meta_client::FrontendMetaClient;
 use crate::scheduler::worker_node_manager::WorkerNodeManagerRef;
 use crate::session::AuthContext;
 use crate::user::user_service::UserInfoReader;
+use crate::user::UserId;
 
 #[expect(dead_code)]
 pub struct SysCatalogReaderImpl {
@@ -92,17 +95,104 @@ impl SysCatalogReader for SysCatalogReaderImpl {
     }
 }
 
+/// get acl items of `object` in string, ignore public.
+fn get_acl_items(
+    object: &Object,
+    users: &Vec<UserInfo>,
+    username_map: &HashMap<UserId, String>,
+) -> String {
+    let mut res = String::new();
+    for user in users {
+        let privileges = user
+            .get_grant_privileges()
+            .iter()
+            .filter(|&privilege| privilege.object.as_ref().unwrap() == object)
+            .collect_vec();
+        if privileges.is_empty() {
+            continue;
+        };
+        if !res.is_empty() {
+            res.push('\n');
+        }
+        let mut grantor_map = HashMap::new();
+        privileges.iter().for_each(|&privilege| {
+            privilege.action_with_opts.iter().for_each(|ao| {
+                grantor_map.entry(ao.granted_by).or_insert_with(Vec::new);
+                grantor_map
+                    .get_mut(&ao.granted_by)
+                    .unwrap()
+                    .push((ao.action, ao.with_grant_option));
+            })
+        });
+        for key in grantor_map.keys() {
+            grantor_map
+                .get(key)
+                .unwrap()
+                .iter()
+                .for_each(|(action, option)| {
+                    let str = match Action::from_i32(*action).unwrap() {
+                        Action::Select => "r",
+                        Action::Insert => "a",
+                        Action::Update => "w",
+                        Action::Delete => "d",
+                        Action::Create => "C",
+                        Action::Connect => "c",
+                        _ => "",
+                    };
+                    res.push_str(str);
+                    if *option {
+                        res.push('*');
+                    }
+                });
+            res.push('/');
+            // should be able to query grantor's name
+            res.push_str(username_map.get(key).as_ref().unwrap());
+        }
+    }
+    res
+}
 impl SysCatalogReaderImpl {
     fn read_namespace(&self) -> Result<Vec<Row>> {
+        let privileges = {
+            let user_reader = self.user_info_reader.read_guard();
+            user_reader
+                .get_user_by_name(&self.auth_context.user_name)
+                .unwrap()
+                .grant_privileges
+                .clone()
+        };
+        let mut allowed_schemas = HashSet::new();
+        privileges.iter().for_each(|privilege| {
+            if let Some(Object::SchemaId(id)) = privilege.object {
+                if privilege
+                    .action_with_opts
+                    .iter()
+                    .any(|action| action.action == Action::Select as i32)
+                {
+                    allowed_schemas.insert(id);
+                }
+            }
+        });
         let reader = self.catalog_reader.read_guard();
         let schemas = reader.get_all_schema_info(&self.auth_context.database)?;
+        let user_reader = self.user_info_reader.read_guard();
+        let users = &user_reader.get_all_users();
+        let username_map = user_reader.get_user_name_map();
         Ok(schemas
             .iter()
+            .filter(|&schema| {
+                schema.owner == self.auth_context.user_id || allowed_schemas.contains(&schema.id)
+            })
             .map(|schema| {
                 Row::new(vec![
                     Some(ScalarImpl::Int32(schema.id as i32)),
                     Some(ScalarImpl::Utf8(schema.name.clone())),
                     Some(ScalarImpl::Int32(schema.owner as i32)),
+                    Some(ScalarImpl::Utf8(get_acl_items(
+                        &Object::SchemaId(schema.id),
+                        users,
+                        username_map,
+                    ))),
                 ])
             })
             .collect_vec())

src/frontend/src/catalog/pg_catalog/pg_namespace.rs

@@ -23,5 +23,6 @@ pub const PG_NAMESPACE_TABLE_NAME: &str = "pg_namespace";
 pub const PG_NAMESPACE_COLUMNS: &[PgCatalogColumnsDef] = &[
     (DataType::Int32, "oid"),
     (DataType::Varchar, "nspname"),
-    (DataType::Int32, "nspowner"), // TODO: support ACL here.
+    (DataType::Int32, "nspowner"),
+    (DataType::Varchar, "nspacl"),
 ];

src/frontend/src/user/user_manager.rs

@@ -55,6 +55,10 @@ impl UserInfoManager {
         self.user_name_by_id.get(&id).cloned()
     }
 
+    pub fn get_user_name_map(&self) -> &HashMap<UserId, String> {
+        &self.user_name_by_id
+    }
+
     pub fn create_user(&mut self, user_info: UserInfo) {
         let id = user_info.id;
         let name = user_info.name.clone();

src/frontend/test_runner/tests/testdata/pg_catalog.yaml

@@ -9,10 +9,10 @@
 - sql: |
     select * from pg_catalog.pg_namespace
   logical_plan: |
-    LogicalProject { exprs: [pg_namespace.oid, pg_namespace.nspname, pg_namespace.nspowner] }
-      LogicalScan { table: pg_namespace, columns: [oid, nspname, nspowner] }
+    LogicalProject { exprs: [pg_namespace.oid, pg_namespace.nspname, pg_namespace.nspowner, pg_namespace.nspacl] }
+      LogicalScan { table: pg_namespace, columns: [oid, nspname, nspowner, nspacl] }
   batch_plan: |
-    BatchScan { table: pg_namespace, columns: [oid, nspname, nspowner] }
+    BatchScan { table: pg_namespace, columns: [oid, nspname, nspowner, nspacl] }
 - sql: |
     select * from pg_catalog.pg_cast
   logical_plan: |