common_meta/ddl/
alter_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 executor;
16mod update_metadata;
17mod validator;
18
19use api::region::RegionResponse;
20use async_trait::async_trait;
21use common_catalog::format_full_table_name;
22use common_procedure::error::{FromJsonSnafu, Result as ProcedureResult, ToJsonSnafu};
23use common_procedure::{Context, EventContext, EventTrigger, LockKey, Procedure, Status};
24use common_telemetry::{debug, error, info, warn};
25pub use executor::make_alter_region_request;
26use serde::{Deserialize, Serialize};
27use snafu::{ResultExt, ensure};
28use store_api::metadata::ColumnMetadata;
29use store_api::metric_engine_consts::ALTER_PHYSICAL_EXTENSION_KEY;
30use strum::AsRefStr;
31use table::metadata::TableId;
32
33use crate::cache_invalidator::Context as CacheContext;
34use crate::ddl::DdlContext;
35use crate::ddl::alter_logical_tables::executor::AlterLogicalTablesExecutor;
36use crate::ddl::alter_logical_tables::validator::{
37    AlterLogicalTableValidator, ValidatorResult, retain_unskipped,
38};
39use crate::ddl::event::table::{
40    TableDdlEvent, TableDdlEventType, TableDdlLocator, alter_table_kind_name,
41};
42use crate::ddl::utils::{extract_column_metadatas, map_to_procedure_error, sync_follower_regions};
43use crate::error::{self, Result};
44use crate::instruction::CacheIdent;
45use crate::key::DeserializedValueWithBytes;
46use crate::key::table_info::TableInfoValue;
47use crate::key::table_route::PhysicalTableRouteValue;
48use crate::lock_key::{CatalogLock, SchemaLock, TableLock};
49use crate::metrics;
50use crate::rpc::ddl::AlterTableTask;
51use crate::rpc::router::RegionRoute;
52
53pub struct AlterLogicalTablesProcedure {
54    pub context: DdlContext,
55    pub data: AlterTablesData,
56    /// Physical table route cache.
57    pub physical_table_route: Option<PhysicalTableRouteValue>,
58}
59
60/// Builds the validator from the [`AlterTablesData`].
61fn build_validator_from_alter_table_data<'a>(
62    data: &'a AlterTablesData,
63) -> AlterLogicalTableValidator<'a> {
64    let physical_table_id = data.physical_table_id;
65    let alters = data
66        .tasks
67        .iter()
68        .map(|task| &task.alter_table)
69        .collect::<Vec<_>>();
70    AlterLogicalTableValidator::new(physical_table_id, alters)
71}
72
73/// Builds the executor from the [`AlterTablesData`].
74fn build_executor_from_alter_expr<'a>(data: &'a AlterTablesData) -> AlterLogicalTablesExecutor<'a> {
75    debug_assert_eq!(data.tasks.len(), data.table_info_values.len());
76    let alters = data
77        .tasks
78        .iter()
79        .zip(data.table_info_values.iter())
80        .map(|(task, table_info)| (table_info.table_info.ident.table_id, &task.alter_table))
81        .collect::<Vec<_>>();
82    AlterLogicalTablesExecutor::new(alters)
83}
84
85impl AlterLogicalTablesProcedure {
86    pub const TYPE_NAME: &'static str = "metasrv-procedure::AlterLogicalTables";
87
88    pub fn new(
89        tasks: Vec<AlterTableTask>,
90        physical_table_id: TableId,
91        logical_table_ids: Vec<TableId>,
92        context: DdlContext,
93    ) -> Self {
94        Self {
95            context,
96            data: AlterTablesData {
97                state: AlterTablesState::Prepare,
98                tasks,
99                table_info_values: vec![],
100                physical_table_id,
101                logical_table_ids,
102                physical_table_info: None,
103                physical_columns: vec![],
104                table_cache_keys_to_invalidate: vec![],
105            },
106            physical_table_route: None,
107        }
108    }
109
110    pub fn from_json(json: &str, context: DdlContext) -> ProcedureResult<Self> {
111        let data = serde_json::from_str(json).context(FromJsonSnafu)?;
112        Ok(Self {
113            context,
114            data,
115            physical_table_route: None,
116        })
117    }
118
119    pub(crate) async fn on_prepare(&mut self) -> Result<Status> {
120        let validator = build_validator_from_alter_table_data(&self.data);
121        let ValidatorResult {
122            num_skipped,
123            skip_alter,
124            table_info_values,
125            physical_table_info,
126            physical_table_route,
127        } = validator
128            .validate(&self.context.table_metadata_manager)
129            .await?;
130
131        let num_tasks = self.data.tasks.len();
132        if num_skipped == num_tasks {
133            info!("All the alter tasks are finished, will skip the procedure.");
134            let cache_ident_keys = AlterLogicalTablesExecutor::build_cache_ident_keys(
135                &physical_table_info,
136                &table_info_values
137                    .iter()
138                    .map(|v| v.get_inner_ref())
139                    .collect::<Vec<_>>(),
140            );
141            self.data.table_cache_keys_to_invalidate = cache_ident_keys;
142            // Re-invalidate the table cache
143            self.data.state = AlterTablesState::InvalidateTableCache;
144            return Ok(Status::executing(true));
145        } else if num_skipped > 0 {
146            info!(
147                "There are {} alter tasks, {} of them were already finished.",
148                num_tasks, num_skipped
149            );
150        }
151
152        // Locks are fixed at submission. A logical table that resolves to an
153        // id outside the locked set here was dropped and recreated in between,
154        // so this procedure holds no lock for it. Procedures restored from
155        // pre-upgrade state have no recorded ids and keep the old behavior.
156        if !self.data.logical_table_ids.is_empty() {
157            for value in &table_info_values {
158                let table_id = value.get_inner_ref().table_info.ident.table_id;
159                ensure!(
160                    self.data.logical_table_ids.contains(&table_id),
161                    error::UnexpectedSnafu {
162                        err_msg: format!(
163                            "logical table {} (id {table_id}) is not covered by the \
164                             procedure locks; retry the statement",
165                            value.get_inner_ref().table_info.name
166                        ),
167                    }
168                );
169            }
170        }
171
172        // Updates the procedure state.
173        retain_unskipped(&mut self.data.tasks, &skip_alter);
174        self.data.physical_table_info = Some(physical_table_info);
175        self.data.table_info_values = table_info_values;
176        debug_assert_eq!(self.data.tasks.len(), self.data.table_info_values.len());
177        self.physical_table_route = Some(physical_table_route);
178        self.data.state = AlterTablesState::SubmitAlterRegionRequests;
179        Ok(Status::executing(true))
180    }
181
182    pub(crate) async fn on_submit_alter_region_requests(&mut self) -> Result<Status> {
183        self.fetch_physical_table_route_if_non_exist().await?;
184        // Safety: fetched in `fetch_physical_table_route_if_non_exist`.
185        let region_routes = &self.physical_table_route.as_ref().unwrap().region_routes;
186
187        let executor = build_executor_from_alter_expr(&self.data);
188        let mut results = executor
189            .on_alter_regions(
190                &self.context.node_manager,
191                // Avoid double-borrowing self by extracting the region_routes first
192                region_routes,
193            )
194            .await?;
195
196        if let Some(column_metadatas) =
197            extract_column_metadatas(&mut results, ALTER_PHYSICAL_EXTENSION_KEY)?
198        {
199            self.data.physical_columns = column_metadatas;
200        } else {
201            warn!(
202                "altering logical table result doesn't contains extension key `{ALTER_PHYSICAL_EXTENSION_KEY}`,leaving the physical table's schema unchanged"
203            );
204        }
205        self.submit_sync_region_requests(results, region_routes)
206            .await;
207        self.data.state = AlterTablesState::UpdateMetadata;
208        Ok(Status::executing(true))
209    }
210
211    async fn submit_sync_region_requests(
212        &self,
213        results: Vec<RegionResponse>,
214        region_routes: &[RegionRoute],
215    ) {
216        let table_info = &self.data.physical_table_info.as_ref().unwrap().table_info;
217        if let Err(err) = sync_follower_regions(
218            &self.context,
219            self.data.physical_table_id,
220            &results,
221            region_routes,
222            table_info.meta.engine.as_str(),
223        )
224        .await
225        {
226            error!(err; "Failed to sync regions for table {}, table_id: {}",
227                        format_full_table_name(&table_info.catalog_name, &table_info.schema_name, &table_info.name),
228                        self.data.physical_table_id
229            );
230        }
231    }
232
233    pub(crate) async fn on_update_metadata(&mut self) -> Result<Status> {
234        self.update_physical_table_metadata().await?;
235        self.update_logical_tables_metadata().await?;
236
237        let logical_table_info_values = self
238            .data
239            .table_info_values
240            .iter()
241            .map(|v| v.get_inner_ref())
242            .collect::<Vec<_>>();
243
244        let cache_ident_keys = AlterLogicalTablesExecutor::build_cache_ident_keys(
245            self.data.physical_table_info.as_ref().unwrap(),
246            &logical_table_info_values,
247        );
248        self.data.table_cache_keys_to_invalidate = cache_ident_keys;
249        self.data.clear_metadata_fields();
250
251        self.data.state = AlterTablesState::InvalidateTableCache;
252        Ok(Status::executing(true))
253    }
254
255    pub(crate) async fn on_invalidate_table_cache(&mut self) -> Result<Status> {
256        let to_invalidate = &self.data.table_cache_keys_to_invalidate;
257
258        let ctx = CacheContext {
259            subject: Some(format!(
260                "Invalidate table cache by altering logical tables, physical_table_id: {}",
261                self.data.physical_table_id,
262            )),
263        };
264
265        self.context
266            .cache_invalidator
267            .invalidate(&ctx, to_invalidate)
268            .await?;
269        Ok(Status::done())
270    }
271
272    /// Fetches the physical table route if it is not already fetched.
273    async fn fetch_physical_table_route_if_non_exist(&mut self) -> Result<()> {
274        if self.physical_table_route.is_none() {
275            let (_, physical_table_route) = self
276                .context
277                .table_metadata_manager
278                .table_route_manager()
279                .get_physical_table_route(self.data.physical_table_id)
280                .await?;
281            self.physical_table_route = Some(physical_table_route);
282        }
283
284        Ok(())
285    }
286}
287
288#[async_trait]
289impl Procedure for AlterLogicalTablesProcedure {
290    fn type_name(&self) -> &str {
291        Self::TYPE_NAME
292    }
293
294    async fn execute(&mut self, _ctx: &Context) -> ProcedureResult<Status> {
295        let state = &self.data.state;
296
297        let step = state.as_ref();
298
299        let _timer = metrics::METRIC_META_PROCEDURE_ALTER_TABLE
300            .with_label_values(&[step])
301            .start_timer();
302        debug!(
303            "Executing alter logical tables procedure, state: {:?}",
304            state
305        );
306
307        match state {
308            AlterTablesState::Prepare => self.on_prepare().await,
309            AlterTablesState::SubmitAlterRegionRequests => {
310                self.on_submit_alter_region_requests().await
311            }
312            AlterTablesState::UpdateMetadata => self.on_update_metadata().await,
313            AlterTablesState::InvalidateTableCache => self.on_invalidate_table_cache().await,
314        }
315        .inspect_err(|_| {
316            // Reset the physical table route cache.
317            self.physical_table_route = None;
318        })
319        .map_err(map_to_procedure_error)
320    }
321
322    fn dump(&self) -> ProcedureResult<String> {
323        serde_json::to_string(&self.data).context(ToJsonSnafu)
324    }
325
326    fn lock_key(&self) -> LockKey {
327        // CatalogLock, SchemaLock,
328        // TableLock
329        // TableNameLock(s)
330        let mut lock_key = Vec::with_capacity(2 + 1 + self.data.logical_table_ids.len());
331        let table_ref = self.data.tasks[0].table_ref();
332        lock_key.push(CatalogLock::Read(table_ref.catalog).into());
333        lock_key.push(SchemaLock::read(table_ref.catalog, table_ref.schema).into());
334        lock_key.push(TableLock::Write(self.data.physical_table_id).into());
335        if self.data.logical_table_ids.is_empty() {
336            // Pre-upgrade procedure state has no `logical_table_ids`, but a
337            // dump taken after `Prepare` still carries the resolved table
338            // snapshots — recover the logical locks from them.
339            lock_key.extend(
340                self.data
341                    .table_info_values
342                    .iter()
343                    .map(|table| TableLock::Write(table.table_info.ident.table_id).into()),
344            );
345        } else {
346            lock_key.extend(
347                self.data
348                    .logical_table_ids
349                    .iter()
350                    .map(|table_id| TableLock::Write(*table_id).into()),
351            );
352        }
353
354        LockKey::new(lock_key)
355    }
356
357    fn event(&self, ctx: &EventContext<'_>) -> Option<Box<dyn common_event_recorder::Event>> {
358        if !ctx
359            .event_type_filter
360            .allows(TableDdlEventType::AlterLogicalTables.as_str())
361        {
362            return None;
363        }
364        if ctx.trigger != EventTrigger::Submitted {
365            return Some(Box::new(TableDdlEvent::lifecycle(
366                TableDdlEventType::AlterLogicalTables,
367                self.data.tasks.iter().map(|task| {
368                    let table_ref = task.table_ref();
369                    TableDdlLocator::new(table_ref.catalog, table_ref.schema, table_ref.table)
370                        .with_physical_table_id(self.data.physical_table_id)
371                }),
372            )));
373        }
374
375        let locators = self.data.tasks.iter().map(|task| {
376            let table_ref = task.table_ref();
377            TableDdlLocator::new(table_ref.catalog, table_ref.schema, table_ref.table)
378                .with_physical_table_id(self.data.physical_table_id)
379        });
380        let kinds = self
381            .data
382            .tasks
383            .iter()
384            .filter_map(|task| task.alter_table.kind.as_ref())
385            .filter_map(alter_table_kind_name);
386        Some(Box::new(TableDdlEvent::alter_logical_tables_submitted(
387            locators,
388            self.data.tasks.len(),
389            kinds,
390        )))
391    }
392}
393
394#[derive(Debug, Serialize, Deserialize)]
395pub struct AlterTablesData {
396    state: AlterTablesState,
397    tasks: Vec<AlterTableTask>,
398    /// Table info values before the alter operation.
399    /// Corresponding one-to-one with the AlterTableTask in tasks.
400    table_info_values: Vec<DeserializedValueWithBytes<TableInfoValue>>,
401    /// Physical table info
402    physical_table_id: TableId,
403    /// Logical table ids resolved at submission time, so `lock_key` can name
404    /// them before `Prepare` runs (procedure locks are fixed at submission).
405    /// Empty when restored from pre-upgrade procedure state.
406    #[serde(default)]
407    logical_table_ids: Vec<TableId>,
408    physical_table_info: Option<DeserializedValueWithBytes<TableInfoValue>>,
409    physical_columns: Vec<ColumnMetadata>,
410    table_cache_keys_to_invalidate: Vec<CacheIdent>,
411}
412
413impl AlterTablesData {
414    /// Clears metadata snapshots after the update metadata step.
415    ///
416    /// Keep the tasks and physical table ID until the procedure finishes: lifecycle
417    /// events use them to retain the logical table locator.
418    fn clear_metadata_fields(&mut self) {
419        self.table_info_values.clear();
420        self.physical_table_info = None;
421        self.physical_columns.clear();
422    }
423}
424
425#[derive(Debug, Serialize, Deserialize, AsRefStr)]
426enum AlterTablesState {
427    /// Prepares to alter the table
428    Prepare,
429    SubmitAlterRegionRequests,
430    /// Updates table metadata.
431    UpdateMetadata,
432    /// Broadcasts the invalidating table cache instruction.
433    InvalidateTableCache,
434}
common_meta/ddl/alter_logical_tables.rs

common_meta/ddl/
alter_logical_tables.rs
