feat(catalog): add has_table_privilege, has_schema_privilege, has_any…

e2e_test/batch/catalog/has_privilege.slt.part

@@ -0,0 +1,228 @@
+statement ok
+CREATE USER test_user;
+
+statement ok
+CREATE SCHEMA test_schema;
+
+statement ok
+CREATE TABLE foo (id INT, name VARCHAR);
+
+statement ok
+CREATE VIEW foo_view AS SELECT * FROM foo;
+
+statement ok
+CREATE INDEX foo_index ON foo(id);
+
+statement ok
+CREATE MATERIALIZED VIEW foo_mv AS SELECT * FROM foo;
+
+statement ok
+CREATE SOURCE foo_source (a int, b int) with (
+    connector = 'datagen',
+    datagen.rows.per.second = '1',
+    datagen.split.num = '1'
+);
+
+statement ok
+CREATE TABLE bar (id INT);
+
+statement ok
+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 SELECT ON ALL TABLES IN SCHEMA public TO test_user WITH GRANT OPTION GRANTED BY root;
+
+statement ok
+GRANT SELECT ON ALL MATERIALIZED VIEWS IN SCHEMA public TO test_user WITH GRANT OPTION GRANTED BY root;
+
+statement ok
+GRANT SELECT ON ALL SOURCES IN SCHEMA public TO test_user WITH GRANT OPTION GRANTED BY root;
+
+statement ok
+GRANT CREATE ON SCHEMA test_schema TO test_user;
+
+query error Invalid parameter user: User test_user_err not found
+SELECT has_table_privilege('test_user_err', 'foo', 'SELECT');
+
+query error Invalid parameter name: class not found: foo_err
+SELECT has_table_privilege('test_user', 'foo_err', 'SELECT');
+
+query error Invalid parameter privilege: unrecognized privilege type: "SELE CT"
+SELECT has_table_privilege('test_user', 'foo', 'SELE CT');
+
+query error Invalid parameter privilege: unrecognized privilege type: "SELECT INSERT"
+SELECT has_table_privilege('test_user', 'foo', 'SELECT INSERT');
+
+query error Invalid parameter privilege
+SELECT has_table_privilege('test_user', 'foo', 'SELECT, INSERT  WITH GRANT OPTION');
+
+query error Invalid parameter user: User test_user_err not found
+SELECT has_schema_privilege('test_user_err', 'test_schema', 'CREATE');
+
+query error Invalid parameter schema: schema not found: test_schema_err
+SELECT has_schema_privilege('test_user', 'test_schema_err', 'CREATE');
+
+query error Invalid parameter privilege: unrecognized privilege type: "INSERT"
+SELECT has_schema_privilege('test_user', 'test_schema', 'INSERT');
+
+query error Invalid parameter privilege: unrecognized privilege type: "DELETE"
+SELECT has_any_column_privilege('test_user', 'foo_mv'::regclass, 'DELETE');
+
+query I
+SELECT has_table_privilege('test_user', 'foo', 'SELECT');
+----
+t
+
+query I
+SELECT has_table_privilege('test_user', 'foo', 'SELECT WITH GRANT OPTION');
+----
+t
+
+query I
+SELECT has_table_privilege('test_user', 'foo', 'INSERT WITH GRANT OPTION');
+----
+f
+
+query I
+SELECT has_table_privilege('test_user', 'foo', 'INSERT, SELECT WITH GRANT OPTION');
+----
+t
+
+query I
+SELECT has_table_privilege('test_user', 'foo', 'DELETE, INSERT, SELECT WITH GRANT OPTION');
+----
+t
+
+query I
+SELECT has_table_privilege('test_user', 'foo', 'DELETE WITH GRANT OPTION, INSERT, SELECT 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
+
+query I
+SELECT has_table_privilege('test_user', 'foo_view'::regclass, 'INSERT');
+----
+f
+
+query I
+SELECT has_any_column_privilege('test_user', 'foo_view'::regclass, 'INSERT');
+----
+f
+
+query I
+SELECT has_table_privilege('test_user', 'foo_mv', 'SELECT');
+----
+t
+
+query I
+SELECT has_table_privilege('test_user', 'foo_mv'::regclass, 'SELECT WITH GRANT OPTION');
+----
+t
+
+query I
+SELECT has_any_column_privilege('test_user', 'foo_mv'::regclass, 'SELECT WITH GRANT OPTION');
+----
+t
+
+query I
+SELECT has_table_privilege('test_user', 'foo_mv', 'INSERT');
+----
+f
+
+query I
+SELECT has_table_privilege('test_user', 'foo_source'::regclass, 'SELECT');
+----
+t
+
+query I
+SELECT has_table_privilege('test_user', 'foo_source', 'INSERT');
+----
+f
+
+# Indexes are granted by `GRANT SELECT ON ALL MATERIALIZED VIEWS`
+query I
+SELECT has_table_privilege('test_user', 'foo_index'::regclass, 'SELECT');
+----
+t
+
+query I
+SELECT has_table_privilege('test_user', 'foo_index', 'INSERT');
+----
+f
+
+query I
+SELECT has_table_privilege('test_user', 'bar', 'INSERT');
+----
+t
+
+query I
+SELECT has_table_privilege('bar', 'INSERT');
+----
+t
+
+query I
+SELECT has_table_privilege('bar'::regclass, 'SELECT');
+----
+t
+
+query I
+SELECT has_table_privilege('bar'::regclass, 'SELECT');
+----
+t
+
+query I
+SELECT has_table_privilege('test_user', 'bar', 'UPDATE');
+----
+f
+
+query I
+SELECT has_table_privilege('test_user', 'bar'::regclass, 'INSERT WITH GRANT OPTION');
+----
+t
+
+query I
+SELECT has_schema_privilege('public', 'USAGE');
+----
+t
+
+query I
+SELECT has_schema_privilege('test_user', 'test_schema', 'USAGE');
+----
+f
+
+query I
+SELECT has_schema_privilege('test_user', 'test_schema', 'CREATE');
+----
+t
+
+statement ok
+DROP SOURCE foo_source;
+
+statement ok
+DROP MATERIALIZED VIEW foo_mv;
+
+statement ok
+DROP INDEX foo_index;
+
+statement ok
+DROP VIEW foo_view;
+
+statement ok
+DROP TABLE foo;
+
+statement ok
+DROP TABLE bar;
+
+statement ok
+DROP SCHEMA test_schema;
+
+statement ok
+DROP USER test_user;

proto/expr.proto

@@ -299,6 +299,9 @@ message ExprNode {
     PG_INDEXES_SIZE = 2404;
     PG_RELATION_SIZE = 2405;
     PG_GET_SERIAL_SEQUENCE = 2406;
+    HAS_TABLE_PRIVILEGE = 2407;
+    HAS_ANY_COLUMN_PRIVILEGE = 2408;
+    HAS_SCHEMA_PRIVILEGE = 2409;
 
     // EXTERNAL
     ICEBERG_TRANSFORM = 2201;

src/frontend/src/binder/expr/function.rs

@@ -1303,6 +1303,51 @@ impl Binder {
                 ("pg_get_partkeydef", raw_literal(ExprImpl::literal_null(DataType::Varchar))),
                 ("pg_encoding_to_char", raw_literal(ExprImpl::literal_varchar("UTF8".into()))),
                 ("has_database_privilege", raw_literal(ExprImpl::literal_bool(true))),
+                ("has_table_privilege", raw(|binder, mut inputs|{
+                    if inputs.len() == 2 {
+                        inputs.insert(0, ExprImpl::literal_varchar(binder.auth_context.user_name.clone()));
+                    }
+                    if inputs.len() == 3 {
+                        if inputs[1].return_type() == DataType::Varchar {
+                            inputs[1].cast_to_regclass_mut()?;
+                        }
+                        Ok(FunctionCall::new(ExprType::HasTablePrivilege, inputs)?.into())
+                    } else {
+                        Err(ErrorCode::ExprError(
+                            "Too many/few arguments for pg_catalog.has_table_privilege()".into(),
+                        )
+                        .into())
+                    }
+                })),
+                ("has_any_column_privilege", raw(|binder, mut inputs|{
+                    if inputs.len() == 2 {
+                        inputs.insert(0, ExprImpl::literal_varchar(binder.auth_context.user_name.clone()));
+                    }
+                    if inputs.len() == 3 {
+                        if inputs[1].return_type() == DataType::Varchar {
+                            inputs[1].cast_to_regclass_mut()?;
+                        }
+                        Ok(FunctionCall::new(ExprType::HasAnyColumnPrivilege, inputs)?.into())
+                    } else {
+                        Err(ErrorCode::ExprError(
+                            "Too many/few arguments for pg_catalog.has_any_column_privilege()".into(),
+                        )
+                        .into())
+                    }
+                })),
+                ("has_schema_privilege", raw(|binder, mut inputs|{
+                    if inputs.len() == 2 {
+                        inputs.insert(0, ExprImpl::literal_varchar(binder.auth_context.user_name.clone()));
+                    }
+                    if inputs.len() == 3 {
+                        Ok(FunctionCall::new(ExprType::HasSchemaPrivilege, inputs)?.into())
+                    } else {
+                        Err(ErrorCode::ExprError(
+                            "Too many/few arguments for pg_catalog.has_schema_privilege()".into(),
+                        )
+                        .into())
+                    }
+                })),
                 ("pg_stat_get_numscans", raw_literal(ExprImpl::literal_bigint(0))),
                 ("pg_backend_pid", raw(|binder, _inputs| {
                     // FIXME: the session id is not global unique in multi-frontend env.

src/frontend/src/catalog/database_catalog.rs

@@ -17,6 +17,7 @@ use std::collections::HashMap;
 use itertools::Itertools;
 use risingwave_common::catalog::PG_CATALOG_SCHEMA_NAME;
 use risingwave_pb::catalog::{PbDatabase, PbSchema};
+use risingwave_pb::user::grant_privilege::Object;
 
 use super::OwnedByUserCatalog;
 use crate::catalog::schema_catalog::SchemaCatalog;
@@ -99,6 +100,16 @@ impl DatabaseCatalog {
             .find(|schema| schema.get_table_by_id(table_id).is_some())
     }
 
+    pub fn get_grant_object_by_oid(&self, oid: u32) -> Option<Object> {
+        for schema in self.schema_by_name.values() {
+            let object = schema.get_grant_object_by_oid(oid);
+            if object.is_some() {
+                return object;
+            }
+        }
+        None
+    }
+
     pub fn update_schema(&mut self, prost: &PbSchema) {
         let id = prost.id;
         let name = prost.name.clone();

src/frontend/src/catalog/schema_catalog.rs

@@ -24,6 +24,7 @@ pub use risingwave_expr::sig::*;
 use risingwave_pb::catalog::{
     PbConnection, PbFunction, PbIndex, PbSchema, PbSink, PbSource, PbSubscription, PbTable, PbView,
 };
+use risingwave_pb::user::grant_privilege::Object;
 
 use super::subscription_catalog::SubscriptionCatalog;
 use super::{OwnedByUserCatalog, SubscriptionId};
@@ -703,6 +704,23 @@ impl SchemaCatalog {
             .map(|s| s.to_owned())
     }
 
+    pub fn get_grant_object_by_oid(&self, oid: u32) -> Option<Object> {
+        #[allow(clippy::manual_map)]
+        if self.get_table_by_id(&TableId::new(oid)).is_some()
+            || self.get_index_by_id(&IndexId::new(oid)).is_some()
+        {
+            Some(Object::TableId(oid))
+        } else if self.get_source_by_id(&oid).is_some() {
+            Some(Object::SourceId(oid))
+        } else if self.get_sink_by_id(&oid).is_some() {
+            Some(Object::SinkId(oid))
+        } else if self.get_view_by_id(&oid).is_some() {
+            Some(Object::ViewId(oid))
+        } else {
+            None
+        }
+    }
+
     pub fn id(&self) -> SchemaId {
         self.id
     }