[BugFix] Retain lake vacuum files by version interval across tablet reshard

Auto-vacuum stalled and could lose snapshot data on shared-data
range-distribution partitions after a tablet split/merge when a cluster
snapshot pinned a pre-reshard version V:

- TabletRetainInfo::init read {tablet}_{V}.meta for each retained version to
  build its protect set. On a reshard child, a pinned pre-reshard version V
  predates the child's earliest metadata, so the read returned NotFound and
  init failed. vacuum_tablet_metadata runs init before any deletion, so the
  whole partition's vacuum RPC failed every cycle and expired metadata / data
  files / txn logs accumulated for as long as the snapshot existed.
- Skipping the NotFound version, or only suppressing shared-file cleanup, is
  unsafe: the incremental vacuum can delete a pre-reshard file the snapshot
  still needs, and the reshard "identical tablet" path does not mark inherited
  files shared, so a shared-flag-based guard misses them.

Decide retention by version interval instead of by reading the pinned version's
metadata. A garbage file is kept iff it was live at a retained version, i.e.
there is a retained version V with create_version <= V < remove_version, where
remove_version is the metadata version that moved it into compaction_inputs /
orphan_files.

- TabletRetainInfo now holds only the retained-version set (no metadata reads,
  so the NotFound stall cannot occur) and answers
  retained_by_version(create, remove) and contains_version(v).
- collect_garbage_files decides each garbage file's creation version:
  - compaction_inputs segments share the rowset's version and are retained by
    rowset.version(). Lake partial-segment compaction can carry older segments
    into a higher-versioned rowset, but it is config-gated (off by default),
    non-PK, and being retired in favor of parallel compaction, so it is
    deliberately not special-cased;
  - del files can be transferred by cloud-native PK compaction onto a
    higher-versioned rowset, so they carry a new optional DelfileWithRowsetId
    .version and are retained per file;
  - orphan files carry a new optional FileMetaPB.version, stamped at each
    orphaning site from the source object's own version.
  An unset/0 version means "unknown -> retain conservatively", never a false
  delete. Keyed on version rather than the per-file shared flag, this protects
  both shared and exclusive/identical inherited files.

No user-facing SQL/config/API/metric change; adds two backward-compatible
optional fields to the storage metadata proto. Range distribution is a new,
new-tables-only feature with no on-disk compatibility burden.

Signed-off-by: xiangguangyxg <xiangguangyxg@gmail.com>

Author: xiangguangyxg

be/src/storage/lake/meta_file.cpp

@@ -191,6 +191,9 @@ void MetaFileBuilder::append_dcg(uint32_t rssid,
             if (dcg_ver.shared_files_size() > 0 && i < dcg_ver.shared_files_size()) {
                 file_meta.set_shared(dcg_ver.shared_files(i));
             }
+            if (i < dcg_ver.versions_size()) {
+                file_meta.set_version(dcg_ver.versions(i));
+            }
             _tablet_meta->mutable_orphan_files()->Add(std::move(file_meta));
         }
     }
