feat: adds Requester to process table flush and compaction request

src/client/src/region.rs

@@ -197,7 +197,7 @@ impl RegionRequester {
 
         check_response_header(header)?;
 
-        Ok(affected_rows)
+        Ok(affected_rows as _)
     }
 
     pub async fn handle(&self, request: RegionRequest) -> Result<AffectedRows> {

src/common/base/src/lib.rs

@@ -21,6 +21,8 @@ pub mod readable_size;
 use core::any::Any;
 use std::sync::{Arc, Mutex, MutexGuard};
 
+pub type AffectedRows = usize;
+
 pub use bit_vec::BitVec;
 
 /// [`Plugins`] is a wrapper of Arc contents.

src/common/function/Cargo.toml

@@ -12,6 +12,7 @@ api.workspace = true
 arc-swap = "1.0"
 async-trait.workspace = true
 chrono-tz = "0.6"
+common-base.workspace = true
 common-catalog.workspace = true
 common-error.workspace = true
 common-macro.workspace = true

src/common/function/src/handlers.rs

@@ -15,12 +15,11 @@
 use std::sync::Arc;
 
 use async_trait::async_trait;
+use common_base::AffectedRows;
 use common_meta::rpc::procedure::{MigrateRegionRequest, ProcedureStateResponse};
 use common_query::error::Result;
 use session::context::QueryContextRef;
-use table::requests::{DeleteRequest, InsertRequest};
-
-pub type AffectedRows = usize;
+use table::requests::{CompactTableRequest, DeleteRequest, FlushTableRequest, InsertRequest};
 
 /// A trait for handling table mutations in `QueryEngine`.
 #[async_trait]
@@ -30,6 +29,17 @@ pub trait TableMutationHandler: Send + Sync {
 
     /// Delete rows from the table.
     async fn delete(&self, request: DeleteRequest, ctx: QueryContextRef) -> Result<AffectedRows>;
+
+    /// Trigger a flush task for table.
+    async fn flush(&self, request: FlushTableRequest, ctx: QueryContextRef)
+        -> Result<AffectedRows>;
+
+    /// Trigger a compaction task for table.
+    async fn compact(
+        &self,
+        request: CompactTableRequest,
+        ctx: QueryContextRef,
+    ) -> Result<AffectedRows>;
 }
 
 /// A trait for handling procedure service requests in `QueryEngine`.

src/common/meta/Cargo.toml

@@ -18,6 +18,7 @@ async-trait.workspace = true
 base64.workspace = true
 bytes.workspace = true
 chrono.workspace = true
+common-base.workspace = true
 common-catalog.workspace = true
 common-error.workspace = true
 common-grpc-expr.workspace = true

src/common/meta/src/datanode_manager.rs

@@ -15,23 +15,25 @@
 use std::sync::Arc;
 
 use api::v1::region::{QueryRequest, RegionRequest};
+pub use common_base::AffectedRows;
 use common_recordbatch::SendableRecordBatchStream;
 
 use crate::error::Result;
 use crate::peer::Peer;
 
-pub type AffectedRows = u64;
-
+/// The trait for handling requests to datanode.
 #[async_trait::async_trait]
 pub trait Datanode: Send + Sync {
     /// Handles DML, and DDL requests.
     async fn handle(&self, request: RegionRequest) -> Result<AffectedRows>;
 
+    /// Handles query requests
     async fn handle_query(&self, request: QueryRequest) -> Result<SendableRecordBatchStream>;
 }
 
 pub type DatanodeRef = Arc<dyn Datanode>;
 
+/// Datanode manager
 #[async_trait::async_trait]
 pub trait DatanodeManager: Send + Sync {
     /// Retrieves a target `datanode`.

src/common/meta/src/test_util.rs

@@ -15,6 +15,7 @@
 use std::sync::Arc;
 
 use api::v1::region::{QueryRequest, RegionRequest};
+pub use common_base::AffectedRows;
 use common_recordbatch::SendableRecordBatchStream;
 
 use crate::cache_invalidator::DummyCacheInvalidator;
@@ -29,8 +30,6 @@ use crate::region_keeper::MemoryRegionKeeper;
 use crate::sequence::SequenceBuilder;
 use crate::wal_options_allocator::WalOptionsAllocator;
 
-pub type AffectedRows = u64;
-
 #[async_trait::async_trait]
 pub trait MockDatanodeHandler: Sync + Send + Clone {
     async fn handle(&self, peer: &Peer, request: RegionRequest) -> Result<AffectedRows>;

src/frontend/src/instance/builder.rs

@@ -25,6 +25,7 @@ use common_meta::kv_backend::KvBackendRef;
 use operator::delete::Deleter;
 use operator::insert::Inserter;
 use operator::procedure::ProcedureServiceOperator;
+use operator::request::Requester;
 use operator::statement::StatementExecutor;
 use operator::table::TableMutationOperator;
 use partition::manager::PartitionRuleManager;
@@ -105,13 +106,19 @@ impl FrontendBuilder {
             datanode_manager.clone(),
         ));
         let deleter = Arc::new(Deleter::new(
+            catalog_manager.clone(),
+            partition_manager.clone(),
+            datanode_manager.clone(),
+        ));
+        let requester = Arc::new(Requester::new(
             catalog_manager.clone(),
             partition_manager,
             datanode_manager.clone(),
         ));
         let table_mutation_handler = Arc::new(TableMutationOperator::new(
             inserter.clone(),
             deleter.clone(),
+            requester,
         ));
 
         let procedure_service_handler = Arc::new(ProcedureServiceOperator::new(

src/frontend/src/instance/standalone.rs

@@ -79,7 +79,7 @@ impl Datanode for RegionInvoker {
         check_response_header(response.header)
             .map_err(BoxedError::new)
             .context(meta_error::ExternalSnafu)?;
-        Ok(response.affected_rows)
+        Ok(response.affected_rows as _)
     }
 
     async fn handle_query(&self, request: QueryRequest) -> MetaResult<SendableRecordBatchStream> {

src/operator/src/delete.rs

@@ -98,7 +98,7 @@ impl Deleter {
         &self,
         request: TableDeleteRequest,
         ctx: QueryContextRef,
-    ) -> Result<usize> {
+    ) -> Result<AffectedRows> {
         let catalog = request.catalog_name.as_str();
         let schema = request.schema_name.as_str();
         let table = request.table_name.as_str();
@@ -143,8 +143,8 @@ impl Deleter {
             });
         let results = future::try_join_all(tasks).await.context(JoinTaskSnafu)?;
 
-        let affected_rows = results.into_iter().sum::<Result<u64>>()?;
-        crate::metrics::DIST_DELETE_ROW_COUNT.inc_by(affected_rows);
+        let affected_rows = results.into_iter().sum::<Result<AffectedRows>>()?;
+        crate::metrics::DIST_DELETE_ROW_COUNT.inc_by(affected_rows as u64);
         Ok(affected_rows)
     }
 

src/operator/src/insert.rs

@@ -156,7 +156,7 @@ impl Inserter {
         &self,
         request: TableInsertRequest,
         ctx: QueryContextRef,
-    ) -> Result<usize> {
+    ) -> Result<AffectedRows> {
         let catalog = request.catalog_name.as_str();
         let schema = request.schema_name.as_str();
         let table_name = request.table_name.as_str();
@@ -219,8 +219,8 @@ impl Inserter {
             });
         let results = future::try_join_all(tasks).await.context(JoinTaskSnafu)?;
 
-        let affected_rows = results.into_iter().sum::<Result<u64>>()?;
-        crate::metrics::DIST_INGEST_ROW_COUNT.inc_by(affected_rows);
+        let affected_rows = results.into_iter().sum::<Result<AffectedRows>>()?;
+        crate::metrics::DIST_INGEST_ROW_COUNT.inc_by(affected_rows as u64);
         Ok(affected_rows)
     }
 

src/operator/src/lib.rs

@@ -20,6 +20,7 @@ pub mod metrics;
 pub mod procedure;
 pub mod region_req_factory;
 pub mod req_convert;
+pub mod request;
 pub mod statement;
 pub mod table;
 #[cfg(test)]

src/operator/src/region_req_factory.rs

@@ -40,4 +40,11 @@ impl RegionRequestFactory {
             body: Some(Body::Deletes(requests)),
         }
     }
+
+    pub fn build_request(&self, body: Body) -> RegionRequest {
+        RegionRequest {
+            header: Some(self.header.clone()),
+            body: Some(body),
+        }
+    }
 }