feat: introduce user and privilege functions based on sql meta store

src/meta/model_v2/migration/src/m20230908_072257_init.rs

@@ -124,13 +124,12 @@ impl MigrationTrait for Migration {
                             .primary_key()
                             .auto_increment(),
                     )
-                    .col(ColumnDef::new(User::Name).string().not_null())
+                    .col(ColumnDef::new(User::Name).string().unique_key().not_null())
                     .col(ColumnDef::new(User::IsSuper).boolean().not_null())
                     .col(ColumnDef::new(User::CanCreateDb).boolean().not_null())
                     .col(ColumnDef::new(User::CanCreateUser).boolean().not_null())
                     .col(ColumnDef::new(User::CanLogin).boolean().not_null())
-                    .col(ColumnDef::new(User::AuthType).string())
-                    .col(ColumnDef::new(User::AuthValue).string())
+                    .col(ColumnDef::new(User::AuthInfo).json())
                     .to_owned(),
             )
             .await?;
@@ -197,19 +196,28 @@ impl MigrationTrait for Migration {
                             .primary_key()
                             .auto_increment(),
                     )
+                    .col(ColumnDef::new(UserPrivilege::DependentId).integer())
                     .col(ColumnDef::new(UserPrivilege::UserId).integer().not_null())
                     .col(ColumnDef::new(UserPrivilege::Oid).integer().not_null())
                     .col(
                         ColumnDef::new(UserPrivilege::GrantedBy)
                             .integer()
                             .not_null(),
                     )
-                    .col(ColumnDef::new(UserPrivilege::Actions).string().not_null())
+                    .col(ColumnDef::new(UserPrivilege::Action).string().not_null())
                     .col(
                         ColumnDef::new(UserPrivilege::WithGrantOption)
                             .boolean()
                             .not_null(),
                     )
+                    .foreign_key(
+                        &mut ForeignKey::create()
+                            .name("FK_user_privilege_dependent_id")
+                            .from(UserPrivilege::Table, UserPrivilege::DependentId)
+                            .to(UserPrivilege::Table, UserPrivilege::Id)
+                            .on_delete(ForeignKeyAction::Cascade)
+                            .to_owned(),
+                    )
                     .foreign_key(
                         &mut ForeignKey::create()
                             .name("FK_user_privilege_user_id")
@@ -230,6 +238,7 @@ impl MigrationTrait for Migration {
                             .name("FK_user_privilege_oid")
                             .from(UserPrivilege::Table, UserPrivilege::Oid)
                             .to(Object::Table, Object::Oid)
+                            .on_delete(ForeignKeyAction::Cascade)
                             .to_owned(),
                     )
                     .to_owned(),
@@ -651,6 +660,19 @@ impl MigrationTrait for Migration {
                     .to_owned(),
             )
             .await?;
+        manager
+            .create_index(
+                MigrationIndex::create()
+                    .table(UserPrivilege::Table)
+                    .name("idx_user_privilege_item")
+                    .unique()
+                    .col(UserPrivilege::UserId)
+                    .col(UserPrivilege::Oid)
+                    .col(UserPrivilege::Action)
+                    .col(UserPrivilege::GrantedBy)
+                    .to_owned(),
+            )
+            .await?;
 
         // 4. initialize data.
         let insert_cluster_id = Query::insert()
@@ -799,18 +821,18 @@ enum User {
     CanCreateDb,
     CanCreateUser,
     CanLogin,
-    AuthType,
-    AuthValue,
+    AuthInfo,
 }
 
 #[derive(DeriveIden)]
 enum UserPrivilege {
     Table,
     Id,
+    DependentId,
     UserId,
     Oid,
     GrantedBy,
-    Actions,
+    Action,
     WithGrantOption,
 }
 

src/meta/model_v2/src/connection.rs

@@ -15,7 +15,7 @@
 use risingwave_pb::catalog::connection::PbInfo;
 use risingwave_pb::catalog::PbConnection;
 use sea_orm::entity::prelude::*;
-use sea_orm::ActiveValue;
+use sea_orm::ActiveValue::Set;
 
 use crate::{ConnectionId, PrivateLinkService};
 
@@ -71,9 +71,9 @@ impl From<PbConnection> for ActiveModel {
         };
 
         Self {
-            connection_id: ActiveValue::Set(conn.id as _),
-            name: ActiveValue::Set(conn.name),
-            info: ActiveValue::Set(PrivateLinkService(private_link_srv)),
+            connection_id: Set(conn.id as _),
+            name: Set(conn.name),
+            info: Set(PrivateLinkService(private_link_srv)),
         }
     }
 }

src/meta/model_v2/src/database.rs

@@ -14,7 +14,7 @@
 
 use risingwave_pb::catalog::PbDatabase;
 use sea_orm::entity::prelude::*;
