standalone/
procedure.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;
16use std::time::Duration;
17
18use common_error::ext::BoxedError;
19use common_meta::ddl::DdlContext;
20use common_meta::ddl_manager::{RepartitionProcedureFactory, RepartitionSource};
21use common_meta::key::runtime_switch::RuntimeSwitchManager;
22use common_meta::kv_backend::KvBackendRef;
23use common_meta::state_store::KvStateStore;
24use common_procedure::local::{EventRecorderHandle, LocalManager, ManagerConfig};
25use common_procedure::options::ProcedureConfig;
26use common_procedure::{BoxedProcedure, ProcedureManagerRef};
27use store_api::storage::TableId;
28use table::table_name::TableName;
29
30use crate::error::NoSupportRepartitionProcedureSnafu;
31
32/// Builds the procedure manager.
33pub fn build_procedure_manager(
34    kv_backend: KvBackendRef,
35    procedure_config: ProcedureConfig,
36) -> (ProcedureManagerRef, EventRecorderHandle) {
37    let kv_state_store = Arc::new(KvStateStore::new(kv_backend.clone()));
38
39    let manager_config = ManagerConfig {
40        max_retry_times: procedure_config.max_retry_times,
41        retry_delay: procedure_config.retry_delay,
42        max_running_procedures: procedure_config.max_running_procedures,
43        ..Default::default()
44    };
45    let runtime_switch_manager = Arc::new(RuntimeSwitchManager::new(kv_backend));
46    let procedure_manager = LocalManager::new(
47        manager_config,
48        kv_state_store.clone(),
49        kv_state_store,
50        Some(runtime_switch_manager),
51        None,
52    );
53    let event_recorder_handle = procedure_manager.event_recorder_handle();
54
55    (Arc::new(procedure_manager), event_recorder_handle)
56}
57
58/// No-op implementation of [`RepartitionProcedureFactory`] for standalone mode.
59///
60/// In standalone deployments, repartition operations are not supported, so
61/// this factory always returns a `NoSupportRepartitionProcedure` error
62/// from [`RepartitionProcedureFactory::create`] and performs no registration
63/// work in [`RepartitionProcedureFactory::register_loaders`].
64pub struct StandaloneRepartitionProcedureFactory;
65
66#[async_trait::async_trait]
67impl RepartitionProcedureFactory for StandaloneRepartitionProcedureFactory {
68    fn create(
69        &self,
70        _ddl_ctx: &DdlContext,
71        _table_name: TableName,
72        _table_id: TableId,
73        _source: RepartitionSource,
74        _to_exprs: Vec<String>,
75        _timeout: Option<Duration>,
76    ) -> std::result::Result<BoxedProcedure, BoxedError> {
77        Err(BoxedError::new(NoSupportRepartitionProcedureSnafu.build()))
78    }
79
80    fn register_loaders(
81        &self,
82        _ddl_ctx: &DdlContext,
83        _procedure_manager: &ProcedureManagerRef,
84    ) -> std::result::Result<(), BoxedError> {
85        Ok(())
86    }
87
88    async fn ensure_gc_requirement(&self) -> std::result::Result<(), BoxedError> {
89        Err(BoxedError::new(NoSupportRepartitionProcedureSnafu.build()))
90    }
91}
standalone/procedure.rs

standalone/
procedure.rs
