Merge pull request #5030 from lichuang/transaction_api

Feature: transaction api
databendlabs · Apr 26, 2022 · 66e5104 · 66e5104
2 parents d947b2b + 80a31c6
commit 66e5104
Show file tree

Hide file tree

Showing 25 changed files with 979 additions and 33 deletions.
diff --git a/common/management/tests/it/user.rs b/common/management/tests/it/user.rs
@@ -29,6 +29,8 @@ use common_meta_types::Operation;
 use common_meta_types::PasswordHashMethod;
 use common_meta_types::PrefixListReply;
 use common_meta_types::SeqV;
+use common_meta_types::TxnReply;
+use common_meta_types::TxnRequest;
 use common_meta_types::UpsertKVAction;
 use common_meta_types::UpsertKVActionReply;
 use common_meta_types::UserIdentity;
@@ -53,6 +55,9 @@ mock! {
         ) -> Result<MGetKVActionReply,MetaError>;
 
         async fn prefix_list_kv(&self, prefix: &str) -> Result<PrefixListReply, MetaError>;
+
+        async fn transaction(&self, txn: TxnRequest) -> Result<TxnReply, MetaError>;
+
         }
 }
 

diff --git a/common/meta/api/src/kv_api.rs b/common/meta/api/src/kv_api.rs
@@ -20,6 +20,8 @@ use common_meta_types::GetKVActionReply;
 use common_meta_types::MGetKVActionReply;
 use common_meta_types::MetaError;
 use common_meta_types::PrefixListReply;
+use common_meta_types::TxnReply;
+use common_meta_types::TxnRequest;
 use common_meta_types::UpsertKVAction;
 use common_meta_types::UpsertKVActionReply;
 
@@ -44,6 +46,8 @@ pub trait KVApi: Send + Sync {
     async fn mget_kv(&self, key: &[String]) -> Result<MGetKVActionReply, MetaError>;
 
     async fn prefix_list_kv(&self, prefix: &str) -> Result<PrefixListReply, MetaError>;
+
+    async fn transaction(&self, txn: TxnRequest) -> Result<TxnReply, MetaError>;
 }
 
 #[async_trait]
@@ -63,4 +67,8 @@ impl<U: KVApi, T: Deref<Target = U> + Send + Sync> KVApi for T {
     async fn prefix_list_kv(&self, prefix: &str) -> Result<PrefixListReply, MetaError> {
         self.deref().prefix_list_kv(prefix).await
     }
+
+    async fn transaction(&self, txn: TxnRequest) -> Result<TxnReply, MetaError> {
+        self.deref().transaction(txn).await
+    }
 }