feat(meta-service): Add PutSequential operation for ordered key generation (#18312)

Introduce `PutSequential` operation that creates sequential keys using a
specified `prefix` and sequence generator. This enables applications to
build ordered data structures like queues, which are essential for
implementing distributed locks and semaphores.

Operation details:
- User provides a key `prefix` and sequence generator key
- Sequence generator stores JSON-encoded `u64` values for `FetchAddU64` compatibility
- Target key format: `<prefix>{next_val(<sequence_key>)}`
- Key generation happens atomically during state machine application

Use cases:
- Distributed queue implementations
- Ordered task scheduling
- Distributed lock/semaphore mechanisms
- Any scenario requiring guaranteed key ordering

The operation leverages the existing conflict-free sequence generation
infrastructure to ensure high concurrency and consistent ordering across
distributed environments.

Author: drmingdrmer

src/meta/binaries/metabench/main.rs

@@ -333,7 +333,7 @@ async fn benchmark_table_copy_file(
             &copied_file_ident,
             serialize_struct(&copied_file_value).unwrap(),
         )
-        .with_ttl(param.ttl_ms);
+        .with_ttl(param.ttl_ms.map(Duration::from_millis));
 
         txn.if_then.push(put_op);
     }

src/meta/client/src/lib.rs

@@ -162,6 +162,9 @@ pub static METACLI_COMMIT_SEMVER: LazyLock<Version> = LazyLock::new(|| {
 /// - 2025-07-03: since TODO: add when merged
 ///   🖥 server: adaptive `expire_at` support both seconds and milliseconds.
 ///
+/// - 2025-07-04: since TODO: add when merged
+///   🖥 server: add `PutSequential`.
+///
 /// Server feature set:
 /// ```yaml
 /// server_features:

src/meta/kvapi-test-suite/src/test_suite.rs renamed to src/meta/kvapi-test-suite/src/kvapi_test_suite.rs

@@ -18,6 +18,7 @@ use databend_common_meta_kvapi::kvapi;
 use databend_common_meta_types::protobuf as pb;
 use databend_common_meta_types::protobuf::BooleanExpression;
 use databend_common_meta_types::protobuf::FetchAddU64Response;
+use databend_common_meta_types::protobuf::KvMeta;
 use databend_common_meta_types::seq_value::KVMeta;
 use databend_common_meta_types::seq_value::SeqV;
 use databend_common_meta_types::txn_condition;
@@ -98,6 +99,9 @@ impl TestSuite {
             .await?;
         self.kv_transaction_fetch_add_u64_match_seq(&builder.build().await)
             .await?;
+        self.kv_txn_put_sequential(&builder.build().await).await?;
+        self.kv_txn_put_sequential_expire_and_ttl(&builder.build().await)
+            .await?;
         self.kv_transaction_with_ttl(&builder.build().await).await?;
         self.kv_transaction_delete_match_seq_none(&builder.build().await)
             .await?;
@@ -1348,6 +1352,125 @@ impl TestSuite {
         Ok(())
     }
 
+    /// Tests match_seq must match the record seq to take place the operation.
+    #[fastrace::trace]
+    pub async fn kv_txn_put_sequential<KV: kvapi::KVApi>(&self, kv: &KV) -> anyhow::Result<()> {
+        info!("--- {}", func_path!());
+
+        let txn = TxnRequest::new(vec![], vec![
+            TxnOp::put_sequential("k1/", "seq1", b("v1")),
+            TxnOp::put_sequential("k2/", "seq1", b("v2")),
+        ]);
+
+        let resp = kv.transaction(txn).await?;
+
+        assert_eq!(resp.responses.len(), 2);
+
+        assert_eq!(resp.responses[0].try_as_put().unwrap(), &TxnPutResponse {
+            key: "k1/000_000_000_000_000_000_000".to_string(),
+            prev_value: None,
+            current: Some(pb::SeqV::with_meta(
+                2,
+                Some(KvMeta::default()),
+                b("v1").to_vec()
+            )),
+        });
+        assert_eq!(resp.responses[1].try_as_put().unwrap(), &TxnPutResponse {
+            key: "k2/000_000_000_000_000_000_001".to_string(),
+            prev_value: None,
+            current: Some(pb::SeqV::with_meta(
+                4,
+                Some(KvMeta::default()),
+                b("v2").to_vec()
+            )),
+        });
+
+        // insert again
+
+        let txn = TxnRequest::new(vec![], vec![
+            TxnOp::put_sequential("k1/", "seq1", b("v1")),
+            TxnOp::put_sequential("k2/", "seq1", b("v2")),
+        ]);
+
+        let resp = kv.transaction(txn).await?;
+
+        assert_eq!(resp.responses.len(), 2);
+
+        assert_eq!(resp.responses[0].try_as_put().unwrap(), &TxnPutResponse {
+            key: "k1/000_000_000_000_000_000_002".to_string(),
+            prev_value: None,
+            current: Some(pb::SeqV::with_meta(
+                6,
+                Some(KvMeta::default()),
+                b("v1").to_vec()
+            )),
+        });
+        assert_eq!(resp.responses[1].try_as_put().unwrap(), &TxnPutResponse {
+            key: "k2/000_000_000_000_000_000_003".to_string(),
+            prev_value: None,
+            current: Some(pb::SeqV::with_meta(
+                8,
+                Some(KvMeta::default()),
+                b("v2").to_vec()
+            )),
+        });
+
+        Ok(())
+    }
+
+    #[fastrace::trace]
+    pub async fn kv_txn_put_sequential_expire_and_ttl<KV: kvapi::KVApi>(
+        &self,
+        kv: &KV,
+    ) -> anyhow::Result<()> {
+        // - Add a record via transaction with ttl
+
+        info!("--- {}", func_path!());
+
+        let now_ms = SeqV::<()>::now_ms();
+
+        let txn = TxnRequest::new(vec![], vec![
+            TxnOp::put_sequential("k1/", "seq1", b("v1")).with_expires_at_ms(Some(now_ms + 1000)),
+            TxnOp::put_sequential("k2/", "seq1", b("v2"))
+                .with_ttl(Some(Duration::from_millis(1000))),
+        ]);
+
+        let resp = kv.transaction(txn).await?;
+
+        assert_eq!(resp.responses.len(), 2);
+
+        assert_eq!(resp.responses[0].try_as_put().unwrap(), &TxnPutResponse {
+            key: "k1/000_000_000_000_000_000_000".to_string(),
+            prev_value: None,
+            current: Some(pb::SeqV::with_meta(
+                2,
+                Some(KvMeta::new_expire(now_ms + 1000)),
+                b("v1").to_vec()
+            )),
+        });
+
+        let got = resp.responses[1].try_as_put().unwrap();
+        assert_eq!(got.key, "k2/000_000_000_000_000_000_001".to_string());
+        assert_eq!(got.prev_value, None);
+        let current = got.current.clone().unwrap();
+        assert_eq!(current.seq, 4);
+        assert_eq!(current.data, b("v2"));
+        assert!(current.meta.unwrap().close_to(
+            &KvMeta::new_expire(now_ms + 1000),
+            Duration::from_millis(1000)
+        ));
+
+        sleep(Duration::from_millis(2000)).await;
+
+        let got = kv.get_kv("k1/000_000_000_000_000_000_000").await?;
+        assert!(got.is_none(), "k1 should be expired");
+
+        let got = kv.get_kv("k2/000_000_000_000_000_000_001").await?;
+        assert!(got.is_none(), "k2 should be expired");
+
+        Ok(())
+    }
+
     #[fastrace::trace]
     pub async fn kv_transaction_with_ttl<KV: kvapi::KVApi>(&self, kv: &KV) -> anyhow::Result<()> {
         // - Add a record via transaction with ttl

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

@@ -14,5 +14,5 @@
 
 #![allow(clippy::uninlined_format_args)]
 
-mod test_suite;
-pub use test_suite::TestSuite;
+mod kvapi_test_suite;
+pub use kvapi_test_suite::TestSuite;

src/meta/process/src/kv_processor.rs

@@ -214,6 +214,7 @@ where F: Fn(&str, Vec<u8>) -> Result<Vec<u8>, anyhow::Error>
             Request::Delete(_) => {}
             Request::DeleteByPrefix(_) => {}
             Request::FetchAddU64(_) => {}
+            Request::PutSequential(_) => {}
         }
 
         Ok(TxnOp { request: Some(req) })