Merge remote-tracking branch 'origin/main' into never-see-again

Author: Xuanwo

.github/workflows/build-sqllogic.yml

@@ -0,0 +1,38 @@
+name: Build Sqllogic Test Image
+
+on:
+  push:
+    branches:
+      - main
+    paths:
+      - "tests/logictest/**"
+
+permissions:
+  id-token: write
+  contents: read
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v3
+
+      - uses: ./.github/actions/setup_docker
+        id: login
+        with:
+          repo: build-tool
+          ecr_role_arn: ${{ secrets.ECR_ROLE_ARN }}
+          dockerhub_user: ${{ secrets.DOCKERHUB_USERNAME }}
+          dockerhub_token: ${{ secrets.DOCKERHUB_TOKEN }}
+
+      - name: Build and push
+        id: docker_build
+        uses: docker/build-push-action@v3
+        with:
+          push: true
+          tags: |
+            ${{ steps.login.outputs.dockerhub_repo }}:sqllogic
+            ${{ steps.login.outputs.ecr_repo }}:sqllogic
+          platforms: linux/amd64
+          context: ./tests/logictest/
+          file: ./tests/logictest/Dockerfile

.github/workflows/databend-release.yml

@@ -191,6 +191,24 @@ jobs:
           target: ${{ steps.target.outputs.target }}
           repo_role_arn: ${{ secrets.REPO_ROLE_ARN }}
 
