refactor: make ListKV non-blocking by returning stream directly (#16868)

perf: make ListKV non-blocking by returning stream directly

Return Stream to client instead of collecting results into Vec first.
This change eliminates unnecessary buffering of ListKV results in
databend-meta server, reducing memory usage and improving response
times. The direct streaming approach better matches current usage
patterns and removes legacy collection behavior.

Although a large ListKV response body won't block the databend-meta
server, but it will still block on the client side.

src/meta/raft-store/src/state_machine_api_ext.rs

@@ -107,11 +107,6 @@ pub trait StateMachineApiExt: StateMachineApi {
                 future::ready(Ok(res))
             });
 
-        // Make it static
-
-        let vs = strm.collect::<Vec<_>>().await;
-        let strm = futures::stream::iter(vs);
-
         Ok(strm.boxed())
     }
 

src/meta/service/Cargo.toml

@@ -59,6 +59,7 @@ semver = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }
 serfig = { workspace = true }
+tokio = { workspace = true }
 tokio-stream = { workspace = true }
 tonic = { workspace = true }
 tonic-reflection = { workspace = true }

src/meta/service/src/meta_service/meta_leader.rs

@@ -36,11 +36,13 @@ use databend_common_meta_types::MetaOperationError;
 use databend_common_meta_types::Node;
 use databend_common_metrics::count::Count;
 use futures::StreamExt;
+use futures::TryStreamExt;
 use log::debug;
 use log::info;
 use maplit::btreemap;
 use maplit::btreeset;
 use tonic::codegen::BoxStream;
+use tonic::Status;
 
 use crate::message::ForwardRequest;
 use crate::message::ForwardRequestBody;
@@ -147,12 +149,14 @@ impl<'a> Handler<MetaGrpcReadReq> for MetaLeader<'a> {
             }
 
             MetaGrpcReadReq::ListKV(req) => {
-                // safe unwrap(): Infallible
-                let kvs = kv_api.prefix_list_kv(&req.prefix).await.unwrap();
-
-                let kv_iter = kvs.into_iter().map(|kv| Ok(StreamItem::from(kv)));
-
-                let strm = futures::stream::iter(kv_iter);
+                let strm =
+                    kv_api.list_kv(&req.prefix).await.map_err(|e| {
+                        MetaOperationError::DataError(MetaDataError::ReadError(
+                            MetaDataReadError::new("list_kv", &req.prefix, &e),
+                        ))
+                    })?;
+
+                let strm = strm.map_err(|e| Status::internal(e.to_string()));
 
                 Ok(strm.boxed())
             }