feat(meta): introduce user service in meta, to support user and privilege management

proto/user.proto

@@ -20,15 +20,14 @@ message AuthInfo {
 
 /// User defines a user in the system.
 message UserInfo {
-  uint32 id = 1;
-  string name = 2;
-  bool is_supper = 3;
-  bool can_create_db = 4;
-  bool can_login = 5;
-  AuthInfo auth_info = 6;
+  string name = 1;
+  bool is_supper = 2;
+  bool can_create_db = 3;
+  bool can_login = 4;
+  AuthInfo auth_info = 5;
 
   /// Granted privileges will be only updated through the command of GRANT/REVOKE.
-  repeated GrantPrivilege privileges = 7;
+  repeated GrantPrivilege grant_privileges = 6;
 }
 
 /// GrantPrivilege defines a privilege granted to a user.
@@ -48,12 +47,23 @@ message GrantPrivilege {
     uint32 table_id = 3;
   }
 
+  message GrantSource {
+    uint32 database_id = 1;
+    uint32 schema_id = 2;
+    uint32 source_id = 3;
+  }
+
   /// To support grant privilege on ALL TABLES IN SCHEMA schema_name.
   message GrantAllTables {
     uint32 database_id = 1;
     uint32 schema_id = 2;
   }
 
+  message GrantAllSources {
+    uint32 database_id = 1;
+    uint32 schema_id = 2;
+  }
+
   enum Privilege {
     UNKNOWN = 0;
     SELECT = 1;
@@ -62,16 +72,22 @@ message GrantPrivilege {
     DELETE = 4;
     CREATE = 5;
     CONNECT = 6;
-    ALL = 20;
   }
+
+  message PrivilegeWithGrantOption {
+    Privilege privilege = 1;
+    bool with_grant_option = 2;
+  }
+
   oneof target {
     GrantDatabase grant_database = 1;
     GrantSchema grant_schema = 2;
     GrantTable grant_table = 3;
-    GrantAllTables grant_all_tables = 4;
+    GrantSource grant_source = 4;
+    GrantAllTables grant_all_tables = 5;
+    GrantAllSources grant_all_sources = 6;
   }
-  repeated Privilege privileges = 5;
-  bool with_grant_option = 6;
+  repeated PrivilegeWithGrantOption privilege_with_opts = 7;
 }
 
 message CreateUserRequest {
@@ -80,12 +96,11 @@ message CreateUserRequest {
 
 message CreateUserResponse {
   common.Status status = 1;
-  uint32 user_id = 2;
-  uint64 version = 3;
+  uint64 version = 2;
 }
 
 message DropUserRequest {
-  uint32 user_id = 1;
+  string name = 1;
 }
 
 message DropUserResponse {
@@ -94,8 +109,9 @@ message DropUserResponse {
 }
 
 message GrantPrivilegeRequest {
-  uint32 user_id = 1;
-  GrantPrivilege privilege = 2;
+  string user_name = 1;
+  repeated GrantPrivilege privileges = 2;
+  bool with_grant_option = 3;
 }
 
 message GrantPrivilegeResponse {
@@ -104,8 +120,9 @@ message GrantPrivilegeResponse {
 }
 
 message RevokePrivilegeRequest {
-  uint32 user_id = 1;
-  GrantPrivilege privilege = 2;
+  string user_name = 1;
+  repeated GrantPrivilege privileges = 2;
+  bool revoke_grant_option = 3;
 }
 
 message RevokePrivilegeResponse {

src/common/src/catalog/mod.rs

@@ -25,6 +25,9 @@ pub use schema::{test_utils as schema_test_utils, Field, Schema};
 pub const DEFAULT_DATABASE_NAME: &str = "dev";
 pub const DEFAULT_SCHEMA_NAME: &str = "dev";
 
+pub const DEFAULT_SUPPER_USER: &str = "risingwave";
+pub const DEFAULT_SUPPER_USER_PASSWORD: &str = "risingwave";
+
 pub type CatalogVersion = u64;
 
 pub enum CatalogId {

src/meta/src/manager/catalog.rs

@@ -570,6 +570,34 @@ where
             ))),
         }
     }
+
+    pub async fn list_tables(
+        &self,
+        database_id: DatabaseId,
+        schema_id: SchemaId,
+    ) -> Result<Vec<TableId>> {
+        let core = self.core.lock().await;
+        let tables = Table::list(core.env.meta_store()).await?;
+        Ok(tables
+            .iter()
+            .filter(|t| t.database_id == database_id && t.schema_id == schema_id)
+            .map(|t| t.id)
+            .collect())
+    }
+
+    pub async fn list_sources(
+        &self,
+        database_id: DatabaseId,
+        schema_id: SchemaId,
+    ) -> Result<Vec<SourceId>> {
+        let core = self.core.lock().await;
+        let sources = Source::list(core.env.meta_store()).await?;
+        Ok(sources
+            .iter()
+            .filter(|s| s.database_id == database_id && s.schema_id == schema_id)
+            .map(|s| s.id)
+            .collect())
+    }
 }
 
 type DatabaseKey = String;

src/meta/src/manager/mod.rs

@@ -17,9 +17,11 @@ mod env;
 mod hash_mapping;
 mod id;
 mod notification;
+mod user;
 
 pub use catalog::*;
 pub use env::*;
 pub use hash_mapping::*;
 pub use id::*;
 pub use notification::*;
+pub use user::*;