feat(rbac): procedure object support rbac (#18730)

* eat(rbac): procedure object support rbac

* 1. replace p_id to procedure_id
2. add new function create_id_value_with_cleanup process ownership kv
3. modify test

* no need to assert the seq of ownership key

* refactor create_id_value

* refactor cleanup_old_fn

* fix

Author: TCeason

src/meta/api/src/catalog_api.rs

@@ -71,6 +71,7 @@ where
                     )]
                 },
                 |_, _| Ok(vec![]),
+                |_, _| {},
             )
             .await?;
 

src/meta/api/src/dictionary_api.rs

@@ -73,6 +73,7 @@ where
                 false,
                 |_| vec![],
                 |_, _| Ok(vec![]),
+                |_, _| {},
             )
             .await?;
 

src/meta/api/src/index_api.rs

@@ -110,6 +110,7 @@ where
                     mark_index_as_deleted(name_ident.tenant(), value.table_id, *index_id)
                         .map(|(k, v)| vec![(k, v)])
                 },
+                |_, _| {},
             )
             .await?;
 

src/meta/api/src/name_id_value_api.rs

@@ -65,7 +65,7 @@ where
     IdRsc: TenantResource + Send + Sync + 'static,
     IdRsc::ValueType: FromToProto + Send + Sync + 'static,
 {
-    /// Create a two level `name -> id -> value` mapping.
+    /// Create a two level `name -> id -> value` mapping with cleanup support for old resources.
     ///
     /// `associated_records` is used to generate additional key-values to add or remove along with the main operation.
     /// Such operations do not have any condition constraints.
@@ -76,17 +76,25 @@ where
     ///
     /// If there is already a `name_ident` exists, return the existing id in a `Ok(Err(exist))`.
     /// Otherwise, create `name -> id -> value` and returns the created id in a `Ok(Ok(created))`.
-    async fn create_id_value<A, M>(
+    ///
+    /// `on_override_fn` is called with the old id and txn when override_exist is true and an existing
+    /// resource is being replaced. It can add custom operations to the transaction.
+    ///
+    /// This eliminates the need for callers to manually handle cleanup logic after the fact.
+    /// For example, when a procedure is dropped by `override_exist`, `__fd_object_owners/tenant1/procedure-by-id/<procedure_id>` will be delete
+    async fn create_id_value<A, M, O>(
         &self,
         name_ident: &K,
         value: &IdRsc::ValueType,
         override_exist: bool,
         associated_records: A,
         mark_delete_records: M,
+        on_override_fn: O,
     ) -> Result<Result<DataId<IdRsc>, SeqV<DataId<IdRsc>>>, MetaTxnError>
     where
         A: Fn(DataId<IdRsc>) -> Vec<(String, Vec<u8>)> + Send,
         M: Fn(DataId<IdRsc>, &IdRsc::ValueType) -> Result<Vec<(String, Vec<u8>)>, MetaError> + Send,
+        O: Fn(DataId<IdRsc>, &mut TxnRequest) + Send,
     {
         debug!(name_ident :? =name_ident; "NameIdValueApi: {}", func_name!());
 
@@ -124,6 +132,8 @@ where
                         for (k, v) in kvs {
                             txn.if_then.push(TxnOp::put(k, v));
                         }
+                        // Apply override function if provided
+                        on_override_fn(seq_id.data, &mut txn);
                     } else {
                         return Ok(Err(seq_id));
                     }

src/meta/app/src/principal/ownership_object.rs

@@ -64,6 +64,10 @@ pub enum OwnershipObject {
     Sequence {
         name: String,
     },
+
+    Procedure {
+        procedure_id: u64,
+    },
 }
 
 impl OwnershipObject {
@@ -95,6 +99,7 @@ impl fmt::Display for OwnershipObject {
             OwnershipObject::Warehouse { id } => write!(f, "WAREHOUSE {id}"),
             OwnershipObject::Connection { name } => write!(f, "CONNECTION {name}"),
             OwnershipObject::Sequence { name } => write!(f, "SEQUENCE {name}"),
+            OwnershipObject::Procedure { procedure_id } => write!(f, "PROCEDURE {procedure_id}"),
         }
     }
 }
@@ -137,6 +142,9 @@ impl KeyCodec for OwnershipObject {
             OwnershipObject::Warehouse { id } => b.push_raw("warehouse-by-id").push_str(id),
             OwnershipObject::Connection { name } => b.push_raw("connection-by-name").push_str(name),
             OwnershipObject::Sequence { name } => b.push_raw("sequence-by-name").push_str(name),
+            OwnershipObject::Procedure { procedure_id } => {
+                b.push_raw("procedure-by-id").push_u64(*procedure_id)
+            }
         }
     }
 
