fix(query): guard metadata consistency for flashback, time travel, and DDL column operations (#19653)

* fix

fix

fix

* fix review comments

Author: zhyass

src/query/service/src/interpreters/interpreter_table_drop_column.rs

@@ -21,6 +21,8 @@ use databend_common_expression::DataSchema;
 use databend_common_meta_app::schema::DatabaseType;
 use databend_common_sql::ApproxDistinctColumns;
 use databend_common_sql::BloomIndexColumns;
+use databend_common_sql::binder::validate_constraints_by_schema;
+use databend_common_sql::binder::validate_table_indexes_not_referencing_columns;
 use databend_common_sql::plans::DropTableColumnPlan;
 use databend_common_storages_basic::view_table::VIEW_ENGINE;
 use databend_common_storages_stream::stream_table::STREAM_ENGINE;
@@ -159,6 +161,17 @@ impl Interpreter for DropTableColumnInterpreter {
         }
         let new_schema = new_table_meta.schema.as_ref().clone();
 
+        let dropped_column_ids = field.column_ids().into_iter().collect();
+        validate_table_indexes_not_referencing_columns(
+            self.ctx.clone(),
+            catalog.as_ref(),
+            &self.ctx.get_tenant(),
+            table.get_id(),
+            &dropped_column_ids,
+        )
+        .await?;
+        validate_constraints_by_schema(self.ctx.clone(), &new_table_meta.constraints, &new_schema)?;
+
         commit_table_meta(
             &self.ctx,
             table.as_ref(),

src/query/service/src/interpreters/interpreter_table_modify_column.rs

@@ -44,6 +44,8 @@ use databend_common_sql::BloomIndexColumns;
 use databend_common_sql::DefaultExprBinder;
 use databend_common_sql::Planner;
 use databend_common_sql::analyze_cluster_keys;
+use databend_common_sql::binder::validate_constraints_by_schema;
+use databend_common_sql::binder::validate_table_indexes_not_referencing_columns;
 use databend_common_sql::plans::ModifyColumnAction;
 use databend_common_sql::plans::ModifyTableColumnPlan;
 use databend_common_sql::plans::Plan;
@@ -200,6 +202,7 @@ impl ModifyTableColumnInterpreter {
         let table_info = table.get_table_info();
         let mut new_schema = schema.as_ref().clone();
         let mut modified_cols = HashSet::with_capacity(field_and_comments.len());
+        let mut modified_column_ids = HashSet::new();
         // first check default expr before lock table
         for (field, _comment) in field_and_comments {
             if let Some((i, old_field)) = schema.column_with_name(&field.name) {
@@ -224,6 +227,10 @@ impl ModifyTableColumnInterpreter {
 
                 if old_field.data_type != field.data_type {
                     modified_cols.insert(field.name.clone());
+                    // Aggregating indexes still bind to the existing table column ids, so
+                    // MODIFY COLUMN must check the ids from the current schema instead of the
+                    // freshly analyzed field definition.
+                    modified_column_ids.extend(old_field.column_ids());
                     // Check if this column is referenced by computed columns.
                     let data_schema = DataSchema::from(&new_schema);
                     check_referenced_computed_columns(
@@ -265,6 +272,20 @@ impl ModifyTableColumnInterpreter {
         let catalog_name = table_info.catalog();
         let catalog = self.ctx.get_catalog(catalog_name).await?;
 
+        validate_table_indexes_not_referencing_columns(
+            self.ctx.clone(),
+            catalog.as_ref(),
+            &self.ctx.get_tenant(),
+            table.get_id(),
+            &modified_column_ids,
+        )
+        .await?;
+        validate_constraints_by_schema(
+            self.ctx.clone(),
+            &table_info.meta.constraints,
+            new_schema.as_ref(),
+        )?;
+
         let base_snapshot = fuse_table.read_table_snapshot().await?;
         let prev_snapshot_id = base_snapshot.snapshot_id().map(|(id, _)| id);
         let table_meta_timestamps = self

src/query/service/src/interpreters/interpreter_table_set_options.rs

@@ -278,6 +278,7 @@ async fn set_segment_format(
         table_snapshot.summary.clone(),
         new_segment_locations,
         fuse_table.cluster_key_meta(),
+        fuse_table.cluster_type(),
         table_snapshot.table_statistics_location(),
         table_meta_timestamps,
     )?;

src/query/service/src/test_kits/fuse.rs

@@ -276,6 +276,7 @@ pub async fn generate_snapshots(fixture: &TestFixture) -> Result<()> {
         locations,
         None,
         None,
+        None,
         TestFixture::default_table_meta_timestamps(),
     )?;
     snapshot_1.timestamp = Some(now - Duration::hours(12));
@@ -460,6 +461,7 @@ pub async fn generate_snapshot_v4(
         segments.iter().map(|s| s.0.clone()).collect(),
         None,
         None,
+        None,
         TestFixture::default_table_meta_timestamps(),
     )?;
     let new_snapshot_location =

src/query/service/tests/it/storages/fuse/conflict.rs

@@ -63,6 +63,7 @@ fn test_unresolvable_delete_conflict() {
     let result = generator.do_generate_new_snapshot(
         &TableInfo::default(),
         None,
+        None,
         &Some(Arc::new(latest_snapshot)),
         TestFixture::default_table_meta_timestamps(),
         Default::default(),
@@ -193,6 +194,7 @@ fn test_resolvable_delete_conflict() {
     let result = generator.do_generate_new_snapshot(
         &TableInfo::default(),
         None,
+        None,
         &Some(Arc::new(latest_snapshot)),
         TestFixture::default_table_meta_timestamps(),
         Default::default(),
@@ -349,6 +351,7 @@ fn test_resolvable_replace_conflict() {
     let result = generator.do_generate_new_snapshot(
         &TableInfo::default(),
         None,
+        None,
         &Some(Arc::new(latest_snapshot)),
         TestFixture::default_table_meta_timestamps(),
         Default::default(),

src/query/service/tests/it/storages/fuse/meta/snapshot.rs

@@ -36,6 +36,7 @@ fn default_snapshot() -> TableSnapshot {
         vec![],
         None,
         None,
+        None,
         TestFixture::default_table_meta_timestamps(),
     )
     .unwrap()
@@ -59,6 +60,7 @@ fn snapshot_timestamp_monotonic_increase() {
         vec![],
         None,
         None,
+        None,
         TestFixture::default_table_meta_timestamps(),
     )
     .unwrap();
@@ -84,6 +86,7 @@ fn snapshot_timestamp_time_skew_tolerance() {
         vec![],
         None,
         None,
+        None,
         table_meta_timestamps,
     )
     .unwrap();

src/query/service/tests/it/storages/fuse/operations/commit.rs

@@ -292,6 +292,7 @@ async fn test_commit_to_meta_server() -> anyhow::Result<()> {
                 new_segments,
                 None,
                 None,
+                None,
                 TestFixture::default_table_meta_timestamps(),
             )
             .unwrap();

src/query/service/tests/it/storages/fuse/operations/mutation/block_compact_mutator.rs

@@ -232,6 +232,7 @@ async fn test_safety() -> anyhow::Result<()> {
             locations.clone(),
             None,
             None,
+            None,
             TestFixture::default_table_meta_timestamps(),
         )?;
 

src/query/service/tests/it/storages/fuse/operations/mutation/recluster_mutator.rs

@@ -254,6 +254,7 @@ async fn test_safety_for_recluster() -> anyhow::Result<()> {
             locations.clone(),
             None,
             None,
+            None,
             TestFixture::default_table_meta_timestamps(),
         )?);
 

src/query/service/tests/it/storages/fuse/utils.rs

@@ -34,6 +34,7 @@ pub fn new_empty_snapshot(schema: TableSchema) -> TableSnapshot {
         vec![],
         None,
         None,
+        None,
         TableMetaTimestamps::new(None, Duration::hours(1)),
     )
     .unwrap()