common_meta/ddl/alter_logical_tables/
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::alter_table_expr::Kind;
19use api::v1::region::{
20    alter_request, region_request, AddColumn, AddColumns, AlterRequest, AlterRequests,
21    RegionColumnDef, RegionRequest, RegionRequestHeader,
22};
23use api::v1::{self, AlterTableExpr};
24use common_telemetry::tracing_context::TracingContext;
25use common_telemetry::{debug, warn};
26use futures::future;
27use store_api::metadata::ColumnMetadata;
28use store_api::storage::{RegionId, RegionNumber, TableId};
29
30use crate::ddl::utils::{add_peer_context_if_needed, raw_table_info};
31use crate::error::Result;
32use crate::instruction::CacheIdent;
33use crate::key::table_info::TableInfoValue;
34use crate::key::{DeserializedValueWithBytes, RegionDistribution, TableMetadataManagerRef};
35use crate::node_manager::NodeManagerRef;
36use crate::rpc::router::{find_leaders, region_distribution, RegionRoute};
37
38/// [AlterLogicalTablesExecutor] performs:
39/// - Alters logical regions on the datanodes.
40/// - Updates table metadata for alter table operation.
41pub struct AlterLogicalTablesExecutor<'a> {
42    /// The alter table expressions.
43    ///
44    /// The first element is the logical table id, the second element is the alter table expression.
45    alters: Vec<(TableId, &'a AlterTableExpr)>,
46}
47
48impl<'a> AlterLogicalTablesExecutor<'a> {
49    pub fn new(alters: Vec<(TableId, &'a AlterTableExpr)>) -> Self {
50        Self { alters }
51    }
52
53    /// Alters logical regions on the datanodes.
54    pub(crate) async fn on_alter_regions(
55        &self,
56        node_manager: &NodeManagerRef,
57        region_routes: &[RegionRoute],
58    ) -> Result<Vec<RegionResponse>> {
59        let region_distribution = region_distribution(region_routes);
60        let leaders = find_leaders(region_routes)
61            .into_iter()
62            .map(|p| (p.id, p))
63            .collect::<HashMap<_, _>>();
64        let mut alter_region_tasks = Vec::with_capacity(leaders.len());
65        for (datanode_id, region_role_set) in region_distribution {
66            if region_role_set.leader_regions.is_empty() {
67                continue;
68            }
69            // Safety: must exists.
70            let peer = leaders.get(&datanode_id).unwrap();
71            let requester = node_manager.datanode(peer).await;
72            let requests = self.make_alter_region_request(&region_role_set.leader_regions);
73            let requester = requester.clone();
74            let peer = peer.clone();
75
76            debug!("Sending alter region requests to datanode {}", peer);
77            alter_region_tasks.push(async move {
78                requester
79                    .handle(make_request(requests))
80                    .await
81                    .map_err(add_peer_context_if_needed(peer))
82            });
83        }
84
85        future::join_all(alter_region_tasks)
86            .await
87            .into_iter()
88            .collect::<Result<Vec<_>>>()
89    }
90
91    fn make_alter_region_request(&self, region_numbers: &[RegionNumber]) -> AlterRequests {
92        let mut requests = Vec::with_capacity(region_numbers.len() * self.alters.len());
93        for (table_id, alter) in self.alters.iter() {
94            for region_number in region_numbers {
95                let region_id = RegionId::new(*table_id, *region_number);
96                let request = make_alter_region_request(region_id, alter);
97                requests.push(request);
98            }
99        }
100
101        AlterRequests { requests }
102    }
103
104    /// Updates table metadata for alter table operation.
105    ///
106    /// ## Panic:
107    /// - If the region distribution is not set when updating table metadata.
108    pub(crate) async fn on_alter_metadata(
109        physical_table_id: TableId,
110        table_metadata_manager: &TableMetadataManagerRef,
111        current_table_info_value: &DeserializedValueWithBytes<TableInfoValue>,
112        region_distribution: RegionDistribution,
113        physical_columns: &[ColumnMetadata],
114    ) -> Result<()> {
115        if physical_columns.is_empty() {
116            warn!("No physical columns found, leaving the physical table's schema unchanged when altering logical tables");
117            return Ok(());
118        }
119
120        let table_ref = current_table_info_value.table_ref();
121        let table_id = physical_table_id;
122
123        // Generates new table info
124        let old_raw_table_info = current_table_info_value.table_info.clone();
125        let new_raw_table_info =
126            raw_table_info::build_new_physical_table_info(old_raw_table_info, physical_columns);
127
128        debug!(
129            "Starting update table: {} metadata, table_id: {}, new table info: {:?}",
130            table_ref, table_id, new_raw_table_info
131        );
132
133        table_metadata_manager
134            .update_table_info(
135                current_table_info_value,
136                Some(region_distribution),
137                new_raw_table_info,
138            )
139            .await?;
140
141        Ok(())
142    }
143
144    /// Builds the cache ident keys for the alter logical tables.
145    ///
146    /// The cache ident keys are:
147    /// - The table id of the logical tables.
148    /// - The table name of the logical tables.
149    /// - The table id of the physical table.
150    pub(crate) fn build_cache_ident_keys(
151        physical_table_info: &TableInfoValue,
152        logical_table_info_values: &[&TableInfoValue],
153    ) -> Vec<CacheIdent> {
154        let mut cache_keys = Vec::with_capacity(logical_table_info_values.len() * 2 + 2);
155        cache_keys.extend(logical_table_info_values.iter().flat_map(|table| {
156            vec![
157                CacheIdent::TableId(table.table_info.ident.table_id),
158                CacheIdent::TableName(table.table_name()),
159            ]
160        }));
161        cache_keys.push(CacheIdent::TableId(
162            physical_table_info.table_info.ident.table_id,
163        ));
164        cache_keys.push(CacheIdent::TableName(physical_table_info.table_name()));
165
166        cache_keys
167    }
168}
169
170fn make_request(alter_requests: AlterRequests) -> RegionRequest {
171    RegionRequest {
172        header: Some(RegionRequestHeader {
173            tracing_context: TracingContext::from_current_span().to_w3c(),
174            ..Default::default()
175        }),
176        body: Some(region_request::Body::Alters(alter_requests)),
177    }
178}
179
180/// Makes an alter region request.
181pub fn make_alter_region_request(
182    region_id: RegionId,
183    alter_table_expr: &AlterTableExpr,
184) -> AlterRequest {
185    let region_id = region_id.as_u64();
186    let kind = match &alter_table_expr.kind {
187        Some(Kind::AddColumns(add_columns)) => Some(alter_request::Kind::AddColumns(
188            to_region_add_columns(add_columns),
189        )),
190        _ => unreachable!(), // Safety: we have checked the kind in check_input_tasks
191    };
192
193    AlterRequest {
194        region_id,
195        schema_version: 0,
196        kind,
197    }
198}
199
200fn to_region_add_columns(add_columns: &v1::AddColumns) -> AddColumns {
201    let add_columns = add_columns
202        .add_columns
203        .iter()
204        .map(|add_column| {
205            let region_column_def = RegionColumnDef {
206                column_def: add_column.column_def.clone(),
207                ..Default::default() // other fields are not used in alter logical table
208            };
209            AddColumn {
210                column_def: Some(region_column_def),
211                ..Default::default() // other fields are not used in alter logical table
212            }
213        })
214        .collect();
215    AddColumns { add_columns }
216}
common_meta/ddl/alter_logical_tables/executor.rs

common_meta/ddl/alter_logical_tables/
executor.rs