@@ -195,6 +203,10 @@ impl KeyCodec for OwnershipObject {
                 let name = p.next_str()?;
                 Ok(OwnershipObject::Sequence { name })
             }
+            "procedure-by-id" => {
+                let procedure_id = p.next_u64()?;
+                Ok(OwnershipObject::Procedure { procedure_id })
+            }
             _ => Err(kvapi::KeyError::InvalidSegment {
                 i: p.index(),
                 expect: "database-by-id|database-by-catalog-id|table-by-id|table-by-catalog-id|stage-by-name|udf-by-name|warehouse-by-id|connection-by-name"

src/meta/app/src/principal/tenant_ownership_object_ident.rs

@@ -288,6 +288,19 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_ownership_procedure_list_key() {
+        use databend_common_meta_kvapi::kvapi::Key;
+        let obj = OwnershipObject::Procedure { procedure_id: 1 };
+
+        let ident = TenantOwnershipObjectIdent::new(Tenant::new_literal("tenant1"), obj);
+        let dir_name = kvapi::DirName::new(ident);
+        assert_eq!(
+            dir_name.to_string_key(),
+            "__fd_object_owners/tenant1/procedure-by-id"
+        );
+    }
+
     #[test]
     fn test_tenant_ownership_object_with_key_space() {
         // TODO(xp): implement this test

src/meta/app/src/principal/user_grant.rs

@@ -62,6 +62,7 @@ pub enum GrantObject {
     Warehouse(String),
     Connection(String),
     Sequence(String),
+    Procedure(u64),
 }
 
 impl GrantObject {
@@ -97,6 +98,7 @@ impl GrantObject {
             (GrantObject::Warehouse(w), GrantObject::Warehouse(rw)) => w == rw,
             (GrantObject::Connection(c), GrantObject::Connection(rc)) => c == rc,
             (GrantObject::Sequence(s), GrantObject::Sequence(rs)) => s == rs,
+            (GrantObject::Procedure(p), GrantObject::Procedure(rp)) => p == rp,
             _ => false,
         }
     }
@@ -126,6 +128,9 @@ impl GrantObject {
             GrantObject::Sequence(_) => {
                 UserPrivilegeSet::available_privileges_on_sequence(available_ownership)
             }
+            GrantObject::Procedure(_) => {
+                UserPrivilegeSet::available_privileges_on_procedure(available_ownership)
+            }
         }
     }
 
@@ -136,6 +141,7 @@ impl GrantObject {
             | GrantObject::UDF(_)
             | GrantObject::Warehouse(_)
             | GrantObject::Sequence(_)
+            | GrantObject::Procedure(_)
             | GrantObject::Connection(_) => None,
             GrantObject::Database(cat, _) | GrantObject::DatabaseById(cat, _) => Some(cat.clone()),
             GrantObject::Table(cat, _, _) | GrantObject::TableById(cat, _, _) => Some(cat.clone()),
@@ -160,6 +166,7 @@ impl fmt::Display for GrantObject {
             GrantObject::Warehouse(w) => write!(f, "WAREHOUSE {w}"),
             GrantObject::Connection(c) => write!(f, "CONNECTION {c}"),
             GrantObject::Sequence(s) => write!(f, "SEQUENCE {s}"),
+            GrantObject::Procedure(p) => write!(f, "PROCEDURE {p}"),
         }
     }
 }

src/meta/app/src/principal/user_privilege.rs

@@ -86,6 +86,10 @@ pub enum UserPrivilegeType {
     CreateSequence = 1 << 24,
     // Privilege to Access Sequence
     AccessSequence = 1 << 25,
+    // Privilege to Create Procedure
+    CreateProcedure = 1 << 26,
+    // Privilege to Access Procedure
+    AccessProcedure = 1 << 27,
     // Discard Privilege Type
     Set = 1 << 4,
 }
@@ -147,6 +151,8 @@ impl Display for UserPrivilegeType {
             UserPrivilegeType::AccessConnection => "ACCESS CONNECTION",
             UserPrivilegeType::CreateSequence => "CREATE SEQUENCE",
             UserPrivilegeType::AccessSequence => "ACCESS SEQUENCE",
+            UserPrivilegeType::CreateProcedure => "CREATE PROCEDURE",
+            UserPrivilegeType::AccessProcedure => "ACCESS PROCEDURE",
         })
     }
 }
