common_meta/ddl/
create_logical_tables.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
15mod check;
16mod metadata;
17mod region_request;
18mod update_metadata;
19
20use api::region::RegionResponse;
21use api::v1::CreateTableExpr;
22use async_trait::async_trait;
23use common_catalog::consts::METRIC_ENGINE;
24use common_event_recorder::Event;
25use common_procedure::error::{FromJsonSnafu, Result as ProcedureResult, ToJsonSnafu};
26use common_procedure::{
27    Context as ProcedureContext, EventContext, EventTrigger, LockKey, Procedure, ProcedureState,
28    Status,
29};
30use common_telemetry::{debug, error, warn};
31use futures::future;
32pub use region_request::create_region_request_builder;
33use serde::{Deserialize, Serialize};
34use snafu::ResultExt;
35use store_api::metadata::ColumnMetadata;
36use store_api::metric_engine_consts::ALTER_PHYSICAL_EXTENSION_KEY;
37use store_api::storage::RegionNumber;
38use strum::AsRefStr;
39use table::metadata::{TableId, TableInfo};
40
41use crate::ddl::DdlContext;
42use crate::ddl::event::table::{TableDdlEvent, TableDdlEventType, TableDdlLocator};
43use crate::ddl::utils::{
44    add_peer_context_if_needed, extract_column_metadatas, map_to_procedure_error,
45    sync_follower_regions,
46};
47use crate::error::Result;
48use crate::key::table_route::TableRouteValue;
49use crate::lock_key::{CatalogLock, SchemaLock, TableLock, TableNameLock};
50use crate::metrics;
51use crate::rpc::ddl::CreateTableTask;
52use crate::rpc::router::{RegionRoute, find_leaders};
53
54pub struct CreateLogicalTablesProcedure {
55    pub context: DdlContext,
56    pub data: CreateTablesData,
57}
58
59impl CreateLogicalTablesProcedure {
60    pub const TYPE_NAME: &'static str = "metasrv-procedure::CreateLogicalTables";
61
62    pub fn new(
63        tasks: Vec<CreateTableTask>,
64        physical_table_id: TableId,
65        context: DdlContext,
66    ) -> Self {
67        Self {
68            context,
69            data: CreateTablesData {
70                state: CreateTablesState::Prepare,
71                tasks,
72                table_ids_already_exists: vec![],
73                physical_table_id,
74                physical_region_numbers: vec![],
75                physical_columns: vec![],
76                physical_partition_columns: vec![],
77            },
78        }
79    }
80
81    pub fn from_json(json: &str, context: DdlContext) -> ProcedureResult<Self> {
82        let data = serde_json::from_str(json).context(FromJsonSnafu)?;
83        Ok(Self { context, data })
84    }
85
86    /// On the prepares step, it performs:
87    /// - Checks whether physical table exists.
88    /// - Checks whether logical tables exist.
89    /// - Allocates the table ids.
90    /// - Modify tasks to sort logical columns on their names.
91    ///
92    /// Abort(non-retry):
93    /// - The physical table does not exist.
94    /// - Failed to check whether tables exist.
95    /// - One of logical tables has existing, and the table creation task without setting `create_if_not_exists`.
96    pub(crate) async fn on_prepare(&mut self) -> Result<Status> {
97        self.check_input_tasks()?;
98        // Sets physical region numbers
99        self.fill_physical_table_info().await?;
100        // Add partition columns from physical table to logical table schemas
101        self.merge_partition_columns_into_logical_tables()?;
102        // Checks if the tables exist
103        self.check_tables_already_exist().await?;
104
105        // If all tables already exist, returns the table_ids.
106        if self
107            .data
108            .table_ids_already_exists
109            .iter()
110            .all(Option::is_some)
111        {
112            return Ok(Status::done_with_output(
113                self.data
114                    .table_ids_already_exists
115                    .drain(..)
116                    .flatten()
117                    .collect::<Vec<_>>(),
118            ));
119        }
120
121        // Allocates table ids and sort columns on their names.
122        self.allocate_table_ids().await?;
123
124        self.data.state = CreateTablesState::DatanodeCreateRegions;
125        Ok(Status::executing(true))
126    }
127
128    pub async fn on_datanode_create_regions(&mut self) -> Result<Status> {
129        let (_, physical_table_route) = self
130            .context
131            .table_metadata_manager
132            .table_route_manager()
133            .get_physical_table_route(self.data.physical_table_id)
134            .await?;
135
136        self.create_regions(&physical_table_route.region_routes)
137            .await
138    }
139
140    /// Creates table metadata for logical tables and update corresponding physical
141    /// table's metadata.
142    ///
143    /// Abort(not-retry):
144    /// - Failed to create table metadata.
145    pub async fn on_create_metadata(&mut self) -> Result<Status> {
146        self.update_physical_table_metadata().await?;
147        let table_ids = self.create_logical_tables_metadata().await?;
148
149        Ok(Status::done_with_output(table_ids))
150    }
151
152    async fn create_regions(&mut self, region_routes: &[RegionRoute]) -> Result<Status> {
153        let leaders = find_leaders(region_routes);
154        let mut create_region_tasks = Vec::with_capacity(leaders.len());
155
156        for peer in leaders {
157            let requester = self.context.node_manager.datanode(&peer).await;
158            let Some(request) = self.make_request(&peer, region_routes)? else {
159                debug!("no region request to send to datanode {}", peer);
160                // We can skip the rest of the datanodes,
161                // the rest of the datanodes should have the same result.
162                break;
163            };
164
165            create_region_tasks.push(async move {
166                requester
167                    .handle(request)
168                    .await
169                    .map_err(add_peer_context_if_needed(peer))
170            });
171        }
172
173        let mut results = future::join_all(create_region_tasks)
174            .await
175            .into_iter()
176            .collect::<Result<Vec<_>>>()?;
177
178        if let Some(column_metadatas) =
179            extract_column_metadatas(&mut results, ALTER_PHYSICAL_EXTENSION_KEY)?
180        {
181            self.data.physical_columns = column_metadatas;
182        } else {
183            warn!(
184                "creating logical table result doesn't contains extension key `{ALTER_PHYSICAL_EXTENSION_KEY}`,leaving the physical table's schema unchanged"
185            );
186        }
187
188        self.submit_sync_region_requests(&results, region_routes)
189            .await;
190        self.data.state = CreateTablesState::CreateMetadata;
191        Ok(Status::executing(true))
192    }
193
194    async fn submit_sync_region_requests(
195        &self,
196        results: &[RegionResponse],
197        region_routes: &[RegionRoute],
198    ) {
199        if let Err(err) = sync_follower_regions(
200            &self.context,
201            self.data.physical_table_id,
202            results,
203            region_routes,
204            METRIC_ENGINE,
205        )
206        .await
207        {
208            error!(err; "Failed to sync regions for physical table_id: {}",self.data.physical_table_id);
209        }
210    }
211}
212
213impl CreateLogicalTablesProcedure {
214    fn event_locators(&self) -> impl Iterator<Item = TableDdlLocator> + '_ {
215        self.data.tasks.iter().map(|task| {
216            TableDdlLocator::new(
217                &task.create_table.catalog_name,
218                &task.create_table.schema_name,
219                &task.create_table.table_name,
220            )
221            .with_physical_table_id(self.data.physical_table_id)
222        })
223    }
224}
225
226#[async_trait]
227impl Procedure for CreateLogicalTablesProcedure {
228    fn type_name(&self) -> &str {
229        Self::TYPE_NAME
230    }
231
232    async fn execute(&mut self, _ctx: &ProcedureContext) -> ProcedureResult<Status> {
233        let state = &self.data.state;
234
235        let _timer = metrics::METRIC_META_PROCEDURE_CREATE_TABLES
236            .with_label_values(&[state.as_ref()])
237            .start_timer();
238
239        match state {
240            CreateTablesState::Prepare => self.on_prepare().await,
241            CreateTablesState::DatanodeCreateRegions => self.on_datanode_create_regions().await,
242            CreateTablesState::CreateMetadata => self.on_create_metadata().await,
243        }
244        .map_err(map_to_procedure_error)
245    }
246
247    fn dump(&self) -> ProcedureResult<String> {
248        serde_json::to_string(&self.data).context(ToJsonSnafu)
249    }
250
251    fn lock_key(&self) -> LockKey {
252        // CatalogLock, SchemaLock,
253        // TableLock
254        // TableNameLock(s)
255        let mut lock_key = Vec::with_capacity(2 + 1 + self.data.tasks.len());
256        let table_ref = self.data.tasks[0].table_ref();
257        lock_key.push(CatalogLock::Read(table_ref.catalog).into());
258        lock_key.push(SchemaLock::read(table_ref.catalog, table_ref.schema).into());
259        lock_key.push(TableLock::Write(self.data.physical_table_id).into());
260
261        for task in &self.data.tasks {
262            lock_key.push(
263                TableNameLock::new(
264                    &task.create_table.catalog_name,
265                    &task.create_table.schema_name,
266                    &task.create_table.table_name,
267                )
268                .into(),
269            );
270        }
271        LockKey::new(lock_key)
272    }
273
274    fn event(&self, ctx: &EventContext<'_>) -> Option<Box<dyn Event>> {
275        if !ctx
276            .event_type_filter
277            .allows(TableDdlEventType::CreateLogicalTables.as_str())
278        {
279            return None;
280        }
281        let event = match &ctx.trigger {
282            EventTrigger::Submitted => TableDdlEvent::create_logical_tables_submitted(
283                self.event_locators(),
284                self.data.tasks.len(),
285            ),
286            EventTrigger::Succeeded => match ctx.lifecycle_state {
287                ProcedureState::Done {
288                    output: Some(output),
289                } => output
290                    .downcast_ref::<Vec<TableId>>()
291                    .map(|table_ids| {
292                        debug_assert_eq!(self.data.tasks.len(), table_ids.len());
293                        let locators = self
294                            .event_locators()
295                            .zip(table_ids)
296                            .map(|(locator, table_id)| locator.with_table_id(*table_id));
297                        TableDdlEvent::create_logical_tables_succeeded(locators)
298                    })
299                    .unwrap_or_else(|| {
300                        TableDdlEvent::lifecycle(
301                            TableDdlEventType::CreateLogicalTables,
302                            self.event_locators(),
303                        )
304                    }),
305                _ => TableDdlEvent::lifecycle(
306                    TableDdlEventType::CreateLogicalTables,
307                    self.event_locators(),
308                ),
309            },
310            _ => TableDdlEvent::lifecycle(
311                TableDdlEventType::CreateLogicalTables,
312                self.event_locators(),
313            ),
314        };
315
316        Some(Box::new(event))
317    }
318}
319
320#[derive(Debug, Serialize, Deserialize)]
321pub struct CreateTablesData {
322    state: CreateTablesState,
323    tasks: Vec<CreateTableTask>,
324    table_ids_already_exists: Vec<Option<TableId>>,
325    physical_table_id: TableId,
326    physical_region_numbers: Vec<RegionNumber>,
327    physical_columns: Vec<ColumnMetadata>,
328    physical_partition_columns: Vec<String>,
329}
330
331impl CreateTablesData {
332    pub fn state(&self) -> &CreateTablesState {
333        &self.state
334    }
335
336    fn all_create_table_exprs(&self) -> Vec<&CreateTableExpr> {
337        self.tasks
338            .iter()
339            .map(|task| &task.create_table)
340            .collect::<Vec<_>>()
341    }
342
343    /// Returns the remaining tasks.
344    /// The length of tasks must be greater than 0.
345    fn remaining_tasks(&self) -> Vec<(TableInfo, TableRouteValue)> {
346        self.tasks
347            .iter()
348            .zip(self.table_ids_already_exists.iter())
349            .flat_map(|(task, table_id)| {
350                if table_id.is_none() {
351                    let table_info = task.table_info.clone();
352                    let table_route = TableRouteValue::logical(self.physical_table_id);
353                    Some((table_info, table_route))
354                } else {
355                    None
356                }
357            })
358            .collect::<Vec<_>>()
359    }
360}
361
362#[derive(Debug, Clone, Serialize, Deserialize, AsRefStr)]
363pub enum CreateTablesState {
364    /// Prepares to create the tables
365    Prepare,
366    /// Creates regions on the Datanode
367    DatanodeCreateRegions,
368    /// Creates metadata
369    CreateMetadata,
370}
common_meta/ddl/create_logical_tables.rs

common_meta/ddl/
create_logical_tables.rs
