refactor: delete user_procedure and refactor procedure interpreter

src/meta/api/src/meta_txn_error.rs

@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use databend_common_exception::ErrorCode;
+use databend_common_meta_app::app_error::AppErrorMessage;
 use databend_common_meta_app::app_error::TxnRetryMaxTimes;
 use databend_common_meta_types::InvalidArgument;
 use databend_common_meta_types::MetaError;
@@ -33,3 +35,18 @@ impl From<InvalidArgument> for MetaTxnError {
         Self::MetaError(MetaError::from(network_error))
     }
 }
+
+impl From<MetaNetworkError> for MetaTxnError {
+    fn from(value: MetaNetworkError) -> Self {
+        Self::MetaError(MetaError::from(value))
+    }
+}
+
+impl From<MetaTxnError> for ErrorCode {
+    fn from(meta_err: MetaTxnError) -> Self {
+        match meta_err {
+            MetaTxnError::TxnRetryMaxTimes(err) => ErrorCode::TxnRetryMaxTimes(err.message()),
+            MetaTxnError::MetaError(err) => ErrorCode::from(err),
+        }
+    }
+}

src/query/management/src/procedure/procedure_mgr.rs

@@ -14,12 +14,12 @@
 
 use std::sync::Arc;
 
-use databend_common_meta_api::kv_app_error::KVAppError;
+use databend_common_meta_api::meta_txn_error::MetaTxnError;
 use databend_common_meta_api::name_id_value_api::NameIdValueApi;
 use databend_common_meta_api::serialize_struct;
-use databend_common_meta_app::app_error::AppError;
 use databend_common_meta_app::principal::procedure::ProcedureInfo;
 use databend_common_meta_app::principal::procedure_id_ident::ProcedureIdIdent;
+use databend_common_meta_app::principal::procedure_name_ident::ProcedureName;
 use databend_common_meta_app::principal::CreateProcedureReply;
 use databend_common_meta_app::principal::CreateProcedureReq;
 use databend_common_meta_app::principal::GetProcedureReply;
@@ -30,6 +30,7 @@ use databend_common_meta_app::principal::ProcedureIdToNameIdent;
 use databend_common_meta_app::principal::ProcedureIdentity;
 use databend_common_meta_app::principal::ProcedureMeta;
 use databend_common_meta_app::principal::ProcedureNameIdent;
+use databend_common_meta_app::tenant_key::errors::ExistError;
 use databend_common_meta_app::KeyWithTenant;
 use databend_common_meta_kvapi::kvapi;
 use databend_common_meta_kvapi::kvapi::DirName;
