get table lvt

Author: zhyass

src/meta/api/src/schema_api_test_suite.rs

@@ -1453,6 +1453,7 @@ impl SchemaApiTestSuite {
 
         info!("--- test lvt");
         {
+            let time_zero = DateTime::<Utc>::from_timestamp(0, 0).unwrap();
             let time_small = DateTime::<Utc>::from_timestamp(102, 0).unwrap();
             let time_big = DateTime::<Utc>::from_timestamp(1024, 0).unwrap();
             let time_bigger = DateTime::<Utc>::from_timestamp(1025, 0).unwrap();
@@ -1463,25 +1464,25 @@ impl SchemaApiTestSuite {
 
             let lvt_name_ident = LeastVisibleTimeIdent::new(&tenant, table_id);
 
-            let res = mt.get_pb(&lvt_name_ident).await?;
-            assert!(res.is_none());
+            let res = mt.get_table_lvt(&lvt_name_ident).await?;
+            assert_eq!(res.time, time_zero);
 
             let res = mt.set_table_lvt(&lvt_name_ident, &lvt_big).await?;
             assert_eq!(res.time, time_big);
-            let res = mt.get_pb(&lvt_name_ident).await?;
-            assert_eq!(res.unwrap().data.time, time_big);
+            let res = mt.get_table_lvt(&lvt_name_ident).await?;
+            assert_eq!(res.time, time_big);
 
             // test lvt never fall back
 
             let res = mt.set_table_lvt(&lvt_name_ident, &lvt_small).await?;
             assert_eq!(res.time, time_big);
-            let res = mt.get_pb(&lvt_name_ident).await?;
-            assert_eq!(res.unwrap().data.time, time_big);
+            let res = mt.get_table_lvt(&lvt_name_ident).await?;
+            assert_eq!(res.time, time_big);
 
             let res = mt.set_table_lvt(&lvt_name_ident, &lvt_bigger).await?;
             assert_eq!(res.time, time_bigger);
-            let res = mt.get_pb(&lvt_name_ident).await?;
-            assert_eq!(res.unwrap().data.time, time_bigger);
+            let res = mt.get_table_lvt(&lvt_name_ident).await?;
+            assert_eq!(res.time, time_bigger);
         }
 
         Ok(())

src/meta/api/src/table_api.rs

@@ -1895,6 +1895,18 @@ where
 
         return Ok(transition.unwrap().result.into_value().unwrap_or_default());
     }
+
+    #[logcall::logcall]
+    #[fastrace::trace]
+    async fn get_table_lvt(
+        &self,
+        name_ident: &LeastVisibleTimeIdent,
+    ) -> Result<LeastVisibleTime, KVAppError> {
+        debug!(req :? =(&name_ident); "TableApi: {}", func_name!());
+
+        let seq_v = self.get_pb(name_ident).await?;
+        Ok(seq_v.map(|v| v.data).unwrap_or_default())
+    }
 }
 
 #[async_trait::async_trait]

src/query/catalog/src/catalog/interface.rs

@@ -619,6 +619,10 @@ pub trait Catalog: DynClone + Send + Sync + Debug {
         unimplemented!()
     }
 
