common_function/
handlers.rs

1// Copyright 2023 Greptime Team
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::sync::Arc;
16
17use api::v1::meta::ReconcileRequest;
18use async_trait::async_trait;
19use catalog::CatalogManagerRef;
20use common_base::AffectedRows;
21use common_meta::rpc::procedure::{
22    GcRegionsRequest as MetaGcRegionsRequest, GcResponse as MetaGcResponse,
23    GcTableRequest as MetaGcTableRequest, ManageRegionFollowerRequest, MigrateRegionRequest,
24    ProcedureStateResponse,
25};
26use common_query::Output;
27use common_query::error::Result;
28use session::context::QueryContextRef;
29use store_api::storage::RegionId;
30use table::requests::{
31    BuildIndexTableRequest, CompactTableRequest, DeleteRequest, FlushTableRequest, InsertRequest,
32};
33use table::table_name::TableName;
34
35/// A trait for handling table mutations in `QueryEngine`.
36#[async_trait]
37pub trait TableMutationHandler: Send + Sync {
38    /// Inserts rows into the table.
39    async fn insert(&self, request: InsertRequest, ctx: QueryContextRef) -> Result<Output>;
40
41    /// Delete rows from the table.
42    async fn delete(&self, request: DeleteRequest, ctx: QueryContextRef) -> Result<AffectedRows>;
43
44    /// Trigger a flush task for table.
45    async fn flush(&self, request: FlushTableRequest, ctx: QueryContextRef)
46    -> Result<AffectedRows>;
47
48    /// Trigger a compaction task for table.
49    async fn compact(
50        &self,
51        request: CompactTableRequest,
52        ctx: QueryContextRef,
53    ) -> Result<AffectedRows>;
54
55    /// Trigger an index build task for the table.
56    async fn build_index(
57        &self,
58        request: BuildIndexTableRequest,
59        ctx: QueryContextRef,
60    ) -> Result<AffectedRows>;
61
62    /// Trigger a flush task for a table region.
63    async fn flush_region(&self, region_id: RegionId, ctx: QueryContextRef)
64    -> Result<AffectedRows>;
65
66    /// Trigger a compaction task for a table region.
67    async fn compact_region(
68        &self,
69        region_id: RegionId,
70        ctx: QueryContextRef,
71    ) -> Result<AffectedRows>;
72
73    /// Discard all unflushed data from a table region.
74    async fn discard_unflushed_data(
75        &self,
76        region_id: RegionId,
77        ctx: QueryContextRef,
78    ) -> Result<AffectedRows>;
79
80    /// Discard all unflushed data from all regions of a table.
81    async fn discard_unflushed_data_by_table(
82        &self,
83        table_name: TableName,
84        ctx: QueryContextRef,
85    ) -> Result<AffectedRows>;
86}
87
88/// A trait for handling procedure service requests in `QueryEngine`.
89#[async_trait]
90pub trait ProcedureServiceHandler: Send + Sync {
91    /// Permanently purge a dropped table.
92    async fn purge_table(&self, query_ctx: QueryContextRef, table_name: TableName) -> Result<()>;
93
94    /// Migrate a region from source peer to target peer, returns the procedure id if success.
95    async fn migrate_region(
96        &self,
97        query_ctx: QueryContextRef,
98        request: MigrateRegionRequest,
99    ) -> Result<Option<String>>;
100
101    /// Reconcile a table, database or catalog, returns the procedure id if success.
102    async fn reconcile(&self, request: ReconcileRequest) -> Result<Option<String>>;
103
104    /// Query the procedure' state by its id
105    async fn query_procedure_state(&self, pid: &str) -> Result<ProcedureStateResponse>;
106
107    /// Manage a region follower to a region.
108    async fn manage_region_follower(&self, request: ManageRegionFollowerRequest) -> Result<()>;
109
110    /// Get the catalog manager
111    fn catalog_manager(&self) -> &CatalogManagerRef;
112
113    /// Manually trigger GC for specific regions.
114    async fn gc_regions(
115        &self,
116        query_ctx: QueryContextRef,
117        request: MetaGcRegionsRequest,
118    ) -> Result<MetaGcResponse>;
119
120    /// Manually trigger GC for a table.
121    async fn gc_table(
122        &self,
123        query_ctx: QueryContextRef,
124        request: MetaGcTableRequest,
125    ) -> Result<MetaGcResponse>;
126}
127
128/// This flow service handler is only use for flush flow for now.
129#[async_trait]
130pub trait FlowServiceHandler: Send + Sync {
131    async fn flush(
132        &self,
133        catalog: &str,
134        flow: &str,
135        ctx: QueryContextRef,
136    ) -> Result<api::v1::flow::FlowResponse>;
137}
138
139pub type TableMutationHandlerRef = Arc<dyn TableMutationHandler>;
140
141pub type ProcedureServiceHandlerRef = Arc<dyn ProcedureServiceHandler>;
142
143pub type FlowServiceHandlerRef = Arc<dyn FlowServiceHandler>;
common_function/handlers.rs

common_function/
handlers.rs
