fix: grant default CONNECT action for new created user

src/frontend/src/handler/create_user.rs

@@ -15,19 +15,21 @@
 use pgwire::pg_response::{PgResponse, StatementType};
 use risingwave_common::error::ErrorCode::PermissionDenied;
 use risingwave_common::error::Result;
-use risingwave_pb::user::UserInfo;
+use risingwave_pb::user::grant_privilege::{Action, ActionWithGrantOption, Object};
+use risingwave_pb::user::{GrantPrivilege, UserInfo};
 use risingwave_sqlparser::ast::{CreateUserStatement, UserOption, UserOptions};
 
 use super::RwPgResponse;
 use crate::binder::Binder;
-use crate::catalog::CatalogError;
+use crate::catalog::{CatalogError, DatabaseId};
 use crate::handler::HandlerArgs;
 use crate::user::user_authentication::encrypted_password;
 
 fn make_prost_user_info(
     user_name: String,
     options: &UserOptions,
     session_user: &UserInfo,
+    database_id: DatabaseId,
 ) -> Result<UserInfo> {
     if !session_user.is_super {
         let require_super = options
@@ -45,10 +47,22 @@ fn make_prost_user_info(
         }
     }
 
+    // Since we don't have concept of PUBLIC group yet, here we simply grant new user with CONNECT
+    // action of session database.
+    let grant_privileges = vec![GrantPrivilege {
+        action_with_opts: vec![ActionWithGrantOption {
+            action: Action::Connect as i32,
+            with_grant_option: true,
+            granted_by: session_user.id,
+        }],
+        object: Some(Object::DatabaseId(database_id)),
+    }];
+
     let mut user_info = UserInfo {
         name: user_name,
         // the LOGIN option is implied if it is not explicitly specified.
         can_login: true,
+        grant_privileges,
         ..Default::default()
     };
 
@@ -85,6 +99,13 @@ pub async fn handle_create_user(
     stmt: CreateUserStatement,
 ) -> Result<RwPgResponse> {
     let session = handler_args.session;
+    let database_id = {
+        let catalog_reader = session.env().catalog_reader().read_guard();
+        catalog_reader
+            .get_database_by_name(session.database())
+            .expect("session database should exist")
+            .id()
+    };
     let user_info = {
         let user_name = Binder::resolve_user_name(stmt.user_name)?;
         let user_reader = session.env().user_info_reader().read_guard();
@@ -96,7 +117,7 @@ pub async fn handle_create_user(
             .get_user_by_name(session.user_name())
             .ok_or_else(|| CatalogError::NotFound("user", session.user_name().to_string()))?;
 
-        make_prost_user_info(user_name, &stmt.with_options, session_user)?
+        make_prost_user_info(user_name, &stmt.with_options, session_user, database_id)?
     };
 
     let user_info_writer = session.env().user_info_writer();

src/frontend/src/handler/handle_privilege.rs

@@ -241,10 +241,16 @@ mod tests {
             .await
             .unwrap();
 
-        let database_id = {
+        let (session_database_id, database_id) = {
             let catalog_reader = session.env().catalog_reader();
             let reader = catalog_reader.read_guard();
-            reader.get_database_by_name("db1").unwrap().id()
+            (
+                reader
+                    .get_database_by_name(session.database())
+                    .unwrap()
+                    .id(),
+                reader.get_database_by_name("db1").unwrap().id(),
+            )
         };
 
         {
@@ -253,21 +259,31 @@ mod tests {
             let user_info = reader.get_user_by_name("user1").unwrap();
             assert_eq!(
                 user_info.grant_privileges,
-                vec![ProstPrivilege {
-                    action_with_opts: vec![
-                        ActionWithGrantOption {
+                vec![
+                    ProstPrivilege {
+                        action_with_opts: vec![ActionWithGrantOption {
                             action: Action::Connect as i32,
                             with_grant_option: true,
-                            granted_by: DEFAULT_SUPER_USER_ID,
-                        },
-                        ActionWithGrantOption {
-                            action: Action::Create as i32,
-                            with_grant_option: true,
-                            granted_by: DEFAULT_SUPER_USER_ID,
-                        }
-                    ],
-                    object: Some(ProstObject::DatabaseId(database_id)),
-                }]
+                            granted_by: session.user_id(),
+                        }],
+                        object: Some(ProstObject::DatabaseId(session_database_id)),
+                    },
+                    ProstPrivilege {
+                        action_with_opts: vec![
+                            ActionWithGrantOption {
+                                action: Action::Connect as i32,
+                                with_grant_option: true,
+                                granted_by: DEFAULT_SUPER_USER_ID,
+                            },
+                            ActionWithGrantOption {
+                                action: Action::Create as i32,
+                                with_grant_option: true,
+                                granted_by: DEFAULT_SUPER_USER_ID,
+                            }
+                        ],
+                        object: Some(ProstObject::DatabaseId(database_id)),
+                    }
+                ]
             );
         }
 
@@ -282,6 +298,7 @@ mod tests {
             assert!(user_info
                 .grant_privileges
                 .iter()
+                .filter(|gp| gp.object == Some(ProstObject::DatabaseId(database_id)))
                 .all(|p| p.action_with_opts.iter().all(|ao| !ao.with_grant_option)));
         }
 
@@ -293,7 +310,18 @@ mod tests {
             let user_reader = session.env().user_info_reader();
             let reader = user_reader.read_guard();
             let user_info = reader.get_user_by_name("user1").unwrap();
-            assert!(user_info.grant_privileges.is_empty());
+            assert_eq!(
+                user_info.grant_privileges,
+                vec![ProstPrivilege {
+                    action_with_opts: vec![ActionWithGrantOption {
+                        action: Action::Connect as i32,
+                        with_grant_option: true,
+                        granted_by: session.user_id(),
+                    }],
+                    object: Some(ProstObject::DatabaseId(session_database_id)),
+                }]
+            );
         }
+        frontend.run_sql("DROP USER user1").await.unwrap();
     }
 }

src/meta/src/manager/catalog/mod.rs

@@ -1614,12 +1614,6 @@ where
                 user.name
             )));
         }
-        if !user.grant_privileges.is_empty() {
-            return Err(MetaError::permission_denied(format!(
-                "Cannot drop user {} with privileges",
-                id
-            )));
-        }
         if user_core
             .user_grant_relation
             .get(&id)

src/meta/src/rpc/service/user_service.rs

@@ -27,8 +27,6 @@ use crate::manager::{CatalogManagerRef, IdCategory, MetaSrvEnv};
 use crate::storage::MetaStore;
 use crate::MetaResult;
 
-// TODO: Change user manager as a part of the catalog manager, to ensure that operations on Catalog
-// and User are transactional.
 pub struct UserServiceImpl<S: MetaStore> {
     env: MetaSrvEnv<S>,