@@ -273,14 +276,24 @@ void MetaFileBuilder::apply_opwrite(const TxnLogPB_OpWrite& op_write,
         if (del_meta.has_shared()) {
             del_file_with_rid.set_shared(del_meta.shared());
         }
+        // Freshly created del file: stamp its creation version so lake vacuum can retain it by its
+        // own version after it is later transferred onto a higher-versioned compaction output rowset.
+        del_file_with_rid.set_version(rowset->version());
         rowset->add_del_files()->CopyFrom(del_file_with_rid);
     }
     // if rowset don't contain segment files, still inc next_rowset_id
     _tablet_meta->set_next_rowset_id(_tablet_meta->next_rowset_id() + get_rowset_id_step(*rowset));
     // collect trash files: replaced partial-update segments and their segment-name-keyed .vi files
     for (const auto& orphan_file : orphan_files) {
         DCHECK(is_segment(orphan_file.name()) || is_vector_index(orphan_file.name()));
-        _tablet_meta->mutable_orphan_files()->Add()->CopyFrom(orphan_file);
+        auto* added_orphan = _tablet_meta->mutable_orphan_files()->Add();
+        added_orphan->CopyFrom(orphan_file);
+        // These replaced segment/.vi files are produced by this txn's partial-update rewrite, so
+        // their creation version is the metadata version being built. Respect an accurate version if
+        // the producer already set one.
+        if (!added_orphan->has_version()) {
+            added_orphan->set_version(_tablet_meta->version());
+        }
     }
     if (!_tablet_meta->rowset_to_schema().empty()) {
         auto schema_id = _tablet_meta->schema().id();
@@ -301,6 +314,9 @@ void MetaFileBuilder::apply_column_mode_partial_update(const TxnLogPB_OpWrite& o
         // sibling tablets. The orphan FileMetaPB only carries `shared`, so encode bundling into it
         // too, otherwise vacuum deletes the shared bundle file while a sibling still references it.
         file_meta.set_shared(segment_meta.shared() || segment_meta.has_bundle_file_offset());
+        // These segments are produced and discarded within this txn, so their creation version is
+        // the metadata version being built.
+        file_meta.set_version(_tablet_meta->version());
         _tablet_meta->mutable_orphan_files()->Add(std::move(file_meta));
     }
 }
@@ -502,6 +518,7 @@ void MetaFileBuilder::apply_drop_index(const TxnLogPB_OpDropIndex& op) {
                     file_meta.set_name(entry.index_file());
                     if (entry.has_file_size()) file_meta.set_size(entry.file_size());
                     if (entry.has_shared_file()) file_meta.set_shared(entry.shared_file());
+                    file_meta.set_version(entry.version());
                     _tablet_meta->mutable_orphan_files()->Add(std::move(file_meta));
                 }
             }
@@ -636,6 +653,7 @@ void MetaFileBuilder::remove_compacted_sst(const TxnLogPB_OpCompaction& op_compa
             }
         }
         file_meta.set_shared(shared);
+        file_meta.set_version(input_sstable.version());
         _tablet_meta->mutable_orphan_files()->Add(std::move(file_meta));
     }
 }
@@ -652,7 +670,12 @@ void MetaFileBuilder::remove_lcrm_file(const TxnLogPB_OpCompaction& op_compactio
     // tablet metadata GC process, which runs periodically and handles failures gracefully.
     // This approach is safe, non-blocking, and handles distributed cleanup correctly.
     if (op_compaction.has_lcrm_file()) {
-        _tablet_meta->add_orphan_files()->CopyFrom(op_compaction.lcrm_file());
+        auto* added = _tablet_meta->add_orphan_files();
+        added->CopyFrom(op_compaction.lcrm_file());
+        // The mapper file is produced and orphaned by this compaction and is never referenced by any
+        // visible tablet metadata, so its creation version is the metadata version being built.
+        // Stamping it lets vacuum reclaim it instead of over-retaining it under a covering snapshot.
+        added->set_version(_tablet_meta->version());
     }
 }
 
@@ -737,6 +760,9 @@ Status MetaFileBuilder::apply_opcompaction(const TxnLogPB_OpCompaction& op_compa
                 if (dcg.shared_files_size() > 0) {
                     file_meta.set_shared(dcg.shared_files(i));
                 }
+                if (i < dcg.versions_size()) {
+                    file_meta.set_version(dcg.versions(i));
+                }
                 // Put useless `.cols` files into orphan files
                 _tablet_meta->mutable_orphan_files()->Add(std::move(file_meta));
             }
@@ -761,6 +787,7 @@ Status MetaFileBuilder::apply_opcompaction(const TxnLogPB_OpCompaction& op_compa
                     file_meta.set_name(entry.index_file());
                     if (entry.has_file_size()) file_meta.set_size(entry.file_size());
                     if (entry.has_shared_file()) file_meta.set_shared(entry.shared_file());
+                    file_meta.set_version(entry.version());
                     _tablet_meta->mutable_orphan_files()->Add(std::move(file_meta));
                 }
             }
