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::{PARTITION_INFO_CACHE_NAME, 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;
36#[cfg(feature = "enterprise")]
37use operator::statement::CreateDatabaseHandlerRef;
38use operator::statement::{
39    ExecutorConfigureContext, StatementExecutor, StatementExecutorConfiguratorRef,
40    StatementExecutorRef,
41};
42use operator::table::TableMutationOperator;
43use partition::cache::PartitionInfoCacheRef;
44use partition::manager::PartitionRuleManager;
45use pipeline::pipeline_operator::PipelineOperator;
46use query::QueryEngineFactory;
47use query::region_query::RegionQueryHandlerFactoryRef;
48use snafu::{OptionExt, ResultExt};
49
50use crate::error::{self, DataFusionSnafu, ExternalSnafu, Result};
51use crate::events::EventHandlerImpl;
52use crate::frontend::FrontendOptions;
53use crate::heartbeat::frontend_peer_addr;
54use crate::instance::Instance;
55use crate::instance::region_query::FrontendRegionQueryHandler;
56
57/// The frontend [`Instance`] builder.
58pub struct FrontendBuilder {
59    options: FrontendOptions,
60    kv_backend: KvBackendRef,
61    layered_cache_registry: LayeredCacheRegistryRef,
62    local_cache_invalidator: Option<CacheInvalidatorRef>,
63    catalog_manager: CatalogManagerRef,
64    node_manager: NodeManagerRef,
65    plugins: Option<Plugins>,
66    procedure_executor: ProcedureExecutorRef,
67    process_manager: ProcessManagerRef,
68}
69
70impl FrontendBuilder {
71    #[allow(clippy::too_many_arguments)]
72    pub fn new(
73        options: FrontendOptions,
74        kv_backend: KvBackendRef,
75        layered_cache_registry: LayeredCacheRegistryRef,
76        catalog_manager: CatalogManagerRef,
77        node_manager: NodeManagerRef,
78        procedure_executor: ProcedureExecutorRef,
79        process_manager: ProcessManagerRef,
80    ) -> Self {
81        Self {
82            options,
83            kv_backend,
84            layered_cache_registry,
85            local_cache_invalidator: None,
86            catalog_manager,
87            node_manager,
88            plugins: None,
89            procedure_executor,
90            process_manager,
91        }
92    }
93
94    #[cfg(test)]
95    pub(crate) fn new_test(
96        options: &FrontendOptions,
97        meta_client: meta_client::MetaClientRef,
98    ) -> Self {
99        use cache::{build_fundamental_cache_registry, with_default_composite_cache_registry};
100        use common_meta::cache::LayeredCacheRegistryBuilder;
101        use common_meta::kv_backend::memory::MemoryKvBackend;
102
103        let kv_backend = Arc::new(MemoryKvBackend::new());
104
105        // Builds cache registry
106        let layered_cache_builder = LayeredCacheRegistryBuilder::default();
107        let fundamental_cache_registry = build_fundamental_cache_registry(kv_backend.clone());
108        let layered_cache_registry = Arc::new(
109            with_default_composite_cache_registry(
110                layered_cache_builder.add_cache_registry(fundamental_cache_registry),
111            )
112            .unwrap()
113            .build(),
114        );
115
116        Self::new(
117            options.clone(),
118            kv_backend,
119            layered_cache_registry,
120            catalog::memory::MemoryCatalogManager::with_default_setup(),
121            Arc::new(client::client_manager::NodeClients::default()),
122            meta_client,
123            Arc::new(catalog::process_manager::ProcessManager::new(
124                "".to_string(),
125                None,
126            )),
127        )
128    }
129
130    pub fn with_local_cache_invalidator(self, cache_invalidator: CacheInvalidatorRef) -> Self {
131        Self {
132            local_cache_invalidator: Some(cache_invalidator),
133            ..self
134        }
135    }
136
137    pub fn options(&self) -> &FrontendOptions {
138        &self.options
139    }
140
141    pub fn kv_backend(&self) -> &KvBackendRef {
142        &self.kv_backend
143    }
144
145    pub fn layered_cache_registry(&self) -> &LayeredCacheRegistryRef {
146        &self.layered_cache_registry
147    }
148
149    pub fn catalog_manager(&self) -> &CatalogManagerRef {
150        &self.catalog_manager
151    }
152
153    pub fn node_manager(&self) -> &NodeManagerRef {
154        &self.node_manager
155    }
156
157    pub fn procedure_executor(&self) -> &ProcedureExecutorRef {
158        &self.procedure_executor
159    }
160
161    pub fn process_manager(&self) -> &ProcessManagerRef {
162        &self.process_manager
163    }
164
165    pub fn with_plugin(self, plugins: Plugins) -> Self {
166        Self {
167            plugins: Some(plugins),
168            ..self
169        }
170    }
171
172    pub async fn try_build(self) -> Result<Instance> {
173        let kv_backend = self.kv_backend;
174        let node_manager = self.node_manager;
175        let plugins = self.plugins.unwrap_or_default();
176        let process_manager = self.process_manager;
177        let table_route_cache: TableRouteCacheRef =
178            self.layered_cache_registry
179                .get()
180                .context(error::CacheRequiredSnafu {
181                    name: TABLE_ROUTE_CACHE_NAME,
182                })?;
183        let partition_info_cache: PartitionInfoCacheRef = self
184            .layered_cache_registry
185            .get()
186            .context(error::CacheRequiredSnafu {
187                name: PARTITION_INFO_CACHE_NAME,
188            })?;
189        let partition_manager = Arc::new(PartitionRuleManager::new(
190            kv_backend.clone(),
191            table_route_cache.clone(),
192            partition_info_cache.clone(),
193        ));
194
195        let local_cache_invalidator = self
196            .local_cache_invalidator
197            .unwrap_or_else(|| Arc::new(DummyCacheInvalidator));
198
199        let region_query_handler =
200            if let Some(factory) = plugins.get::<RegionQueryHandlerFactoryRef>() {
201                factory.build(partition_manager.clone(), node_manager.clone())
202            } else {
203                FrontendRegionQueryHandler::arc(partition_manager.clone(), node_manager.clone())
204            };
205
206        let table_flownode_cache =
207            self.layered_cache_registry
208                .get()
209                .context(error::CacheRequiredSnafu {
210                    name: TABLE_FLOWNODE_SET_CACHE_NAME,
211                })?;
212
213        let inserter = Arc::new(Inserter::new(
214            self.catalog_manager.clone(),
215            partition_manager.clone(),
216            node_manager.clone(),
217            table_flownode_cache,
218            self.options.auto_create_table,
219        ));
220        let deleter = Arc::new(Deleter::new(
221            self.catalog_manager.clone(),
222            partition_manager.clone(),
223            node_manager.clone(),
224        ));
225        let requester = Arc::new(Requester::new(
226            self.catalog_manager.clone(),
227            partition_manager.clone(),
228            node_manager.clone(),
229        ));
230        let table_mutation_handler = Arc::new(TableMutationOperator::new(
231            inserter.clone(),
232            deleter.clone(),
233            requester,
234        ));
235
236        let table_metadata_manager = Arc::new(TableMetadataManager::new(kv_backend.clone()));
237        let procedure_service_handler = Arc::new(ProcedureServiceOperator::new(
238            self.procedure_executor.clone(),
239            self.catalog_manager.clone(),
240            table_metadata_manager.clone(),
241        ));
242
243        let flow_metadata_manager: Arc<FlowMetadataManager> =
244            Arc::new(FlowMetadataManager::new(kv_backend.clone()));
245        let flow_service = FlowServiceOperator::new(flow_metadata_manager, node_manager.clone());
246
247        let mut query_options = self.options.query.clone();
248        query_options.enable_per_region_metrics = self.options.logging.enable_per_region_metrics;
249        let query_engine = QueryEngineFactory::try_new_with_plugins(
250            self.catalog_manager.clone(),
251            Some(partition_manager.clone()),
252            Some(region_query_handler.clone()),
253            Some(table_mutation_handler),
254            Some(procedure_service_handler),
255            Some(Arc::new(flow_service)),
256            true,
257            plugins.clone(),
258            query_options,
259        )
260        .context(DataFusionSnafu)?
261        .query_engine();
262
263        let frontend_peer_addr = frontend_peer_addr(&self.options);
264        let statement_executor = StatementExecutor::new(
265            self.catalog_manager.clone(),
266            query_engine.clone(),
267            self.procedure_executor,
268            kv_backend.clone(),
269            local_cache_invalidator,
270            inserter.clone(),
271            partition_manager,
272            Some(process_manager.clone()),
273            frontend_peer_addr.clone(),
274        );
275
276        let statement_executor =
277            if let Some(configurator) = plugins.get::<StatementExecutorConfiguratorRef>() {
278                let ctx = ExecutorConfigureContext {
279                    kv_backend: kv_backend.clone(),
280                };
281                configurator
282                    .configure(statement_executor, ctx)
283                    .await
284                    .context(ExternalSnafu)?
285            } else {
286                statement_executor
287            };
288
289        #[cfg(feature = "enterprise")]
290        let statement_executor = if let Some(handler) = plugins.get::<CreateDatabaseHandlerRef>() {
291            statement_executor.with_create_database_handler(handler)
292        } else {
293            statement_executor
294        };
295
296        let statement_executor = Arc::new(statement_executor);
297
298        let pipeline_operator = Arc::new(PipelineOperator::new(
299            inserter.clone(),
300            statement_executor.clone(),
301            self.catalog_manager.clone(),
302            query_engine.clone(),
303        ));
304
305        plugins.insert::<StatementExecutorRef>(statement_executor.clone());
306
307        let event_recorder = Arc::new(EventRecorderImpl::new(Box::new(EventHandlerImpl::new(
308            statement_executor.clone(),
309            self.options.slow_query.ttl,
310            self.options.event_recorder.ttl,
311        ))));
312
313        Ok(Instance {
314            frontend_peer_addr,
315            catalog_manager: self.catalog_manager,
316            pipeline_operator,
317            statement_executor,
318            query_engine,
319            plugins,
320            inserter,
321            deleter,
322            table_metadata_manager,
323            event_recorder: Some(event_recorder),
324            process_manager,
325            otlp_metrics_table_legacy_cache: DashMap::new(),
326            slow_query_options: self.options.slow_query.clone(),
327            influxdb_default_merge_mode: self.options.influxdb.default_merge_mode,
328            trace_ingest_chunk_size: self.options.otlp.trace_ingest_chunk_size,
329            suspend: Arc::new(AtomicBool::new(false)),
330        })
331    }
332}
frontend/instance/builder.rs

frontend/instance/
builder.rs
