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;
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::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        context: DdlContext,
92    ) -> Self {
93        Self {
94            context,
95            data: AlterTablesData {
96                state: AlterTablesState::Prepare,
97                tasks,
98                table_info_values: vec![],
99                physical_table_id,
100                physical_table_info: None,
101                physical_columns: vec![],
102                table_cache_keys_to_invalidate: vec![],
103            },
104            physical_table_route: None,
105        }
106    }
107
108    pub fn from_json(json: &str, context: DdlContext) -> ProcedureResult<Self> {
109        let data = serde_json::from_str(json).context(FromJsonSnafu)?;
110        Ok(Self {
111            context,
112            data,
113            physical_table_route: None,
114        })
115    }
116
117    pub(crate) async fn on_prepare(&mut self) -> Result<Status> {
118        let validator = build_validator_from_alter_table_data(&self.data);
119        let ValidatorResult {
120            num_skipped,
121            skip_alter,
122            table_info_values,
123            physical_table_info,
124            physical_table_route,
125        } = validator
126            .validate(&self.context.table_metadata_manager)
127            .await?;
128
129        let num_tasks = self.data.tasks.len();
130        if num_skipped == num_tasks {
131            info!("All the alter tasks are finished, will skip the procedure.");
132            let cache_ident_keys = AlterLogicalTablesExecutor::build_cache_ident_keys(
133                &physical_table_info,
134                &table_info_values
135                    .iter()
136                    .map(|v| v.get_inner_ref())
137                    .collect::<Vec<_>>(),
138            );
139            self.data.table_cache_keys_to_invalidate = cache_ident_keys;
140            // Re-invalidate the table cache
141            self.data.state = AlterTablesState::InvalidateTableCache;
142            return Ok(Status::executing(true));
143        } else if num_skipped > 0 {
144            info!(
145                "There are {} alter tasks, {} of them were already finished.",
146                num_tasks, num_skipped
147            );
148        }
149
150        // Updates the procedure state.
151        retain_unskipped(&mut self.data.tasks, &skip_alter);
152        self.data.physical_table_info = Some(physical_table_info);
153        self.data.table_info_values = table_info_values;
154        debug_assert_eq!(self.data.tasks.len(), self.data.table_info_values.len());
155        self.physical_table_route = Some(physical_table_route);
156        self.data.state = AlterTablesState::SubmitAlterRegionRequests;
157        Ok(Status::executing(true))
158    }
159
160    pub(crate) async fn on_submit_alter_region_requests(&mut self) -> Result<Status> {
161        self.fetch_physical_table_route_if_non_exist().await?;
162        // Safety: fetched in `fetch_physical_table_route_if_non_exist`.
163        let region_routes = &self.physical_table_route.as_ref().unwrap().region_routes;
164
165        let executor = build_executor_from_alter_expr(&self.data);
166        let mut results = executor
167            .on_alter_regions(
168                &self.context.node_manager,
169                // Avoid double-borrowing self by extracting the region_routes first
170                region_routes,
171            )
172            .await?;
173
174        if let Some(column_metadatas) =
175            extract_column_metadatas(&mut results, ALTER_PHYSICAL_EXTENSION_KEY)?
176        {
177            self.data.physical_columns = column_metadatas;
178        } else {
179            warn!(
180                "altering logical table result doesn't contains extension key `{ALTER_PHYSICAL_EXTENSION_KEY}`,leaving the physical table's schema unchanged"
181            );
182        }
183        self.submit_sync_region_requests(results, region_routes)
184            .await;
185        self.data.state = AlterTablesState::UpdateMetadata;
186        Ok(Status::executing(true))
187    }
188
189    async fn submit_sync_region_requests(
190        &self,
191        results: Vec<RegionResponse>,
192        region_routes: &[RegionRoute],
193    ) {
194        let table_info = &self.data.physical_table_info.as_ref().unwrap().table_info;
195        if let Err(err) = sync_follower_regions(
196            &self.context,
197            self.data.physical_table_id,
198            &results,
199            region_routes,
200            table_info.meta.engine.as_str(),
201        )
202        .await
203        {
204            error!(err; "Failed to sync regions for table {}, table_id: {}",
205                        format_full_table_name(&table_info.catalog_name, &table_info.schema_name, &table_info.name),
206                        self.data.physical_table_id
207            );
208        }
209    }
210
211    pub(crate) async fn on_update_metadata(&mut self) -> Result<Status> {
212        self.update_physical_table_metadata().await?;
213        self.update_logical_tables_metadata().await?;
214
215        let logical_table_info_values = self
216            .data
217            .table_info_values
218            .iter()
219            .map(|v| v.get_inner_ref())
220            .collect::<Vec<_>>();
221
222        let cache_ident_keys = AlterLogicalTablesExecutor::build_cache_ident_keys(
223            self.data.physical_table_info.as_ref().unwrap(),
224            &logical_table_info_values,
225        );
226        self.data.table_cache_keys_to_invalidate = cache_ident_keys;
227        self.data.clear_metadata_fields();
228
229        self.data.state = AlterTablesState::InvalidateTableCache;
230        Ok(Status::executing(true))
231    }
232
233    pub(crate) async fn on_invalidate_table_cache(&mut self) -> Result<Status> {
234        let to_invalidate = &self.data.table_cache_keys_to_invalidate;
235
236        let ctx = CacheContext {
237            subject: Some(format!(
238                "Invalidate table cache by altering logical tables, physical_table_id: {}",
239                self.data.physical_table_id,
240            )),
241        };
242
243        self.context
244            .cache_invalidator
245            .invalidate(&ctx, to_invalidate)
246            .await?;
247        Ok(Status::done())
248    }
249
250    /// Fetches the physical table route if it is not already fetched.
251    async fn fetch_physical_table_route_if_non_exist(&mut self) -> Result<()> {
252        if self.physical_table_route.is_none() {
253            let (_, physical_table_route) = self
254                .context
255                .table_metadata_manager
256                .table_route_manager()
257                .get_physical_table_route(self.data.physical_table_id)
258                .await?;
259            self.physical_table_route = Some(physical_table_route);
260        }
261
262        Ok(())
263    }
264}
265
266#[async_trait]
267impl Procedure for AlterLogicalTablesProcedure {
268    fn type_name(&self) -> &str {
269        Self::TYPE_NAME
270    }
271
272    async fn execute(&mut self, _ctx: &Context) -> ProcedureResult<Status> {
273        let state = &self.data.state;
274
275        let step = state.as_ref();
276
277        let _timer = metrics::METRIC_META_PROCEDURE_ALTER_TABLE
278            .with_label_values(&[step])
279            .start_timer();
280        debug!(
281            "Executing alter logical tables procedure, state: {:?}",
282            state
283        );
284
285        match state {
286            AlterTablesState::Prepare => self.on_prepare().await,
287            AlterTablesState::SubmitAlterRegionRequests => {
288                self.on_submit_alter_region_requests().await
289            }
290            AlterTablesState::UpdateMetadata => self.on_update_metadata().await,
291            AlterTablesState::InvalidateTableCache => self.on_invalidate_table_cache().await,
292        }
293        .inspect_err(|_| {
294            // Reset the physical table route cache.
295            self.physical_table_route = None;
296        })
297        .map_err(map_to_procedure_error)
298    }
299
300    fn dump(&self) -> ProcedureResult<String> {
301        serde_json::to_string(&self.data).context(ToJsonSnafu)
302    }
303
304    fn lock_key(&self) -> LockKey {
305        // CatalogLock, SchemaLock,
306        // TableLock
307        // TableNameLock(s)
308        let mut lock_key = Vec::with_capacity(2 + 1 + self.data.tasks.len());
309        let table_ref = self.data.tasks[0].table_ref();
310        lock_key.push(CatalogLock::Read(table_ref.catalog).into());
311        lock_key.push(SchemaLock::read(table_ref.catalog, table_ref.schema).into());
312        lock_key.push(TableLock::Write(self.data.physical_table_id).into());
313        lock_key.extend(
314            self.data
315                .table_info_values
316                .iter()
317                .map(|table| TableLock::Write(table.table_info.ident.table_id).into()),
318        );
319
320        LockKey::new(lock_key)
321    }
322
323    fn event(&self, ctx: &EventContext<'_>) -> Option<Box<dyn common_event_recorder::Event>> {
324        if !ctx
325            .event_type_filter
326            .allows(TableDdlEventType::AlterLogicalTables.as_str())
327        {
328            return None;
329        }
330        if ctx.trigger != EventTrigger::Submitted {
331            return Some(Box::new(TableDdlEvent::lifecycle(
332                TableDdlEventType::AlterLogicalTables,
333                self.data.tasks.iter().map(|task| {
334                    let table_ref = task.table_ref();
335                    TableDdlLocator::new(table_ref.catalog, table_ref.schema, table_ref.table)
336                        .with_physical_table_id(self.data.physical_table_id)
337                }),
338            )));
339        }
340
341        let locators = self.data.tasks.iter().map(|task| {
342            let table_ref = task.table_ref();
343            TableDdlLocator::new(table_ref.catalog, table_ref.schema, table_ref.table)
344                .with_physical_table_id(self.data.physical_table_id)
345        });
346        let kinds = self
347            .data
348            .tasks
349            .iter()
350            .filter_map(|task| task.alter_table.kind.as_ref())
351            .filter_map(alter_table_kind_name);
352        Some(Box::new(TableDdlEvent::alter_logical_tables_submitted(
353            locators,
354            self.data.tasks.len(),
355            kinds,
356        )))
357    }
358}
359
360#[derive(Debug, Serialize, Deserialize)]
361pub struct AlterTablesData {
362    state: AlterTablesState,
363    tasks: Vec<AlterTableTask>,
364    /// Table info values before the alter operation.
365    /// Corresponding one-to-one with the AlterTableTask in tasks.
366    table_info_values: Vec<DeserializedValueWithBytes<TableInfoValue>>,
367    /// Physical table info
368    physical_table_id: TableId,
369    physical_table_info: Option<DeserializedValueWithBytes<TableInfoValue>>,
370    physical_columns: Vec<ColumnMetadata>,
371    table_cache_keys_to_invalidate: Vec<CacheIdent>,
372}
373
374impl AlterTablesData {
375    /// Clears metadata snapshots after the update metadata step.
376    ///
377    /// Keep the tasks and physical table ID until the procedure finishes: lifecycle
378    /// events use them to retain the logical table locator.
379    fn clear_metadata_fields(&mut self) {
380        self.table_info_values.clear();
381        self.physical_table_info = None;
382        self.physical_columns.clear();
383    }
384}
385
386#[derive(Debug, Serialize, Deserialize, AsRefStr)]
387enum AlterTablesState {
388    /// Prepares to alter the table
389    Prepare,
390    SubmitAlterRegionRequests,
391    /// Updates table metadata.
392    UpdateMetadata,
393    /// Broadcasts the invalidating table cache instruction.
394    InvalidateTableCache,
395}
common_meta/ddl/alter_logical_tables.rs

common_meta/ddl/
alter_logical_tables.rs
