common_meta/reconciliation/reconcile_table/
update_table_info.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 store_api::metadata::ColumnMetadata;
19use tonic::async_trait;
20
21use crate::cache_invalidator::Context as CacheContext;
22use crate::error::Result;
23use crate::instruction::CacheIdent;
24use crate::key::DeserializedValueWithBytes;
25use crate::key::table_info::TableInfoValue;
26use crate::reconciliation::reconcile_table::reconciliation_end::ReconciliationEnd;
27use crate::reconciliation::reconcile_table::{ReconcileTableContext, State};
28use crate::rpc::router::region_distribution;
29
30/// Updates the table info with the new column metadatas.
31#[derive(Debug, Serialize, Deserialize)]
32pub struct UpdateTableInfo {
33    table_info_value: DeserializedValueWithBytes<TableInfoValue>,
34    column_metadatas: Vec<ColumnMetadata>,
35}
36
37impl UpdateTableInfo {
38    pub fn new(
39        table_info_value: DeserializedValueWithBytes<TableInfoValue>,
40        column_metadatas: Vec<ColumnMetadata>,
41    ) -> Self {
42        Self {
43            table_info_value,
44            column_metadatas,
45        }
46    }
47}
48
49#[async_trait]
50#[typetag::serde]
51impl State for UpdateTableInfo {
52    async fn next(
53        &mut self,
54        ctx: &mut ReconcileTableContext,
55        _procedure_ctx: &ProcedureContext,
56    ) -> Result<(Box<dyn State>, Status)> {
57        let new_table_meta = match &ctx.volatile_ctx.table_meta {
58            Some(table_meta) => table_meta.clone(),
59            None => ctx.build_table_meta(&self.column_metadatas)?,
60        };
61
62        let region_routes = &ctx
63            .persistent_ctx
64            .physical_table_route
65            .as_ref()
66            .unwrap()
67            .region_routes;
68        let region_distribution = region_distribution(region_routes);
69        let current_table_info_value = ctx.persistent_ctx.table_info_value.as_ref().unwrap();
70        let new_table_info = {
71            let mut new_table_info = current_table_info_value.table_info.clone();
72            new_table_info.meta = new_table_meta;
73            new_table_info
74        };
75
76        if new_table_info.meta == current_table_info_value.table_info.meta {
77            info!(
78                "Table info is already up to date for table: {}, table_id: {}",
79                ctx.table_name(),
80                ctx.table_id()
81            );
82            ctx.volatile_ctx
83                .result_summary
84                .mark_table_info_phase_completed();
85            return Ok((Box::new(ReconciliationEnd), Status::executing(true)));
86        }
87
88        info!(
89            "Updating table info for table: {}, table_id: {}. new table meta: {:?}, current table meta: {:?}",
90            ctx.table_name(),
91            ctx.table_id(),
92            new_table_info.meta,
93            current_table_info_value.table_info.meta,
94        );
95        ctx.table_metadata_manager
96            .update_table_info(
97                current_table_info_value,
98                Some(region_distribution),
99                new_table_info,
100            )
101            .await?;
102        ctx.volatile_ctx.result_summary.mark_table_info_updated();
103
104        let table_ref = ctx.table_name().table_ref();
105        let table_id = ctx.table_id();
106        let cache_ctx = CacheContext {
107            subject: Some(format!(
108                "Invalidate table cache by reconciling table {}, table_id: {}",
109                table_ref, table_id,
110            )),
111        };
112        ctx.cache_invalidator
113            .invalidate(
114                &cache_ctx,
115                &[
116                    CacheIdent::TableName(table_ref.into()),
117                    CacheIdent::TableId(table_id),
118                ],
119            )
120            .await?;
121        // Update metrics.
122        let metrics = ctx.mut_metrics();
123        metrics.update_table_info = true;
124        ctx.volatile_ctx
125            .result_summary
126            .mark_table_info_phase_completed();
127
128        Ok((Box::new(ReconciliationEnd), Status::executing(true)))
129    }
130}
common_meta/reconciliation/reconcile_table/update_table_info.rs

common_meta/reconciliation/reconcile_table/
update_table_info.rs
