fix(catalog): cherry-pick privilege fixes #16674 and #16699

e2e_test/batch/catalog/has_privilege.slt.part

@@ -0,0 +1,264 @@
+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 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;
+
+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 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');
+
+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
+
+query I
+SELECT has_table_privilege('test_user', 'foo_view', 'SELECT');
+----
+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');
+----
+t
+
+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
+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;
+
+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;

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/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();