common_meta/
procedure_executor.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::{ProcedureDetailResponse, ReconcileRequest, ReconcileResponse};
18use common_event_recorder::ProcedureEventInput;
19use common_procedure::{ProcedureId, ProcedureManagerRef};
20use common_telemetry::tracing_context::W3cTrace;
21use snafu::{OptionExt, ResultExt};
22
23use crate::ddl_manager::DdlManagerRef;
24use crate::error::{
25    ParseProcedureIdSnafu, ProcedureNotFoundSnafu, QueryProcedureSnafu, Result, UnsupportedSnafu,
26};
27use crate::rpc::ddl::{QueryContext, SubmitDdlTaskRequest, SubmitDdlTaskResponse};
28use crate::rpc::procedure::{
29    self, GcRegionsRequest, GcResponse, GcTableRequest, ManageRegionFollowerRequest,
30    MigrateRegionRequest, MigrateRegionResponse, ProcedureStateResponse,
31};
32
33/// The context of procedure executor.
34#[derive(Debug, Clone, Default)]
35pub struct ExecutorContext {
36    pub tracing_context: Option<W3cTrace>,
37    /// Query execution data available at the frontend/standalone submission boundary.
38    ///
39    /// Only DDL serializes the full context. Migration and GC use its typed
40    /// channel to derive protocol without sending the query context itself.
41    pub query_context: Option<QueryContext>,
42    pub actor: Option<String>,
43    /// Caller-supplied event metadata. Protocol is derived by the submission adapter.
44    pub event_input: Option<ProcedureEventInput>,
45}
46
47/// The procedure executor that accepts ddl, region migration task etc.
48#[async_trait::async_trait]
49pub trait ProcedureExecutor: Send + Sync {
50    /// Submit a ddl task
51    async fn submit_ddl_task(
52        &self,
53        ctx: ExecutorContext,
54        request: SubmitDdlTaskRequest,
55    ) -> Result<SubmitDdlTaskResponse>;
56
57    /// Submit ad manage region follower task
58    async fn manage_region_follower(
59        &self,
60        _ctx: &ExecutorContext,
61        _request: ManageRegionFollowerRequest,
62    ) -> Result<()> {
63        UnsupportedSnafu {
64            operation: "manage_region_follower",
65        }
66        .fail()
67    }
68
69    /// Submit a region migration task
70    async fn migrate_region(
71        &self,
72        ctx: &ExecutorContext,
73        request: MigrateRegionRequest,
74    ) -> Result<MigrateRegionResponse>;
75
76    /// Submit a reconcile task.
77    async fn reconcile(
78        &self,
79        _ctx: &ExecutorContext,
80        request: ReconcileRequest,
81    ) -> Result<ReconcileResponse>;
82
83    /// Query the procedure state by its id
84    async fn query_procedure_state(
85        &self,
86        ctx: &ExecutorContext,
87        pid: &str,
88    ) -> Result<ProcedureStateResponse>;
89
90    /// Manually trigger GC for the specified regions.
91    async fn gc_regions(
92        &self,
93        _ctx: &ExecutorContext,
94        _request: GcRegionsRequest,
95    ) -> Result<GcResponse> {
96        UnsupportedSnafu {
97            operation: "gc_regions",
98        }
99        .fail()
100    }
101
102    /// Manually trigger GC for the specified table.
103    async fn gc_table(
104        &self,
105        _ctx: &ExecutorContext,
106        _request: GcTableRequest,
107    ) -> Result<GcResponse> {
108        UnsupportedSnafu {
109            operation: "gc_table",
110        }
111        .fail()
112    }
113
114    async fn list_procedures(&self, ctx: &ExecutorContext) -> Result<ProcedureDetailResponse>;
115}
116
117pub type ProcedureExecutorRef = Arc<dyn ProcedureExecutor>;
118
119/// The local procedure executor that accepts ddl, region migration task etc.
120pub struct LocalProcedureExecutor {
121    pub ddl_manager: DdlManagerRef,
122    pub procedure_manager: ProcedureManagerRef,
123}
124
125impl LocalProcedureExecutor {
126    pub fn new(ddl_manager: DdlManagerRef, procedure_manager: ProcedureManagerRef) -> Self {
127        Self {
128            ddl_manager,
129            procedure_manager,
130        }
131    }
132}
133
134#[async_trait::async_trait]
135impl ProcedureExecutor for LocalProcedureExecutor {
136    async fn submit_ddl_task(
137        &self,
138        ctx: ExecutorContext,
139        request: SubmitDdlTaskRequest,
140    ) -> Result<SubmitDdlTaskResponse> {
141        self.ddl_manager.submit_ddl_task(ctx, request).await
142    }
143
144    async fn migrate_region(
145        &self,
146        _ctx: &ExecutorContext,
147        _request: MigrateRegionRequest,
148    ) -> Result<MigrateRegionResponse> {
149        UnsupportedSnafu {
150            operation: "migrate_region",
151        }
152        .fail()
153    }
154
155    async fn reconcile(
156        &self,
157        _ctx: &ExecutorContext,
158        _request: ReconcileRequest,
159    ) -> Result<ReconcileResponse> {
160        UnsupportedSnafu {
161            operation: "reconcile",
162        }
163        .fail()
164    }
165
166    async fn query_procedure_state(
167        &self,
168        _ctx: &ExecutorContext,
169        pid: &str,
170    ) -> Result<ProcedureStateResponse> {
171        let pid =
172            ProcedureId::parse_str(pid).with_context(|_| ParseProcedureIdSnafu { key: pid })?;
173
174        let state = self
175            .procedure_manager
176            .procedure_state(pid)
177            .await
178            .context(QueryProcedureSnafu)?
179            .with_context(|| ProcedureNotFoundSnafu {
180                pid: pid.to_string(),
181            })?;
182
183        Ok(procedure::procedure_state_to_pb_response(&state))
184    }
185
186    async fn list_procedures(&self, _ctx: &ExecutorContext) -> Result<ProcedureDetailResponse> {
187        let metas = self
188            .procedure_manager
189            .list_procedures()
190            .await
191            .context(QueryProcedureSnafu)?;
192        Ok(procedure::procedure_details_to_pb_response(metas))
193    }
194}
common_meta/procedure_executor.rs

common_meta/
procedure_executor.rs
