feat: impl kv backend for raft engine (#2280)

* feat: kv backend on raft-engine

* feat: raft-engine kvbackend

* fix: toml

* fix: some review comments

* chore: optimize delete

* fix: lift lock in batch_delete

src/common/meta/src/kv_backend/memory.rs

@@ -318,9 +318,8 @@ impl<T: ErrorExt + Send + Sync> TxnService for MemoryKvBackend<T> {
 
         let do_txn = |txn_op| match txn_op {
             TxnOp::Put(key, value) => {
-                let prev_value = kvs.insert(key.clone(), value);
-                let prev_kv = prev_value.map(|value| KeyValue { key, value });
-                TxnOpResponse::ResponsePut(PutResponse { prev_kv })
+                kvs.insert(key.clone(), value);
+                TxnOpResponse::ResponsePut(PutResponse { prev_kv: None })
             }
 
             TxnOp::Get(key) => {
@@ -337,16 +336,11 @@ impl<T: ErrorExt + Send + Sync> TxnService for MemoryKvBackend<T> {
 
             TxnOp::Delete(key) => {
                 let prev_value = kvs.remove(&key);
-                let deleted = prev_value.as_ref().map(|x| x.len()).unwrap_or(0) as i64;
-
-                let prev_kvs = prev_value
-                    .into_iter()
-                    .map(|value| KeyValue {
-                        key: key.clone(),
-                        value,
-                    })
-                    .collect();
-                TxnOpResponse::ResponseDelete(DeleteRangeResponse { deleted, prev_kvs })
+                let deleted = if prev_value.is_some() { 1 } else { 0 };
+                TxnOpResponse::ResponseDelete(DeleteRangeResponse {
+                    deleted,
+                    prev_kvs: vec![],
+                })
             }
         };
 

src/log-store/Cargo.toml

@@ -17,6 +17,7 @@ byteorder = "1.4"
 bytes = "1.1"
 common-base = { workspace = true }
 common-error = { workspace = true }
+common-meta = { workspace = true }
 common-runtime = { workspace = true }
 common-telemetry = { workspace = true }
 futures-util.workspace = true

src/log-store/src/lib.rs

@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#![feature(let_chains)]
+
 mod config;
 pub mod error;
 mod noop;

src/log-store/src/raft_engine.rs

@@ -20,6 +20,7 @@ use store_api::logstore::namespace::Namespace;
 use crate::error::Error;
 use crate::raft_engine::protos::logstore::{EntryImpl, NamespaceImpl};
 
+mod backend;
 pub mod log_store;
 
 pub mod protos {