frontend/instance/
builder.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::sync::atomic::AtomicBool;
17
18use cache::{TABLE_FLOWNODE_SET_CACHE_NAME, TABLE_ROUTE_CACHE_NAME};
19use catalog::CatalogManagerRef;
20use catalog::process_manager::ProcessManagerRef;
21use common_base::Plugins;
22use common_event_recorder::EventRecorderImpl;
23use common_meta::cache::{LayeredCacheRegistryRef, TableRouteCacheRef};
24use common_meta::cache_invalidator::{CacheInvalidatorRef, DummyCacheInvalidator};
25use common_meta::key::TableMetadataManager;
26use common_meta::key::flow::FlowMetadataManager;
27use common_meta::kv_backend::KvBackendRef;
28use common_meta::node_manager::NodeManagerRef;
29use common_meta::procedure_executor::ProcedureExecutorRef;
30use dashmap::DashMap;
31use operator::delete::Deleter;
32use operator::flow::FlowServiceOperator;
33use operator::insert::Inserter;
34use operator::procedure::ProcedureServiceOperator;
35use operator::request::Requester;
36use operator::statement::{
37    ExecutorConfigureContext, StatementExecutor, StatementExecutorConfiguratorRef,
38    StatementExecutorRef,
39};
40use operator::table::TableMutationOperator;
41use partition::manager::PartitionRuleManager;
42use pipeline::pipeline_operator::PipelineOperator;
43use query::QueryEngineFactory;
44use query::region_query::RegionQueryHandlerFactoryRef;
45use snafu::{OptionExt, ResultExt};
46
47use crate::error::{self, ExternalSnafu, Result};
48use crate::events::EventHandlerImpl;
49use crate::frontend::FrontendOptions;
50use crate::instance::Instance;
51use crate::instance::region_query::FrontendRegionQueryHandler;
52
53/// The frontend [`Instance`] builder.
54pub struct FrontendBuilder {
55    options: FrontendOptions,
56    kv_backend: KvBackendRef,
57    layered_cache_registry: LayeredCacheRegistryRef,
58    local_cache_invalidator: Option<CacheInvalidatorRef>,
59    catalog_manager: CatalogManagerRef,
60    node_manager: NodeManagerRef,
61    plugins: Option<Plugins>,
62    procedure_executor: ProcedureExecutorRef,
63    process_manager: ProcessManagerRef,
64}
65
66impl FrontendBuilder {
67    #[allow(clippy::too_many_arguments)]
68    pub fn new(
69        options: FrontendOptions,
70        kv_backend: KvBackendRef,
71        layered_cache_registry: LayeredCacheRegistryRef,
72        catalog_manager: CatalogManagerRef,
73        node_manager: NodeManagerRef,
74        procedure_executor: ProcedureExecutorRef,
75        process_manager: ProcessManagerRef,
76    ) -> Self {
77        Self {
78            options,
79            kv_backend,
80            layered_cache_registry,
81            local_cache_invalidator: None,
82            catalog_manager,
83            node_manager,
84            plugins: None,
85            procedure_executor,
86            process_manager,
87        }
88    }
89
90    #[cfg(test)]
91    pub(crate) fn new_test(
92        options: &FrontendOptions,
93        meta_client: meta_client::MetaClientRef,
94    ) -> Self {
95        let kv_backend = Arc::new(common_meta::kv_backend::memory::MemoryKvBackend::new());
96
97        let layered_cache_registry = Arc::new(
98            common_meta::cache::LayeredCacheRegistryBuilder::default()
99                .add_cache_registry(cache::build_fundamental_cache_registry(kv_backend.clone()))
100                .build(),
101        );
102
103        Self::new(
104            options.clone(),
105            kv_backend,
106            layered_cache_registry,
107            catalog::memory::MemoryCatalogManager::with_default_setup(),
108            Arc::new(client::client_manager::NodeClients::default()),
109            meta_client,
110            Arc::new(catalog::process_manager::ProcessManager::new(
111                "".to_string(),
112                None,
113            )),
114        )
115    }
116
117    pub fn with_local_cache_invalidator(self, cache_invalidator: CacheInvalidatorRef) -> Self {
118        Self {
119            local_cache_invalidator: Some(cache_invalidator),
120            ..self
121        }
122    }
123
124    pub fn with_plugin(self, plugins: Plugins) -> Self {
125        Self {
126            plugins: Some(plugins),
127            ..self
128        }
129    }
130
131    pub async fn try_build(self) -> Result<Instance> {
132        let kv_backend = self.kv_backend;
133        let node_manager = self.node_manager;
134        let plugins = self.plugins.unwrap_or_default();
135        let process_manager = self.process_manager;
136        let table_route_cache: TableRouteCacheRef =
137            self.layered_cache_registry
138                .get()
139                .context(error::CacheRequiredSnafu {
140                    name: TABLE_ROUTE_CACHE_NAME,
141                })?;
142        let partition_manager = Arc::new(PartitionRuleManager::new(
143            kv_backend.clone(),
144            table_route_cache.clone(),
145        ));
146
147        let local_cache_invalidator = self
148            .local_cache_invalidator
149            .unwrap_or_else(|| Arc::new(DummyCacheInvalidator));
150
151        let region_query_handler =
152            if let Some(factory) = plugins.get::<RegionQueryHandlerFactoryRef>() {
153                factory.build(partition_manager.clone(), node_manager.clone())
154            } else {
155                FrontendRegionQueryHandler::arc(partition_manager.clone(), node_manager.clone())
156            };
157
158        let table_flownode_cache =
159            self.layered_cache_registry
160                .get()
161                .context(error::CacheRequiredSnafu {
162                    name: TABLE_FLOWNODE_SET_CACHE_NAME,
163                })?;
164
165        let inserter = Arc::new(Inserter::new(
166            self.catalog_manager.clone(),
167            partition_manager.clone(),
168            node_manager.clone(),
169            table_flownode_cache,
170        ));
171        let deleter = Arc::new(Deleter::new(
172            self.catalog_manager.clone(),
173            partition_manager.clone(),
174            node_manager.clone(),
175        ));
176        let requester = Arc::new(Requester::new(
177            self.catalog_manager.clone(),
178            partition_manager.clone(),
179            node_manager.clone(),
180        ));
181        let table_mutation_handler = Arc::new(TableMutationOperator::new(
182            inserter.clone(),
183            deleter.clone(),
184            requester,
185        ));
186
187        let procedure_service_handler = Arc::new(ProcedureServiceOperator::new(
188            self.procedure_executor.clone(),
189            self.catalog_manager.clone(),
190        ));
191
192        let flow_metadata_manager: Arc<FlowMetadataManager> =
193            Arc::new(FlowMetadataManager::new(kv_backend.clone()));
194        let flow_service = FlowServiceOperator::new(flow_metadata_manager, node_manager.clone());
195
196        let query_engine = QueryEngineFactory::new_with_plugins(
197            self.catalog_manager.clone(),
198            Some(partition_manager.clone()),
199            Some(region_query_handler.clone()),
200            Some(table_mutation_handler),
201            Some(procedure_service_handler),
202            Some(Arc::new(flow_service)),
203            true,
204            plugins.clone(),
205            self.options.query.clone(),
206        )
207        .query_engine();
208
209        let statement_executor = StatementExecutor::new(
210            self.catalog_manager.clone(),
211            query_engine.clone(),
212            self.procedure_executor,
213            kv_backend.clone(),
214            local_cache_invalidator,
215            inserter.clone(),
216            table_route_cache,
217            Some(process_manager.clone()),
218        );
219
220        let statement_executor =
221            if let Some(configurator) = plugins.get::<StatementExecutorConfiguratorRef>() {
222                let ctx = ExecutorConfigureContext {
223                    kv_backend: kv_backend.clone(),
224                };
225                configurator
226                    .configure(statement_executor, ctx)
227                    .await
228                    .context(ExternalSnafu)?
229            } else {
230                statement_executor
231            };
232
233        let statement_executor = Arc::new(statement_executor);
234
235        let pipeline_operator = Arc::new(PipelineOperator::new(
236            inserter.clone(),
237            statement_executor.clone(),
238            self.catalog_manager.clone(),
239            query_engine.clone(),
240        ));
241
242        plugins.insert::<StatementExecutorRef>(statement_executor.clone());
243
244        let event_recorder = Arc::new(EventRecorderImpl::new(Box::new(EventHandlerImpl::new(
245            statement_executor.clone(),
246            self.options.slow_query.ttl,
247            self.options.event_recorder.ttl,
248        ))));
249
250        Ok(Instance {
251            catalog_manager: self.catalog_manager,
252            pipeline_operator,
253            statement_executor,
254            query_engine,
255            plugins,
256            inserter,
257            deleter,
258            table_metadata_manager: Arc::new(TableMetadataManager::new(kv_backend)),
259            event_recorder: Some(event_recorder),
260            process_manager,
261            otlp_metrics_table_legacy_cache: DashMap::new(),
262            slow_query_options: self.options.slow_query.clone(),
263            suspend: Arc::new(AtomicBool::new(false)),
264        })
265    }
266}
frontend/instance/builder.rs

frontend/instance/
builder.rs