+    async fn get_table_lvt(&self, _name_ident: &LeastVisibleTimeIdent) -> Result<LeastVisibleTime> {
+        unimplemented!()
+    }
+
     async fn rename_dictionary(&self, req: RenameDictionaryReq) -> Result<()>;
 
     fn transform_udtf_as_table_function(

src/query/ee/src/storages/fuse/operations/vacuum_table_v2.rs

@@ -44,6 +44,7 @@ use databend_storages_common_table_meta::meta::CompactSegmentInfo;
 use databend_storages_common_table_meta::meta::TableSnapshot;
 use databend_storages_common_table_meta::meta::VACUUM2_OBJECT_KEY_PREFIX;
 use log::info;
+use log::warn;
 use opendal::Entry;
 use opendal::ErrorKind;
 
@@ -612,24 +613,35 @@ async fn process_snapshot_refs(
     ctx: &Arc<dyn TableContext>,
 ) -> Result<RefVacuumInfo> {
     let start = std::time::Instant::now();
+    let table_info = fuse_table.get_table_info();
+    let mut files_to_gc = match fuse_table.cleanup_orphan_ref_dirs().await {
+        Ok(paths) => paths,
+        Err(e) => {
+            warn!(
+                "Failed to clean orphan refs for table {}: {}",
+                table_info.desc, e
+            );
+            Vec::new()
+        }
+    };
     let op = fuse_table.get_operator();
     // Refs that expired and should be cleaned up
     let mut expired_refs = HashSet::new();
+    let mut expired_ref_names = Vec::new();
     // Ref snapshot paths to be cleaned up
     let mut ref_snapshots_to_gc = Vec::new();
     let mut ref_gc_roots = Vec::new();
     let mut gc_root_meta_ts: Option<DateTime<Utc>> = None;
     let mut gc_root_timestamp: Option<DateTime<Utc>> = None;
-    let mut files_to_gc = Vec::new();
 
     let now = Utc::now();
     let (retention_time, num_snapshots_to_keep) =
         fuse_table.get_refs_retention_policy(ctx.as_ref(), now)?;
-    let table_info = fuse_table.get_table_info();
     // Process active refs
     for (ref_name, snapshot_ref) in table_info.meta.refs.iter() {
         if snapshot_ref.expire_at.is_some_and(|v| v < now) {
-            expired_refs.insert(ref_name);
+            expired_refs.insert(snapshot_ref.id);
+            expired_ref_names.push(ref_name);
             continue;
         }
 
@@ -701,9 +713,10 @@ async fn process_snapshot_refs(
 
     if !expired_refs.is_empty() {
         let start_update = std::time::Instant::now();
-        files_to_gc = fuse_table
+        let expired_ref_dirs = fuse_table
             .update_table_refs_meta(ctx, &expired_refs)
             .await?;
+        files_to_gc.extend(expired_ref_dirs);
         ctx.set_status_info(&format!(
             "Updated table meta for table {}, elapsed: {:?}",
             table_info.desc,
@@ -717,12 +730,11 @@ async fn process_snapshot_refs(
         file_op.remove_file_in_batch(&ref_snapshots_to_gc).await?;
     }
 
-    let expired_vec = expired_refs.into_iter().collect::<Vec<_>>();
     ctx.set_status_info(&format!(
         "Processed snapshot refs for table {}, elapsed: {:?}, expire_refs: {}, ref_snapshots_to_gc: {}, ref_gc_root_meta_ts: {:?}",
         table_info.desc,
         start.elapsed(),
-        slice_summary(&expired_vec),
+        slice_summary(&expired_ref_names),
         slice_summary(&ref_snapshots_to_gc),
         gc_root_meta_ts
     ));

src/query/service/src/catalogs/default/database_catalog.rs

@@ -921,6 +921,10 @@ impl Catalog for DatabaseCatalog {
         self.mutable_catalog.set_table_lvt(name_ident, value).await
     }
 
+    async fn get_table_lvt(&self, name_ident: &LeastVisibleTimeIdent) -> Result<LeastVisibleTime> {
+        self.mutable_catalog.get_table_lvt(name_ident).await
+    }
+
     async fn rename_dictionary(&self, req: RenameDictionaryReq) -> Result<()> {
         self.mutable_catalog.rename_dictionary(req).await
     }

src/query/service/src/catalogs/default/mutable_catalog.rs

@@ -971,6 +971,10 @@ impl Catalog for MutableCatalog {
         Ok(self.ctx.meta.set_table_lvt(name_ident, value).await?)
     }
 
+    async fn get_table_lvt(&self, name_ident: &LeastVisibleTimeIdent) -> Result<LeastVisibleTime> {
+        Ok(self.ctx.meta.get_table_lvt(name_ident).await?)
+    }
+
     #[async_backtrace::framed]
     async fn rename_dictionary(&self, req: RenameDictionaryReq) -> Result<()> {
         let res = self.ctx.meta.rename_dictionary(req).await?;

src/query/service/src/catalogs/default/session_catalog.rs

@@ -791,6 +791,10 @@ impl Catalog for SessionCatalog {
         self.inner.set_table_lvt(name_ident, value).await
     }
 
+    async fn get_table_lvt(&self, name_ident: &LeastVisibleTimeIdent) -> Result<LeastVisibleTime> {
+        self.inner.get_table_lvt(name_ident).await
+    }
+
     async fn rename_dictionary(&self, req: RenameDictionaryReq) -> Result<()> {
         self.inner.rename_dictionary(req).await
     }

src/query/storages/fuse/src/operations/gc.rs

@@ -895,6 +895,15 @@ impl FuseTable {
         counter: &mut PurgeCounter,
         dry_run: bool,
     ) -> Result<HashSet<Location>> {
+        if !dry_run {
+            if let Err(e) = self.cleanup_orphan_ref_dirs().await {
+                warn!(
+                    "Failed to clean orphan refs for table {} before purge: {}",
+                    self.table_info.desc, e
+                );
+            }
+        }
+
         let now = Utc::now();
         let table_info = self.get_table_info();
         let op = self.get_operator();
@@ -909,7 +918,7 @@ impl FuseTable {
         for (ref_name, snapshot_ref) in table_info.meta.refs.iter() {
             // Check if ref is expired
             if snapshot_ref.expire_at.is_some_and(|v| v < now) {
-                expired_refs.insert(ref_name);
+                expired_refs.insert(snapshot_ref.id);
                 continue;
             }
 

src/query/storages/fuse/src/operations/vacuum.rs

@@ -439,7 +439,7 @@ impl FuseTable {
     pub async fn update_table_refs_meta(
         &self,
         ctx: &Arc<dyn TableContext>,
-        expired_refs: &HashSet<&String>,
+        expired_refs: &HashSet<u64>,
     ) -> Result<Vec<String>> {
         let catalog = ctx.get_default_catalog()?;
 
@@ -453,7 +453,7 @@ impl FuseTable {
             let mut new_table_meta = latest_table_info.meta.clone();
             new_table_meta
                 .refs
-                .retain(|ref_name, _| !expired_refs.contains(ref_name));
+                .retain(|_, val| !expired_refs.contains(&val.id));
             let req = UpdateTableMetaReq {
                 table_id: latest_table_info.ident.table_id,
                 seq: MatchSeq::Exact(latest_table_info.ident.seq),
@@ -505,6 +505,54 @@ impl FuseTable {
         }
         Ok(dir_to_gc)
     }
+
+    pub async fn cleanup_orphan_ref_dirs(&self) -> Result<Vec<String>> {
+        let prefix = self
+            .meta_location_generator()
+            .ref_snapshot_location_prefix();
+        let op = self.get_operator();
+        let table_info = self.get_table_info();
+        let table_seq = table_info.ident.seq;
+        let active_ids = table_info
+            .meta
+            .refs
+            .values()
+            .map(|snapshot_ref| snapshot_ref.id)
+            .collect::<HashSet<_>>();
+        let mut removed = Vec::new();
+        let mut lister = op.lister(prefix).await?;
+        while let Some(entry) = lister.try_next().await? {
+            if !entry.metadata().is_dir() {
+                continue;
+            }
+
+            let dir_path = entry.path();
+            let id_str = dir_path.trim_end_matches('/').rsplit('/').next().unwrap();
+            let Ok(ref_id) = id_str.parse::<u64>() else {
+                continue;
+            };
+            if table_seq <= ref_id || active_ids.contains(&ref_id) {
+                continue;
+            }
+
+            match op.remove_all(dir_path).await {
+                Ok(_) => {
+                    info!(
+                        "Removed orphan ref directory '{}' for table {}",
+                        dir_path, table_info.desc
+                    );
+                    removed.push(dir_path.to_string());
+                }
+                Err(e) => {
+                    warn!(
+                        "Failed to remove orphan ref directory '{}' for table {}: {}",
+                        dir_path, table_info.desc, e
+                    );
+                }
+            }
+        }
+        Ok(removed)
+    }
 }
 
 pub async fn vacuum_tables_from_info(