mito2/worker/
handle_create.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
15//! Handling create request.
16
17use common_telemetry::info;
18use store_api::logstore::LogStore;
19use store_api::metadata::RegionMetadataBuilder;
20use store_api::region_request::{AffectedRows, RegionCreateRequest};
21use store_api::storage::RegionId;
22
23use crate::error::Result;
24use crate::region::opener::{RegionOpener, check_recovered_region};
25use crate::worker::RegionWorkerLoop;
26
27impl<S: LogStore> RegionWorkerLoop<S> {
28    pub(crate) async fn handle_create_request(
29        &mut self,
30        region_id: RegionId,
31        request: RegionCreateRequest,
32    ) -> Result<AffectedRows> {
33        // Checks whether the table exists.
34        if let Some(region) = self.regions.get_region(region_id) {
35            // Region already exists.
36            check_recovered_region(
37                &region.metadata(),
38                region_id,
39                &request.column_metadatas,
40                &request.primary_key,
41            )?;
42
43            return Ok(0);
44        }
45
46        // Convert the request into a RegionMetadata and validate it.
47        let mut builder = RegionMetadataBuilder::new(region_id);
48        for column in request.column_metadatas {
49            builder.push_column_metadata(column);
50        }
51        builder.primary_key(request.primary_key);
52        if let Some(expr_json) = request.partition_expr_json.as_ref() {
53            builder.partition_expr_json(Some(expr_json.clone()));
54        }
55
56        // Create a MitoRegion from the RegionMetadata.
57        let requirements = request.requirements;
58        let opener = RegionOpener::new(
59            region_id,
60            &request.table_dir,
61            request.path_type,
62            self.memtable_builder_provider.clone(),
63            self.object_store_manager.clone(),
64            self.purge_scheduler.clone(),
65            self.puffin_manager_factory.clone(),
66            self.intermediate_manager.clone(),
67            self.time_provider.clone(),
68            self.file_ref_manager.clone(),
69            self.partition_expr_fetcher.clone(),
70        )
71        .metadata_builder(builder)
72        .parse_options(request.options)?
73        .cache(Some(self.cache_manager.clone()))
74        .series_index_store(self.series_index_store.clone())
75        .series_index_purger(self.series_index_purger.clone())
76        .hook(self.plugins.get());
77
78        opener.ensure_region_requirements(requirements)?;
79
80        let region = opener.create_or_open(&self.config, &self.wal).await?;
81
82        info!(
83            "A new region created with requirement {:?}, worker: {}, region: {:?}",
84            requirements,
85            self.id,
86            region.metadata()
87        );
88
89        self.region_count.inc();
90
91        // Notify the region hook that the region has been opened (created).
92        // Fires before registration; allocates nothing when no hook is registered.
93        if let Some(hook) = region.manifest_ctx.hook() {
94            hook.on_region_opened(region_id, &region.metadata()).await;
95        }
96
97        // Insert the MitoRegion into the RegionMap.
98        self.regions.insert_region(region);
99
100        Ok(0)
101    }
102}
mito2/worker/handle_create.rs

mito2/worker/
handle_create.rs
