fix: vacuum tables that are dropped by create or replace statement (#18751)

* fix: vacuum tables dropped by `create or replace` statement

Tables implicitly dropped by the `create or replace` statement are not handled correctly:

The vacuum operation collects candidate tables according to the `drop_on` mark of table meta data,
unfortunately, during `create or replace`, the table being replaced is NOT marked with `drop_on`.

In this PR, for table ids that have no `drop_on` marks, will be selected as the candidates for
vacuum even, except it is the largest table id (of the same table name), which is the live table that
should be kept.

* fix typo

* tweak integration test

* fix lint

* refactor: include all tables of a given dropped db

Author: dantengsky

Cargo.lock

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::collections::HashSet;
 use std::ops::Range;
 
 use chrono::DateTime;
@@ -179,6 +180,7 @@ pub async fn get_history_tables_for_gc(
     drop_time_range: Range<Option<DateTime<Utc>>>,
     db_id: u64,
     limit: usize,
+    db_is_dropped: bool,
 ) -> Result<Vec<TableNIV>, KVAppError> {
     info!(
         "get_history_tables_for_gc: db_id {}, limit {}",
@@ -194,9 +196,37 @@ pub async fn get_history_tables_for_gc(
 
     let mut args = vec![];
 
+    // For table ids of each table with the same name, we keep the largest one, e.g.
+    // ```
+    //  create or replace table t ... ; -- table_id 1
+    //  create or replace table t ... ; -- table_id 2
+    // ```
+    // table_id 2 will be kept for table t (we do not care about the table name though),
+    //
+    // Please note that the largest table id might be "invisible", e.g.
+    // ```
+    //  create or replace table t ... ; -- table_id 1
+    //  create or replace table t ... ; -- table_id 2
+    //  drop table t ... ;
+    // ```
+    // In this case, table_id 2 is marked as dropped.
+
+    let mut latest_table_ids = HashSet::with_capacity(table_history_kvs.len());
+
     for (ident, table_history) in table_history_kvs {
-        for table_id in table_history.id_list.iter() {
-            args.push((TableId::new(*table_id), ident.table_name.clone()));
+        let id_list = &table_history.id_list;
+        if !id_list.is_empty() {
+            // Make sure that the last table id is also the max one (of each table name).
+            let last_id = id_list.last().unwrap();
+            {
+                let max_id = id_list.iter().max().unwrap();
+                assert_eq!(max_id, last_id);
+            }
+            latest_table_ids.insert(*last_id);
+
+            for table_id in id_list.iter() {
+                args.push((TableId::new(*table_id), ident.table_name.clone()));
+            }
         }
     }
 
@@ -228,10 +258,39 @@ pub async fn get_history_tables_for_gc(
                 continue;
             };
 
-            if !drop_time_range.contains(&seq_meta.data.drop_on) {
-                debug!("table {:?} is not in drop_time_range", seq_meta.data);
-                num_out_of_time_range += 1;
-                continue;
+            if !db_is_dropped {
+                // We shall not vacuum it if the table id is the largest table id of some table, but not marked with drop_on,
+                {
+                    let is_visible_last_active_table = seq_meta.data.drop_on.is_none()
+                        && latest_table_ids.contains(&table_id.table_id);
+
+                    if is_visible_last_active_table {
+                        debug!(
+                            "Table id {:?} of {} is the last visible one, not available for vacuum",
+                            table_id, table_name,
+                        );
+                        num_out_of_time_range += 1;
+                        continue;
+                    }
+                }
+
+                // Now the table id is
+                // - Either marked with drop_on
+                //   We use its `drop_on` to do the time range filtering
+                // - Or has no drop_on, but is not the last visible one of the table
+                //   It is still available for vacuum, and we use its `updated_on` to do the time range filtering
+
+                let time_point = if seq_meta.drop_on.is_none() {
+                    &Some(seq_meta.updated_on)
+                } else {
+                    &seq_meta.drop_on
+                };
+
+                if !drop_time_range.contains(time_point) {
+                    debug!("table {:?} is not in drop_time_range", seq_meta.data);
+                    num_out_of_time_range += 1;
+                    continue;
+                }
             }
 
             filter_tb_infos.push(TableNIV::new(

src/meta/api/src/garbage_collection_api.rs

@@ -1601,6 +1601,7 @@ where
                     table_drop_time_range,
                     db_info.database_id.db_id,
                     capacity,
+                    vacuum_db,
                 )
                 .await?;
 
@@ -1647,9 +1648,14 @@ where
         };
 
         let database_id = seq_db_id.data;
-        let table_nivs =
-            get_history_tables_for_gc(self, drop_time_range.clone(), database_id.db_id, the_limit)
-                .await?;
+        let table_nivs = get_history_tables_for_gc(
+            self,
+            drop_time_range.clone(),
+            database_id.db_id,
+            the_limit,
+            false,
+        )
+        .await?;
 
         let mut drop_ids = vec![];
         let mut vacuum_tables = vec![];

src/meta/api/src/table_api.rs

@@ -66,6 +66,7 @@ uuid = { workspace = true }
 
 [dev-dependencies]
 databend-common-functions = { workspace = true }
+databend-common-meta-kvapi = { workspace = true }
 jsonb = { workspace = true }
 tantivy = { workspace = true }
 walkdir = { workspace = true }

src/query/ee/Cargo.toml

@@ -32,6 +32,7 @@ use databend_common_meta_app::schema::DatabaseId;
 use databend_common_meta_app::schema::TableInfo;
 use databend_common_meta_app::schema::TableMeta;
 use databend_common_meta_app::storage::StorageParams;
+use databend_common_meta_kvapi::kvapi::KVApi;
 use databend_common_meta_store::MetaStore;
 use databend_common_meta_store::MetaStoreProvider;
 use databend_common_meta_types::TxnRequest;
@@ -713,17 +714,101 @@ async fn test_gc_in_progress_db_not_undroppable() -> Result<()> {
     Ok(())
 }
 
+#[tokio::test(flavor = "multi_thread")]
+async fn test_vacuum_drop_create_or_replace() -> Result<()> {
+    // vacuum dropped tables by specific database names
+    test_vacuum_drop_create_or_replace_impl(&[
+        "vacuum drop table from db1",
+        "vacuum drop table from db2",
+    ])
+    .await?;
+
+    // vacuum dropped tables all
+    test_vacuum_drop_create_or_replace_impl(&["vacuum drop table"]).await?;
+    Ok(())
+}
+
+async fn test_vacuum_drop_create_or_replace_impl(vacuum_stmts: &[&str]) -> Result<()> {
+    // Setup
+    let meta = new_local_meta().await;
+    let endpoints = meta.endpoints.clone();
+
+    let mut ee_setup = EESetup::new();
+    let config = ee_setup.config_mut();
+    config.meta.endpoints = endpoints.clone();
+
+    let fixture = TestFixture::setup_with_custom(ee_setup).await?;
+
+    // Adjust retention period to 0, so that dropped tables will be available for vacuum immediately
+    let session = fixture.default_session();
+    session.get_settings().set_data_retention_time_in_days(0)?;
+
+    // Prepare test dbs and tables
+    //   - 2 db ids
+    //   - 6 table ids
+    //   - db2.t1 explicitly dropped
+
+    let sqls = vec![
+        "create database db1",
+        "create or replace table db1.t1 (a int)",
+        "create or replace table db1.t1 (a int) as select 1",
+        "create or replace table db1.t1 (a int) as select 2",
+        "create database db2",
+        "create or replace table db2.t1 (a int) as select 1",
+        "create or replace table db2.t1 (a int) as select 2",
+        "create or replace table db2.t1 (a int) as select 3",
+        "DROP TABLE DB2.T1",
+    ];
+
+    for sql in sqls {
+        fixture.execute_command(sql).await?;
+    }
+
+    // there should be 6 table ids : create or replace table 6 times
+    let prefix = "__fd_table_by_id";
+    let items = meta.list_kv_collect(prefix).await?;
+    assert_eq!(items.len(), 6);
+
+    // there should be ownerships for 2 db ids and 5 table ids, one of the ownership is revoked by drop table
+    let prefix = "__fd_object_owners";
+    let items = meta.list_kv_collect(prefix).await?;
+    assert_eq!(items.len(), 7);
+
+    for sql in vacuum_stmts {
+        fixture.execute_command(sql).await?;
+    }
+
+    // After vacuum, 1 table ids left
+    let prefix = "__fd_table_by_id";
+    let items = meta.list_kv_collect(prefix).await?;
+    assert_eq!(items.len(), 1);
+
+    let prefix = "__fd_table_id_to_name";
+    let items = meta.list_kv_collect(prefix).await?;
+    assert_eq!(items.len(), 1);
+
+    // There are ownership objs of  2 dbs and 1 tables
+    let prefix = "__fd_object_owners";
+    let items = meta.list_kv_collect(prefix).await?;
+    assert_eq!(items.len(), 3);
+
+    // db1.t1 should still be accessible
+    fixture.execute_command("select * from db1.t1").await?;
+    // db2.t1 should not exist
+    assert!(fixture
+        .execute_command("select * from db2.t1")
+        .await
+        .is_err());
+    Ok(())
+}
+
 async fn new_local_meta() -> MetaStore {
     let version = &BUILD_INFO;
     let meta_config = MetaConfig::default();
     let meta = {
         let config = meta_config.to_meta_grpc_client_conf(version);
         let provider = Arc::new(MetaStoreProvider::new(config));
-        provider
-            .create_meta_store()
-            .await
-            .map_err(|e| ErrorCode::MetaServiceError(format!("Failed to create meta store: {}", e)))
-            .unwrap()
+        provider.create_meta_store().await.unwrap()
     };
     meta
 }