Merge remote-tracking branch 'upstream/main' into recluster_dis

Cargo.toml

@@ -116,7 +116,7 @@ opendal = { version = "0.41", features = [
 ] }
 ethnum = { version = "1.3.2" }
 ordered-float = { version = "3.6.0", default-features = false }
-jsonb = { git = "https://github.com/datafuselabs/jsonb", rev = "b29f2ea" }
+jsonb = { version = "0.3.0" }
 
 # openraft = { version = "0.8.2", features = ["compat-07"] }
 # For debugging

docs/doc/14-sql-commands/00-ddl/50-udf/index.md

@@ -75,7 +75,7 @@ def gcd(x: int, y: int) -> int:
 
 if __name__ == '__main__':
     # create a UDF server listening at '0.0.0.0:8815'
-    server = UdfServer("0.0.0.0:8815")
+    server = UDFServer("0.0.0.0:8815")
     # add defined functions
     server.add_function(gcd)
     # start the UDF server
@@ -122,4 +122,4 @@ python3 udf_server.py
 
 ```sql
 CREATE FUNCTION gcd (INT, INT) RETURNS INT LANGUAGE python HANDLER = 'gcd' ADDRESS = 'http://0.0.0.0:8815'；
-```
+```

src/meta/client/src/grpc_action.rs

@@ -26,6 +26,7 @@ use common_meta_kvapi::kvapi::UpsertKVReq;
 use common_meta_types::protobuf::meta_service_client::MetaServiceClient;
 use common_meta_types::protobuf::ClientInfo;
 use common_meta_types::protobuf::RaftRequest;
+use common_meta_types::protobuf::StreamItem;
 use common_meta_types::protobuf::WatchRequest;
 use common_meta_types::protobuf::WatchResponse;
 use common_meta_types::InvalidArgument;
@@ -36,12 +37,14 @@ use log::debug;
 use tonic::codegen::InterceptedService;
 use tonic::transport::Channel;
 use tonic::Request;
+use tonic::Streaming;
 
 use crate::grpc_client::AuthInterceptor;
 use crate::message::ExportReq;
 use crate::message::GetClientInfo;
 use crate::message::GetEndpoints;
 use crate::message::MakeClient;
+use crate::message::Streamed;
 
 /// Bind a request type to its corresponding response type.
 pub trait RequestFor {
@@ -86,18 +89,62 @@ impl MetaGrpcReq {
     }
 }
 
+#[derive(
+    serde::Serialize,
+    serde::Deserialize,
+    Debug,
+    Clone,
+    PartialEq,
+    Eq,
+    derive_more::From,
+    derive_more::TryInto,
+)]
+pub enum MetaGrpcReadReq {
+    GetKV(GetKVReq),
+    MGetKV(MGetKVReq),
+    ListKV(ListKVReq),
+}
+
+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"))?,
+        };
+
+        debug!(
+            req = as_debug!(&raft_request);
+            "build raft_request"
+        );
+
+        Ok(raft_request)
+    }
+}
+
 impl RequestFor for GetKVReq {
     type Reply = GetKVReply;
 }
 
+impl RequestFor for Streamed<GetKVReq> {
+    type Reply = Streaming<StreamItem>;
+}
+
 impl RequestFor for MGetKVReq {
     type Reply = MGetKVReply;
 }
 
+impl RequestFor for Streamed<MGetKVReq> {
+    type Reply = Streaming<StreamItem>;
+}
+
 impl RequestFor for ListKVReq {
     type Reply = ListKVReply;
 }
 
+impl RequestFor for Streamed<ListKVReq> {
+    type Reply = Streaming<StreamItem>;
+}
+
 impl RequestFor for UpsertKVReq {
     type Reply = UpsertKVReply;
 }