common_meta/reconciliation/reconcile_table/
reconcile_regions.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, HashSet};
16
17use api::v1::column_def::try_as_column_def;
18use api::v1::region::region_request::Body;
19use api::v1::region::{
20    AlterRequest, RegionColumnDef, RegionRequest, RegionRequestHeader, SyncColumns, alter_request,
21};
22use api::v1::{ColumnDef, SemanticType};
23use async_trait::async_trait;
24use common_procedure::{Context as ProcedureContext, Status};
25use common_telemetry::info;
26use common_telemetry::tracing_context::TracingContext;
27use futures::future;
28use serde::{Deserialize, Serialize};
29use snafu::{OptionExt, ResultExt};
30use store_api::metadata::ColumnMetadata;
31use store_api::metric_engine_consts::TABLE_COLUMN_METADATA_EXTENSION_KEY;
32use store_api::storage::{ColumnId, RegionId};
33
34use crate::ddl::utils::{add_peer_context_if_needed, extract_column_metadatas};
35use crate::error::{ConvertColumnDefSnafu, Result, UnexpectedSnafu};
36use crate::reconciliation::reconcile_table::reconciliation_end::ReconciliationEnd;
37use crate::reconciliation::reconcile_table::update_table_info::UpdateTableInfo;
38use crate::reconciliation::reconcile_table::{ReconcileTableContext, State};
39use crate::rpc::router::{find_leaders, region_distribution};
40
41#[derive(Debug, Serialize, Deserialize)]
42pub struct ReconcileRegions {
43    column_metadatas: Vec<ColumnMetadata>,
44    region_ids: HashSet<RegionId>,
45}
46
47impl ReconcileRegions {
48    pub fn new(column_metadatas: Vec<ColumnMetadata>, region_ids: Vec<RegionId>) -> Self {
49        Self {
50            column_metadatas,
51            region_ids: region_ids.into_iter().collect(),
52        }
53    }
54}
55
56#[async_trait]
57#[typetag::serde]
58impl State for ReconcileRegions {
59    async fn next(
60        &mut self,
61        ctx: &mut ReconcileTableContext,
62        _procedure_ctx: &ProcedureContext,
63    ) -> Result<(Box<dyn State>, Status)> {
64        let table_meta = ctx.build_table_meta(&self.column_metadatas)?;
65        ctx.volatile_ctx.table_meta = Some(table_meta);
66        let table_id = ctx.table_id();
67
68        let primary_keys = self
69            .column_metadatas
70            .iter()
71            .filter(|c| c.semantic_type == SemanticType::Tag)
72            .map(|c| c.column_schema.name.clone())
73            .collect::<HashSet<_>>();
74        let column_defs = self
75            .column_metadatas
76            .iter()
77            .map(|c| {
78                let column_def = try_as_column_def(
79                    &c.column_schema,
80                    primary_keys.contains(&c.column_schema.name),
81                )
82                .context(ConvertColumnDefSnafu {
83                    column: &c.column_schema.name,
84                })?;
85
86                Ok((c.column_id, column_def))
87            })
88            .collect::<Result<Vec<_>>>()?;
89
90        // Sends sync column metadatas to datanode.
91        // Safety: The physical table route is set in `ReconciliationStart` state.
92        let region_routes = &ctx
93            .persistent_ctx
94            .physical_table_route
95            .as_ref()
96            .unwrap()
97            .region_routes;
98        let region_distribution = region_distribution(region_routes);
99        let leaders = find_leaders(region_routes)
100            .into_iter()
101            .map(|p| (p.id, p))
102            .collect::<HashMap<_, _>>();
103        let mut sync_column_tsks = Vec::with_capacity(self.region_ids.len());
104        for (datanode_id, region_role_set) in region_distribution {
105            if region_role_set.leader_regions.is_empty() {
106                continue;
107            }
108            // Safety: It contains all leaders in the region routes.
109            let peer = leaders.get(&datanode_id).unwrap();
110            for region_id in region_role_set.leader_regions {
111                let region_id = RegionId::new(ctx.persistent_ctx.table_id, region_id);
112                if self.region_ids.contains(&region_id) {
113                    let requester = ctx.node_manager.datanode(peer).await;
114                    let request = make_alter_region_request(region_id, &column_defs);
115                    let peer = peer.clone();
116
117                    sync_column_tsks.push(async move {
118                        requester
119                            .handle(request)
120                            .await
121                            .map_err(add_peer_context_if_needed(peer))
122                    });
123                }
124            }
125        }
126
127        let results = future::join_all(sync_column_tsks).await;
128        let updated_region_count = results.iter().filter(|result| result.is_ok()).count();
129        ctx.volatile_ctx
130            .result_summary
131            .record_updated_regions(updated_region_count);
132        let mut results = results.into_iter().collect::<Result<Vec<_>>>()?;
133
134        // Ensures all the column metadatas are the same.
135        let column_metadatas =
136            extract_column_metadatas(&mut results, TABLE_COLUMN_METADATA_EXTENSION_KEY)?.context(
137                UnexpectedSnafu {
138                    err_msg: format!(
139                        "The table column metadata schemas from datanodes are not the same, table: {}, table_id: {}",
140                        ctx.table_name(),
141                        table_id
142                    ),
143                },
144            )?;
145
146        ctx.volatile_ctx
147            .result_summary
148            .mark_region_phase_completed();
149
150        // Checks all column metadatas are consistent, and updates the table info if needed.
151        if column_metadatas != self.column_metadatas {
152            info!(
153                "Datanode column metadatas are not consistent with metasrv, updating metasrv's column metadatas, table: {}, table_id: {}",
154                ctx.table_name(),
155                table_id
156            );
157            // Safety: fetched in the above.
158            let table_info_value = ctx.persistent_ctx.table_info_value.clone().unwrap();
159            return Ok((
160                Box::new(UpdateTableInfo::new(table_info_value, column_metadatas)),
161                Status::executing(true),
162            ));
163        }
164
165        Ok((Box::new(ReconciliationEnd), Status::executing(false)))
166    }
167}
168
169/// Makes an alter region request to sync columns.
170fn make_alter_region_request(
171    region_id: RegionId,
172    column_defs: &[(ColumnId, ColumnDef)],
173) -> RegionRequest {
174    let kind = alter_request::Kind::SyncColumns(to_region_sync_columns(column_defs));
175
176    let alter_request = AlterRequest {
177        region_id: region_id.as_u64(),
178        schema_version: 0,
179        kind: Some(kind),
180    };
181
182    RegionRequest {
183        header: Some(RegionRequestHeader {
184            tracing_context: TracingContext::from_current_span().to_w3c(),
185            ..Default::default()
186        }),
187        body: Some(Body::Alter(alter_request)),
188    }
189}
190
191fn to_region_sync_columns(column_defs: &[(ColumnId, ColumnDef)]) -> SyncColumns {
192    let region_column_defs = column_defs
193        .iter()
194        .map(|(column_id, column_def)| RegionColumnDef {
195            column_id: *column_id,
196            column_def: Some(column_def.clone()),
197        })
198        .collect::<Vec<_>>();
199
200    SyncColumns {
201        column_defs: region_column_defs,
202    }
203}
common_meta/reconciliation/reconcile_table/reconcile_regions.rs

common_meta/reconciliation/reconcile_table/
reconcile_regions.rs