@@ -866,6 +893,9 @@ void MetaFileBuilder::apply_opcompaction_with_conflict(const TxnLogPB_OpCompacti
         // tablets. Encode bundling into the orphan's `shared` flag so vacuum's alive-check protects
         // it instead of deleting a bundle file a sibling still references.
         file_meta.set_shared(segment_meta.shared() || segment_meta.has_bundle_file_offset());
+        // These segments are produced and discarded within this txn, so their creation version is
+        // the metadata version being built.
+        file_meta.set_version(_tablet_meta->version());
         _tablet_meta->mutable_orphan_files()->Add(std::move(file_meta));
     }
 }
@@ -982,6 +1012,7 @@ Status MetaFileBuilder::_finalize_delvec(int64_t version, int64_t txn_id) {
         if (refered_versions.find(itr->first) == refered_versions.end()) {
             VLOG(2) << "Remove delvec file record from delvec meta, version: " << itr->first
                     << ", file: " << itr->second.name();
+            itr->second.set_version(itr->first);
             _tablet_meta->mutable_orphan_files()->Add(std::move(itr->second));
             itr = _tablet_meta->mutable_delvec_meta()->mutable_version_to_file()->erase(itr);
         } else {
@@ -1005,6 +1036,7 @@ void MetaFileBuilder::_sstable_meta_clean_after_alter_type() {
             file_meta.set_name(sstable.filename());
             file_meta.set_size(sstable.filesize());
             file_meta.set_shared(sstable.shared());
+            file_meta.set_version(sstable.version());
             _tablet_meta->mutable_orphan_files()->Add(std::move(file_meta));
         }
         // Clear the SSTable metadata.
@@ -1370,6 +1402,8 @@ Status MetaFileBuilder::set_final_rowset() {
             del_file_with_rid.set_encryption_meta(del.encryption_meta());
         }
         del_file_with_rid.set_shared(del.shared());
+        // Freshly created del file: stamp its creation version (see apply_opwrite).
+        del_file_with_rid.set_version(rowset->version());
         rowset->add_del_files()->CopyFrom(del_file_with_rid);
     }
 
@@ -1379,7 +1413,14 @@ Status MetaFileBuilder::set_final_rowset() {
     // Collect orphan files: replaced partial-update segments and their segment-name-keyed .vi files
     for (const auto& orphan_file : _pending_rowset_data.orphan_files) {
         DCHECK(is_segment(orphan_file.name()) || is_vector_index(orphan_file.name()));
-        _tablet_meta->mutable_orphan_files()->Add()->CopyFrom(orphan_file);
+        auto* added_orphan = _tablet_meta->mutable_orphan_files()->Add();
+        added_orphan->CopyFrom(orphan_file);
+        // These replaced segment/.vi files are produced by this txn's partial-update rewrite, so
+        // their creation version is the metadata version being built. Respect an accurate version if
+        // the producer already set one.
+        if (!added_orphan->has_version()) {
+            added_orphan->set_version(_tablet_meta->version());
+        }
     }
 
     // Handle schema mapping (same logic as apply_opwrite)

be/src/storage/lake/tablet_merger.cpp

@@ -1465,6 +1465,7 @@ Status merge_idg_meta(const std::vector<TabletMergeContext>& merge_contexts, Tab
                 fm.set_name(e.index_file());
                 if (e.has_file_size()) fm.set_size(e.file_size());
                 if (tgt_shared) fm.set_shared(true);
+                fm.set_version(e.version());
             }
         }
         if (ver.entries_size() == 0) continue;

be/src/storage/lake/tablet_retain_info.cpp

@@ -14,52 +14,23 @@
 
 #include "storage/lake/tablet_retain_info.h"
 
-#include "common/status.h"
-#include "storage/lake/tablet_manager.h"
-#include "storage/lake/tablet_metadata.h"
-
 namespace starrocks::lake {
 
-Status TabletRetainInfo::init(int64_t tablet_id, const std::unordered_set<int64_t>& retain_versions,
-                              TabletManager* tablet_mgr) {
-    _tablet_id = tablet_id;
-    for (const auto& version : retain_versions) {
-        auto res = tablet_mgr->get_tablet_metadata(tablet_id, version, false);
-        if (!res.status().ok()) {
-            return res.status();
-        }
-        auto metadata = std::move(res).value();
-        for (const auto& rowset : metadata->rowsets()) {
-            _rowset_ids.insert(rowset.id());
-        }
+void TabletRetainInfo::init(const std::unordered_set<int64_t>& retain_versions) {
+    _retain_versions = retain_versions;
+}
 
-        for (const auto& [_, dcg] : (*metadata).dcg_meta().dcgs()) {
-            for (const auto& column_file : dcg.column_files()) {
-                _files.insert(column_file);
-            }
-        }
-        for (const auto& [_, file] : (*metadata).delvec_meta().version_to_file()) {
-            _files.insert(file.name());
-        }
-        for (const auto& sstable : (*metadata).sstable_meta().sstables()) {
-            _files.insert(sstable.filename());
+bool TabletRetainInfo::retained_by_version(int64_t create_version, int64_t remove_version) const {
+    for (auto version : _retain_versions) {
+        if (create_version <= version && version < remove_version) {
+            return true;
         }
-
-        _versions.insert(version);
     }
-    return Status::OK();
-}
-
-bool TabletRetainInfo::contains_file(const std::string& file_name) const {
-    return _files.find(file_name) != _files.end();
+    return false;
 }
 
 bool TabletRetainInfo::contains_version(int64_t version) const {
-    return _versions.find(version) != _versions.end();
-}
-
-bool TabletRetainInfo::contains_rowset(uint32_t rowset_id) const {
-    return _rowset_ids.find(rowset_id) != _rowset_ids.end();
+    return _retain_versions.find(version) != _retain_versions.end();
 }
 
 } // namespace starrocks::lake

be/src/storage/lake/tablet_retain_info.h

@@ -15,41 +15,40 @@
 #pragma once
 
 #include <cstdint>
-#include <string>
 #include <unordered_set>
 
-namespace starrocks {
-class Status;
-}
-
 namespace starrocks::lake {
 
-class TabletManager;
-
 /*
- * TabletRetainInfo is used to collect all files name, rowsets id for the specifed versions
- * that used in vacuum process to determine which files(data or meta) can be deleted.
+ * TabletRetainInfo carries the set of tablet metadata versions that lake vacuum must retain for a
+ * partition (e.g. versions pinned by a cluster snapshot). It answers two questions used while
+ * collecting garbage:
+ *   - whether a tablet metadata file at a given version must be kept (contains_version);
+ *   - whether a data file must be kept because it was live at a retained version
+ *     (retained_by_version), decided purely from the file's own [create, remove) version interval.
+ *
+ * It deliberately reads no per-version tablet metadata: a retained version may predate the tablet
+ * (e.g. a split/merge child asked to retain a pre-reshard version it never had), so reading it would
+ * fail with NotFound. Keying on the file's own version instead makes the decision independent of
+ * both that read and the per-file `shared` flag (which reshard does not set uniformly).
 */
 class TabletRetainInfo {
 public:
     TabletRetainInfo() = default;
     ~TabletRetainInfo() = default;
 
-    Status init(int64_t tablet_id, const std::unordered_set<int64_t>& retain_versions, TabletManager* tablet_mgr);
+    void init(const std::unordered_set<int64_t>& retain_versions);
 
-    bool contains_file(const std::string& file_name) const;
+    // Whether a data file created at |create_version| and removed at |remove_version| was live at
+    // some retained version, i.e. there exists a retained version V with
+    // create_version <= V < remove_version. A |create_version| of 0 means "unknown" and is treated
+    // as the earliest possible version, so the file is retained conservatively.
+    bool retained_by_version(int64_t create_version, int64_t remove_version) const;
 
     bool contains_version(int64_t version) const;
 
-    bool contains_rowset(uint32_t rowset_id) const;
-
-    int64_t tablet_id() const { return _tablet_id; }
-
 private:
-    int64_t _tablet_id;
-    std::unordered_set<int64_t> _versions;
-    std::unordered_set<std::string> _files;
-    std::unordered_set<uint32_t> _rowset_ids;
+    std::unordered_set<int64_t> _retain_versions;
 };
 
 } // namespace starrocks::lake

be/src/storage/lake/txn_log_applier.cpp

@@ -235,6 +235,9 @@ void collect_dcg_orphan_files(const DeltaColumnGroupMetadataPB& old_dcg_meta,
             if (dcg_ver.shared_files_size() > 0 && i < dcg_ver.shared_files_size()) {
                 file_meta.set_shared(dcg_ver.shared_files(i));
             }
+            if (i < dcg_ver.versions_size()) {
+                file_meta.set_version(dcg_ver.versions(i));
+            }
             metadata->mutable_orphan_files()->Add(std::move(file_meta));
         }
     }
@@ -257,6 +260,7 @@ void collect_idg_orphan_files(const IndexDeltaGroupMetadataPB& old_idg_meta,
             file_meta.set_name(entry.index_file());
             if (entry.has_file_size()) file_meta.set_size(entry.file_size());
             file_meta.set_shared(entry.shared_file());
+            file_meta.set_version(entry.version());
             metadata->mutable_orphan_files()->Add(std::move(file_meta));
         }
     }
@@ -311,6 +315,7 @@ class PrimaryKeyTxnLogApplier : public TxnLogApplier {
                 file_meta.set_name(sstable.filename());
                 file_meta.set_size(sstable.filesize());
                 file_meta.set_shared(sstable.shared());
+                file_meta.set_version(sstable.version());
                 _metadata->mutable_orphan_files()->Add(std::move(file_meta));
             }
             _metadata->clear_sstable_meta();
@@ -637,7 +642,12 @@ class PrimaryKeyTxnLogApplier : public TxnLogApplier {
         // Cleanup orphan lcrm files from merged large rowset split subtasks
         // These files are no longer valid after merging (segment IDs changed)
         for (const auto& lcrm_file : op_parallel.orphan_lcrm_files()) {
-            _metadata->add_orphan_files()->CopyFrom(lcrm_file);
+            auto* added = _metadata->add_orphan_files();
+            added->CopyFrom(lcrm_file);
+            // Transient mapper file produced and orphaned by this compaction, never referenced by
+            // visible metadata; stamp its creation version so vacuum can reclaim it rather than
+            // over-retaining it under a covering snapshot.
+            added->set_version(_new_version);
         }
 
         return Status::OK();
@@ -827,6 +837,7 @@ class PrimaryKeyTxnLogApplier : public TxnLogApplier {
                 file_meta.set_name(file.name());
                 file_meta.set_size(file.size());
                 file_meta.set_shared(file.shared());
+                file_meta.set_version(version);
                 _metadata->mutable_orphan_files()->Add(std::move(file_meta));
             }
             // Clear sstable_meta and add to orphan files.
@@ -836,6 +847,7 @@ class PrimaryKeyTxnLogApplier : public TxnLogApplier {
                 file_meta.set_name(sstable.filename());
                 file_meta.set_size(sstable.filesize());
                 file_meta.set_shared(sstable.shared());
+                file_meta.set_version(sstable.version());
                 _metadata->mutable_orphan_files()->Add(std::move(file_meta));
             }
             collect_dcg_orphan_files(old_dcg_meta, new_referenced_files, _metadata.get());