feat: add server-side KvTransaction gRPC API (client gated by feature flag)

The existing `Transaction` endpoint stores protobuf `TxnRequest` directly in the
raft log, coupling wire format to persistence. Any proto schema change risks
breaking raft log deserialization. The new `KvTransaction` endpoint introduces a
three-layer architecture: protobuf transport types for the wire, Rust-native
serde types (`kv_transaction::Transaction`) for raft log storage, and conversions
between them. This lets the proto schema evolve freely without affecting stored
data.

The new API uses a cleaner branch-based model where each branch has an optional
predicate and a list of operations, evaluated in order. This replaces the legacy
dual execution paths, stringly-typed `execution_path`, and misleading `success`
field of `TxnRequest`.

The server now exposes the `KvTransaction` RPC endpoint, but the client does not
use it by default — `ClientHandle::transaction_v2()` is gated behind the
`transaction-v2` feature flag and only enabled in tests. Production client code
continues to use the old `Transaction` API. The new endpoint will be adopted on
the client side in a future change once the server has been deployed widely
enough.

The applier is rewritten to operate on `kv_transaction::*` types. The old
`Cmd::Transaction(TxnRequest)` path converts to `kv_transaction::Transaction`
internally and delegates to the same code. Both paths return
`AppliedState::KvTxnReply(pb::KvTransactionReply)`.

Changes:
- Add `kv_transaction` module with `Transaction`, `Branch`, `Predicate`,
  `Condition`, `Operation`, `Operand`, `CompareOperator` storage types
- Add `Cmd::KvTransaction(Transaction)` variant
- Replace `AppliedState::TxnReply` with `AppliedState::KvTxnReply`
- Add `KvTransaction` RPC using redirect pattern (not forward), returning
  leader endpoint in gRPC metadata for client-side redirect
- Gate `ClientHandle::transaction_v2()` behind `transaction-v2` feature flag
- Register `Feature::KvTransaction` at version 260217.0.0

Author: Zhang Yanpo

Cargo.toml

@@ -55,7 +55,7 @@ byteorder = "1.5.0"
 chrono = { version = "0.4.40", features = ["serde"] }
 deepsize = "0.2.0"
 derive_more = { version = "2.1.1", features = ["full"] }
-display-more = "0.2.1"
+display-more = "0.2.5"
 env_logger = "0.11"
 fastrace = { version = "0.7.14", features = ["enable"] }
 feature-set = "0.1.1"

crates/client/client/Cargo.toml

@@ -38,6 +38,9 @@ thiserror = { workspace = true }
 tokio = { workspace = true }
 tonic = { workspace = true }
 
+[features]
+transaction-v2 = []
+
 [dev-dependencies]
 anyhow = { workspace = true }
 databend-meta-version = { workspace = true }

crates/client/client/src/client_handle.rs

@@ -162,6 +162,14 @@ impl<RT: SpawnApi> ClientHandle<RT> {
         self.request(txn).await.map_err(MetaError::from)
     }
 
