update

Author: zhyass

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

@@ -227,7 +227,10 @@ pub async fn do_vacuum2(
 
     let start = std::time::Instant::now();
     let gc_root_timestamp = gc_root.timestamp.unwrap();
-    // Persist the LVT only if we have not written it before.
+    // NOTE: when using the time-based retention policy we already persisted an LVT above to
+    // constrain snapshot listing, and that value can legitimately be newer than the gc_root.
+    // The meta side only ever moves LVT forward, so only retention policies that have not
+    // written an LVT in this run (e.g. keep-N-snapshots) need to persist it here.
     if need_update_lvt {
         let cat = ctx.get_default_catalog()?;
         cat.set_table_lvt(

src/query/storages/common/session/src/transaction.rs

@@ -67,7 +67,7 @@ pub struct TxnBuffer {
     table_desc_to_id: HashMap<String, u64>,
     mutated_tables: HashMap<u64, TableInfo>,
     base_snapshot_location: HashMap<u64, Option<String>>,
-    snapshot_ts: HashMap<u64, Option<DateTime<Utc>>>,
+    snapshot_ts: HashMap<u64, DateTime<Utc>>,
     tenant: HashMap<u64, Tenant>,
     copied_files: HashMap<u64, Vec<UpsertTableCopiedFileReq>>,
     update_stream_meta: HashMap<u64, UpdateStreamMetaReq>,
@@ -109,7 +109,9 @@ impl TxnBuffer {
                 .entry(table_id)
                 .or_insert(req.base_snapshot_location);
 
-            self.snapshot_ts.entry(table_id).or_insert(req.snapshot_ts);
+            if let Some(ts) = req.snapshot_ts {
+                self.snapshot_ts.entry(table_id).or_insert(ts);
+            }
 
             self.tenant.entry(table_id).or_insert(req.tenant);
         }
@@ -321,7 +323,7 @@ impl TxnManager {
                             seq: MatchSeq::Exact(info.ident.seq),
                             new_table_meta: info.meta.clone(),
                             base_snapshot_location: None,
-                            snapshot_ts: self.txn_buffer.snapshot_ts.get(id).cloned().flatten(),
+                            snapshot_ts: self.txn_buffer.snapshot_ts.get(id).cloned(),
                         },
                         info.clone(),
                     )