common_meta/ddl/
create_table.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
15pub mod executor;
16pub mod template;
17
18use std::collections::HashMap;
19
20use api::v1::CreateTableExpr;
21use async_trait::async_trait;
22use common_error::ext::BoxedError;
23use common_procedure::error::{
24    ExternalSnafu, FromJsonSnafu, Result as ProcedureResult, ToJsonSnafu,
25};
26use common_procedure::{Context as ProcedureContext, LockKey, Procedure, ProcedureId, Status};
27use common_telemetry::info;
28use serde::{Deserialize, Serialize};
29use snafu::{OptionExt, ResultExt};
30use store_api::metadata::ColumnMetadata;
31use store_api::storage::RegionNumber;
32use strum::AsRefStr;
33use table::metadata::{RawTableInfo, TableId};
34use table::table_name::TableName;
35use table::table_reference::TableReference;
36pub(crate) use template::{CreateRequestBuilder, build_template_from_raw_table_info};
37
38use crate::ddl::create_table::executor::CreateTableExecutor;
39use crate::ddl::create_table::template::build_template;
40use crate::ddl::utils::map_to_procedure_error;
41use crate::ddl::{DdlContext, TableMetadata};
42use crate::error::{self, Result};
43use crate::key::table_route::PhysicalTableRouteValue;
44use crate::lock_key::{CatalogLock, SchemaLock, TableNameLock};
45use crate::metrics;
46use crate::region_keeper::OperatingRegionGuard;
47use crate::rpc::ddl::CreateTableTask;
48use crate::rpc::router::{RegionRoute, operating_leader_regions};
49
50pub struct CreateTableProcedure {
51    pub context: DdlContext,
52    /// The serializable data.
53    pub data: CreateTableData,
54    /// The guards of opening.
55    pub opening_regions: Vec<OperatingRegionGuard>,
56    /// The executor of the procedure.
57    pub executor: CreateTableExecutor,
58}
59
60fn build_executor_from_create_table_data(
61    create_table_expr: &CreateTableExpr,
62) -> Result<CreateTableExecutor> {
63    let template = build_template(create_table_expr)?;
64    let builder = CreateRequestBuilder::new(template, None);
65    let table_name = TableName::new(
66        create_table_expr.catalog_name.clone(),
67        create_table_expr.schema_name.clone(),
68        create_table_expr.table_name.clone(),
69    );
70    let executor =
71        CreateTableExecutor::new(table_name, create_table_expr.create_if_not_exists, builder);
72    Ok(executor)
73}
74
75impl CreateTableProcedure {
76    pub const TYPE_NAME: &'static str = "metasrv-procedure::CreateTable";
77
78    pub fn new(task: CreateTableTask, context: DdlContext) -> Result<Self> {
79        let executor = build_executor_from_create_table_data(&task.create_table)?;
80
81        Ok(Self {
82            context,
83            data: CreateTableData::new(task),
84            opening_regions: vec![],
85            executor,
86        })
87    }
88
89    pub fn from_json(json: &str, context: DdlContext) -> ProcedureResult<Self> {
90        let data: CreateTableData = serde_json::from_str(json).context(FromJsonSnafu)?;
91        let create_table_expr = &data.task.create_table;
92        let executor = build_executor_from_create_table_data(create_table_expr)
93            .map_err(BoxedError::new)
94            .context(ExternalSnafu {
95                clean_poisons: false,
96            })?;
97
98        Ok(CreateTableProcedure {
99            context,
100            data,
101            opening_regions: vec![],
102            executor,
103        })
104    }
105
106    fn table_info(&self) -> &RawTableInfo {
107        &self.data.task.table_info
108    }
109
110    pub(crate) fn table_id(&self) -> TableId {
111        self.table_info().ident.table_id
112    }
113
114    fn region_wal_options(&self) -> Result<&HashMap<RegionNumber, String>> {
115        self.data
116            .region_wal_options
117            .as_ref()
118            .context(error::UnexpectedSnafu {
119                err_msg: "region_wal_options is not allocated",
120            })
121    }
122
123    fn table_route(&self) -> Result<&PhysicalTableRouteValue> {
124        self.data
125            .table_route
126            .as_ref()
127            .context(error::UnexpectedSnafu {
128                err_msg: "table_route is not allocated",
129            })
130    }
131
132    /// On the prepare step, it performs:
133    /// - Checks whether the table exists.
134    /// - Allocates the table id.
135    ///
136    /// Abort(non-retry):
137    /// - TableName exists and `create_if_not_exists` is false.
138    /// - Failed to allocate [TableMetadata].
139    pub(crate) async fn on_prepare(&mut self) -> Result<Status> {
140        let table_id = self
141            .executor
142            .on_prepare(&self.context.table_metadata_manager)
143            .await?;
144        // Return the table id if the table already exists.
145        if let Some(table_id) = table_id {
146            return Ok(Status::done_with_output(table_id));
147        }
148
149        self.data.state = CreateTableState::DatanodeCreateRegions;
150        let TableMetadata {
151            table_id,
152            table_route,
153            region_wal_options,
154        } = self
155            .context
156            .table_metadata_allocator
157            .create(&self.data.task)
158            .await?;
159        self.set_allocated_metadata(table_id, table_route, region_wal_options);
160
161        Ok(Status::executing(true))
162    }
163
164    /// Creates regions on datanodes
165    ///
166    /// Abort(non-retry):
167    /// - Failed to create [CreateRequestBuilder].
168    /// - Failed to get the table route of physical table (for logical table).
169    ///
170    /// Retry:
171    /// - If the underlying servers returns one of the following [Code](tonic::status::Code):
172    ///   - [Code::Cancelled](tonic::status::Code::Cancelled)
173    ///   - [Code::DeadlineExceeded](tonic::status::Code::DeadlineExceeded)
174    ///   - [Code::Unavailable](tonic::status::Code::Unavailable)
175    pub async fn on_datanode_create_regions(&mut self) -> Result<Status> {
176        let table_route = self.table_route()?.clone();
177        // Registers opening regions
178        let guards = self.register_opening_regions(&self.context, &table_route.region_routes)?;
179        if !guards.is_empty() {
180            self.opening_regions = guards;
181        }
182        self.create_regions(&table_route.region_routes).await
183    }
184
185    async fn create_regions(&mut self, region_routes: &[RegionRoute]) -> Result<Status> {
186        let table_id = self.table_id();
187        let region_wal_options = self.region_wal_options()?;
188        let column_metadatas = self
189            .executor
190            .on_create_regions(
191                &self.context.node_manager,
192                table_id,
193                region_routes,
194                region_wal_options,
195            )
196            .await?;
197
198        self.data.column_metadatas = column_metadatas;
199        self.data.state = CreateTableState::CreateMetadata;
200        Ok(Status::executing(true))
201    }
202
203    /// Creates table metadata
204    ///
205    /// Abort(not-retry):
206    /// - Failed to create table metadata.
207    async fn on_create_metadata(&mut self, pid: ProcedureId) -> Result<Status> {
208        let table_id = self.table_id();
209        let table_ref = self.data.table_ref();
210        let manager = &self.context.table_metadata_manager;
211
212        let raw_table_info = self.table_info().clone();
213        // Safety: the region_wal_options must be allocated.
214        let region_wal_options = self.region_wal_options()?.clone();
215        // Safety: the table_route must be allocated.
216        let physical_table_route = self.table_route()?.clone();
217        self.executor
218            .on_create_metadata(
219                manager,
220                &self.context.region_failure_detector_controller,
221                raw_table_info,
222                &self.data.column_metadatas,
223                physical_table_route,
224                region_wal_options,
225            )
226            .await?;
227
228        info!(
229            "Successfully created table: {}, table_id: {}, procedure_id: {}",
230            table_ref, table_id, pid
231        );
232
233        self.opening_regions.clear();
234        Ok(Status::done_with_output(table_id))
235    }
236
237    /// Registers and returns the guards of the opening region if they don't exist.
238    fn register_opening_regions(
239        &self,
240        context: &DdlContext,
241        region_routes: &[RegionRoute],
242    ) -> Result<Vec<OperatingRegionGuard>> {
243        let opening_regions = operating_leader_regions(region_routes);
244        if self.opening_regions.len() == opening_regions.len() {
245            return Ok(vec![]);
246        }
247
248        let mut opening_region_guards = Vec::with_capacity(opening_regions.len());
249
250        for (region_id, datanode_id) in opening_regions {
251            let guard = context
252                .memory_region_keeper
253                .register(datanode_id, region_id)
254                .context(error::RegionOperatingRaceSnafu {
255                    region_id,
256                    peer_id: datanode_id,
257                })?;
258            opening_region_guards.push(guard);
259        }
260        Ok(opening_region_guards)
261    }
262
263    pub fn set_allocated_metadata(
264        &mut self,
265        table_id: TableId,
266        table_route: PhysicalTableRouteValue,
267        region_wal_options: HashMap<RegionNumber, String>,
268    ) {
269        self.data.task.table_info.ident.table_id = table_id;
270        self.data.table_route = Some(table_route);
271        self.data.region_wal_options = Some(region_wal_options);
272    }
273}
274
275#[async_trait]
276impl Procedure for CreateTableProcedure {
277    fn type_name(&self) -> &str {
278        Self::TYPE_NAME
279    }
280
281    fn recover(&mut self) -> ProcedureResult<()> {
282        // Only registers regions if the table route is allocated.
283        if let Some(x) = &self.data.table_route {
284            self.opening_regions = self
285                .register_opening_regions(&self.context, &x.region_routes)
286                .map_err(BoxedError::new)
287                .context(ExternalSnafu {
288                    clean_poisons: false,
289                })?;
290        }
291
292        Ok(())
293    }
294
295    async fn execute(&mut self, ctx: &ProcedureContext) -> ProcedureResult<Status> {
296        let state = &self.data.state;
297
298        let _timer = metrics::METRIC_META_PROCEDURE_CREATE_TABLE
299            .with_label_values(&[state.as_ref()])
300            .start_timer();
301
302        match state {
303            CreateTableState::Prepare => self.on_prepare().await,
304            CreateTableState::DatanodeCreateRegions => self.on_datanode_create_regions().await,
305            CreateTableState::CreateMetadata => self.on_create_metadata(ctx.procedure_id).await,
306        }
307        .map_err(map_to_procedure_error)
308    }
309
310    fn dump(&self) -> ProcedureResult<String> {
311        serde_json::to_string(&self.data).context(ToJsonSnafu)
312    }
313
314    fn lock_key(&self) -> LockKey {
315        let table_ref = &self.data.table_ref();
316
317        LockKey::new(vec![
318            CatalogLock::Read(table_ref.catalog).into(),
319            SchemaLock::read(table_ref.catalog, table_ref.schema).into(),
320            TableNameLock::new(table_ref.catalog, table_ref.schema, table_ref.table).into(),
321        ])
322    }
323}
324
325#[derive(Debug, Clone, Serialize, Deserialize, AsRefStr, PartialEq)]
326pub enum CreateTableState {
327    /// Prepares to create the table
328    Prepare,
329    /// Creates regions on the Datanode
330    DatanodeCreateRegions,
331    /// Creates metadata
332    CreateMetadata,
333}
334
335#[derive(Debug, Serialize, Deserialize)]
336pub struct CreateTableData {
337    pub state: CreateTableState,
338    pub task: CreateTableTask,
339    #[serde(default)]
340    pub column_metadatas: Vec<ColumnMetadata>,
341    /// None stands for not allocated yet.
342    table_route: Option<PhysicalTableRouteValue>,
343    /// None stands for not allocated yet.
344    pub region_wal_options: Option<HashMap<RegionNumber, String>>,
345}
346
347impl CreateTableData {
348    pub fn new(task: CreateTableTask) -> Self {
349        CreateTableData {
350            state: CreateTableState::Prepare,
351            column_metadatas: vec![],
352            task,
353            table_route: None,
354            region_wal_options: None,
355        }
356    }
357
358    fn table_ref(&self) -> TableReference<'_> {
359        self.task.table_ref()
360    }
361}
common_meta/ddl/create_table.rs

common_meta/ddl/
create_table.rs
