refactor(meta): replace manual struct instantiation with constructor methods (#18393)

* refactor(meta): replace manual LogEntry construction with constructor methods

Replace manual LogEntry struct instantiation with LogEntry::new(cmd) for cases
where time_ms is None, and introduce LogEntry::new_with_time(cmd, time_ms) for
cases that need to preserve existing timestamps.

* refactor(meta): replace manual struct instantiation with constructor methods

Replace manual struct instantiation with constructor methods across meta types
to improve code consistency and maintainability following the LogEntry pattern:

- UpsertKV: Use ::new(), ::update(), ::delete() instead of manual instantiation
- WatchRequest: Use ::new() with ::with_filter(), ::with_initial_flush() builders
- RaftRequest: Use GrpcHelper::encode_raft_request() for consistent serialization
- TxnPutRequest: Add ::new() constructor for (key, value, prev_value, expire_at, ttl_ms)
- TxnGetRequest: Add ::new() constructor for (key)
- TxnDeleteRequest: Add ::new() constructor for (key, prev_value, match_seq)

This provides better type safety, readability, and ensures field changes only
require updating constructor methods rather than scattered manual instantiations.

* chore: fmt

* chore(meta): resolve clippy warnings for unused imports and private module access

- Remove unused MatchSeq and Operation imports from util.rs
- Fix grpc_helper module access by using re-exported GrpcHelper
- Ensure all constructor method refactoring passes clippy lints

* chore: fix fmt

Author: drmingdrmer

src/meta/api/src/schema_api_impl.rs

@@ -2188,9 +2188,7 @@ impl<KV: kvapi::KVApi<Error = MetaError> + ?Sized> SchemaApi for KV {
             txn.if_then
                 .push(txn_op_put(&tbid, serialize_struct(&new_table_meta)?));
             txn.else_then.push(TxnOp {
-                request: Some(Request::Get(TxnGetRequest {
-                    key: tbid.to_string_key(),
-                })),
+                request: Some(Request::Get(TxnGetRequest::new(tbid.to_string_key()))),
             });
 
             new_table_meta_map.insert(req.0.table_id, new_table_meta);

src/meta/api/src/schema_api_test_suite.rs

@@ -135,7 +135,6 @@ use databend_common_meta_kvapi::kvapi;
 use databend_common_meta_kvapi::kvapi::Key;
 use databend_common_meta_types::MatchSeq;
 use databend_common_meta_types::MetaError;
-use databend_common_meta_types::Operation;
 use databend_common_meta_types::UpsertKV;
 use fastrace::func_name;
 use log::debug;
@@ -240,12 +239,7 @@ async fn upsert_test_data(
     value: Vec<u8>,
 ) -> Result<u64, KVAppError> {
     let res = kv_api
-        .upsert_kv(UpsertKV {
-            key: key.to_string_key(),
-            seq: MatchSeq::GE(0),
-            value: Operation::Update(value),
-            value_meta: None,
-        })
+        .upsert_kv(UpsertKV::update(key.to_string_key(), &value))
         .await?;
 
     let seq_v = res.result.unwrap();
@@ -257,12 +251,7 @@ async fn delete_test_data(
     key: &impl kvapi::Key,
 ) -> Result<(), KVAppError> {
     let _res = kv_api
-        .upsert_kv(UpsertKV {
-            key: key.to_string_key(),
-            seq: MatchSeq::GE(0),
-            value: Operation::Delete,
-            value_meta: None,
-        })
+        .upsert_kv(UpsertKV::delete(key.to_string_key()))
         .await?;
 
     Ok(())

src/meta/api/src/util.rs

@@ -28,10 +28,8 @@ use databend_common_meta_types::txn_condition::Target;
 use databend_common_meta_types::ConditionResult;
 use databend_common_meta_types::InvalidArgument;
 use databend_common_meta_types::InvalidReply;
-use databend_common_meta_types::MatchSeq;
 use databend_common_meta_types::MetaError;
 use databend_common_meta_types::MetaNetworkError;
-use databend_common_meta_types::Operation;
 use databend_common_meta_types::SeqV;
 use databend_common_meta_types::TxnCondition;
 use databend_common_meta_types::TxnGetResponse;
@@ -200,12 +198,7 @@ pub async fn fetch_id<T: kvapi::Key>(
     generator: T,
 ) -> Result<u64, MetaError> {
     let res = kv_api
-        .upsert_kv(UpsertKV {
-            key: generator.to_string_key(),
-            seq: MatchSeq::GE(0),
-            value: Operation::Update(b"".to_vec()),
-            value_meta: None,
-        })
+        .upsert_kv(UpsertKV::update(generator.to_string_key(), b""))
         .await?;
 
     // seq: MatchSeq::Any always succeeds

src/meta/binaries/meta/entry.rs

@@ -208,14 +208,11 @@ async fn do_register(meta_node: &Arc<MetaNode>, conf: &Config) -> Result<(), Met
     println!("Register this node: {{{}}}", node);
     println!();
 
-    let ent = LogEntry {
-        time_ms: None,
-        cmd: Cmd::AddNode {
-            node_id,
-            node,
-            overriding: true,
-        },
-    };
+    let ent = LogEntry::new(Cmd::AddNode {
+        node_id,
+        node,
+        overriding: true,
+    });
     info!("Raft log entry for updating node: {:?}", ent);
 
     meta_node.write(ent).await?;

src/meta/client/src/grpc_action.rs

@@ -29,6 +29,7 @@ use databend_common_meta_types::protobuf::RaftRequest;
 use databend_common_meta_types::protobuf::StreamItem;
 use databend_common_meta_types::protobuf::WatchRequest;
 use databend_common_meta_types::protobuf::WatchResponse;
+use databend_common_meta_types::GrpcHelper;
 use databend_common_meta_types::InvalidArgument;
 use databend_common_meta_types::TxnReply;
 use databend_common_meta_types::TxnRequest;
@@ -71,10 +72,7 @@ impl TryInto<MetaGrpcReq> for Request<RaftRequest> {
 
 impl From<MetaGrpcReq> for RaftRequest {
     fn from(v: MetaGrpcReq) -> Self {
-        let raft_request = RaftRequest {
-            // Safe unwrap(): serialize to string must be ok.
-            data: serde_json::to_string(&v).unwrap(),
-        };
+        let raft_request = GrpcHelper::encode_raft_request(&v).expect("fail to serialize");
 
         debug!(
             req :? =(&raft_request);
@@ -87,10 +85,8 @@ impl From<MetaGrpcReq> for RaftRequest {
 
 impl MetaGrpcReq {
     pub fn to_raft_request(&self) -> Result<RaftRequest, InvalidArgument> {
-        let raft_request = RaftRequest {
-            data: serde_json::to_string(self)
-                .map_err(|e| InvalidArgument::new(e, "fail to encode request"))?,
-        };
+        let raft_request = GrpcHelper::encode_raft_request(self)
+            .map_err(|e| InvalidArgument::new(e, "fail to encode request"))?;
 
         debug!(
             req :? =(&raft_request);
@@ -124,10 +120,7 @@ impl RequestFor for MetaGrpcReadReq {
 
 impl From<MetaGrpcReadReq> for RaftRequest {
     fn from(v: MetaGrpcReadReq) -> Self {
-        let raft_request = RaftRequest {
-            // Safe unwrap(): serialize to string must be ok.
-            data: serde_json::to_string(&v).unwrap(),
-        };
+        let raft_request = GrpcHelper::encode_raft_request(&v).expect("fail to serialize");
 
         debug!(
             req :? =(&raft_request);
@@ -140,10 +133,8 @@ impl From<MetaGrpcReadReq> for RaftRequest {
 
 impl MetaGrpcReadReq {
     pub fn to_raft_request(&self) -> Result<RaftRequest, InvalidArgument> {
-        let raft_request = RaftRequest {
-            data: serde_json::to_string(self)
-                .map_err(|e| InvalidArgument::new(e, "fail to encode request"))?,
-        };
+        let raft_request = GrpcHelper::encode_raft_request(self)
+            .map_err(|e| InvalidArgument::new(e, "fail to encode request"))?;
 
         debug!(
             req :? =(&raft_request);

src/meta/control/src/import.rs

@@ -251,7 +251,7 @@ async fn init_new_cluster(
 
             let entry: Entry = Entry {
                 log_id,
-                payload: EntryPayload::Normal(LogEntry { time_ms: None, cmd }),
+                payload: EntryPayload::Normal(LogEntry::new(cmd)),
             };
 
             sto.blocking_append([entry]).await?;

src/meta/kvapi-test-suite/src/kvapi_test_suite.rs

@@ -834,23 +834,19 @@ impl TestSuite {
                 TxnOp::put(txn_key2.clone(), b("new_v2")),
                 // get k1
                 TxnOp {
-                    request: Some(txn_op::Request::Get(TxnGetRequest {
-                        key: txn_key1.clone(),
-                    })),
+                    request: Some(txn_op::Request::Get(TxnGetRequest::new(txn_key1.clone()))),
                 },
                 // delete k1
                 TxnOp {
-                    request: Some(txn_op::Request::Delete(TxnDeleteRequest {
-                        key: txn_key1.clone(),
-                        prev_value: true,
-                        match_seq: None,
-                    })),
+                    request: Some(txn_op::Request::Delete(TxnDeleteRequest::new(
+                        txn_key1.clone(),
+                        true,
+                        None,
+                    ))),
                 },
                 // get k1
                 TxnOp {
-                    request: Some(txn_op::Request::Get(TxnGetRequest {
-                        key: txn_key1.clone(),
-                    })),
+                    request: Some(txn_op::Request::Get(TxnGetRequest::new(txn_key1.clone()))),
                 },
             ];
 

src/meta/process/src/kv_processor.rs

@@ -138,10 +138,10 @@ where F: Fn(&str, Vec<u8>) -> Result<Vec<u8>, anyhow::Error>
             Cmd::RemoveNode { .. } => Ok(None),
             Cmd::SetFeature { .. } => Ok(None),
             Cmd::UpsertKV(ups) => {
-                let x = LogEntry {
-                    time_ms: log_entry.time_ms,
-                    cmd: Cmd::UpsertKV(unwrap_or_return!(self.proc_upsert_kv(ups)?)),
-                };
+                let x = LogEntry::new_with_time(
+                    Cmd::UpsertKV(unwrap_or_return!(self.proc_upsert_kv(ups)?)),
+                    log_entry.time_ms,
+                );
                 Ok(Some(x))
             }
             Cmd::Transaction(tx) => {
@@ -160,10 +160,10 @@ where F: Fn(&str, Vec<u8>) -> Result<Vec<u8>, anyhow::Error>
                     else_then.push(self.proc_txop(op)?);
                 }
 
-                Ok(Some(LogEntry {
-                    time_ms: log_entry.time_ms,
-                    cmd: Cmd::Transaction(TxnRequest::new(condition, if_then).with_else(else_then)),
-                }))
+                Ok(Some(LogEntry::new_with_time(
+                    Cmd::Transaction(TxnRequest::new(condition, if_then).with_else(else_then)),
+                    log_entry.time_ms,
+                )))
             }
         }
     }
@@ -173,12 +173,12 @@ where F: Fn(&str, Vec<u8>) -> Result<Vec<u8>, anyhow::Error>
             Operation::Update(v) => {
                 let buf = (self.process_pb)(&ups.key, v)?;
 
-                Ok(Some(UpsertKV {
-                    key: ups.key,
-                    seq: ups.seq,
-                    value: Operation::Update(buf),
-                    value_meta: ups.value_meta,
-                }))
+                Ok(Some(UpsertKV::new(
+                    ups.key,
+                    ups.seq,
+                    Operation::Update(buf),
+                    ups.value_meta,
+                )))
             }
             Operation::Delete => Ok(None),
             #[allow(deprecated)]
@@ -220,13 +220,7 @@ where F: Fn(&str, Vec<u8>) -> Result<Vec<u8>, anyhow::Error>
     fn proc_tx_put_request(&self, p: TxnPutRequest) -> Result<TxnPutRequest, anyhow::Error> {
         let value = (self.process_pb)(&p.key, p.value)?;
 
-        let pr = TxnPutRequest {
-            key: p.key,
-            value,
-            prev_value: p.prev_value,
-            expire_at: p.expire_at,
-            ttl_ms: p.ttl_ms,
-        };
+        let pr = TxnPutRequest::new(p.key, value, p.prev_value, p.expire_at, p.ttl_ms);
 
         Ok(pr)
     }

src/meta/raft-store/src/state_machine/testing.rs

@@ -37,10 +37,9 @@ pub fn snapshot_logs() -> (Vec<Entry>, Vec<String>) {
         Entry::new_blank(new_log_id(1, 0, 3)),
         Entry {
             log_id: new_log_id(1, 0, 4),
-            payload: EntryPayload::Normal(LogEntry {
-                time_ms: None,
-                cmd: Cmd::UpsertKV(UpsertKV::update("a", b"A")),
-            }),
+            payload: EntryPayload::Normal(LogEntry::new(Cmd::UpsertKV(UpsertKV::update(
+                "a", b"A",
+            )))),
         },
         Entry {
             log_id: new_log_id(1, 0, 5),
@@ -60,14 +59,11 @@ pub fn snapshot_logs() -> (Vec<Entry>, Vec<String>) {
         },
         Entry {
             log_id: new_log_id(1, 0, 9),
-            payload: EntryPayload::Normal(LogEntry {
-                time_ms: None,
-                cmd: Cmd::AddNode {
-                    node_id: 5,
-                    node: Default::default(),
-                    overriding: false,
-                },
-            }),
+            payload: EntryPayload::Normal(LogEntry::new(Cmd::AddNode {
+                node_id: 5,
+                node: Default::default(),
+                overriding: false,
+            })),
         },
     ];
     let want = [ //

src/meta/service/src/message.rs

@@ -24,6 +24,7 @@ use databend_common_meta_types::protobuf::RaftRequest;
 use databend_common_meta_types::raft_types::NodeId;
 use databend_common_meta_types::AppliedState;
 use databend_common_meta_types::Endpoint;
+use databend_common_meta_types::GrpcHelper;
 use databend_common_meta_types::LogEntry;
 use databend_common_meta_types::MetaAPIError;
 
@@ -150,18 +151,14 @@ pub enum ForwardResponse {
 
 impl tonic::IntoRequest<RaftRequest> for ForwardRequest<ForwardRequestBody> {
     fn into_request(self) -> tonic::Request<RaftRequest> {
-        let mes = RaftRequest {
-            data: serde_json::to_string(&self).expect("fail to serialize"),
-        };
+        let mes = GrpcHelper::encode_raft_request(&self).expect("fail to serialize");
         tonic::Request::new(mes)
     }
 }
 
 impl tonic::IntoRequest<RaftRequest> for ForwardRequest<MetaGrpcReadReq> {
     fn into_request(self) -> tonic::Request<RaftRequest> {
-        let mes = RaftRequest {
-            data: serde_json::to_string(&self).expect("fail to serialize"),
-        };
+        let mes = GrpcHelper::encode_raft_request(&self).expect("fail to serialize");
         tonic::Request::new(mes)
     }
 }