Merge branch 'main' into agg-v2

Author: sundy-li

Cargo.lock

@@ -34,7 +34,7 @@ members = [
     "src/query/pipeline/sinks",
     "src/query/pipeline/sources",
     "src/query/pipeline/transforms",
-    "src/query/planners",
+    "src/query/legacy-planners",
     "src/query/settings",
     "src/query/storages/fuse",
     "src/query/storages/fuse-meta",

Cargo.toml

@@ -68,36 +68,16 @@ pub enum MetaGrpcReq {
     ListKV(ListKVReq),
 }
 
-impl From<MetaGrpcWriteReq> for MetaGrpcReq {
-    fn from(r: MetaGrpcWriteReq) -> Self {
-        match r {
-            MetaGrpcWriteReq::UpsertKV(x) => x.into(),
-        }
-    }
-}
-
-impl From<MetaGrpcReadReq> for MetaGrpcReq {
-    fn from(r: MetaGrpcReadReq) -> Self {
-        match r {
-            MetaGrpcReadReq::GetKV(x) => x.into(),
-            MetaGrpcReadReq::MGetKV(x) => x.into(),
-            MetaGrpcReadReq::ListKV(x) => x.into(),
-        }
-    }
-}
-
-/// Try convert tonic::Request<RaftRequest> to DoActionAction.
-impl TryInto<MetaGrpcWriteReq> for Request<RaftRequest> {
+impl TryInto<MetaGrpcReq> for Request<RaftRequest> {
     type Error = tonic::Status;
 
-    fn try_into(self) -> Result<MetaGrpcWriteReq, Self::Error> {
+    fn try_into(self) -> Result<MetaGrpcReq, Self::Error> {
         let raft_request = self.into_inner();
 
-        // Decode DoActionAction from flight request body.
         let json_str = raft_request.data.as_str();
-        let action = serde_json::from_str::<MetaGrpcWriteReq>(json_str)
+        let req = serde_json::from_str::<MetaGrpcReq>(json_str)
             .map_err(|e| tonic::Status::internal(e.to_string()))?;
-        Ok(action)
+        Ok(req)
     }
 }
 
@@ -110,7 +90,6 @@ impl tonic::IntoRequest<RaftRequest> for MetaGrpcWriteReq {
     }
 }
 
-/// Try convert DoActionAction to tonic::Request<RaftRequest>.
 impl TryInto<Request<RaftRequest>> for MetaGrpcWriteReq {
     type Error = serde_json::Error;
 
@@ -124,19 +103,6 @@ impl TryInto<Request<RaftRequest>> for MetaGrpcWriteReq {
     }
 }
 
-impl TryInto<MetaGrpcReadReq> for Request<RaftRequest> {
-    type Error = tonic::Status;
-
-    fn try_into(self) -> Result<MetaGrpcReadReq, Self::Error> {
-        let raft_req = self.into_inner();
-
-        let json_str = raft_req.data.as_str();
-        let action = serde_json::from_str::<MetaGrpcReadReq>(json_str)
-            .map_err(|e| tonic::Status::internal(e.to_string()))?;
-        Ok(action)
-    }
-}
-
 impl TryInto<Request<RaftRequest>> for MetaGrpcReadReq {
     type Error = serde_json::Error;
 

src/meta/client/src/grpc_action.rs

@@ -55,6 +55,10 @@ impl MetaService for GrpcServiceForTestImpl {
         Ok(Response::new(Box::pin(output)))
     }
 
+    async fn kv_api(&self, _request: Request<RaftRequest>) -> Result<Response<RaftReply>, Status> {
+        Err(Status::unimplemented("Not yet implemented"))
+    }
+
     async fn write_msg(
         &self,
         _request: Request<RaftRequest>,