@@ -199,6 +205,12 @@ impl From<databend_common_ast::ast::UserPrivilegeType> for UserPrivilegeType {
             databend_common_ast::ast::UserPrivilegeType::AccessSequence => {
                 UserPrivilegeType::AccessSequence
             }
+            databend_common_ast::ast::UserPrivilegeType::CreateProcedure => {
+                UserPrivilegeType::CreateProcedure
+            }
+            databend_common_ast::ast::UserPrivilegeType::AccessProcedure => {
+                UserPrivilegeType::AccessProcedure
+            }
             databend_common_ast::ast::UserPrivilegeType::Set => UserPrivilegeType::Set,
         }
     }
@@ -235,7 +247,7 @@ impl UserPrivilegeSet {
         let wh_privs_without_ownership = Self::available_privileges_on_warehouse(false);
         let connection_privs_without_ownership = Self::available_privileges_on_connection(false);
         let seq_privs_without_ownership = Self::available_privileges_on_sequence(false);
-        let privs = make_bitflags!(UserPrivilegeType::{ Usage | Super | CreateUser | DropUser | CreateRole | DropRole | CreateDatabase | Grant | CreateDataMask | CreateWarehouse | CreateConnection | CreateSequence });
+        let privs = make_bitflags!(UserPrivilegeType::{ Usage | Super | CreateUser | DropUser | CreateRole | DropRole | CreateDatabase | Grant | CreateDataMask | CreateWarehouse | CreateConnection | CreateSequence | CreateProcedure });
         (database_privs.privileges
             | privs
             | stage_privs_without_ownership.privileges
@@ -297,6 +309,14 @@ impl UserPrivilegeSet {
         }
     }
 
+    pub fn available_privileges_on_procedure(available_ownership: bool) -> Self {
+        if available_ownership {
+            make_bitflags!(UserPrivilegeType::{ AccessProcedure | Ownership }).into()
+        } else {
+            make_bitflags!(UserPrivilegeType::{ AccessProcedure }).into()
+        }
+    }
+
     pub fn available_privileges_on_udf(available_ownership: bool) -> Self {
         if available_ownership {
             make_bitflags!(UserPrivilegeType::{ Usage | Ownership }).into()

src/meta/proto-conv/src/ownership_from_to_protobuf_impl.rs

@@ -97,6 +97,9 @@ impl FromToProto for mt::principal::OwnershipObject {
             pb::ownership_object::Object::Sequence(
                 pb::ownership_object::OwnershipSequenceObject { sequence },
             ) => Ok(mt::principal::OwnershipObject::Sequence { name: sequence }),
+            pb::ownership_object::Object::Procedure(
+                pb::ownership_object::OwnershipProcedureObject { procedure_id },
+            ) => Ok(mt::principal::OwnershipObject::Procedure { procedure_id }),
         }
     }
 
@@ -151,6 +154,13 @@ impl FromToProto for mt::principal::OwnershipObject {
                     },
                 ))
             }
+            mt::principal::OwnershipObject::Procedure { procedure_id } => {
+                Some(pb::ownership_object::Object::Procedure(
+                    pb::ownership_object::OwnershipProcedureObject {
+                        procedure_id: *procedure_id,
+                    },
+                ))
+            }
         };
         Ok(pb::OwnershipObject {
             ver: VER,

src/meta/proto-conv/src/user_from_to_protobuf_impl.rs

@@ -200,6 +200,9 @@ impl FromToProto for mt::principal::GrantObject {
             pb::grant_object::Object::Sequence(pb::grant_object::GrantSequenceObject {
                 sequence,
             }) => Ok(mt::principal::GrantObject::Sequence(sequence)),
+            pb::grant_object::Object::Procedure(pb::grant_object::GrantProcedureObject {
+                procedure_id,
+            }) => Ok(mt::principal::GrantObject::Procedure(procedure_id)),
         }
     }
 
@@ -257,6 +260,9 @@ impl FromToProto for mt::principal::GrantObject {
                     sequence: s.clone(),
                 },
             )),
+            mt::principal::GrantObject::Procedure(p) => Some(pb::grant_object::Object::Procedure(
+                pb::grant_object::GrantProcedureObject { procedure_id: *p },
+            )),
         };
         Ok(pb::GrantObject {
             ver: VER,