feat: 2/7 add role membership read contract

proto/user.proto

@@ -35,6 +35,7 @@ message UserInfo {
   repeated GrantPrivilege grant_privileges = 8;
 
   bool is_admin = 9;
+  bool can_inherit = 10;
 }
 
 enum Action {
@@ -100,6 +101,7 @@ message UpdateUserRequest {
     RENAME = 5;
     CREATE_USER = 6;
     ADMIN = 7;
+    INHERIT = 8;
   }
   UserInfo user = 1;
   repeated UpdateField update_fields = 2;
@@ -136,6 +138,24 @@ message RevokePrivilegeResponse {
   uint64 version = 2;
 }
 
+message RoleMembership {
+  uint32 role_id = 1;
+  uint32 member_id = 2;
+  uint32 granted_by = 3;
+  bool admin_option = 4;
+  bool inherit_option = 5;
+  bool set_option = 6;
+  uint32 id = 7;
+}
+
+message ListRoleMembershipsRequest {
+  repeated uint32 member_ids = 1;
+}
+
+message ListRoleMembershipsResponse {
+  repeated RoleMembership memberships = 1;
+}
+
 message AlterDefaultPrivilegeRequest {
   repeated uint32 user_ids = 1;
   uint32 database_id = 2;
@@ -175,4 +195,6 @@ service UserService {
   rpc RevokePrivilege(RevokePrivilegeRequest) returns (RevokePrivilegeResponse);
   // AlterDefaultPrivilege alters the default privileges.
   rpc AlterDefaultPrivilege(AlterDefaultPrivilegeRequest) returns (AlterDefaultPrivilegeResponse);
+  // ListRoleMemberships lists direct role memberships for the requested members.
+  rpc ListRoleMemberships(ListRoleMembershipsRequest) returns (ListRoleMembershipsResponse);
 }

src/frontend/src/handler/create_user.rs

@@ -45,6 +45,7 @@ pub async fn handle_create_user(
         name: user_name.clone(),
         // the LOGIN option is implied if it is not explicitly specified.
         can_login: true,
+        can_inherit: true,
         ..Default::default()
     };
     let mut notices = vec![];

src/frontend/src/meta_client.rs

@@ -15,7 +15,7 @@
 use std::collections::{BTreeMap, HashMap, HashSet};
 
 use anyhow::Context;
-use risingwave_common::id::{ConnectionId, JobId, SourceId, TableId, WorkerId};
+use risingwave_common::id::{ConnectionId, JobId, SourceId, TableId, UserId, WorkerId};
 use risingwave_common::session_config::SessionConfig;
 use risingwave_common::system_param::reader::SystemParamsReader;
 use risingwave_common::util::cluster_limit::ClusterLimit;
@@ -46,6 +46,7 @@ use risingwave_pb::meta::{
     RefreshRequest, RefreshResponse, list_sink_log_store_tables_response,
 };
 use risingwave_pb::secret::PbSecretRef;
+use risingwave_pb::user::RoleMembership as PbRoleMembership;
 use risingwave_rpc_client::error::Result;
 use risingwave_rpc_client::{HummockMetaClient, MetaClient};
 
@@ -210,6 +211,9 @@ pub trait FrontendMetaClient: Send + Sync {
 
     async fn list_hosted_iceberg_tables(&self) -> Result<Vec<IcebergTable>>;
 
+    async fn list_role_memberships(&self, member_ids: Vec<UserId>)
+    -> Result<Vec<PbRoleMembership>>;
+
     async fn get_fragment_by_id(
         &self,
         fragment_id: FragmentId,
@@ -543,6 +547,13 @@ impl FrontendMetaClient for FrontendMetaClientImpl {
         self.0.list_hosted_iceberg_tables().await
     }
 
+    async fn list_role_memberships(
+        &self,
+        member_ids: Vec<UserId>,
+    ) -> Result<Vec<PbRoleMembership>> {
+        self.0.list_role_memberships(member_ids).await
+    }
+
     async fn get_fragment_by_id(
         &self,
         fragment_id: FragmentId,

src/frontend/src/test_utils.rs

@@ -75,7 +75,7 @@ use risingwave_pb::secret::PbSecretRef;
 use risingwave_pb::stream_plan::StreamFragmentGraph;
 use risingwave_pb::user::alter_default_privilege_request::Operation as AlterDefaultPrivilegeOperation;
 use risingwave_pb::user::update_user_request::UpdateField;
-use risingwave_pb::user::{GrantPrivilege, UserInfo};
+use risingwave_pb::user::{GrantPrivilege, RoleMembership, UserInfo};
 use risingwave_rpc_client::error::Result as RpcResult;
 use tempfile::{Builder, NamedTempFile};
 
@@ -1050,6 +1050,7 @@ impl UserInfoWriter for MockUserInfoWriter {
             UpdateField::AuthInfo => user_info.auth_info.clone_from(&update_user.auth_info),
             UpdateField::Rename => user_info.name.clone_from(&update_user.name),
             UpdateField::Admin => user_info.is_admin = update_user.is_admin,
+            UpdateField::Inherit => user_info.can_inherit = update_user.can_inherit,
             UpdateField::Unspecified => unreachable!(),
         });
         lock.update_user(update_user);
@@ -1122,6 +1123,7 @@ impl MockUserInfoWriter {
             can_create_db: true,
             can_create_user: true,
             can_login: true,
+            can_inherit: true,
             ..Default::default()
         });
         user_info.write().create_user(UserInfo {
@@ -1132,6 +1134,7 @@ impl MockUserInfoWriter {
             can_create_user: true,
             can_login: true,
             is_admin: true,
+            can_inherit: true,
             ..Default::default()
         });
         Self {
@@ -1383,6 +1386,13 @@ impl FrontendMetaClient for MockFrontendMetaClient {
         unimplemented!()
     }
 
+    async fn list_role_memberships(
+        &self,
+        _member_ids: Vec<UserId>,
+    ) -> RpcResult<Vec<RoleMembership>> {
+        Ok(vec![])
+    }
+
     async fn list_iceberg_compaction_status(&self) -> RpcResult<Vec<IcebergCompactionStatus>> {
         Ok(vec![])
     }

src/frontend/src/user/user_catalog.rs

@@ -31,6 +31,7 @@ pub struct UserCatalog {
     pub can_create_user: bool,
     pub can_login: bool,
     pub is_admin: bool,
+    pub can_inherit: bool,
     pub auth_info: Option<PbAuthInfo>,
     pub grant_privileges: Vec<PbGrantPrivilege>,
 
@@ -51,6 +52,7 @@ impl From<PbUserInfo> for UserCatalog {
             can_create_user: user.can_create_user,
             can_login: user.can_login,
             is_admin: user.is_admin,
+            can_inherit: user.can_inherit,
             auth_info: user.auth_info,
             grant_privileges: user.grant_privileges,
             database_acls: Default::default(),
@@ -73,6 +75,7 @@ impl UserCatalog {
             can_create_user: self.can_create_user,
             can_login: self.can_login,
             is_admin: self.is_admin,
+            can_inherit: self.can_inherit,
             auth_info: self.auth_info.clone(),
             grant_privileges: self.grant_privileges.clone(),
         }

src/meta/model/migration/src/lib.rs

@@ -68,6 +68,8 @@ mod m20251224_142321_sink_schema_change;
 mod m20251231_000000_sink_ignore_delete;
 mod m20260119_153927_streaming_job_is_serverless_backfill;
 mod m20260120_120000_streaming_job_backfill_orders;
+mod m20260422_000001_role_membership;
+mod m20260422_000002_user_inherit_flag;
 mod utils;
 
 pub struct Migrator;
@@ -174,6 +176,8 @@ impl MigratorTrait for Migrator {
             Box::new(m20251231_000000_sink_ignore_delete::Migration),
             Box::new(m20260119_153927_streaming_job_is_serverless_backfill::Migration),
             Box::new(m20260120_120000_streaming_job_backfill_orders::Migration),
+            Box::new(m20260422_000001_role_membership::Migration),
+            Box::new(m20260422_000002_user_inherit_flag::Migration),
         ]
     }
 }

src/meta/model/migration/src/m20260422_000001_role_membership.rs

@@ -0,0 +1,131 @@
+use sea_orm_migration::prelude::*;
+
+#[derive(DeriveMigrationName)]
+pub struct Migration;
+
+#[async_trait::async_trait]
+impl MigrationTrait for Migration {
+    async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+        manager
+            .create_table(
+                Table::create()
+                    .table(UserRoleMembership::Table)
+                    .if_not_exists()
+                    .col(
+                        ColumnDef::new(UserRoleMembership::Id)
+                            .integer()
+                            .primary_key()
+                            .auto_increment(),
+                    )
+                    .col(
+                        ColumnDef::new(UserRoleMembership::RoleId)
+                            .integer()
+                            .not_null(),
+                    )
+                    .col(
+                        ColumnDef::new(UserRoleMembership::MemberId)
+                            .integer()
+                            .not_null(),
+                    )
+                    .col(
+                        ColumnDef::new(UserRoleMembership::GrantedBy)
+                            .integer()
+                            .not_null(),
+                    )
+                    .col(
+                        ColumnDef::new(UserRoleMembership::AdminOption)
+                            .boolean()
+                            .not_null()
+                            .default(false),
+                    )
+                    .col(
+                        ColumnDef::new(UserRoleMembership::InheritOption)
+                            .boolean()
+                            .not_null()
+                            .default(true),
+                    )
+                    .col(
+                        ColumnDef::new(UserRoleMembership::SetOption)
+                            .boolean()
+                            .not_null()
+                            .default(true),
+                    )
+                    .foreign_key(
+                        &mut ForeignKey::create()
+                            .name("FK_user_role_membership_role_id")
+                            .from(UserRoleMembership::Table, UserRoleMembership::RoleId)
+                            .to(User::Table, User::UserId)
+                            .on_delete(ForeignKeyAction::Cascade)
+                            .to_owned(),
+                    )
+                    .foreign_key(
+                        &mut ForeignKey::create()
+                            .name("FK_user_role_membership_member_id")
+                            .from(UserRoleMembership::Table, UserRoleMembership::MemberId)
+                            .to(User::Table, User::UserId)
+                            .on_delete(ForeignKeyAction::Cascade)
+                            .to_owned(),
+                    )
+                    .foreign_key(
+                        &mut ForeignKey::create()
+                            .name("FK_user_role_membership_granted_by")
+                            .from(UserRoleMembership::Table, UserRoleMembership::GrantedBy)
+                            .to(User::Table, User::UserId)
+                            .to_owned(),
+                    )
+                    .to_owned(),
+            )
+            .await?;
+
+        manager
+            .create_index(
+                Index::create()
+                    .name("idx_user_role_membership_item")
+                    .table(UserRoleMembership::Table)
+                    .unique()
+                    .col(UserRoleMembership::RoleId)
+                    .col(UserRoleMembership::MemberId)
+                    .col(UserRoleMembership::GrantedBy)
+                    .to_owned(),
+            )
+            .await?;
+
+        Ok(())
+    }
+
+    async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+        manager
+            .drop_index(
+                Index::drop()
+                    .name("idx_user_role_membership_item")
+                    .table(UserRoleMembership::Table)
+                    .to_owned(),
+            )
+            .await?;
+
+        manager
+            .drop_table(Table::drop().table(UserRoleMembership::Table).to_owned())
+            .await?;
+
+        Ok(())
+    }
+}
+
+#[derive(DeriveIden)]
+enum UserRoleMembership {
+    Table,
+    Id,
+    RoleId,
+    MemberId,
+    GrantedBy,
+    AdminOption,
+    InheritOption,
+    SetOption,
+}
+
+#[derive(DeriveIden)]
+enum User {
+    #[sea_orm(iden = "user")]
+    Table,
+    UserId,
+}

src/meta/model/migration/src/m20260422_000002_user_inherit_flag.rs

@@ -0,0 +1,41 @@
+use sea_orm_migration::prelude::*;
+
+#[derive(DeriveMigrationName)]
+pub struct Migration;
+
+#[async_trait::async_trait]
+impl MigrationTrait for Migration {
+    async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+        manager
+            .alter_table(
+                Table::alter()
+                    .table(User::Table)
+                    .add_column(
+                        ColumnDef::new(User::CanInherit)
+                            .boolean()
+                            .not_null()
+                            .default(true),
+                    )
+                    .to_owned(),
+            )
+            .await
+    }
+
+    async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+        manager
+            .alter_table(
+                Table::alter()
+                    .table(User::Table)
+                    .drop_column(User::CanInherit)
+                    .to_owned(),
+            )
+            .await
+    }
+}
+
+#[derive(DeriveIden)]
+enum User {
+    #[sea_orm(iden = "user")]
+    Table,
+    CanInherit,
+}

src/meta/model/src/lib.rs

@@ -69,6 +69,7 @@ pub mod table;
 pub mod user;
 pub mod user_default_privilege;
 pub mod user_privilege;
+pub mod user_role_membership;
 pub mod view;
 pub mod worker;
 pub mod worker_property;
@@ -77,6 +78,7 @@ pub type TransactionId = i32;
 
 pub type PrivilegeId = i32;
 pub type DefaultPrivilegeId = i32;
+pub type RoleMembershipId = i32;
 
 pub use risingwave_pb::id::{CompactionGroupId, HummockSstableObjectId, HummockVersionId};
 pub type Epoch = i64;

src/meta/model/src/prelude.rs

@@ -47,6 +47,7 @@ pub use super::table::Entity as Table;
 pub use super::user::Entity as User;
 pub use super::user_default_privilege::Entity as UserDefaultPrivilege;
 pub use super::user_privilege::Entity as UserPrivilege;
+pub use super::user_role_membership::Entity as UserRoleMembership;
 pub use super::view::Entity as View;
 pub use super::worker::Entity as Worker;
 pub use super::worker_property::Entity as WorkerProperty;