+    #[cfg(any(test, feature = "transaction-v2"))]
+    pub async fn transaction_v2(
+        &self,
+        txn: databend_meta_types::protobuf::KvTransactionRequest,
+    ) -> Result<databend_meta_types::protobuf::KvTransactionReply, MetaError> {
+        self.request(txn).await.map_err(MetaError::from)
+    }
+
     pub async fn get_kv(&self, key: &str) -> Result<GetKVReply, MetaError> {
         let mut res = self.mget_kv(&[key.to_string()]).await?;
         Ok(res.pop().flatten())

crates/client/client/src/established_client.rs

@@ -306,6 +306,17 @@ impl EstablishedClient {
         self.client.transaction(request).await.update_client(self)
     }
 
+    #[async_backtrace::framed]
+    pub async fn kv_transaction(
+        &mut self,
+        request: impl tonic::IntoRequest<pb::KvTransactionRequest>,
+    ) -> Result<Response<pb::KvTransactionReply>, Status> {
+        self.client
+            .kv_transaction(request)
+            .await
+            .update_client(self)
+    }
+
     #[async_backtrace::framed]
     pub async fn member_list(
         &mut self,

crates/client/client/src/grpc_action.rs

@@ -79,6 +79,8 @@ use databend_meta_types::TxnRequest;
 use databend_meta_types::UpsertKV;
 use databend_meta_types::protobuf::ClientInfo;
 use databend_meta_types::protobuf::ClusterStatus;
+use databend_meta_types::protobuf::KvTransactionReply;
+use databend_meta_types::protobuf::KvTransactionRequest;
 use databend_meta_types::protobuf::MemberListReply;
 use databend_meta_types::protobuf::RaftRequest;
 use databend_meta_types::protobuf::StreamItem;
@@ -256,6 +258,10 @@ impl RequestFor for TxnRequest {
     type Reply = TxnReply;
 }
 
+impl RequestFor for KvTransactionRequest {
+    type Reply = KvTransactionReply;
+}
+
 impl RequestFor for GetClusterStatus {
     type Reply = ClusterStatus;
 }

crates/client/client/src/grpc_client.rs

@@ -349,6 +349,10 @@ impl<RT: RuntimeApi> MetaGrpcClient<RT> {
                 let resp = self.transaction(r).await;
                 Response::Txn(resp)
             }
+            message::Request::KvTransaction(r) => {
+                let resp = self.transaction_v2(r).await;
+                Response::KvTransaction(resp)
+            }
             message::Request::Watch(r) => {
                 let resp = self.watch(r).await;
                 Response::Watch(resp)
@@ -859,6 +863,43 @@ impl<RT: RuntimeApi> MetaGrpcClient<RT> {
         Err(net_err.into())
     }
 
+    #[fastrace::trace]
+    #[async_backtrace::framed]
+    pub(crate) async fn transaction_v2(
+        &self,
+        txn: pb::KvTransactionRequest,
+    ) -> Result<pb::KvTransactionReply, MetaClientError> {
+        debug!("{self}::transaction_v2 request: {txn:?}");
+
+        let mut rpc_handler = RpcHandler::new(self);
+
+        for _i in 0..RPC_RETRIES {
+            let req = RT::prepare_request(Request::new(txn.clone()));
+
+            let established = rpc_handler.new_established_client().await?;
+
+            let result = established
+                .kv_transaction(req)
+                .inspect_elapsed_over(threshold(), info_spent("transaction_v2"))
+                .await;
+
+            let retryable = rpc_handler.process_response_result(&txn, result)?;
+
+            let response = match retryable {
+                ResponseAction::Success(resp) => resp,
+                ResponseAction::ShouldRetry => {
+                    continue;
+                }
+            };
+
+            let reply = response.into_inner();
+            return Ok(reply);
+        }
+
+        let net_err = rpc_handler.create_network_error();
+        Err(net_err.into())
+    }
+
     fn get_current_endpoint(&self) -> Option<String> {
         let es = self.endpoints.lock();
         es.current().map(|x| x.to_string())

crates/client/client/src/message.rs

@@ -22,6 +22,8 @@ use databend_meta_types::TxnRequest;
 use databend_meta_types::protobuf::ClientInfo;
 use databend_meta_types::protobuf::ClusterStatus;
 use databend_meta_types::protobuf::ExportedChunk;
+use databend_meta_types::protobuf::KvTransactionReply;
+use databend_meta_types::protobuf::KvTransactionRequest;
 use databend_meta_types::protobuf::MemberListReply;
 use databend_meta_types::protobuf::StreamItem;
 use databend_meta_types::protobuf::WatchRequest;
@@ -89,6 +91,9 @@ pub enum Request {
     /// Run a transaction on remote
     Txn(TxnRequest),
 
+    /// Run a kv_transaction on remote (new API)
+    KvTransaction(KvTransactionRequest),
+
     /// Watch KV changes, expecting a Stream that reports KV change events
     Watch(WatchRequest),
 
@@ -129,6 +134,7 @@ impl Request {
             Request::StreamedGetMany(_) => "StreamGetMany",
             Request::StreamList(_) => "StreamList",
             Request::Txn(_) => "Txn",
+            Request::KvTransaction(_) => "KvTransaction",
             Request::Watch(_) => "Watch",
             Request::WatchWithInitialization(_) => "WatchWithInitialization",
             Request::Export(_) => "Export",
@@ -153,6 +159,7 @@ pub enum Response {
     ),
     StreamList(Result<BoxStream<StreamItem>, MetaError>),
     Txn(Result<TxnReply, MetaClientError>),
+    KvTransaction(Result<KvTransactionReply, MetaClientError>),
     Watch(Result<tonic::codec::Streaming<WatchResponse>, MetaClientError>),
     WatchWithInitialization(Result<tonic::codec::Streaming<WatchResponse>, MetaClientError>),
     Export(Result<tonic::codec::Streaming<ExportedChunk>, MetaClientError>),
@@ -178,6 +185,9 @@ impl fmt::Debug for Response {
             Response::Txn(x) => {
                 write!(f, "Txn({:?})", x)
             }
+            Response::KvTransaction(x) => {
+                write!(f, "KvTransaction({:?})", x)
+            }
             Response::Watch(x) => {
                 write!(f, "Watch({:?})", x)
             }
@@ -221,6 +231,7 @@ impl Response {
             Response::StreamedGetMany(res) => to_err(res),
             Response::StreamList(res) => to_err(res),
             Response::Txn(res) => to_err(res),
+            Response::KvTransaction(res) => to_err(res),
             Response::Watch(res) => to_err(res),
             Response::WatchWithInitialization(res) => to_err(res),
             Response::Export(res) => to_err(res),

crates/client/client/tests/it/grpc_server.rs

@@ -19,6 +19,7 @@ use std::time::Duration;
 use databend_meta_client::MIN_SERVER_VERSION;
 use databend_meta_runtime_api::SpawnApi;
 use databend_meta_runtime_api::TokioRuntime;
+use databend_meta_types::protobuf as pb;
 use databend_meta_types::protobuf::ClientInfo;
 use databend_meta_types::protobuf::ClusterStatus;
 use databend_meta_types::protobuf::Empty;
@@ -144,6 +145,13 @@ impl MetaService for GrpcServiceForTestImpl {
         unimplemented!()
     }
 
+    async fn kv_transaction(
+        &self,
+        _request: Request<pb::KvTransactionRequest>,
+    ) -> Result<Response<pb::KvTransactionReply>, Status> {
+        unimplemented!()
+    }
+
     async fn member_list(
         &self,
         _request: Request<MemberListRequest>,

crates/common/types/build.rs

@@ -130,6 +130,14 @@ fn build_proto() {
             "TxnReply",
             "#[derive(Eq, serde::Serialize, serde::Deserialize, deepsize::DeepSizeOf)]",
         )
+        .type_attribute(
+            "KvTransactionRequest",
+            "#[derive(Eq, deepsize::DeepSizeOf)]",
+        )
+        .type_attribute(
+            "KvTransactionReply",
+            "#[derive(Eq, serde::Serialize, serde::Deserialize, deepsize::DeepSizeOf)]",
+        )
         .type_attribute(
             "WatchRequest",
             "#[derive(Eq, deepsize::DeepSizeOf)]",

crates/common/types/proto/meta.proto

@@ -261,6 +261,31 @@ message TxnReply {
   string execution_path = 4;
 }
 
+// A transaction request with cleaner branch-based structure.
+//
+// Replaces the legacy `TxnRequest` which couples the wire format to the raft log
+// storage format and carries historical complexity (dual execution paths,
+// stringly-typed `execution_path`, misleading `success` field).
+//
+// Each branch is a ConditionalOperation: optional predicate + operations.
+// Branches are evaluated in order; the first matching branch is executed.
+// A branch with no predicate is unconditional (use as "else" at the end).
+//
+// Since: 2025-02-17, version 260217.0.0
+message KvTransactionRequest {
+  repeated ConditionalOperation branches = 1;
+}
+
+// Response from KvTransaction.
+// Since: 2025-02-17, version 260217.0.0
+message KvTransactionReply {
+  // Index of the branch that was executed, or absent if no branch matched.
+  optional uint32 executed_branch = 1;
+
+  // Responses from the operations in the executed branch.
+  repeated TxnOpResponse responses = 2;
+}
+
 message ClusterStatus {
   uint64 id = 1;
   string binary_version = 2;
@@ -523,6 +548,10 @@ service MetaService {
 
   rpc Transaction(TxnRequest) returns (TxnReply);
 
+  // Branch-based transaction API, replacing the legacy Transaction RPC.
+  // Since: 2025-02-17, version 260217.0.0
+  rpc KvTransaction(KvTransactionRequest) returns (KvTransactionReply);
+
   // Get MetaSrv member list endpoints
   rpc MemberList(MemberListRequest) returns (MemberListReply);