Merge pull request #3370 from drmingdrmer/master

[common/meta/types] refactor: a AddResult returned by meta op should include an ID to indicate what has been changed

Author: BohuTANG

common/management/src/cluster/cluster_mgr.rs

@@ -21,10 +21,10 @@ use std::time::UNIX_EPOCH;
 use common_exception::ErrorCode;
 use common_exception::Result;
 use common_meta_api::KVApi;
-use common_meta_types::AddResult;
 use common_meta_types::KVMeta;
 use common_meta_types::MatchSeq;
 use common_meta_types::NodeInfo;
+use common_meta_types::OkOrExist;
 use common_meta_types::Operation;
 use common_meta_types::SeqV;
 use common_meta_types::UpsertKVAction;
@@ -148,9 +148,9 @@ impl ClusterApi for ClusterMgr {
 
         let res = upsert_node.await?.into_add_result()?;
 
-        match res {
-            AddResult::Ok(v) => Ok(v.seq),
-            AddResult::Exists(v) => Err(ErrorCode::ClusterNodeAlreadyExists(format!(
+        match res.res {
+            OkOrExist::Ok(v) => Ok(v.seq),
+            OkOrExist::Exists(v) => Err(ErrorCode::ClusterNodeAlreadyExists(format!(
                 "Cluster ID already exists, seq [{}]",
                 v.seq
             ))),

common/management/src/stage/stage_mgr.rs

@@ -18,10 +18,10 @@ use std::sync::Arc;
 use common_exception::ErrorCode;
 use common_exception::Result;
 use common_meta_api::KVApi;
-use common_meta_types::AddResult;
 use common_meta_types::IntoSeqV;
 use common_meta_types::MatchSeq;
 use common_meta_types::MatchSeqExt;
+use common_meta_types::OkOrExist;
 use common_meta_types::Operation;
 use common_meta_types::SeqV;
 use common_meta_types::UpsertKVAction;
@@ -58,9 +58,9 @@ impl StageMgrApi for StageMgr {
 
         let res = upsert_info.await?.into_add_result()?;
 
-        match res {
-            AddResult::Ok(v) => Ok(v.seq),
-            AddResult::Exists(v) => Err(ErrorCode::StageAlreadyExists(format!(
+        match res.res {
+            OkOrExist::Ok(v) => Ok(v.seq),
+            OkOrExist::Exists(v) => Err(ErrorCode::StageAlreadyExists(format!(
                 "Stage already exists, seq [{}]",
                 v.seq
             ))),

common/management/src/user/user_mgr.rs

@@ -19,12 +19,12 @@ use common_exception::ErrorCode;
 use common_exception::Result;
 use common_exception::ToErrorCode;
 use common_meta_api::KVApi;
-use common_meta_types::AddResult;
 use common_meta_types::AuthType;
 use common_meta_types::GrantObject;
 use common_meta_types::IntoSeqV;
 use common_meta_types::MatchSeq;
 use common_meta_types::MatchSeqExt;
+use common_meta_types::OkOrExist;
 use common_meta_types::Operation;
 use common_meta_types::SeqV;
 use common_meta_types::UpsertKVAction;
@@ -100,9 +100,9 @@ impl UserMgrApi for UserMgr {
             None,
         ));
         let res = upsert_kv.await?.into_add_result()?;
-        match res {
-            AddResult::Ok(v) => Ok(v.seq),
-            AddResult::Exists(v) => Err(ErrorCode::UserAlreadyExists(format!(
+        match res.res {
+            OkOrExist::Ok(v) => Ok(v.seq),
+            OkOrExist::Exists(v) => Err(ErrorCode::UserAlreadyExists(format!(
                 "User already exists, seq [{}]",
                 v.seq
             ))),

common/meta/raft-store/src/state_machine/applied_state.rs

@@ -12,7 +12,11 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::fmt::Debug;
+
 use async_raft::AppDataResponse;
+use common_exception::ErrorCode;
+use common_meta_types::AddResult;
 use common_meta_types::Change;
 use common_meta_types::DatabaseMeta;
 use common_meta_types::Node;
@@ -50,6 +54,24 @@ pub enum AppliedState {
 
 impl AppDataResponse for AppliedState {}
 
+impl<T, ID> TryInto<AddResult<T, ID>> for AppliedState
+where
+    ID: Clone + PartialEq + Debug,
+    T: Clone + PartialEq + Debug,
+    Change<T, ID>: TryFrom<AppliedState>,
+    <Change<T, ID> as TryFrom<AppliedState>>::Error: Debug,
+{
+    type Error = ErrorCode;
+
+    fn try_into(self) -> Result<AddResult<T, ID>, Self::Error> {
+        let typ = std::any::type_name::<T>();
+
+        let ch = TryInto::<Change<T, ID>>::try_into(self).expect(typ);
+        let add_res = ch.into_add_result()?;
+        Ok(add_res)
+    }
+}
+
 pub enum PrevOrResult<'a> {
     Prev(&'a AppliedState),
     Result(&'a AppliedState),

common/meta/types/src/change.rs

@@ -18,11 +18,16 @@ use serde::Serialize;
 
 use crate::SeqV;
 
-pub enum AddResult<T> {
+pub enum OkOrExist<T> {
     Ok(SeqV<T>),
     Exists(SeqV<T>),
 }
 
+pub struct AddResult<T, ID> {
+    pub id: Option<ID>,
+    pub res: OkOrExist<T>,
+}
+
 /// `Change` describes a state change, including the states before and after a change.
 #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, derive_more::From)]
 pub struct Change<T, ID = u64>
@@ -94,11 +99,15 @@ where
         self.prev != self.result
     }
 
-    pub fn into_add_result(self) -> Result<AddResult<T>, ErrorCode> {
+    pub fn into_add_result(mut self) -> Result<AddResult<T, ID>, ErrorCode> {
+        let id = self.ident.take();
         let (prev, result) = self.unpack();
         if let Some(p) = prev {
             return if result.is_some() {
-                Ok(AddResult::Exists(p))
+                Ok(AddResult {
+                    id,
+                    res: OkOrExist::Exists(p),
+                })
             } else {
                 Err(ErrorCode::UnknownException(format!(
                     "invalid result for add: prev: {:?} result: None",
@@ -108,7 +117,10 @@ where
         }
 
         if let Some(res) = result {
-            Ok(AddResult::Ok(res))
+            Ok(AddResult {
+                id,
+                res: OkOrExist::Ok(res),
+            })
         } else {
             Err(ErrorCode::UnknownException(
                 "invalid result for add: prev: None result: None".to_string(),

common/meta/types/src/lib.rs

@@ -37,6 +37,7 @@ mod user_stage;
 
 pub use change::AddResult;
 pub use change::Change;
+pub use change::OkOrExist;
 pub use cluster::Node;
 pub use cluster::NodeInfo;
 pub use cluster::Slot;

metasrv/src/executor/meta_handlers.rs

@@ -20,6 +20,7 @@ use common_exception::ErrorCode;
 use common_exception::ToErrorCode;
 use common_meta_flight::FlightReq;
 use common_meta_flight::GetTableExtReq;
+use common_meta_types::AddResult;
 use common_meta_types::Change;
 use common_meta_types::Cmd::CreateDatabase;
 use common_meta_types::Cmd::CreateTable;
@@ -41,6 +42,7 @@ use common_meta_types::GetTableReq;
 use common_meta_types::ListDatabaseReq;
 use common_meta_types::ListTableReq;
 use common_meta_types::LogEntry;
+use common_meta_types::OkOrExist;
 use common_meta_types::TableIdent;
 use common_meta_types::TableInfo;
 use common_meta_types::TableMeta;
@@ -187,18 +189,20 @@ impl RequestHandler<FlightReq<CreateTableReq>> for ActionHandler {
             .await
             .map_err(|e| ErrorCode::MetaNodeInternalError(e.to_string()))?;
 
-        let mut ch: Change<TableMeta, u64> = rst.try_into().expect("TableId");
-        let table_id = ch.ident.take().unwrap();
-        let (prev, _) = ch.unpack_data();
+        let add_res: AddResult<TableMeta, u64> = rst.try_into()?;
 
-        if prev.is_some() && !if_not_exists {
-            return Err(ErrorCode::TableAlreadyExists(format!(
-                "table exists: {}",
-                table_name
-            )));
+        if let OkOrExist::Exists(_) = add_res.res {
+            if !if_not_exists {
+                return Err(ErrorCode::TableAlreadyExists(format!(
+                    "table exists: {}",
+                    table_name
+                )));
+            }
         }
 
-        Ok(CreateTableReply { table_id })
+        Ok(CreateTableReply {
+            table_id: add_res.id.unwrap(),
+        })
     }
 }