-use sea_orm::ActiveValue;
+use sea_orm::ActiveValue::Set;
 
 use crate::DatabaseId;
 
@@ -50,8 +50,8 @@ impl ActiveModelBehavior for ActiveModel {}
 impl From<PbDatabase> for ActiveModel {
     fn from(db: PbDatabase) -> Self {
         Self {
-            database_id: ActiveValue::Set(db.id),
-            name: ActiveValue::Set(db.name),
+            database_id: Set(db.id),
+            name: Set(db.name),
         }
     }
 }

src/meta/model_v2/src/function.rs

@@ -15,7 +15,7 @@
 use risingwave_pb::catalog::function::Kind;
 use risingwave_pb::catalog::PbFunction;
 use sea_orm::entity::prelude::*;
-use sea_orm::ActiveValue;
+use sea_orm::ActiveValue::Set;
 
 use crate::{DataType, DataTypeArray, FunctionId};
 
@@ -77,14 +77,14 @@ impl From<Kind> for FunctionKind {
 impl From<PbFunction> for ActiveModel {
     fn from(function: PbFunction) -> Self {
         Self {
-            function_id: ActiveValue::Set(function.id as _),
-            name: ActiveValue::Set(function.name),
-            arg_types: ActiveValue::Set(DataTypeArray(function.arg_types)),
-            return_type: ActiveValue::Set(DataType(function.return_type.unwrap())),
-            language: ActiveValue::Set(function.language),
-            link: ActiveValue::Set(function.link),
-            identifier: ActiveValue::Set(function.identifier),
-            kind: ActiveValue::Set(function.kind.unwrap().into()),
+            function_id: Set(function.id as _),
+            name: Set(function.name),
+            arg_types: Set(DataTypeArray(function.arg_types)),
+            return_type: Set(DataType(function.return_type.unwrap())),
+            language: Set(function.language),
+            link: Set(function.link),
+            identifier: Set(function.identifier),
+            kind: Set(function.kind.unwrap().into()),
         }
     }
 }

src/meta/model_v2/src/lib.rs

@@ -61,6 +61,8 @@ pub type ViewId = ObjectId;
 pub type FunctionId = ObjectId;
 pub type ConnectionId = ObjectId;
 pub type UserId = u32;
+pub type PrivilegeId = u32;
+
 pub type HummockVersionId = u64;
 pub type Epoch = u64;
 pub type CompactionGroupId = u64;
@@ -157,6 +159,7 @@ derive_from_json_struct!(
     PrivateLinkService,
     risingwave_pb::catalog::connection::PbPrivateLinkService
 );
+derive_from_json_struct!(AuthInfo, risingwave_pb::user::PbAuthInfo);
 
 derive_from_json_struct!(StreamNode, risingwave_pb::stream_plan::PbStreamNode);
 derive_from_json_struct!(Dispatchers, Vec<risingwave_pb::stream_plan::Dispatcher>);

src/meta/model_v2/src/schema.rs

@@ -14,7 +14,7 @@
 
 use risingwave_pb::catalog::PbSchema;
 use sea_orm::entity::prelude::*;
-use sea_orm::ActiveValue;
+use sea_orm::ActiveValue::Set;
 
 use crate::SchemaId;
 
@@ -49,8 +49,8 @@ impl ActiveModelBehavior for ActiveModel {}
 impl From<PbSchema> for ActiveModel {
     fn from(schema: PbSchema) -> Self {
         Self {
-            schema_id: ActiveValue::Set(schema.id),
-            name: ActiveValue::Set(schema.name),
+            schema_id: Set(schema.id),
+            name: Set(schema.name),
         }
     }
 }

src/meta/model_v2/src/user.rs

@@ -12,22 +12,25 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use risingwave_pb::user::PbUserInfo;
 use sea_orm::entity::prelude::*;
+use sea_orm::ActiveValue::Set;
+use sea_orm::NotSet;
 
-use crate::UserId;
+use crate::{AuthInfo, UserId};
 
 #[derive(Clone, Debug, PartialEq, DeriveEntityModel, Eq)]
 #[sea_orm(table_name = "user")]
 pub struct Model {
     #[sea_orm(primary_key)]
     pub user_id: UserId,
+    #[sea_orm(unique)]
     pub name: String,
     pub is_super: bool,
     pub can_create_db: bool,
     pub can_create_user: bool,
     pub can_login: bool,
-    pub auth_type: Option<String>,
-    pub auth_value: Option<String>,
+    pub auth_info: Option<AuthInfo>,
 }
 
 #[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)]
@@ -43,3 +46,33 @@ impl Related<super::object::Entity> for Entity {
 }
 
 impl ActiveModelBehavior for ActiveModel {}
+
+impl From<PbUserInfo> for ActiveModel {
+    fn from(user: PbUserInfo) -> Self {
+        let user_id = if user.id == 0 { NotSet } else { Set(user.id) };
+        Self {
+            user_id,
+            name: Set(user.name),
+            is_super: Set(user.is_super),
+            can_create_db: Set(user.can_create_db),
+            can_create_user: Set(user.can_create_user),
+            can_login: Set(user.can_login),
+            auth_info: Set(user.auth_info.map(AuthInfo)),
+        }
+    }
+}
+
+impl From<Model> for PbUserInfo {
+    fn from(val: Model) -> Self {
+        PbUserInfo {
+            id: val.user_id,
+            name: val.name,
+            is_super: val.is_super,
+            can_create_db: val.can_create_db,
+            can_create_user: val.can_create_user,
+            can_login: val.can_login,
+            auth_info: val.auth_info.map(|x| x.into_inner()),
+            grant_privileges: vec![], // fill in later
+        }
+    }
+}

src/meta/model_v2/src/user_privilege.rs

@@ -12,19 +12,69 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use risingwave_pb::user::grant_privilege::PbAction;
 use sea_orm::entity::prelude::*;
 
-use crate::{ObjectId, UserId};
+use crate::{ObjectId, PrivilegeId, UserId};
+
+#[derive(Clone, Debug, Hash, PartialEq, Eq, EnumIter, DeriveActiveEnum)]
+#[sea_orm(rs_type = "String", db_type = "String(None)")]
+pub enum Action {
+    #[sea_orm(string_value = "INSERT")]
+    Insert,
+    #[sea_orm(string_value = "SELECT")]
+    Select,
+    #[sea_orm(string_value = "UPDATE")]
+    Update,
+    #[sea_orm(string_value = "DELETE")]
+    Delete,
+    #[sea_orm(string_value = "USAGE")]
+    Usage,
+    #[sea_orm(string_value = "CREATE")]
+    Create,
+    #[sea_orm(string_value = "CONNECT")]
+    Connect,
+}
+
+impl From<PbAction> for Action {
+    fn from(action: PbAction) -> Self {
+        match action {
+            PbAction::Unspecified => unreachable!("unspecified action"),
+            PbAction::Insert => Self::Insert,
+            PbAction::Select => Self::Select,
+            PbAction::Update => Self::Update,
+            PbAction::Delete => Self::Delete,
+            PbAction::Usage => Self::Usage,
+            PbAction::Create => Self::Create,
+            PbAction::Connect => Self::Connect,
+        }
+    }
+}
+
+impl From<Action> for PbAction {
+    fn from(action: Action) -> Self {
+        match action {
+            Action::Insert => Self::Insert,
+            Action::Select => Self::Select,
+            Action::Update => Self::Update,
+            Action::Delete => Self::Delete,
+            Action::Usage => Self::Usage,
+            Action::Create => Self::Create,
+            Action::Connect => Self::Connect,
+        }
+    }
+}
 
 #[derive(Clone, Debug, PartialEq, DeriveEntityModel, Eq)]
 #[sea_orm(table_name = "user_privilege")]
 pub struct Model {
     #[sea_orm(primary_key)]
-    pub id: i32,
+    pub id: PrivilegeId,
+    pub dependent_id: Option<PrivilegeId>,
     pub user_id: UserId,
     pub oid: ObjectId,
     pub granted_by: UserId,
-    pub actions: String,
+    pub action: Action,
     pub with_grant_option: bool,
 }
 
@@ -54,6 +104,14 @@ pub enum Relation {
         on_delete = "Cascade"
     )]
     User1,