+  publish_sqllogic_testsuites:
+    name: sqllogic testsuites
+    runs-on: ubuntu-latest
+    needs: [create_release]
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v3
+      - name: Get the version
+        id: get_version
+        run: echo ::set-output name=VERSION::${GITHUB_REF/refs\/tags\//}
+      - name: Upload to github release
+        env:
+          GH_TOKEN: ${{ inputs.github_token }}
+        run: |
+          version=${{ needs.create_release.outputs.version }}
+          tar -C ./tests/logictest -czvf testsuites-${version}.tar.gz suites
+          gh release upload ${version} testsuites-${version}.tar.gz --clobber
+
   release_docker_combined:
     name: docker image combined
     runs-on: ubuntu-latest
@@ -305,34 +323,3 @@ jobs:
           context: .
           file: ./docker/${{ matrix.distro }}/${{ matrix.service }}.Dockerfile
           build-args: VERSION=${{ needs.create_release.outputs.version }}
-
-  release_logic_test_docker_image:
-    name: release logic test docker image
-    runs-on: ubuntu-latest
-    needs: [create_release]
-    steps:
-      - name: Checkout
-        uses: actions/checkout@v3
-
-      - uses: ./.github/actions/setup_docker
-        id: login
-        with:
-          repo: sqllogictest
-          ecr_role_arn: ${{ secrets.ECR_ROLE_ARN }}
-          dockerhub_user: ${{ secrets.DOCKERHUB_USERNAME }}
-          dockerhub_token: ${{ secrets.DOCKERHUB_TOKEN }}
-
-      - name: Build and push
-        id: docker_build
-        uses: docker/build-push-action@v3
-        with:
-          push: true
-          tags: |
-            ${{ steps.login.outputs.dockerhub_repo }}:latest
-            ${{ steps.login.outputs.dockerhub_repo }}:${{ needs.create_release.outputs.version }}
-            ${{ steps.login.outputs.ecr_repo }}:latest
-            ${{ steps.login.outputs.ecr_repo }}:${{ needs.create_release.outputs.version }}
-          platforms: linux/amd64
-          context: ./tests/logictest/
-          file: ./tests/logictest/Dockerfile
-          build-args: VERSION=${{ needs.create_release.outputs.version }}

Cargo.lock

@@ -59,6 +59,33 @@ pub enum MetaGrpcReadReq {
     ListKV(ListKVReq), // since 2022-05-23
 }
 
+#[derive(serde::Serialize, serde::Deserialize, Clone, Debug, derive_more::From)]
+pub enum MetaGrpcReq {
+    UpsertKV(UpsertKVReq),
+
+    GetKV(GetKVReq),
+    MGetKV(MGetKVReq),
+    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> {
     type Error = tonic::Status;

src/meta/client/src/grpc_action.rs

@@ -18,6 +18,7 @@ mod kv_api_impl;
 mod message;
 
 pub use grpc_action::MetaGrpcReadReq;
+pub use grpc_action::MetaGrpcReq;
 pub use grpc_action::MetaGrpcWriteReq;
 pub use grpc_action::RequestFor;
 pub use grpc_client::ClientHandle;

src/meta/client/src/lib.rs

@@ -21,7 +21,9 @@ use common_arrow::arrow_format::flight::data::BasicAuth;
 use common_base::base::tokio::sync::mpsc;
 use common_grpc::GrpcClaim;
 use common_grpc::GrpcToken;
+use common_meta_api::KVApi;
 use common_meta_client::MetaGrpcReadReq;
+use common_meta_client::MetaGrpcReq;
 use common_meta_client::MetaGrpcWriteReq;
 use common_meta_types::protobuf::meta_service_server::MetaService;
 use common_meta_types::protobuf::ClientInfo;
@@ -48,11 +50,11 @@ use tonic::Status;
 use tonic::Streaming;
 use tracing::info;
 
-use crate::executor::ActionHandler;
 use crate::meta_service::meta_service_impl::GrpcStream;
 use crate::meta_service::MetaNode;
 use crate::metrics::add_meta_metrics_meta_request_inflights;
 use crate::metrics::incr_meta_metrics_meta_recv_bytes;
+use crate::metrics::incr_meta_metrics_meta_request_result;
 use crate::metrics::incr_meta_metrics_meta_sent_bytes;
 use crate::version::from_digit_ver;
 use crate::version::to_digit_ver;
@@ -61,14 +63,14 @@ use crate::version::MIN_METACLI_SEMVER;
 
 pub struct MetaServiceImpl {
     token: GrpcToken,
-    action_handler: ActionHandler,
+    pub(crate) meta_node: Arc<MetaNode>,
 }
 
 impl MetaServiceImpl {
     pub fn create(meta_node: Arc<MetaNode>) -> Self {
         Self {
             token: GrpcToken::create(),
-            action_handler: ActionHandler::create(meta_node),
+            meta_node,
         }
     }
 
@@ -84,6 +86,47 @@ impl MetaServiceImpl {
         })?;
         Ok(claim)
     }
+
+    pub async fn execute_kv_req(&self, req: MetaGrpcReq) -> RaftReply {
+        // To keep the code IDE-friendly, we manually expand the enum variants and dispatch them one by one
+
+        match req {
+            MetaGrpcReq::UpsertKV(a) => {
+                let res = self.meta_node.upsert_kv(a).await;
+                incr_meta_metrics_meta_request_result(res.is_ok());
+                RaftReply::from(res)
+            }
+            MetaGrpcReq::GetKV(a) => {
+                let res = self.meta_node.get_kv(&a.key).await;
+                incr_meta_metrics_meta_request_result(res.is_ok());
+                RaftReply::from(res)
+            }
+            MetaGrpcReq::MGetKV(a) => {
+                let res = self.meta_node.mget_kv(&a.keys).await;
+                incr_meta_metrics_meta_request_result(res.is_ok());
+                RaftReply::from(res)
+            }
+            MetaGrpcReq::ListKV(a) => {
+                let res = self.meta_node.prefix_list_kv(&a.prefix).await;
+                incr_meta_metrics_meta_request_result(res.is_ok());
+                RaftReply::from(res)
+            }
+        }
+    }
+
+    pub async fn execute_txn(&self, req: TxnRequest) -> TxnReply {
+        let ret = self.meta_node.transaction(req).await;
+        incr_meta_metrics_meta_request_result(ret.is_ok());
+
+        match ret {
+            Ok(resp) => resp,
+            Err(err) => TxnReply {
+                success: false,
+                error: serde_json::to_string(&err).expect("fail to serialize"),
+                responses: vec![],
+            },
+        }
+    }
 }
 
 #[async_trait::async_trait]
@@ -154,13 +197,14 @@ impl MetaService for MetaServiceImpl {
 
         incr_meta_metrics_meta_recv_bytes(request.get_ref().encoded_len() as u64);
 
-        let action: MetaGrpcWriteReq = request.try_into()?;
+        let req: MetaGrpcWriteReq = request.try_into()?;
+        let req: MetaGrpcReq = req.into();
 
         add_meta_metrics_meta_request_inflights(1);
 
-        info!("Receive write_action: {:?}", action);
+        info!("Receive write_action: {:?}", req);
 
-        let body = self.action_handler.execute_write(action).await;
+        let body = self.execute_kv_req(req).await;
 
         add_meta_metrics_meta_request_inflights(-1);
 
@@ -175,13 +219,14 @@ impl MetaService for MetaServiceImpl {
 
         incr_meta_metrics_meta_recv_bytes(request.get_ref().encoded_len() as u64);
 
-        let action: MetaGrpcReadReq = request.try_into()?;
+        let req: MetaGrpcReadReq = request.try_into()?;
+        let req: MetaGrpcReq = req.into();
 
         add_meta_metrics_meta_request_inflights(1);
 
-        info!("Receive read_action: {:?}", action);
+        info!("Receive read_action: {:?}", req);
 
-        let res = self.action_handler.execute_read(action).await;
+        let res = self.execute_kv_req(req).await;
 
         add_meta_metrics_meta_request_inflights(-1);
 
@@ -201,7 +246,7 @@ impl MetaService for MetaServiceImpl {
         &self,
         _request: Request<common_meta_types::protobuf::Empty>,
     ) -> Result<Response<Self::ExportStream>, Status> {
-        let meta_node = &self.action_handler.meta_node;
+        let meta_node = &self.meta_node;
 
         let res = meta_node.sto.export().await?;
 
@@ -222,7 +267,7 @@ impl MetaService for MetaServiceImpl {
     ) -> Result<Response<Self::WatchStream>, Status> {
         let (tx, rx) = mpsc::channel(4);
 
-        let meta_node = &self.action_handler.meta_node;
+        let meta_node = &self.meta_node;
         meta_node.create_watcher_stream(request.into_inner(), tx);
 
         let output_stream = tokio_stream::wrappers::ReceiverStream::new(rx);
@@ -243,7 +288,7 @@ impl MetaService for MetaServiceImpl {
 
         info!("Receive txn_request: {:?}", request);
 
-        let body = self.action_handler.execute_txn(request).await;
+        let body = self.execute_txn(request).await;
         add_meta_metrics_meta_request_inflights(-1);
 
         incr_meta_metrics_meta_sent_bytes(body.encoded_len() as u64);
@@ -256,7 +301,7 @@ impl MetaService for MetaServiceImpl {
         request: Request<MemberListRequest>,
     ) -> Result<Response<MemberListReply>, Status> {
         self.check_token(request.metadata())?;
-        let meta_node = &self.action_handler.meta_node;
+        let meta_node = &self.meta_node;
         let members = meta_node.get_meta_addrs().await.map_err(|e| {
             Status::internal(format!("Cannot get metasrv member list, error: {:?}", e))
         })?;