common_meta/reconciliation/reconcile_table/
reconciliation_start.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 common_procedure::{Context as ProcedureContext, Status};
16use common_telemetry::info;
17use serde::{Deserialize, Serialize};
18use snafu::ensure;
19
20use crate::ddl::utils::region_metadata_lister::RegionMetadataLister;
21use crate::error::{self, Result};
22use crate::metrics::{self};
23use crate::reconciliation::reconcile_table::resolve_column_metadata::ResolveColumnMetadata;
24use crate::reconciliation::reconcile_table::{
25    ReconcileTableContext, ReconcileTableProcedure, State,
26};
27
28/// The start state of the reconciliation procedure.
29///
30/// This state is used to prepare the table for reconciliation.
31/// It will:
32/// 1. Check the table id and table name consistency.
33/// 2. Ensures the table is a physical table.
34/// 3. List the region metadatas for the physical table.
35#[derive(Debug, Serialize, Deserialize)]
36pub struct ReconciliationStart;
37
38#[async_trait::async_trait]
39#[typetag::serde]
40impl State for ReconciliationStart {
41    async fn next(
42        &mut self,
43        ctx: &mut ReconcileTableContext,
44        procedure_ctx: &ProcedureContext,
45    ) -> Result<(Box<dyn State>, Status)> {
46        let table_id = ctx.table_id();
47
48        let (physical_table_id, physical_table_route) = ctx
49            .table_metadata_manager
50            .table_route_manager()
51            .get_physical_table_route(table_id)
52            .await?;
53        ensure!(
54            physical_table_id == table_id,
55            error::UnexpectedSnafu {
56                err_msg: format!(
57                    "Reconcile table only works for physical table, but got logical table: {}, table_id: {}",
58                    ctx.table_name(),
59                    table_id
60                ),
61            }
62        );
63
64        info!(
65            "Reconciling table: {}, table_id: {}, procedure_id: {}",
66            ctx.table_name(),
67            table_id,
68            procedure_ctx.procedure_id
69        );
70        // TODO(weny): Repairs the table route if needed.
71        let region_metadata_lister = RegionMetadataLister::new(ctx.node_manager.clone());
72
73        let region_metadatas = {
74            let _timer = metrics::METRIC_META_RECONCILIATION_LIST_REGION_METADATA_DURATION
75                .with_label_values(&[metrics::TABLE_TYPE_PHYSICAL])
76                .start_timer();
77            // Always list region metadatas for the physical table.
78            region_metadata_lister
79                .list(physical_table_id, &physical_table_route.region_routes)
80                .await?
81        };
82        ctx.volatile_ctx
83            .result_summary
84            .record_scanned_regions(region_metadatas.len());
85
86        ensure!(!region_metadatas.is_empty(), {
87            metrics::METRIC_META_RECONCILIATION_STATS
88                .with_label_values(&[
89                    ReconcileTableProcedure::TYPE_NAME,
90                    metrics::TABLE_TYPE_PHYSICAL,
91                    metrics::STATS_TYPE_NO_REGION_METADATA,
92                ])
93                .inc();
94
95            error::UnexpectedSnafu {
96                err_msg: format!(
97                    "No region metadata found for table: {}, table_id: {}",
98                    ctx.table_name(),
99                    table_id
100                ),
101            }
102        });
103
104        ensure!(region_metadatas.iter().all(|r| r.is_some()), {
105            metrics::METRIC_META_RECONCILIATION_STATS
106                .with_label_values(&[
107                    ReconcileTableProcedure::TYPE_NAME,
108                    metrics::TABLE_TYPE_PHYSICAL,
109                    metrics::STATS_TYPE_REGION_NOT_OPEN,
110                ])
111                .inc();
112
113            error::UnexpectedSnafu {
114                err_msg: format!(
115                    "Some regions are not opened, table: {}, table_id: {}",
116                    ctx.table_name(),
117                    table_id
118                ),
119            }
120        });
121
122        ctx.volatile_ctx.result_summary.mark_start_completed();
123
124        // Persist the physical table route.
125        // TODO(weny): refetch the physical table route if repair is needed.
126        ctx.persistent_ctx.physical_table_route = Some(physical_table_route);
127        let region_metadatas = region_metadatas.into_iter().map(|r| r.unwrap()).collect();
128        Ok((
129            Box::new(ResolveColumnMetadata::new(
130                ctx.persistent_ctx.resolve_strategy,
131                region_metadatas,
132            )),
133            // We don't persist the state of this step.
134            Status::executing(false),
135        ))
136    }
137}
common_meta/reconciliation/reconcile_table/reconciliation_start.rs

common_meta/reconciliation/reconcile_table/
reconciliation_start.rs