+    #[sea_orm(
+        belongs_to = "Entity",
+        from = "Column::DependentId",
+        to = "Column::Id",
+        on_update = "NoAction",
+        on_delete = "Cascade"
+    )]
+    SelfRef,
 }
 
 impl Related<super::object::Entity> for Entity {

src/meta/model_v2/src/view.rs

@@ -14,7 +14,7 @@
 
 use risingwave_pb::catalog::PbView;
 use sea_orm::entity::prelude::*;
-use sea_orm::ActiveValue;
+use sea_orm::ActiveValue::Set;
 
 use crate::{FieldArray, Property, ViewId};
 
@@ -52,11 +52,11 @@ impl ActiveModelBehavior for ActiveModel {}
 impl From<PbView> for ActiveModel {
     fn from(view: PbView) -> Self {
         Self {
-            view_id: ActiveValue::Set(view.id as _),
-            name: ActiveValue::Set(view.name),
-            properties: ActiveValue::Set(Property(view.properties)),
-            definition: ActiveValue::Set(view.sql),
-            columns: ActiveValue::Set(FieldArray(view.columns)),
+            view_id: Set(view.id as _),
+            name: Set(view.name),
+            properties: Set(Property(view.properties)),
+            definition: Set(view.sql),
+            columns: Set(FieldArray(view.columns)),
         }
     }
 }