common_meta/ddl/alter_table/
executor.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
15use std::collections::HashMap;
16
17use api::region::RegionResponse;
18use api::v1::AlterTableExpr;
19use api::v1::region::region_request::Body;
20use api::v1::region::{AlterRequest, RegionRequest, RegionRequestHeader, alter_request};
21use common_catalog::format_full_table_name;
22use common_grpc_expr::alter_expr_to_request;
23use common_telemetry::tracing_context::TracingContext;
24use common_telemetry::{debug, info};
25use futures::future;
26use snafu::{ResultExt, ensure};
27use store_api::metadata::ColumnMetadata;
28use store_api::storage::{RegionId, TableId};
29use table::metadata::TableInfo;
30use table::requests::AlterKind;
31use table::table_name::TableName;
32
33use crate::cache_invalidator::{CacheInvalidatorRef, Context};
34use crate::ddl::utils::{add_peer_context_if_needed, raw_table_info};
35use crate::error::{self, Result, UnexpectedSnafu};
36use crate::instruction::CacheIdent;
37use crate::key::table_info::TableInfoValue;
38use crate::key::table_name::TableNameKey;
39use crate::key::{DeserializedValueWithBytes, RegionDistribution, TableMetadataManagerRef};
40use crate::node_manager::NodeManagerRef;
41use crate::rpc::router::{RegionRoute, find_followers, find_leaders, region_distribution};
42
43/// [AlterTableExecutor] performs:
44/// - Alters the metadata of the table.
45/// - Alters regions on the datanode nodes.
46pub struct AlterTableExecutor {
47    table: TableName,
48    table_id: TableId,
49    /// The new table name if the alter kind is rename table.
50    new_table_name: Option<String>,
51}
52
53impl AlterTableExecutor {
54    /// Creates a new [`AlterTableExecutor`].
55    pub fn new(table: TableName, table_id: TableId, new_table_name: Option<String>) -> Self {
56        Self {
57            table,
58            table_id,
59            new_table_name,
60        }
61    }
62
63    /// Prepares to alter the table.
64    ///
65    /// ## Checks:
66    /// - The new table name doesn't exist (rename).
67    /// - Table exists.
68    pub(crate) async fn on_prepare(
69        &self,
70        table_metadata_manager: &TableMetadataManagerRef,
71    ) -> Result<()> {
72        let catalog = &self.table.catalog_name;
73        let schema = &self.table.schema_name;
74        let table_name = &self.table.table_name;
75
76        let manager = table_metadata_manager;
77        if let Some(new_table_name) = &self.new_table_name {
78            let new_table_name_key = TableNameKey::new(catalog, schema, new_table_name);
79            let exists = manager
80                .table_name_manager()
81                .exists(new_table_name_key)
82                .await?;
83            ensure!(
84                !exists,
85                error::TableAlreadyExistsSnafu {
86                    table_name: format_full_table_name(catalog, schema, new_table_name),
87                }
88            )
89        }
90
91        let table_name_key = TableNameKey::new(catalog, schema, table_name);
92        let exists = manager.table_name_manager().exists(table_name_key).await?;
93        ensure!(
94            exists,
95            error::TableNotFoundSnafu {
96                table_name: format_full_table_name(catalog, schema, table_name),
97            }
98        );
99
100        Ok(())
101    }
102
103    /// Validates the alter table expression and builds the new table info.
104    ///
105    /// This validation is performed early to ensure the alteration is valid before
106    /// proceeding to the `on_alter_metadata` state, where regions have already been altered.
107    /// Building the new table info here allows us to catch any issues with the
108    /// alteration before committing metadata changes.
109    pub(crate) fn validate_alter_table_expr(
110        table_info: &TableInfo,
111        alter_table_expr: AlterTableExpr,
112    ) -> Result<TableInfo> {
113        build_new_table_info(table_info, alter_table_expr)
114    }
115
116    /// Updates table metadata for alter table operation.
117    pub(crate) async fn on_alter_metadata(
118        &self,
119        table_metadata_manager: &TableMetadataManagerRef,
120        current_table_info_value: &DeserializedValueWithBytes<TableInfoValue>,
121        region_distribution: Option<&RegionDistribution>,
122        mut raw_table_info: TableInfo,
123        column_metadatas: &[ColumnMetadata],
124        metadata_only_alter: bool,
125    ) -> Result<()> {
126        let table_ref = self.table.table_ref();
127        let table_id = self.table_id;
128
129        if let Some(new_table_name) = &self.new_table_name {
130            debug!(
131                "Starting update table: {} metadata, table_id: {}, new table info: {:?}, new table name: {}",
132                table_ref, table_id, raw_table_info, new_table_name
133            );
134
135            table_metadata_manager
136                .rename_table(current_table_info_value, new_table_name.clone())
137                .await?;
138        } else {
139            debug!(
140                "Starting update table: {} metadata, table_id: {}, new table info: {:?}",
141                table_ref, table_id, raw_table_info
142            );
143
144            ensure!(
145                metadata_only_alter || region_distribution.is_some(),
146                UnexpectedSnafu {
147                    err_msg: "region distribution is not set when updating table metadata",
148                }
149            );
150
151            if !column_metadatas.is_empty() {
152                raw_table_info::update_table_info_column_ids(&mut raw_table_info, column_metadatas);
153            }
154            table_metadata_manager
155                .update_table_info(
156                    current_table_info_value,
157                    region_distribution.cloned(),
158                    raw_table_info,
159                )
160                .await?;
161        }
162
163        Ok(())
164    }
165
166    /// Alters regions on the datanode nodes.
167    pub(crate) async fn on_alter_regions(
168        &self,
169        node_manager: &NodeManagerRef,
170        region_routes: &[RegionRoute],
171        kind: Option<alter_request::Kind>,
172    ) -> Vec<Result<RegionResponse>> {
173        self.dispatch_alter_region_requests(node_manager, region_routes, kind, false)
174            .await
175    }
176
177    /// Alters all replicas for the irreversible skip-WAL flow.
178    pub(crate) async fn on_alter_skip_wal_regions(
179        &self,
180        node_manager: &NodeManagerRef,
181        region_routes: &[RegionRoute],
182        kind: Option<alter_request::Kind>,
183    ) -> Vec<Result<RegionResponse>> {
184        self.dispatch_alter_region_requests(node_manager, region_routes, kind, true)
185            .await
186    }
187
188    async fn dispatch_alter_region_requests(
189        &self,
190        node_manager: &NodeManagerRef,
191        region_routes: &[RegionRoute],
192        kind: Option<alter_request::Kind>,
193        include_followers: bool,
194    ) -> Vec<Result<RegionResponse>> {
195        let region_distribution = region_distribution(region_routes);
196        let mut peers = find_leaders(region_routes)
197            .into_iter()
198            .map(|p| (p.id, p))
199            .collect::<HashMap<_, _>>();
200        if include_followers {
201            peers.extend(find_followers(region_routes).into_iter().map(|p| (p.id, p)));
202        }
203        let total_num_region = region_distribution
204            .values()
205            .map(|r| {
206                r.leader_regions.len()
207                    + if include_followers {
208                        r.follower_regions.len()
209                    } else {
210                        0
211                    }
212            })
213            .sum::<usize>();
214        let mut alter_region_tasks = Vec::with_capacity(total_num_region);
215        for (datanode_id, region_role_set) in region_distribution {
216            let mut region_ids = region_role_set.leader_regions;
217            if include_followers {
218                region_ids.extend(region_role_set.follower_regions);
219            }
220            if region_ids.is_empty() {
221                continue;
222            }
223            // Safety: must exists.
224            let peer = peers.get(&datanode_id).unwrap();
225            let requester = node_manager.datanode(peer).await;
226
227            for region_id in region_ids {
228                let region_id = RegionId::new(self.table_id, region_id);
229                let request = make_alter_region_request(region_id, kind.clone());
230
231                let requester = requester.clone();
232                let peer = peer.clone();
233
234                alter_region_tasks.push(async move {
235                    requester
236                        .handle(request)
237                        .await
238                        .map_err(add_peer_context_if_needed(peer))
239                });
240            }
241        }
242
243        future::join_all(alter_region_tasks)
244            .await
245            .into_iter()
246            .collect::<Vec<_>>()
247    }
248
249    /// Invalidates cache for the table.
250    pub(crate) async fn invalidate_table_cache(
251        &self,
252        cache_invalidator: &CacheInvalidatorRef,
253    ) -> Result<()> {
254        let ctx = Context {
255            subject: Some(format!(
256                "Invalidate table cache by altering table {}, table_id: {}",
257                self.table.table_ref(),
258                self.table_id,
259            )),
260        };
261
262        cache_invalidator
263            .invalidate(
264                &ctx,
265                &[
266                    CacheIdent::TableName(self.table.clone()),
267                    CacheIdent::TableId(self.table_id),
268                ],
269            )
270            .await?;
271
272        Ok(())
273    }
274}
275
276/// Makes alter region request.
277pub(crate) fn make_alter_region_request(
278    region_id: RegionId,
279    kind: Option<alter_request::Kind>,
280) -> RegionRequest {
281    RegionRequest {
282        header: Some(RegionRequestHeader {
283            tracing_context: TracingContext::from_current_span().to_w3c(),
284            ..Default::default()
285        }),
286        body: Some(Body::Alter(AlterRequest {
287            region_id: region_id.as_u64(),
288            kind,
289            ..Default::default()
290        })),
291    }
292}
293
294/// Builds new table info after alteration.
295///
296/// This function creates a new table info by applying the alter table expression
297/// to the existing table info. For add column operations, it increments the
298/// `next_column_id` by the number of columns being added, which may result in gaps
299/// in the column id sequence.
300fn build_new_table_info(
301    table_info: &TableInfo,
302    alter_table_expr: AlterTableExpr,
303) -> Result<TableInfo> {
304    let table_info = table_info.clone();
305    let schema_name = &table_info.schema_name;
306    let catalog_name = &table_info.catalog_name;
307    let table_name = &table_info.name;
308    let table_id = table_info.ident.table_id;
309    let request = alter_expr_to_request(table_id, alter_table_expr, Some(&table_info.meta))
310        .context(error::ConvertAlterTableRequestSnafu)?;
311
312    let new_meta = table_info
313        .meta
314        .builder_with_alter_kind(table_name, &request.alter_kind)
315        .context(error::TableSnafu)?
316        .build()
317        .with_context(|_| error::BuildTableMetaSnafu {
318            table_name: format_full_table_name(catalog_name, schema_name, table_name),
319        })?;
320
321    let mut new_info = table_info.clone();
322    new_info.meta = new_meta;
323    new_info.ident.version = table_info.ident.version + 1;
324    match request.alter_kind {
325        AlterKind::AddColumns { columns } => {
326            // Bumps the column id for the new columns.
327            // It may bump more than the actual number of columns added if there are
328            // existing columns, but it's fine.
329            new_info.meta.next_column_id += columns.len() as u32;
330        }
331        AlterKind::RenameTable { new_table_name } => {
332            new_info.name = new_table_name.clone();
333        }
334        AlterKind::DropColumns { .. }
335        | AlterKind::ModifyColumnTypes { .. }
336        | AlterKind::SetTableOptions { .. }
337        | AlterKind::UnsetTableOptions { .. }
338        | AlterKind::SetRepartitionColumnHint { .. }
339        | AlterKind::UnsetRepartitionColumnHint
340        | AlterKind::SetIndexes { .. }
341        | AlterKind::UnsetIndexes { .. }
342        | AlterKind::DropDefaults { .. }
343        | AlterKind::SetDefaults { .. } => {}
344    }
345
346    info!(
347        "Built new table info: {:?} for table {}, table_id: {}",
348        new_info.meta, table_name, table_id
349    );
350    Ok(new_info)
351}
common_meta/ddl/alter_table/executor.rs

common_meta/ddl/alter_table/
executor.rs
