feat(meta): add KvGetMany gRPC API; refactor: split chained awaits (#19244)

* feat(meta): add `KvGetMany` gRPC API for batch key lookups

Add bidirectional streaming gRPC API for looking up multiple keys.
The client streams `KvGetManyRequest` messages (one key per message),
and the server responds with a stream of `StreamItem` results.

Changes:
- Add `KvGetManyRequest` proto message and `KvGetMany` RPC
- Implement `MetaLeader::kv_get_many()` for streaming key lookups
- Add leader check in `MetaNode::handle_kv_get_many()`
- Add `MetaHandle::handle_kv_get_many()` cross-runtime bridge
- Implement gRPC service handler with token check and metrics
- Add `EstablishedClient::kv_get_many()` client method
- Add integration tests for leader, follower, and no-quorum scenarios

- Fix: #349

* refactor: split chained await statements for readability

Split chained `.await?.try_collect().await?` and
`.await?.map_ok(...).try_collect().await?` patterns into separate
statements across the codebase for improved readability and debugging.

Changes:
- Split patterns in `schema_api.rs`, `sequence_api_impl.rs`, `database_util.rs`
- Split patterns in `kv_pb_api/mod.rs`, `tag_api.rs`
- Split patterns in `compact_all.rs`, `leveled_map_test.rs`
- Split patterns in `db_scoped_seq_bounded_read_test.rs`, `sm_v003_test.rs`
- Split patterns in `compact_with_db_test.rs`, `store.rs`
- Split patterns in `udf_mgr.rs`, `stage.rs`
- Split patterns in `kvapi_test_suite.rs`
- Split patterns in query service tests: `io.rs`, `commit.rs`, `truncate.rs`,
  `navigate.rs`, `index_scan.rs`, `index_refresh.rs`

Author: drmingdrmer

src/common/storage/src/stage.rs

@@ -228,10 +228,9 @@ impl StageFilesInfo {
         pattern: Option<Regex>,
         max_files: usize,
     ) -> Result<Vec<StageFileInfo>> {
-        Self::list_files_stream_with_pattern(operator, path, pattern, Some(max_files))
-            .await?
-            .try_collect::<Vec<_>>()
-            .await
+        let strm =
+            Self::list_files_stream_with_pattern(operator, path, pattern, Some(max_files)).await?;
+        strm.try_collect::<Vec<_>>().await
     }
 
     #[async_backtrace::framed]

src/meta/api/src/database_util.rs

@@ -154,11 +154,8 @@ pub(crate) async fn drop_database_meta(
         ObjectTagIdRef::new(taggable_object.clone(), 0),
     );
     let obj_tag_dir = DirName::new(obj_tag_prefix);
-    let tag_entries: Vec<_> = kv_api
-        .list_pb(ListOptions::unlimited(&obj_tag_dir))
-        .await?
-        .try_collect()
-        .await?;
+    let strm = kv_api.list_pb(ListOptions::unlimited(&obj_tag_dir)).await?;
+    let tag_entries: Vec<_> = strm.try_collect().await?;
     for entry in tag_entries {
         let tag_id = entry.key.name().tag_id;
         // Delete object -> tag reference

src/meta/api/src/kv_pb_api/mod.rs

@@ -863,28 +863,19 @@ mod tests {
         let dir = DirName::new(CatalogIdIdent::new(&tenant, 0));
 
         // List all with no limit
-        let res: Vec<_> = foo
-            .list_pb_values(ListOptions::unlimited(&dir))
-            .await?
-            .try_collect()
-            .await?;
+        let strm = foo.list_pb_values(ListOptions::unlimited(&dir)).await?;
+        let res: Vec<_> = strm.try_collect().await?;
         assert_eq!(res.len(), 5);
         assert_eq!(res[0].catalog_option, catalog_meta.catalog_option);
 
         // List with limit 3
-        let res: Vec<_> = foo
-            .list_pb_values(ListOptions::limited(&dir, 3))
-            .await?
-            .try_collect()
-            .await?;
+        let strm = foo.list_pb_values(ListOptions::limited(&dir, 3)).await?;
+        let res: Vec<_> = strm.try_collect().await?;
         assert_eq!(res.len(), 3);
 
         // List with limit 0
-        let res: Vec<_> = foo
-            .list_pb_values(ListOptions::limited(&dir, 0))
-            .await?
-            .try_collect()
-            .await?;
+        let strm = foo.list_pb_values(ListOptions::limited(&dir, 0)).await?;
+        let res: Vec<_> = strm.try_collect().await?;
         assert!(res.is_empty());
 
         Ok(())

src/meta/api/src/schema_api.rs

@@ -348,11 +348,8 @@ pub async fn construct_drop_table_txn_operations(
         ObjectTagIdRef::new(taggable_object.clone(), 0),
     );
     let obj_tag_dir = DirName::new(obj_tag_prefix);
-    let tag_entries: Vec<_> = kv_api
-        .list_pb(ListOptions::unlimited(&obj_tag_dir))
-        .await?
-        .try_collect()
-        .await?;
+    let strm = kv_api.list_pb(ListOptions::unlimited(&obj_tag_dir)).await?;
+    let tag_entries: Vec<_> = strm.try_collect().await?;
     for entry in tag_entries {
         let tag_id = entry.key.name().tag_id;
         // Delete object -> tag reference

src/meta/api/src/sequence_api_impl.rs

@@ -134,9 +134,8 @@ impl<KV: kvapi::KVApi<Error = MetaError> + ?Sized> SequenceApi for KV {
     ) -> Result<Vec<(String, SequenceMeta)>, MetaError> {
         let dir_name = DirName::new(SequenceIdent::new(tenant, "dummy"));
 
-        let ident_metas = self
-            .list_pb(ListOptions::unlimited(&dir_name))
-            .await?
+        let strm = self.list_pb(ListOptions::unlimited(&dir_name)).await?;
+        let ident_metas = strm
             .map_ok(|itm| (itm.key, itm.seqv.data))
             .try_collect::<Vec<_>>()
             .await?;
@@ -145,11 +144,8 @@ impl<KV: kvapi::KVApi<Error = MetaError> + ?Sized> SequenceApi for KV {
             .iter()
             .map(|(ident, _)| SequenceStorageIdent::new_from(ident.clone()));
 
-        let storage_values = self
-            .get_pb_values(storage_idents)
-            .await?
-            .try_collect::<Vec<_>>()
-            .await?;
+        let strm = self.get_pb_values(storage_idents).await?;
+        let storage_values = strm.try_collect::<Vec<_>>().await?;
 
         let key_metas = ident_metas
             .into_iter()

src/meta/api/src/tag_api.rs

@@ -188,9 +188,8 @@ where
             }
 
             // Transaction failed. Check if it's due to references or concurrent modification.
-            let refs: Vec<String> = self
-                .list_pb(ListOptions::unlimited(&refs_dir))
-                .await?
+            let strm = self.list_pb(ListOptions::unlimited(&refs_dir)).await?;
+            let refs: Vec<String> = strm
                 .map_ok(|entry| entry.key.name().object.to_string())
                 .try_collect()
                 .await?;
@@ -482,9 +481,8 @@ where
             TagIdObjectRefIdent::new_generic(tenant, TagIdObjectRef::prefix(tag_id)),
             2,
         );
-        let refs = self
-            .list_pb(ListOptions::unlimited(&refs_dir))
-            .await?
+        let strm = self.list_pb(ListOptions::unlimited(&refs_dir)).await?;
+        let refs = strm
             .map_ok(|entry| {
                 let object = entry.key.name().object.clone();
                 let value_key = ObjectTagIdRefIdent::new_generic(

src/meta/client/src/established_client.rs

@@ -32,6 +32,7 @@ use databend_common_meta_types::protobuf::Empty;
 use databend_common_meta_types::protobuf::ExportedChunk;
 use databend_common_meta_types::protobuf::KeysCount;
 use databend_common_meta_types::protobuf::KeysLayoutRequest;
+use databend_common_meta_types::protobuf::KvGetManyRequest;
 use databend_common_meta_types::protobuf::KvListRequest;
 use databend_common_meta_types::protobuf::MemberListReply;
 use databend_common_meta_types::protobuf::MemberListRequest;
@@ -280,6 +281,14 @@ impl EstablishedClient {
         self.client.kv_list(request).await.update_client(self)
     }
 
+    #[async_backtrace::framed]
+    pub async fn kv_get_many(
+        &mut self,
+        request: impl tonic::IntoStreamingRequest<Message = KvGetManyRequest>,
+    ) -> Result<Response<Streaming<StreamItem>>, Status> {
+        self.client.kv_get_many(request).await.update_client(self)
+    }
+
     #[async_backtrace::framed]
     pub async fn export(
         &mut self,

src/meta/client/src/lib.rs

@@ -170,6 +170,9 @@ use semver::Version;
 /// - 2026-01-12: since TODO
 ///   🖥 server: add `kv_list` gRPC API: in protobuf, with `limit`, return stream.
 ///
+/// - 2026-01-13: since TODO
+///   🖥 server: add `kv_get_many` gRPC API: in protobuf, receive stream, return stream.
+///
 /// Server feature set:
 /// ```yaml
 /// server_features:

src/meta/client/tests/it/grpc_server.rs

@@ -28,6 +28,7 @@ use databend_common_meta_types::protobuf::ExportedChunk;
 use databend_common_meta_types::protobuf::HandshakeResponse;
 use databend_common_meta_types::protobuf::KeysCount;
 use databend_common_meta_types::protobuf::KeysLayoutRequest;
+use databend_common_meta_types::protobuf::KvGetManyRequest;
 use databend_common_meta_types::protobuf::KvListRequest;
 use databend_common_meta_types::protobuf::MemberListReply;
 use databend_common_meta_types::protobuf::MemberListRequest;
@@ -97,6 +98,15 @@ impl MetaService for GrpcServiceForTestImpl {
         unimplemented!()
     }
 
+    type KvGetManyStream = BoxStream<StreamItem>;
+
+    async fn kv_get_many(
+        &self,
+        _request: Request<Streaming<KvGetManyRequest>>,
+    ) -> Result<Response<Self::KvGetManyStream>, Status> {
+        unimplemented!()
+    }
+
     type ExportStream =
         Pin<Box<dyn Stream<Item = Result<ExportedChunk, tonic::Status>> + Send + 'static>>;
 

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

@@ -616,12 +616,8 @@ impl TestSuite {
         }
 
         // List all with no limit
-        let res: Vec<_> = kv
-            .list_kv(ListOptions::unlimited("__limit_test/"))
-            .await?
-            .map_ok(|item| item.key)
-            .try_collect()
-            .await?;
+        let strm = kv.list_kv(ListOptions::unlimited("__limit_test/")).await?;
+        let res: Vec<_> = strm.map_ok(|item| item.key).try_collect().await?;
         assert_eq!(res, vec![
             "__limit_test/0",
             "__limit_test/1",
@@ -631,34 +627,24 @@ impl TestSuite {
         ]);
 
         // List with limit 3
-        let res: Vec<_> = kv
-            .list_kv(ListOptions::limited("__limit_test/", 3))
-            .await?
-            .map_ok(|item| item.key)
-            .try_collect()
-            .await?;
+        let strm = kv.list_kv(ListOptions::limited("__limit_test/", 3)).await?;
+        let res: Vec<_> = strm.map_ok(|item| item.key).try_collect().await?;
         assert_eq!(res, vec![
             "__limit_test/0",
             "__limit_test/1",
             "__limit_test/2",
         ]);
 
         // List with limit 0
-        let res: Vec<_> = kv
-            .list_kv(ListOptions::limited("__limit_test/", 0))
-            .await?
-            .map_ok(|item| item.key)
-            .try_collect()
-            .await?;
+        let strm = kv.list_kv(ListOptions::limited("__limit_test/", 0)).await?;
+        let res: Vec<_> = strm.map_ok(|item| item.key).try_collect().await?;
         assert_eq!(res, Vec::<String>::new());
 
         // List with limit larger than result set
-        let res: Vec<_> = kv
+        let strm = kv
             .list_kv(ListOptions::limited("__limit_test/", 100))
-            .await?
-            .map_ok(|item| item.key)
-            .try_collect()
             .await?;
+        let res: Vec<_> = strm.map_ok(|item| item.key).try_collect().await?;
         assert_eq!(res, vec![
             "__limit_test/0",
             "__limit_test/1",