@@ -54,7 +55,10 @@ impl ProcedureMgr {
         &self,
         req: CreateProcedureReq,
         overriding: bool,
-    ) -> Result<CreateProcedureReply, KVAppError> {
+    ) -> Result<
+        Result<CreateProcedureReply, ExistError<ProcedureName, ProcedureIdentity>>,
+        MetaTxnError,
+    > {
         debug!(req :? =(&req); "SchemaApi: {}", func_name!());
         let name_ident = &req.name_ident;
         let meta = &req.meta;
@@ -71,8 +75,8 @@ impl ProcedureMgr {
             .await?;
 
         match create_res {
-            Ok(id) => Ok(CreateProcedureReply { procedure_id: *id }),
-            Err(_) => Err(AppError::from(name_ident.exist_error(func_name!())).into()),
+            Ok(id) => Ok(Ok(CreateProcedureReply { procedure_id: *id })),
+            Err(_) => Ok(Err(name_ident.exist_error(func_name!()))),
         }
     }
 
@@ -81,7 +85,7 @@ impl ProcedureMgr {
     pub async fn drop_procedure(
         &self,
         name_ident: &ProcedureNameIdent,
-    ) -> Result<Option<(SeqV<ProcedureId>, SeqV<ProcedureMeta>)>, KVAppError> {
+    ) -> Result<Option<(SeqV<ProcedureId>, SeqV<ProcedureMeta>)>, MetaTxnError> {
         debug!(name_ident :? =(name_ident); "SchemaApi: {}", func_name!());
         let dropped = self
             .kv_api
@@ -96,7 +100,7 @@ impl ProcedureMgr {
     pub async fn get_procedure(
         &self,
         req: &GetProcedureReq,
-    ) -> Result<Option<GetProcedureReply>, KVAppError> {
+    ) -> Result<Option<GetProcedureReply>, MetaError> {
         debug!(req :? =(req); "SchemaApi: {}", func_name!());
 
         let res = self.kv_api.get_id_value(&req.inner).await?;
@@ -114,7 +118,7 @@ impl ProcedureMgr {
     pub async fn list_procedures(
         &self,
         req: ListProcedureReq,
-    ) -> Result<Vec<ProcedureInfo>, KVAppError> {
+    ) -> Result<Vec<ProcedureInfo>, MetaError> {
         debug!(req :? =(&req); "SchemaApi: {}", func_name!());
 
         // Get procedure id list by `prefix_list` "<prefix>/<tenant>"

src/query/service/src/interpreters/interpreter_procedure_create.rs

@@ -14,9 +14,8 @@
 
 use std::sync::Arc;
 
+use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
-use databend_common_meta_api::kv_app_error::KVAppError;
-use databend_common_meta_app::app_error::AppError;
 use databend_common_meta_app::principal::CreateProcedureReq;
 use databend_common_meta_app::schema::CreateOption;
 use databend_common_sql::plans::CreateProcedurePlan;
@@ -60,22 +59,22 @@ impl Interpreter for CreateProcedureInterpreter {
         let create_procedure_req: CreateProcedureReq = self.plan.clone().into();
         let overriding = self.plan.create_option.is_overriding();
 
-        return if let Err(e) = UserApiProvider::instance()
-            .add_procedure(&tenant, create_procedure_req, overriding)
+        if UserApiProvider::instance()
+            .procedure_api(&tenant)
+            .create_procedure(create_procedure_req, overriding)
             .await?
+            .is_err()
         {
-            match e {
-                KVAppError::AppError(AppError::ProcedureAlreadyExists(_)) => {
-                    if self.plan.create_option != CreateOption::CreateIfNotExists {
-                        Err(e.into())
-                    } else {
-                        Ok(PipelineBuildResult::create())
-                    }
-                }
-                _ => Err(e.into()),
+            if self.plan.create_option != CreateOption::CreateIfNotExists {
+                Err(ErrorCode::ProcedureAlreadyExists(format!(
+                    "Procedure {} already exists",
+                    self.plan.name
+                )))
+            } else {
+                Ok(PipelineBuildResult::create())
             }
         } else {
             Ok(PipelineBuildResult::create())
-        };
+        }
     }
 }

src/query/service/src/interpreters/interpreter_procedure_drop.rs

@@ -14,6 +14,7 @@
 
 use std::sync::Arc;
 
+use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
 use databend_common_meta_app::principal::DropProcedureReq;
 use databend_common_sql::plans::DropProcedurePlan;
@@ -55,9 +56,17 @@ impl Interpreter for DropProcedureInterpreter {
         let tenant = self.plan.tenant.clone();
 
         let drop_procedure_req: DropProcedureReq = self.plan.clone().into();
-        let _ = UserApiProvider::instance()
-            .drop_procedure(&tenant, drop_procedure_req, self.plan.if_exists)
+
+        let dropped = UserApiProvider::instance()
+            .procedure_api(&tenant)
+            .drop_procedure(&drop_procedure_req.name_ident)
             .await?;
+        if dropped.is_none() && !self.plan.if_exists {
+            return Err(ErrorCode::UnknownProcedure(format!(
+                "Unknown procedure '{}' while drop procedure",
+                drop_procedure_req.name_ident
+            )));
+        }
 
         Ok(PipelineBuildResult::create())
     }

src/query/sql/src/planner/binder/ddl/procedure.rs

@@ -21,6 +21,7 @@ use databend_common_ast::ast::ExecuteImmediateStmt;
 use databend_common_ast::ast::ProcedureLanguage;
 use databend_common_ast::ast::ProcedureType;
 use databend_common_ast::ast::ShowOptions;
+use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
 use databend_common_expression::types::DataType;
 use databend_common_meta_app::principal::GetProcedureReq;
@@ -138,26 +139,32 @@ impl Binder {
             }
             arg_types.push(arg_type.to_string());
         }
+        let procedure_ident = ProcedureIdentity::new(name, arg_types.join(","));
         let req = GetProcedureReq {
-            inner: ProcedureNameIdent::new(
-                tenant.clone(),
-                ProcedureIdentity::new(name, arg_types.join(",")),
-            ),
+            inner: ProcedureNameIdent::new(tenant.clone(), procedure_ident.clone()),
         };
 
         let procedure = UserApiProvider::instance()
-            .get_procedure(&tenant, req)
+            .procedure_api(&tenant)
+            .get_procedure(&req)
             .await?;
-        if arg_types.is_empty() {
-            Ok(Plan::ExecuteImmediate(Box::new(ExecuteImmediatePlan {
-                script: procedure.procedure_meta.script,
-            })))
+        if let Some(procedure) = procedure {
+            if arg_types.is_empty() {
+                Ok(Plan::ExecuteImmediate(Box::new(ExecuteImmediatePlan {
+                    script: procedure.procedure_meta.script,
+                })))
+            } else {
+                Ok(Plan::CallProcedure(Box::new(CallProcedurePlan {
+                    script: procedure.procedure_meta.script,
+                    arg_names: procedure.procedure_meta.arg_names,
+                    args: arguments.clone(),
+                })))
+            }
         } else {
-            Ok(Plan::CallProcedure(Box::new(CallProcedurePlan {
-                script: procedure.procedure_meta.script,
-                arg_names: procedure.procedure_meta.arg_names,
-                args: arguments.clone(),
-            })))
+            Err(ErrorCode::UnknownProcedure(format!(
+                "Unknown procedure {}",
+                procedure_ident
+            )))
         }
     }
 

src/query/users/src/lib.rs

@@ -34,7 +34,6 @@ pub mod connection;
 pub mod file_format;
 pub mod role_cache_mgr;
 pub mod role_util;
-mod user_procedure;
 
 pub use jwt::*;
 pub use password_policy::*;