common_meta/ddl/alter_logical_tables/
validator.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::HashSet;
16
17use api::v1::alter_table_expr::Kind;
18use api::v1::AlterTableExpr;
19use snafu::{ensure, OptionExt};
20use store_api::storage::TableId;
21use table::table_reference::TableReference;
22
23use crate::ddl::utils::table_id::get_all_table_ids_by_names;
24use crate::ddl::utils::table_info::{
25    all_logical_table_routes_have_same_physical_id, get_all_table_info_values_by_table_ids,
26};
27use crate::error::{
28    AlterLogicalTablesInvalidArgumentsSnafu, Result, TableInfoNotFoundSnafu,
29    TableRouteNotFoundSnafu,
30};
31use crate::key::table_info::TableInfoValue;
32use crate::key::table_route::{PhysicalTableRouteValue, TableRouteManager, TableRouteValue};
33use crate::key::{DeserializedValueWithBytes, TableMetadataManagerRef};
34
35/// [AlterLogicalTableValidator] validates the alter logical expressions.
36pub struct AlterLogicalTableValidator<'a> {
37    physical_table_id: TableId,
38    alters: Vec<&'a AlterTableExpr>,
39}
40
41impl<'a> AlterLogicalTableValidator<'a> {
42    pub fn new(physical_table_id: TableId, alters: Vec<&'a AlterTableExpr>) -> Self {
43        Self {
44            physical_table_id,
45            alters,
46        }
47    }
48
49    /// Validates all alter table expressions have the same schema and catalog.
50    fn validate_schema(&self) -> Result<()> {
51        let is_same_schema = self.alters.windows(2).all(|pair| {
52            pair[0].catalog_name == pair[1].catalog_name
53                && pair[0].schema_name == pair[1].schema_name
54        });
55
56        ensure!(
57            is_same_schema,
58            AlterLogicalTablesInvalidArgumentsSnafu {
59                err_msg: "Schemas of the alter table expressions are not the same"
60            }
61        );
62
63        Ok(())
64    }
65
66    /// Validates that all alter table expressions are of the supported kind.
67    /// Currently only supports `AddColumns` operations.
68    fn validate_alter_kind(&self) -> Result<()> {
69        for alter in &self.alters {
70            let kind = alter
71                .kind
72                .as_ref()
73                .context(AlterLogicalTablesInvalidArgumentsSnafu {
74                    err_msg: "Alter kind is missing",
75                })?;
76
77            let Kind::AddColumns(_) = kind else {
78                return AlterLogicalTablesInvalidArgumentsSnafu {
79                    err_msg: "Only support add columns operation",
80                }
81                .fail();
82            };
83        }
84
85        Ok(())
86    }
87
88    fn table_names(&self) -> Vec<TableReference> {
89        self.alters
90            .iter()
91            .map(|alter| {
92                TableReference::full(&alter.catalog_name, &alter.schema_name, &alter.table_name)
93            })
94            .collect()
95    }
96
97    /// Validates that the physical table info and route exist.
98    ///
99    /// This method performs the following validations:
100    /// 1. Retrieves the full table info and route for the given physical table id
101    /// 2. Ensures the table info and table route exists
102    /// 3. Verifies that the table route is actually a physical table route, not a logical one
103    ///
104    /// Returns a tuple containing the validated table info and physical table route.
105    async fn validate_physical_table(
106        &self,
107        table_metadata_manager: &TableMetadataManagerRef,
108    ) -> Result<(
109        DeserializedValueWithBytes<TableInfoValue>,
110        PhysicalTableRouteValue,
111    )> {
112        let (table_info, table_route) = table_metadata_manager
113            .get_full_table_info(self.physical_table_id)
114            .await?;
115
116        let table_info = table_info.with_context(|| TableInfoNotFoundSnafu {
117            table: format!("table id - {}", self.physical_table_id),
118        })?;
119
120        let physical_table_route = table_route
121            .context(TableRouteNotFoundSnafu {
122                table_id: self.physical_table_id,
123            })?
124            .into_inner();
125
126        let TableRouteValue::Physical(table_route) = physical_table_route else {
127            return AlterLogicalTablesInvalidArgumentsSnafu {
128                err_msg: format!(
129                    "expected a physical table but got a logical table: {:?}",
130                    self.physical_table_id
131                ),
132            }
133            .fail();
134        };
135
136        Ok((table_info, table_route))
137    }
138
139    /// Validates that all logical table routes have the same physical table id.
140    ///
141    /// This method performs the following validations:
142    /// 1. Retrieves table routes for all the given table ids.
143    /// 2. Ensures that all retrieved routes are logical table routes (not physical)
144    /// 3. Verifies that all logical table routes reference the same physical table id.
145    /// 4. Returns an error if any route is not logical or references a different physical table.
146    async fn validate_logical_table_routes(
147        &self,
148        table_route_manager: &TableRouteManager,
149        table_ids: &[TableId],
150    ) -> Result<()> {
151        let all_logical_table_routes_have_same_physical_id =
152            all_logical_table_routes_have_same_physical_id(
153                table_route_manager,
154                table_ids,
155                self.physical_table_id,
156            )
157            .await?;
158
159        ensure!(
160            all_logical_table_routes_have_same_physical_id,
161            AlterLogicalTablesInvalidArgumentsSnafu {
162                err_msg: "All the tasks should have the same physical table id"
163            }
164        );
165
166        Ok(())
167    }
168
169    /// Validates the alter logical expressions.
170    ///
171    /// This method performs the following validations:
172    /// 1. Validates that all alter table expressions have the same schema and catalog.
173    /// 2. Validates that all alter table expressions are of the supported kind.
174    /// 3. Validates that the physical table info and route exist.
175    /// 4. Validates that all logical table routes have the same physical table id.
176    ///
177    /// Returns a [ValidatorResult] containing the validation results.
178    pub async fn validate(
179        &self,
180        table_metadata_manager: &TableMetadataManagerRef,
181    ) -> Result<ValidatorResult> {
182        self.validate_schema()?;
183        self.validate_alter_kind()?;
184        let (physical_table_info, physical_table_route) =
185            self.validate_physical_table(table_metadata_manager).await?;
186        let table_names = self.table_names();
187        let table_ids =
188            get_all_table_ids_by_names(table_metadata_manager.table_name_manager(), &table_names)
189                .await?;
190        let mut table_info_values = get_all_table_info_values_by_table_ids(
191            table_metadata_manager.table_info_manager(),
192            &table_ids,
193            &table_names,
194        )
195        .await?;
196        self.validate_logical_table_routes(
197            table_metadata_manager.table_route_manager(),
198            &table_ids,
199        )
200        .await?;
201        let skip_alter = self
202            .alters
203            .iter()
204            .zip(table_info_values.iter())
205            .map(|(task, table)| skip_alter_logical_region(task, table))
206            .collect::<Vec<_>>();
207        retain_unskipped(&mut table_info_values, &skip_alter);
208        let num_skipped = skip_alter.iter().filter(|&&x| x).count();
209
210        Ok(ValidatorResult {
211            num_skipped,
212            skip_alter,
213            table_info_values,
214            physical_table_info,
215            physical_table_route,
216        })
217    }
218}
219
220/// The result of the validator.
221pub(crate) struct ValidatorResult {
222    pub(crate) num_skipped: usize,
223    pub(crate) skip_alter: Vec<bool>,
224    pub(crate) table_info_values: Vec<DeserializedValueWithBytes<TableInfoValue>>,
225    pub(crate) physical_table_info: DeserializedValueWithBytes<TableInfoValue>,
226    pub(crate) physical_table_route: PhysicalTableRouteValue,
227}
228
229/// Retains the elements that are not skipped.
230pub(crate) fn retain_unskipped<T>(target: &mut Vec<T>, skipped: &[bool]) {
231    debug_assert_eq!(target.len(), skipped.len());
232    let mut iter = skipped.iter();
233    target.retain(|_| !iter.next().unwrap());
234}
235
236/// Returns true if does not required to alter the logical region.
237fn skip_alter_logical_region(alter: &AlterTableExpr, table: &TableInfoValue) -> bool {
238    let existing_columns = table
239        .table_info
240        .meta
241        .schema
242        .column_schemas
243        .iter()
244        .map(|c| &c.name)
245        .collect::<HashSet<_>>();
246
247    let Some(kind) = alter.kind.as_ref() else {
248        return true; // Never get here since we have checked it in `validate_alter_kind`
249    };
250    let Kind::AddColumns(add_columns) = kind else {
251        return true; // Never get here since we have checked it in `validate_alter_kind`
252    };
253
254    // We only check that all columns have been finished. That is to say,
255    // if one part is finished but another part is not, it will be considered
256    // unfinished.
257    add_columns
258        .add_columns
259        .iter()
260        .map(|add_column| add_column.column_def.as_ref().map(|c| &c.name))
261        .all(|column| {
262            column
263                .map(|c| existing_columns.contains(c))
264                .unwrap_or(false)
265        })
266}
267
268#[cfg(test)]
269mod tests {
270    use super::*;
271
272    #[test]
273    fn test_retain_unskipped() {
274        let mut target = vec![1, 2, 3, 4, 5];
275        let skipped = vec![false, true, false, true, false];
276        retain_unskipped(&mut target, &skipped);
277        assert_eq!(target, vec![1, 3, 5]);
278    }
279}
common_meta/ddl/alter_logical_tables/validator.rs

common_meta/ddl/alter_logical_tables/
validator.rs
