fix(scan): ignore cached manifest entries of a rewritten snapshot

The snapshot live manifest entry cache took a matching snapshot id as
a hit. A rollback deletes the newest snapshots and the commits that
follow reuse their ids, so a process that kept the cache planned the
rewritten snapshot from the entries of the deleted one: reads of the
deleted files failed and rows committed after the rollback were
missing until the ids moved past the cached ones.

Record the delta manifest list the entries were built from with every
cached snapshot and require it to match on a hit. Its name is unique
per commit, so a rewritten snapshot misses and is rebuilt in place.
The serialized form changes with it; the magic moves from SMEC to
SMED so bytes of the old layout are rebuilt rather than misread.

(cherry picked from commit 3b76467)

Author: wangyong9999

src/paimon/core/manifest/snapshot_live_manifest_entries.cpp

@@ -21,6 +21,7 @@
 
 #include <algorithm>
 #include <cstring>
+#include <string>
 #include <utility>
 
 #include "paimon/common/io/memory_segment_output_stream.h"
@@ -34,7 +35,9 @@
 namespace paimon {
 namespace {
 
-constexpr int32_t kMagic = 0x534d4543;  // SMEC
+// Bumped from SMEC when the delta manifest list was added to every cached snapshot; older bytes
+// fail the magic check and are rebuilt.
+constexpr int32_t kMagic = 0x534d4544;  // SMED
 
 size_t NormalizeMaxSnapshots(int32_t max_snapshots) {
     return static_cast<size_t>(std::max(0, max_snapshots));
@@ -68,15 +71,27 @@ std::optional<SnapshotLiveManifestEntries::Entry> SnapshotLiveManifestEntries::L
         return std::optional<Entry>();
     }
     --iter;
-    return Entry{iter->first, iter->second};
+    return Entry{iter->first, iter->second.delta_manifest_list, iter->second.entries};
 }
 
-void SnapshotLiveManifestEntries::Put(int64_t snapshot_id, std::vector<ManifestEntry>&& entries) {
+std::optional<SnapshotLiveManifestEntries::Entry> SnapshotLiveManifestEntries::Find(
+    int64_t snapshot_id, const std::string& delta_manifest_list) const {
+    auto iter = entries_by_snapshot_.find(snapshot_id);
+    if (iter == entries_by_snapshot_.end() ||
+        iter->second.delta_manifest_list != delta_manifest_list) {
+        return std::optional<Entry>();
+    }
+    return Entry{iter->first, iter->second.delta_manifest_list, iter->second.entries};
+}
+
+void SnapshotLiveManifestEntries::Put(int64_t snapshot_id, const std::string& delta_manifest_list,
+                                      std::vector<ManifestEntry>&& entries) {
     if (NormalizeMaxSnapshots(max_snapshots_) == 0) {
         return;
     }
     entries_by_snapshot_[snapshot_id] =
-        std::make_shared<const std::vector<ManifestEntry>>(std::move(entries));
+        Value{delta_manifest_list,
+              std::make_shared<const std::vector<ManifestEntry>>(std::move(entries))};
     EvictIfNeeded();
 }
 
@@ -91,9 +106,10 @@ Result<std::shared_ptr<Bytes>> SnapshotLiveManifestEntries::Serialize(
     out.WriteValue<int32_t>(static_cast<int32_t>(entries_by_snapshot_.size()));
 
     ManifestEntrySerializer serializer(pool);
-    for (const auto& [snapshot_id, entries] : entries_by_snapshot_) {
+    for (const auto& [snapshot_id, value] : entries_by_snapshot_) {
         out.WriteValue<int64_t>(snapshot_id);
-        PAIMON_RETURN_NOT_OK(serializer.SerializeList(*entries, &out));
+        out.WriteString(value.delta_manifest_list);
+        PAIMON_RETURN_NOT_OK(serializer.SerializeList(*value.entries, &out));
     }
     return ToBytes(out, pool);
 }
@@ -121,9 +137,11 @@ Result<SnapshotLiveManifestEntries> SnapshotLiveManifestEntries::Deserialize(
     ManifestEntrySerializer serializer(pool);
     for (int32_t i = 0; i < snapshot_count; i++) {
         PAIMON_ASSIGN_OR_RAISE(int64_t snapshot_id, in.ReadValue<int64_t>());
+        PAIMON_ASSIGN_OR_RAISE(std::string delta_manifest_list, in.ReadString());
         PAIMON_ASSIGN_OR_RAISE(std::vector<ManifestEntry> entries, serializer.DeserializeList(&in));
         snapshot_live_manifest_entries.entries_by_snapshot_[snapshot_id] =
-            std::make_shared<const std::vector<ManifestEntry>>(std::move(entries));
+            Value{std::move(delta_manifest_list),
+                  std::make_shared<const std::vector<ManifestEntry>>(std::move(entries))};
     }
     snapshot_live_manifest_entries.EvictIfNeeded();
     return snapshot_live_manifest_entries;

src/paimon/core/manifest/snapshot_live_manifest_entries.h

@@ -24,6 +24,7 @@
 #include <map>
 #include <memory>
 #include <optional>
+#include <string>
 #include <vector>
 
 #include "paimon/core/manifest/manifest_entry.h"
@@ -38,17 +39,27 @@ class MemorySegment;
 ///
 /// This value object owns merged live manifest entries by snapshot id. It does not own or access a
 /// cache; callers are responsible for storing the serialized bytes in the cache layer.
+///
+/// A snapshot id alone does not identify a snapshot: after a rollback, or after a table is dropped
+/// and recreated at the same path, later commits reuse the ids of the deleted snapshots. Every
+/// cached snapshot therefore also records the delta manifest list it was built from, whose name is
+/// unique per commit, and `Find()` only reports a hit when both match.
 class SnapshotLiveManifestEntries {
  public:
     struct Entry {
         int64_t snapshot_id;
+        std::string delta_manifest_list;
         std::shared_ptr<const std::vector<ManifestEntry>> entries;
     };
 
     explicit SnapshotLiveManifestEntries(int32_t max_snapshots);
 
     std::optional<Entry> LatestBeforeOrEqual(int64_t snapshot_id) const;
-    void Put(int64_t snapshot_id, std::vector<ManifestEntry>&& entries);
+    /// Returns the entries cached for exactly this snapshot: the same id, built from the same delta
+    /// manifest list. A snapshot with the same id but another delta manifest list is a miss.
+    std::optional<Entry> Find(int64_t snapshot_id, const std::string& delta_manifest_list) const;
+    void Put(int64_t snapshot_id, const std::string& delta_manifest_list,
+             std::vector<ManifestEntry>&& entries);
     size_t Size() const;
 
     Result<std::shared_ptr<Bytes>> Serialize(const std::shared_ptr<MemoryPool>& pool) const;
@@ -59,7 +70,12 @@ class SnapshotLiveManifestEntries {
  private:
     void EvictIfNeeded();
 
-    std::map<int64_t, std::shared_ptr<const std::vector<ManifestEntry>>> entries_by_snapshot_;
+    struct Value {
+        std::string delta_manifest_list;
+        std::shared_ptr<const std::vector<ManifestEntry>> entries;
+    };
+
+    std::map<int64_t, Value> entries_by_snapshot_;
     int32_t max_snapshots_;
 };
 

src/paimon/core/operation/file_store_scan.cpp

@@ -325,9 +325,10 @@ Status FileStoreScan::ReadManifestEntries(const std::vector<ManifestFileMeta>& m
 }
 
 // Cache merged live manifest entries for one bucket before applying scan filters. Each cache value
-// keeps a bounded number of snapshot results for the same table/branch/bucket. Exact snapshot hits
-// can be returned directly; cache misses rebuild the target snapshot bucket from the target
-// snapshot's data manifests.
+// keeps a bounded number of snapshot results for the same table/branch/bucket. A hit needs the same
+// snapshot id built from the same delta manifest list, because a rollback (or a table recreated at
+// the same path) reuses snapshot ids for different content; cache misses rebuild the target
+// snapshot bucket from the target snapshot's data manifests.
 Status FileStoreScan::ReadManifestEntriesWithCache(
     const Snapshot& snapshot, const std::vector<ManifestFileMeta>& all_manifest_metas,
     int32_t bucket, std::vector<ManifestEntry>* manifest_entries, bool* cache_hit) const {
@@ -339,8 +340,8 @@ Status FileStoreScan::ReadManifestEntriesWithCache(
     metrics_->ObserveHistogram(ScanMetrics::SNAPSHOT_CACHE_LOAD_DURATION,
                                static_cast<double>(cache_load_duration_ms));
     std::optional<SnapshotLiveManifestEntries::Entry> cached =
-        cached_entries.LatestBeforeOrEqual(snapshot.Id());
-    if (cached && cached->snapshot_id == snapshot.Id()) {
+        cached_entries.Find(snapshot.Id(), snapshot.DeltaManifestList());
+    if (cached) {
         *cache_hit = true;
         *manifest_entries = *cached->entries;
         return Status::OK();
@@ -358,7 +359,7 @@ Status FileStoreScan::ReadManifestEntriesWithCache(
     PAIMON_RETURN_NOT_OK(
         ReadAndMergeBucketFileEntries(bucket_manifest_metas, bucket, manifest_entries));
     std::vector<ManifestEntry> cache_entries = *manifest_entries;
-    cached_entries.Put(snapshot.Id(), std::move(cache_entries));
+    cached_entries.Put(snapshot.Id(), snapshot.DeltaManifestList(), std::move(cache_entries));
     Duration cache_store_duration;
     PAIMON_RETURN_NOT_OK(StoreSnapshotLiveManifestEntries(bucket, cached_entries));
     const uint64_t cache_store_duration_ms = cache_store_duration.Get();

src/paimon/core/operation/file_store_scan_test.cpp

@@ -203,17 +203,33 @@ TEST_F(FileStoreScanTest, TestSnapshotLiveManifestEntries) {
     snapshot1.emplace_back(FileKind::Add(), BinaryRow::EmptyRow(), /*bucket=*/0,
                            /*total_buckets=*/1, file1);
     SnapshotLiveManifestEntries entries(/*max_snapshots=*/2);
-    entries.Put(/*snapshot_id=*/1, std::move(snapshot1));
+    entries.Put(/*snapshot_id=*/1, "manifest-list-1", std::move(snapshot1));
     ASSERT_EQ(entries.Size(), 1);
     auto hit = entries.LatestBeforeOrEqual(/*snapshot_id=*/1);
     ASSERT_TRUE(hit);
     ASSERT_EQ(hit->snapshot_id, 1);
+    ASSERT_EQ(hit->delta_manifest_list, "manifest-list-1");
     ASSERT_EQ(hit->entries->size(), 1);
     ASSERT_EQ((*hit->entries)[0].FileName(), "file-1");
     auto latest_before_2 = entries.LatestBeforeOrEqual(/*snapshot_id=*/2);
     ASSERT_TRUE(latest_before_2);
     ASSERT_EQ(latest_before_2->snapshot_id, 1);
 
+    // Find() needs the same id built from the same delta manifest list.
+    auto found = entries.Find(/*snapshot_id=*/1, "manifest-list-1");
+    ASSERT_TRUE(found);
+    ASSERT_EQ((*found->entries)[0].FileName(), "file-1");
+    ASSERT_FALSE(entries.Find(/*snapshot_id=*/1, "manifest-list-1-rewritten"));
+    ASSERT_FALSE(entries.Find(/*snapshot_id=*/2, "manifest-list-1"));
+
+    // A rewritten snapshot with the same id replaces the stale entries in place.
+    entries.Put(/*snapshot_id=*/1, "manifest-list-1-rewritten", {});
+    ASSERT_EQ(entries.Size(), 1);
+    ASSERT_FALSE(entries.Find(/*snapshot_id=*/1, "manifest-list-1"));
+    auto rewritten = entries.Find(/*snapshot_id=*/1, "manifest-list-1-rewritten");
+    ASSERT_TRUE(rewritten);
+    ASSERT_TRUE(rewritten->entries->empty());
+
     std::vector<ManifestEntry> snapshot3;
     ASSERT_OK_AND_ASSIGN(
         auto file3,
@@ -225,13 +241,13 @@ TEST_F(FileStoreScanTest, TestSnapshotLiveManifestEntries) {
                                 /*write_cols=*/std::nullopt));
     snapshot3.emplace_back(FileKind::Add(), BinaryRow::EmptyRow(), /*bucket=*/0,
                            /*total_buckets=*/1, file3);
-    entries.Put(/*snapshot_id=*/3, std::move(snapshot3));
+    entries.Put(/*snapshot_id=*/3, "manifest-list-3", std::move(snapshot3));
 
     auto latest_before_4 = entries.LatestBeforeOrEqual(/*snapshot_id=*/4);
     ASSERT_TRUE(latest_before_4);
     ASSERT_EQ(latest_before_4->snapshot_id, 3);
 
-    entries.Put(/*snapshot_id=*/5, {});
+    entries.Put(/*snapshot_id=*/5, "manifest-list-5", {});
     ASSERT_EQ(entries.Size(), 2);
     ASSERT_FALSE(entries.LatestBeforeOrEqual(/*snapshot_id=*/1));
     ASSERT_TRUE(entries.LatestBeforeOrEqual(/*snapshot_id=*/3));
@@ -251,8 +267,8 @@ TEST_F(FileStoreScanTest, TestSnapshotLiveManifestEntriesSerialization) {
     manifest_entries.emplace_back(FileKind::Add(), BinaryRow::EmptyRow(), /*bucket=*/0,
                                   /*total_buckets=*/1, file1);
     SnapshotLiveManifestEntries entries(/*max_snapshots=*/2);
-    entries.Put(/*snapshot_id=*/1, std::move(manifest_entries));
-    entries.Put(/*snapshot_id=*/3, {});
+    entries.Put(/*snapshot_id=*/1, "manifest-list-1", std::move(manifest_entries));
+    entries.Put(/*snapshot_id=*/3, "manifest-list-3", {});
 
     ASSERT_OK_AND_ASSIGN(auto bytes, entries.Serialize(GetDefaultPool()));
     ASSERT_OK_AND_ASSIGN(auto deserialized,
@@ -262,9 +278,12 @@ TEST_F(FileStoreScanTest, TestSnapshotLiveManifestEntriesSerialization) {
     auto hit = deserialized.LatestBeforeOrEqual(/*snapshot_id=*/2);
     ASSERT_TRUE(hit);
     ASSERT_EQ(hit->snapshot_id, 1);
+    ASSERT_EQ(hit->delta_manifest_list, "manifest-list-1");
     ASSERT_EQ(hit->entries->size(), 1);
     ASSERT_EQ((*hit->entries)[0].FileName(), "file-1");
     ASSERT_EQ(deserialized.LatestBeforeOrEqual(/*snapshot_id=*/4)->snapshot_id, 3);
+    ASSERT_TRUE(deserialized.Find(/*snapshot_id=*/3, "manifest-list-3"));
+    ASSERT_FALSE(deserialized.Find(/*snapshot_id=*/3, "manifest-list-1"));
 
     ASSERT_OK_AND_ASSIGN(auto evicted_deserialized, SnapshotLiveManifestEntries::Deserialize(
                                                         MemorySegment::Wrap(bytes),

src/paimon/core/table/source/table_scan_test.cpp

@@ -19,16 +19,33 @@
 
 #include "paimon/table/source/table_scan.h"
 
+#include <algorithm>
+#include <cstdint>
+#include <map>
 #include <memory>
 #include <string>
 #include <utility>
 #include <vector>
 
+#include "arrow/api.h"
+#include "fmt/format.h"
 #include "gtest/gtest.h"
+#include "paimon/common/io/cache/lru_cache.h"
+#include "paimon/common/utils/checked_cast.h"
+#include "paimon/common/utils/path_util.h"
+#include "paimon/core/snapshot.h"
+#include "paimon/core/utils/snapshot_manager.h"
 #include "paimon/defs.h"
+#include "paimon/fs/file_system.h"
 #include "paimon/metrics.h"
+#include "paimon/read_context.h"
+#include "paimon/record_batch.h"
 #include "paimon/scan_context.h"
 #include "paimon/status.h"
+#include "paimon/table/source/scan_metrics.h"
+#include "paimon/table/source/table_read.h"
+#include "paimon/testing/utils/read_result_collector.h"
+#include "paimon/testing/utils/test_helper.h"
 #include "paimon/testing/utils/testharness.h"
 
 namespace paimon::test {
@@ -122,4 +139,120 @@ TEST(TableScanTest, TestReadOptimizedPrimaryKeyStreamingScanUnsupported) {
                         "key table");
 }
 
+namespace {
+
+struct CachedScanResult {
+    std::vector<std::string> rows;
+    uint64_t cache_hit = 0;
+};
+
+// Plans bucket 0 through the snapshot live manifest entry cache and reads every row back as
+// "k=v", sorted.
+Result<CachedScanResult> ScanBucketThroughCache(const std::string& table_path,
+                                                const std::map<std::string, std::string>& options,
+                                                const std::shared_ptr<Cache>& cache) {
+    ScanContextBuilder scan_builder(table_path);
+    scan_builder.SetOptions(options).WithCache(cache).SetBucketFilter(0);
+    PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<ScanContext> scan_context, scan_builder.Finish());
+    PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<TableScan> table_scan,
+                           TableScan::Create(std::move(scan_context)));
+    PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<Plan> plan, table_scan->CreatePlan());
+    CachedScanResult result;
+    std::shared_ptr<Metrics> metrics = table_scan->GetMetrics();
+    PAIMON_ASSIGN_OR_RAISE(uint64_t cache_enabled,
+                           metrics->GetCounter(ScanMetrics::LAST_SNAPSHOT_CACHE_ENABLED));
+    if (cache_enabled != 1) {
+        return Status::Invalid("the snapshot live manifest entry cache is not in use");
+    }
+    PAIMON_ASSIGN_OR_RAISE(result.cache_hit,
+                           metrics->GetCounter(ScanMetrics::LAST_SNAPSHOT_CACHE_HIT));
+
+    ReadContextBuilder read_builder(table_path);
+    read_builder.SetOptions(options).WithCache(cache);
+    PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<ReadContext> read_context, read_builder.Finish());
+    PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<TableRead> table_read,
+                           TableRead::Create(std::move(read_context)));
+    PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<BatchReader> reader,
+                           table_read->CreateReader(plan->Splits()));
+    PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::ChunkedArray> chunks,
+                           ReadResultCollector::CollectResult(std::move(reader)));
+    for (const std::shared_ptr<arrow::Array>& chunk : chunks->chunks()) {
+        const auto& rows = checked_cast<const arrow::StructArray&>(*chunk);
+        std::shared_ptr<arrow::Array> key_column = rows.GetFieldByName("k");
+        std::shared_ptr<arrow::Array> value_column = rows.GetFieldByName("v");
+        if (key_column == nullptr || key_column->type_id() != arrow::Type::STRING ||
+            value_column == nullptr || value_column->type_id() != arrow::Type::STRING) {
+            return Status::Invalid("read result misses string columns k and v");
+        }
+        auto keys = checked_pointer_cast<arrow::StringArray>(key_column);
+        auto values = checked_pointer_cast<arrow::StringArray>(value_column);
+        for (int64_t i = 0; i < rows.length(); ++i) {
+            result.rows.push_back(fmt::format("{}={}", keys->GetString(i), values->GetString(i)));
+        }
+    }
+    std::sort(result.rows.begin(), result.rows.end());
+    return result;
+}
+
+}  // namespace
+
+// A rollback deletes the newest snapshots, and the commits that follow reuse their ids for
+// different content. Entries cached for the deleted snapshot must not serve the rewritten one.
+TEST(TableScanTest, TestManifestEntryCacheIgnoresRewrittenSnapshot) {
+    std::unique_ptr<UniqueTestDirectory> dir = UniqueTestDirectory::Create("local");
+    std::map<std::string, std::string> options = {
+        {Options::BUCKET, "1"},
+        {Options::SCAN_MANIFEST_ENTRY_CACHE_MAX_SNAPSHOTS, "2"},
+        {Options::SCAN_MANIFEST_ENTRY_LAZY_DECODE_ENABLED, "true"}};
+    arrow::FieldVector fields = {arrow::field("k", arrow::utf8(), /*nullable=*/false),
+                                 arrow::field("v", arrow::utf8())};
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<TestHelper> helper,
+                         TestHelper::Create(dir->Str(), arrow::schema(fields),
+                                            /*partition_keys=*/{}, /*primary_keys=*/{"k"}, options,
+                                            /*is_streaming_mode=*/true));
+    std::string table_path = PathUtil::JoinPath(dir->Str(), "foo.db/bar");
+    auto row_type = arrow::struct_(fields);
+    auto write = [&](TestHelper* writer, const std::string& rows, int64_t commit_identifier) {
+        ASSERT_OK_AND_ASSIGN(std::unique_ptr<RecordBatch> batch,
+                             TestHelper::MakeRecordBatch(row_type, rows, /*partition_map=*/{},
+                                                         /*bucket=*/0, /*row_kinds=*/{}));
+        ASSERT_OK(writer->WriteAndCommit(std::move(batch), commit_identifier, std::nullopt));
+    };
+    write(helper.get(), R"([["k1", "v1"], ["k2", "v2"]])", 1);
+    write(helper.get(), R"([["k3", "v3"]])", 2);
+    write(helper.get(), R"([["k1", "v1b"]])", 3);
+
+    auto cache = std::make_shared<LruCache>(/*max_weight=*/64 * 1024 * 1024);
+    ASSERT_OK_AND_ASSIGN(CachedScanResult before,
+                         ScanBucketThroughCache(table_path, options, cache));
+    ASSERT_EQ(before.rows, (std::vector<std::string>{"k1=v1b", "k2=v2", "k3=v3"}));
+    ASSERT_EQ(before.cache_hit, 0);
+
+    // Roll back to snapshot 1 the way `rollback_to` does: drop the newer snapshot files and move
+    // the LATEST hint. The next commits create a new snapshot 2 and 3.
+    std::shared_ptr<FileSystem> fs = dir->GetFileSystem();
+    SnapshotManager snapshot_manager(fs, table_path);
+    ASSERT_OK_AND_ASSIGN(Snapshot old_snapshot_3, snapshot_manager.LoadSnapshot(3));
+    ASSERT_OK(fs->Delete(snapshot_manager.SnapshotPath(3), /*recursive=*/false));
+    ASSERT_OK(fs->Delete(snapshot_manager.SnapshotPath(2), /*recursive=*/false));
+    ASSERT_OK(snapshot_manager.CommitLatestHint(1));
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<TestHelper> writer_after_rollback,
+                         TestHelper::Create(table_path, options, /*is_streaming_mode=*/true));
+    write(writer_after_rollback.get(), R"([["k4", "v4"]])", 4);
+    write(writer_after_rollback.get(), R"([["k1", "v1c"]])", 5);
+    ASSERT_OK_AND_ASSIGN(Snapshot new_snapshot_3, snapshot_manager.LoadSnapshot(3));
+    ASSERT_NE(new_snapshot_3.DeltaManifestList(), old_snapshot_3.DeltaManifestList());
+
+    ASSERT_OK_AND_ASSIGN(CachedScanResult after,
+                         ScanBucketThroughCache(table_path, options, cache));
+    ASSERT_EQ(after.rows, (std::vector<std::string>{"k1=v1c", "k2=v2", "k4=v4"}));
+    ASSERT_EQ(after.cache_hit, 0);
+
+    // The rewritten snapshot 3 is cached in its own right afterwards.
+    ASSERT_OK_AND_ASSIGN(CachedScanResult again,
+                         ScanBucketThroughCache(table_path, options, cache));
+    ASSERT_EQ(again.rows, after.rows);
+    ASSERT_EQ(again.cache_hit, 1);
+}
+
 }  // namespace paimon::test