feat: implement the FrontendInvoker (#3824)

* chore: add `common-frontend`

* feat: add `FrontendInvoker` trait

* feat: implement the `FrontendInvoker`

Cargo.toml

@@ -11,6 +11,7 @@ members = [
     "src/common/config",
     "src/common/datasource",
     "src/common/error",
+    "src/common/frontend",
     "src/common/function",
     "src/common/macro",
     "src/common/greptimedb-telemetry",
@@ -181,6 +182,7 @@ common-config = { path = "src/common/config" }
 common-datasource = { path = "src/common/datasource" }
 common-decimal = { path = "src/common/decimal" }
 common-error = { path = "src/common/error" }
+common-frontend = { path = "src/common/frontend" }
 common-function = { path = "src/common/function" }
 common-greptimedb-telemetry = { path = "src/common/greptimedb-telemetry" }
 common-grpc = { path = "src/common/grpc" }

src/common/frontend/Cargo.toml

@@ -0,0 +1,16 @@
+[package]
+name = "common-frontend"
+version.workspace = true
+edition.workspace = true
+license.workspace = true
+
+[dependencies]
+api.workspace = true
+async-trait.workspace = true
+common-base.workspace = true
+common-error.workspace = true
+common-macro.workspace = true
+common-query.workspace = true
+session.workspace = true
+snafu.workspace = true
+sql.workspace = true

src/common/frontend/src/error.rs

@@ -0,0 +1,44 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use common_error::ext::{BoxedError, ErrorExt};
+use common_error::status_code::StatusCode;
+use common_macro::stack_trace_debug;
+use snafu::{Location, Snafu};
+
+#[derive(Snafu)]
+#[snafu(visibility(pub))]
+#[stack_trace_debug]
+pub enum Error {
+    #[snafu(display("External error"))]
+    External {
+        location: Location,
+        source: BoxedError,
+    },
+}
+
+pub type Result<T> = std::result::Result<T, Error>;
+
+impl ErrorExt for Error {
+    fn status_code(&self) -> StatusCode {
+        use Error::*;
+        match self {
+            External { source, .. } => source.status_code(),
+        }
+    }
+
+    fn as_any(&self) -> &dyn std::any::Any {
+        self
+    }
+}

src/common/frontend/src/handler.rs

@@ -0,0 +1,38 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use api::v1::{RowDeleteRequests, RowInsertRequests};
+use async_trait::async_trait;
+use common_query::Output;
+use session::context::QueryContextRef;
+
+use crate::error::Result;
+
+/// [FrontendInvoker] provides the ability to:
+/// - Insert rows
+/// - Delete rows
+#[async_trait]
+pub trait FrontendInvoker {
+    async fn row_inserts(
+        &self,
+        requests: RowInsertRequests,
+        ctx: QueryContextRef,
+    ) -> Result<Output>;
+
+    async fn row_deletes(
+        &self,
+        requests: RowDeleteRequests,
+        ctx: QueryContextRef,
+    ) -> Result<Output>;
+}

src/common/frontend/src/lib.rs

@@ -0,0 +1,16 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+pub mod error;
+pub mod handler;

src/frontend/Cargo.toml

@@ -24,6 +24,7 @@ common-catalog.workspace = true
 common-config.workspace = true
 common-datasource.workspace = true
 common-error.workspace = true
+common-frontend.workspace = true
 common-grpc.workspace = true
 common-macro.workspace = true
 common-meta.workspace = true

src/frontend/src/instance.rs

@@ -25,13 +25,15 @@ pub mod standalone;
 use std::sync::Arc;
 
 use api::v1::meta::Role;
+use api::v1::{RowDeleteRequests, RowInsertRequests};
 use async_trait::async_trait;
 use auth::{PermissionChecker, PermissionCheckerRef, PermissionReq};
 use catalog::CatalogManagerRef;
 use client::OutputData;
 use common_base::Plugins;
 use common_config::KvBackendConfig;
 use common_error::ext::{BoxedError, ErrorExt};
+use common_frontend::handler::FrontendInvoker;
 use common_grpc::channel_manager::{ChannelConfig, ChannelManager};
 use common_meta::key::TableMetadataManagerRef;
 use common_meta::kv_backend::KvBackendRef;
@@ -233,6 +235,33 @@ impl Instance {
     }
 }
 
+#[async_trait]
+impl FrontendInvoker for Instance {
+    async fn row_inserts(
+        &self,
+        requests: RowInsertRequests,
+        ctx: QueryContextRef,
+    ) -> common_frontend::error::Result<Output> {
+        self.inserter
+            .handle_row_inserts(requests, ctx, &self.statement_executor)
+            .await
+            .map_err(BoxedError::new)
+            .context(common_frontend::error::ExternalSnafu)
+    }
+
+    async fn row_deletes(
+        &self,
+        requests: RowDeleteRequests,
+        ctx: QueryContextRef,
+    ) -> common_frontend::error::Result<Output> {
+        self.deleter
+            .handle_row_deletes(requests, ctx)
+            .await
+            .map_err(BoxedError::new)
+            .context(common_frontend::error::ExternalSnafu)
+    }
+}
+
 #[async_trait]
 impl FrontendInstance for Instance {
     async fn start(&self) -> Result<()> {