This is an automated email from the ASF dual-hosted git repository.
lxy-9602 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new 3cc1410a perf(scan): skip manifest lists on exact snapshot cache hits
(#374)
3cc1410a is described below
commit 3cc1410aa43049d78cdb63c8c5ace483028d211d
Author: wangyong9999 <[email protected]>
AuthorDate: Mon Oct 26 18:27:04 2026 +0800
perf(scan): skip manifest lists on exact snapshot cache hits (#374)
---
docs/source/user_guide/manifest_entry_cache.rst | 7 +-
.../manifest/snapshot_live_manifest_entries.cpp | 13 ++-
.../core/manifest/snapshot_live_manifest_entries.h | 6 +-
.../operation/append_only_file_store_scan_test.cpp | 47 ++++++++++-
src/paimon/core/operation/file_store_scan.cpp | 98 +++++++++++++---------
src/paimon/core/operation/file_store_scan.h | 20 +++--
src/paimon/core/operation/file_store_scan_test.cpp | 24 +++---
7 files changed, 147 insertions(+), 68 deletions(-)
diff --git a/docs/source/user_guide/manifest_entry_cache.rst
b/docs/source/user_guide/manifest_entry_cache.rst
index 0f37138c..995d77e3 100644
--- a/docs/source/user_guide/manifest_entry_cache.rst
+++ b/docs/source/user_guide/manifest_entry_cache.rst
@@ -28,8 +28,8 @@ scans that target the same bucket.
The cache stores decoded and merged live manifest entries by table path,
branch,
and bucket for ``ScanMode::ALL``. Each cache value can retain several snapshot
-results for that bucket. Exact snapshot hits are served from the cache; cache
-misses rebuild the target snapshot bucket from the target snapshot's data
+results for that bucket. Exact snapshot hits are served from the cache without
reading or decoding the
+manifest lists again; cache misses rebuild the target snapshot bucket from the
target snapshot's data
manifests and store the rebuilt live entries.
Request-specific filters are not stored in the cache. Partition, level, and
@@ -120,3 +120,6 @@ The scan metrics expose existing counters for the last scan:
- ``lastScannedManifests``: how many manifest files were loaded during this
scan before manifest entry decoding.
+
+On an exact cache hit, ``lastScannedManifests`` is zero. The cached table file
+count preserves the skipped-file metric even when manifest lists are bypassed.
diff --git a/src/paimon/core/manifest/snapshot_live_manifest_entries.cpp
b/src/paimon/core/manifest/snapshot_live_manifest_entries.cpp
index c6d981e1..6f925ba1 100644
--- a/src/paimon/core/manifest/snapshot_live_manifest_entries.cpp
+++ b/src/paimon/core/manifest/snapshot_live_manifest_entries.cpp
@@ -34,7 +34,7 @@
namespace paimon {
namespace {
-constexpr int32_t kMagic = 0x534d4543; // SMEC
+constexpr int32_t kMagic = 0x534d4544; // SMED: includes per-snapshot total
file counts
size_t NormalizeMaxSnapshots(int32_t max_snapshots) {
return static_cast<size_t>(std::max(0, max_snapshots));
@@ -61,12 +61,14 @@ std::shared_ptr<Bytes> ToBytes(const
MemorySegmentOutputStream& out,
SnapshotLiveManifestEntries::SnapshotLiveManifestEntries(int32_t max_snapshots)
: max_snapshots_(max_snapshots) {}
-void SnapshotLiveManifestEntries::Put(int64_t snapshot_id,
std::vector<ManifestEntry>&& entries) {
+void SnapshotLiveManifestEntries::Put(int64_t snapshot_id,
std::vector<ManifestEntry>&& entries,
+ int64_t total_data_files) {
if (NormalizeMaxSnapshots(max_snapshots_) == 0) {
return;
}
entries_by_snapshot_[snapshot_id] =
std::make_shared<const std::vector<ManifestEntry>>(std::move(entries));
+ total_data_files_by_snapshot_[snapshot_id] = total_data_files;
EvictIfNeeded();
}
@@ -83,6 +85,7 @@ Result<std::shared_ptr<Bytes>>
SnapshotLiveManifestEntries::Serialize(
ManifestEntrySerializer serializer(pool);
for (const auto& [snapshot_id, entries] : entries_by_snapshot_) {
out.WriteValue<int64_t>(snapshot_id);
+ out.WriteValue<int64_t>(total_data_files_by_snapshot_.at(snapshot_id));
PAIMON_RETURN_NOT_OK(serializer.SerializeList(*entries, &out));
}
return ToBytes(out, pool);
@@ -111,7 +114,10 @@ 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(int64_t total_data_files,
in.ReadValue<int64_t>());
PAIMON_ASSIGN_OR_RAISE(std::vector<ManifestEntry> entries,
serializer.DeserializeList(&in));
+
snapshot_live_manifest_entries.total_data_files_by_snapshot_[snapshot_id] =
+ total_data_files;
snapshot_live_manifest_entries.entries_by_snapshot_[snapshot_id] =
std::make_shared<const
std::vector<ManifestEntry>>(std::move(entries));
}
@@ -122,6 +128,7 @@ Result<SnapshotLiveManifestEntries>
SnapshotLiveManifestEntries::Deserialize(
void SnapshotLiveManifestEntries::EvictIfNeeded() {
size_t max_snapshots = NormalizeMaxSnapshots(max_snapshots_);
while (entries_by_snapshot_.size() > max_snapshots) {
+
total_data_files_by_snapshot_.erase(entries_by_snapshot_.begin()->first);
entries_by_snapshot_.erase(entries_by_snapshot_.begin());
}
}
@@ -151,6 +158,7 @@ SnapshotLiveManifestEntries::ReadSnapshot(
std::optional<int64_t> previous_id;
for (int32_t i = 0; i < snapshot_count; ++i) {
PAIMON_ASSIGN_OR_RAISE(int64_t id, in.ReadValue<int64_t>());
+ PAIMON_ASSIGN_OR_RAISE(int64_t total_data_files,
in.ReadValue<int64_t>());
// Serialize emits snapshots in ascending order. Validate this before
using the position
// to honor a retention limit smaller than the one used when the cache
was populated.
if (previous_id && id <= previous_id.value()) {
@@ -165,6 +173,7 @@ SnapshotLiveManifestEntries::ReadSnapshot(
if (keep) {
selected.emplace();
selected->candidate_entries = entry_count;
+ selected->total_data_files = total_data_files;
if (!row_filter) {
selected->entries.reserve(entry_count);
}
diff --git a/src/paimon/core/manifest/snapshot_live_manifest_entries.h
b/src/paimon/core/manifest/snapshot_live_manifest_entries.h
index 784d31b2..f2d4343a 100644
--- a/src/paimon/core/manifest/snapshot_live_manifest_entries.h
+++ b/src/paimon/core/manifest/snapshot_live_manifest_entries.h
@@ -45,11 +45,12 @@ class SnapshotLiveManifestEntries {
struct Selection {
std::vector<ManifestEntry> entries;
uint64_t candidate_entries = 0;
+ int64_t total_data_files = 0;
};
explicit SnapshotLiveManifestEntries(int32_t max_snapshots);
- void Put(int64_t snapshot_id, std::vector<ManifestEntry>&& entries);
+ void Put(int64_t snapshot_id, std::vector<ManifestEntry>&& entries,
int64_t total_data_files);
size_t Size() const;
Result<std::shared_ptr<Bytes>> Serialize(const
std::shared_ptr<MemoryPool>& pool) const;
@@ -59,7 +60,7 @@ class SnapshotLiveManifestEntries {
/// Select an exact retained snapshot, applying a row filter before
constructing file metadata.
/// The reader validates the entry version before invoking the filter, as
in
- /// ManifestFile::ReadSelectedEntries. The serialized cache value is
unchanged. A Selection is
+ /// ManifestFile::ReadSelectedEntries. A Selection is
/// deliberately not a cache value: query-filtered entries must never
replace the complete
/// bucket's cached entries.
static Result<std::optional<Selection>> ReadSnapshot(
@@ -71,6 +72,7 @@ class SnapshotLiveManifestEntries {
void EvictIfNeeded();
std::map<int64_t, std::shared_ptr<const std::vector<ManifestEntry>>>
entries_by_snapshot_;
+ std::map<int64_t, int64_t> total_data_files_by_snapshot_;
int32_t max_snapshots_;
};
diff --git a/src/paimon/core/operation/append_only_file_store_scan_test.cpp
b/src/paimon/core/operation/append_only_file_store_scan_test.cpp
index 29d19d7f..b8dfbfce 100644
--- a/src/paimon/core/operation/append_only_file_store_scan_test.cpp
+++ b/src/paimon/core/operation/append_only_file_store_scan_test.cpp
@@ -815,7 +815,9 @@ TEST(AppendOnlyFileStoreScanTest,
TestDropStatsAfterFiltering) {
TEST(AppendOnlyFileStoreScanTest, TestSnapshotLiveManifestCachePath) {
TimezoneGuard guard("Asia/Shanghai");
std::string table_path = paimon::test::GetDataDir() +
"/orc/append_09.db/append_09/";
- auto cache = std::make_shared<LruCache>(/*max_weight=*/16 * 1024 * 1024);
+ auto cache = std::make_shared<CountingRoutingCache>(
+ std::map<CacheKind, int64_t>{{CacheKind::MANIFEST, 16 * 1024 * 1024},
+ {CacheKind::SNAPSHOT_LIVE_MANIFEST, 16 *
1024 * 1024}});
// First scan on snapshot 5: cache miss, entries rebuilt from all
manifests.
auto scan_first = BuildScan(table_path, cache, /*bucket=*/0);
@@ -839,6 +841,11 @@ TEST(AppendOnlyFileStoreScanTest,
TestSnapshotLiveManifestCachePath) {
ASSERT_EQ(first_cache_hit, 0);
ASSERT_EQ(first_cache_misses, 1);
+ const int64_t manifest_gets = cache->GetCount(CacheKind::MANIFEST);
+ ASSERT_GT(manifest_gets, 0);
+ ASSERT_OK_AND_ASSIGN(uint64_t first_skipped,
+
first_metrics->GetCounter(ScanMetrics::LAST_SCAN_SKIPPED_TABLE_FILES));
+
// Second scan on the same snapshot should read the same bucket live
entries from cache.
auto scan_second = BuildScan(table_path, cache, /*bucket=*/0);
scan_second->WithSnapshot(snapshot_5);
@@ -855,9 +862,36 @@ TEST(AppendOnlyFileStoreScanTest,
TestSnapshotLiveManifestCachePath) {
uint64_t materialized_rows,
second_metrics->GetCounter(ScanMetrics::LAST_LAZY_DECODE_MATERIALIZED_ROWS));
ASSERT_EQ(second_cache_hit, 1);
+ ASSERT_EQ(cache->GetCount(CacheKind::MANIFEST), manifest_gets);
+ ASSERT_OK_AND_ASSIGN(uint64_t second_scanned,
+
second_metrics->GetCounter(ScanMetrics::LAST_SCANNED_MANIFESTS));
+ ASSERT_OK_AND_ASSIGN(uint64_t second_skipped,
+
second_metrics->GetCounter(ScanMetrics::LAST_SCAN_SKIPPED_TABLE_FILES));
+ ASSERT_EQ(second_scanned, 0);
+ ASSERT_EQ(second_skipped, first_skipped);
ASSERT_EQ(second_cache_hits, 1);
ASSERT_GE(scanned_rows, materialized_rows);
ASSERT_OK(second_metrics->GetHistogramStats(ScanMetrics::SNAPSHOT_CACHE_LOAD_DURATION));
+
+ // A different predicate must still filter the cached, unfiltered bucket
entries.
+ std::shared_ptr<Predicate> predicate =
+ PredicateBuilder::Equal(0, "f0", FieldType::STRING,
Literal(FieldType::STRING, "David", 5));
+ auto filtered = BuildScan(table_path, cache, /*bucket=*/0, predicate);
+ filtered->WithSnapshot(snapshot_5);
+ ASSERT_OK_AND_ASSIGN(auto filtered_plan, filtered->CreatePlan());
+ auto expected = BuildScan(table_path, nullptr, /*bucket=*/0, predicate);
+ expected->WithSnapshot(snapshot_5);
+ ASSERT_OK_AND_ASSIGN(auto expected_plan, expected->CreatePlan());
+ ASSERT_EQ(SortedFileNames(filtered_plan->Files()),
SortedFileNames(expected_plan->Files()));
+ ASSERT_EQ(cache->GetCount(CacheKind::MANIFEST), manifest_gets);
+
+ // Eviction must rebuild from source; an exact-hit shortcut cannot survive
invalidation.
+ cache->InvalidateAll();
+ auto rebuilt = BuildScan(table_path, cache, /*bucket=*/0);
+ rebuilt->WithSnapshot(snapshot_5);
+ ASSERT_OK_AND_ASSIGN(auto rebuilt_plan, rebuilt->CreatePlan());
+ ASSERT_EQ(first_file_names, SortedFileNames(rebuilt_plan->Files()));
+ ASSERT_GT(cache->GetCount(CacheKind::MANIFEST), manifest_gets);
}
TEST(AppendOnlyFileStoreScanTest, TestSnapshotLiveManifestCacheRebuildOnMiss) {
@@ -998,9 +1032,10 @@ TEST(AppendOnlyFileStoreScanTest,
SnapshotCacheRebuildsCorruptValues) {
segment.PutValue<int32_t>(offset, -1);
}
MemorySegmentOutputStream
out(MemorySegmentOutputStream::DEFAULT_SEGMENT_SIZE, pool);
- out.WriteValue<int32_t>(corruption == "magic" ? 0 : 0x534d4543);
+ out.WriteValue<int32_t>(corruption == "magic" ? 0 : 0x534d4544);
out.WriteValue<int32_t>(1);
out.WriteValue<int64_t>(snapshot.Id());
+ out.WriteValue<int64_t>(1);
out.WriteValue<int32_t>(1);
ASSERT_OK(row_serializer.Serialize(row, &out));
ASSERT_EQ(out.Segments().size(), 1);
@@ -1078,7 +1113,11 @@ TEST(AppendOnlyFileStoreScanTest,
PartitionPruningAcrossCacheModes) {
scan->partition_filter_ = counting;
ASSERT_OK_AND_ASSIGN(auto plan, scan->CreatePlan());
ASSERT_EQ(counting->row_calls_.load(), 2);
- ASSERT_GT(counting->stats_calls_.load(), 0);
+ if (mode == 2 && scans > 0) {
+ ASSERT_EQ(counting->stats_calls_.load(), 0);
+ } else {
+ ASSERT_GT(counting->stats_calls_.load(), 0);
+ }
ASSERT_EQ(plan->Files().size(), partition == 99 ? 0 : 1);
if (partition != 99) {
ASSERT_EQ(plan->Files()[0].Partition().GetInt(0),
partition);
@@ -1156,7 +1195,7 @@ TEST(AppendOnlyFileStoreScanTest,
PartitionPruningSkipsExcludedMetadataInCreateP
if (mode == 2) {
// Seed the complete snapshot directly: rebuilding must decode
every partition.
SnapshotLiveManifestEntries entries(8);
- entries.Put(1, {kept, excluded});
+ entries.Put(1, {kept, excluded}, /*total_data_files=*/2);
ASSERT_OK(scan->StoreSnapshotLiveManifestEntries(0, entries));
}
for (int32_t attempt = 0; attempt < 2; ++attempt) {
diff --git a/src/paimon/core/operation/file_store_scan.cpp
b/src/paimon/core/operation/file_store_scan.cpp
index 4e508bab..ca2de308 100644
--- a/src/paimon/core/operation/file_store_scan.cpp
+++ b/src/paimon/core/operation/file_store_scan.cpp
@@ -158,8 +158,8 @@ Result<std::vector<PartitionEntry>>
FileStoreScan::ReadPartitionEntries() const
PAIMON_RETURN_NOT_OK(
ReadManifests(&snapshot, &all_manifest_file_metas,
&filtered_manifest_file_metas));
std::vector<ManifestEntry> manifest_entries;
- PAIMON_RETURN_NOT_OK(ReadFileEntries(filtered_manifest_file_metas,
&manifest_entries,
- /*apply_scan_filter=*/true));
+ PAIMON_RETURN_NOT_OK(ReadFileEntries(filtered_manifest_file_metas,
/*apply_scan_filter=*/true,
+ &manifest_entries));
std::unordered_map<BinaryRow, PartitionEntry> partitions;
PAIMON_RETURN_NOT_OK(PartitionEntry::Merge(manifest_entries, &partitions));
@@ -178,11 +178,10 @@ Result<std::shared_ptr<FileStoreScan::RawPlan>>
FileStoreScan::CreatePlan() cons
Duration manifest_read_duration;
metrics_->SetCounter(ScanMetrics::LAST_SNAPSHOT_CACHE_LOAD_DURATION, 0);
metrics_->SetCounter(ScanMetrics::LAST_SNAPSHOT_CACHE_STORE_DURATION, 0);
- std::optional<Snapshot> snapshot;
+ PAIMON_ASSIGN_OR_RAISE(std::optional<Snapshot> snapshot, ReadSnapshot());
+ int64_t all_data_files = 0;
std::vector<ManifestFileMeta> all_manifest_file_metas;
std::vector<ManifestFileMeta> filtered_manifest_file_metas;
- PAIMON_RETURN_NOT_OK(
- ReadManifests(&snapshot, &all_manifest_file_metas,
&filtered_manifest_file_metas));
std::vector<ManifestEntry> manifest_entries;
std::optional<int32_t> cache_bucket = bucket_filter_;
@@ -197,11 +196,19 @@ Result<std::shared_ptr<FileStoreScan::RawPlan>>
FileStoreScan::CreatePlan() cons
uint64_t lazy_decode_scanned_rows = 0;
bool snapshot_cache_hit = false;
if (use_snapshot_live_manifest_cache) {
- PAIMON_ASSIGN_OR_RAISE(lazy_decode_scanned_rows,
- ReadManifestEntriesWithCache(
- snapshot.value(), all_manifest_file_metas,
cache_bucket.value(),
- &manifest_entries, &snapshot_cache_hit));
+ PAIMON_ASSIGN_OR_RAISE(
+ lazy_decode_scanned_rows,
+ ReadManifestEntriesWithCache(snapshot.value(),
cache_bucket.value(),
+ &all_manifest_file_metas,
&filtered_manifest_file_metas,
+ &manifest_entries,
&snapshot_cache_hit, &all_data_files));
} else {
+ PAIMON_RETURN_NOT_OK(
+ ReadManifestLists(snapshot, &all_manifest_file_metas,
&filtered_manifest_file_metas));
+ all_data_files =
+ std::accumulate(all_manifest_file_metas.begin(),
all_manifest_file_metas.end(),
+ int64_t{0}, [](int64_t sum, const
ManifestFileMeta& meta) {
+ return sum + meta.NumAddedFiles() -
meta.NumDeletedFiles();
+ });
lazy_decode_scanned_rows = std::accumulate(
filtered_manifest_file_metas.begin(),
filtered_manifest_file_metas.end(), uint64_t{0},
[](uint64_t sum, const ManifestFileMeta& meta) {
@@ -240,11 +247,6 @@ Result<std::shared_ptr<FileStoreScan::RawPlan>>
FileStoreScan::CreatePlan() cons
entry = entry.CopyWithoutStats();
}
}
- const int64_t all_data_files = std::accumulate(
- all_manifest_file_metas.begin(), all_manifest_file_metas.end(),
int64_t{0},
- [](const int64_t sum, const ManifestFileMeta& manifest_file_meta) {
- return sum + manifest_file_meta.NumAddedFiles() -
manifest_file_meta.NumDeletedFiles();
- });
const uint64_t scan_duration_ms = duration.Get();
metrics_->SetCounter(ScanMetrics::LAST_SCAN_DURATION, scan_duration_ms);
metrics_->ObserveHistogram(ScanMetrics::SCAN_DURATION,
static_cast<double>(scan_duration_ms));
@@ -276,27 +278,31 @@ Result<std::shared_ptr<FileStoreScan::RawPlan>>
FileStoreScan::CreatePlan() cons
std::move(manifest_entries));
}
-Status FileStoreScan::ReadManifests(std::optional<Snapshot>* snapshot_ptr,
- std::vector<ManifestFileMeta>*
all_manifests_ptr,
- std::vector<ManifestFileMeta>*
filter_manifests_ptr) const {
- auto& snapshot = *snapshot_ptr;
- auto& all_manifests = *all_manifests_ptr;
- auto& filtered_manifests = *filter_manifests_ptr;
- if (specified_snapshot_ != std::nullopt) {
- snapshot = specified_snapshot_;
- } else {
- PAIMON_ASSIGN_OR_RAISE(snapshot, snapshot_manager_->LatestSnapshot());
+Result<std::optional<Snapshot>> FileStoreScan::ReadSnapshot() const {
+ if (specified_snapshot_) {
+ return specified_snapshot_;
}
- if (snapshot == std::nullopt) {
- all_manifests = std::vector<ManifestFileMeta>();
- filtered_manifests = std::vector<ManifestFileMeta>();
+ return snapshot_manager_->LatestSnapshot();
+}
+
+Status FileStoreScan::ReadManifests(std::optional<Snapshot>* snapshot,
+ std::vector<ManifestFileMeta>*
all_manifests,
+ std::vector<ManifestFileMeta>*
filtered_manifests) const {
+ PAIMON_ASSIGN_OR_RAISE(*snapshot, ReadSnapshot());
+ return ReadManifestLists(*snapshot, all_manifests, filtered_manifests);
+}
+
+Status FileStoreScan::ReadManifestLists(const std::optional<Snapshot>&
snapshot,
+ std::vector<ManifestFileMeta>*
all_manifests,
+ std::vector<ManifestFileMeta>*
filtered_manifests) const {
+ if (!snapshot) {
return Status::OK();
}
- PAIMON_RETURN_NOT_OK(ReadManifestsWithSnapshot(snapshot.value(),
&all_manifests));
- for (const auto& meta : all_manifests) {
- PAIMON_ASSIGN_OR_RAISE(bool filter_meta_result,
FilterManifestFileMeta(meta));
- if (filter_meta_result) {
- filtered_manifests.push_back(meta);
+ PAIMON_RETURN_NOT_OK(ReadManifestsWithSnapshot(snapshot.value(),
all_manifests));
+ for (const auto& meta : *all_manifests) {
+ PAIMON_ASSIGN_OR_RAISE(bool keep, FilterManifestFileMeta(meta));
+ if (keep) {
+ filtered_manifests->push_back(meta);
}
}
return Status::OK();
@@ -343,8 +349,8 @@ Status FileStoreScan::ReadManifestsWithSnapshot(const
Snapshot& snapshot,
}
Status FileStoreScan::ReadFileEntries(const std::vector<ManifestFileMeta>&
manifest_metas,
- std::vector<ManifestEntry>*
manifest_entries,
- bool apply_scan_filter) const {
+ bool apply_scan_filter,
+ std::vector<ManifestEntry>*
manifest_entries) const {
std::vector<std::future<Result<std::vector<ManifestEntry>>>> futures;
for (const auto& meta : manifest_metas) {
auto read_meta_task = [this, meta,
@@ -389,8 +395,9 @@ Status FileStoreScan::ReadManifestEntries(const
std::vector<ManifestFileMeta>& m
// can be returned directly; cache misses rebuild the target snapshot bucket
from the target
// snapshot's data manifests.
Result<uint64_t> FileStoreScan::ReadManifestEntriesWithCache(
- const Snapshot& snapshot, const std::vector<ManifestFileMeta>&
all_manifest_metas,
- int32_t bucket, std::vector<ManifestEntry>* manifest_entries, bool*
cache_hit) const {
+ const Snapshot& snapshot, int32_t bucket, std::vector<ManifestFileMeta>*
all_manifest_metas,
+ std::vector<ManifestFileMeta>* filtered_manifest_metas,
+ std::vector<ManifestEntry>* manifest_entries, bool* cache_hit, int64_t*
all_data_files) const {
Duration cache_load_duration;
const int32_t max_snapshots =
core_options_.GetScanManifestEntryCacheMaxSnapshots();
SnapshotLiveManifestEntries cached_entries(max_snapshots);
@@ -413,6 +420,7 @@ Result<uint64_t>
FileStoreScan::ReadManifestEntriesWithCache(
}
if (selected.ok() && selected.value()) {
*cache_hit = true;
+ *all_data_files = selected.value()->total_data_files;
partition_filtered = static_cast<bool>(row_filter);
candidate_entries = selected.value()->candidate_entries;
*manifest_entries = std::move(selected.value()->entries);
@@ -432,8 +440,16 @@ Result<uint64_t>
FileStoreScan::ReadManifestEntriesWithCache(
metrics_->ObserveHistogram(ScanMetrics::SNAPSHOT_CACHE_LOAD_DURATION,
static_cast<double>(cache_load_duration_ms));
if (!*cache_hit) {
+ // The selected snapshot is unchanged. Exact hits skip both lists and
their decoding.
+ PAIMON_RETURN_NOT_OK(
+ ReadManifestLists(snapshot, all_manifest_metas,
filtered_manifest_metas));
+ *all_data_files =
+ std::accumulate(all_manifest_metas->begin(),
all_manifest_metas->end(), int64_t{0},
+ [](int64_t sum, const ManifestFileMeta& meta) {
+ return sum + meta.NumAddedFiles() -
meta.NumDeletedFiles();
+ });
std::vector<ManifestFileMeta> bucket_manifest_metas;
- for (const auto& meta : all_manifest_metas) {
+ for (const auto& meta : *all_manifest_metas) {
if ((!bucket_filter_ && bucket_selector_) ||
MayContainBucket(meta, bucket)) {
bucket_manifest_metas.push_back(meta);
}
@@ -442,7 +458,7 @@ Result<uint64_t>
FileStoreScan::ReadManifestEntriesWithCache(
ReadAndMergeBucketFileEntries(bucket_manifest_metas, bucket,
manifest_entries));
candidate_entries = manifest_entries->size();
std::vector<ManifestEntry> cache_entries = *manifest_entries;
- cached_entries.Put(snapshot.Id(), std::move(cache_entries));
+ cached_entries.Put(snapshot.Id(), std::move(cache_entries),
*all_data_files);
Duration cache_store_duration;
PAIMON_RETURN_NOT_OK(StoreSnapshotLiveManifestEntries(bucket,
cached_entries));
const uint64_t cache_store_duration_ms = cache_store_duration.Get();
@@ -534,7 +550,7 @@ Status FileStoreScan::ReadAndMergeBucketFileEntries(
std::vector<ManifestEntry> unmerged_entries;
std::vector<ManifestEntry> entries;
- PAIMON_RETURN_NOT_OK(ReadFileEntries(manifest_metas, &entries,
/*apply_scan_filter=*/false));
+ PAIMON_RETURN_NOT_OK(ReadFileEntries(manifest_metas,
/*apply_scan_filter=*/false, &entries));
unmerged_entries.reserve(entries.size());
for (auto& entry : entries) {
if (entry.Bucket() == bucket ||
@@ -566,14 +582,14 @@ Status FileStoreScan::ReadAndMergeFileEntries(const
std::vector<ManifestFileMeta
std::vector<ManifestEntry>*
merged_entries) const {
std::vector<ManifestEntry> unmerged_entries;
PAIMON_RETURN_NOT_OK(
- ReadFileEntries(manifest_metas, &unmerged_entries,
/*apply_scan_filter=*/true));
+ ReadFileEntries(manifest_metas, /*apply_scan_filter=*/true,
&unmerged_entries));
return MergeLiveEntries(unmerged_entries, merged_entries);
}
Status FileStoreScan::ReadAndNoMergeFileEntries(
const std::vector<ManifestFileMeta>& manifest_metas,
std::vector<ManifestEntry>* manifest_entries) const {
- return ReadFileEntries(manifest_metas, manifest_entries,
/*apply_scan_filter=*/true);
+ return ReadFileEntries(manifest_metas, /*apply_scan_filter=*/true,
manifest_entries);
}
bool FileStoreScan::MayContainBucket(const ManifestFileMeta& manifest, int32_t
bucket) const {
diff --git a/src/paimon/core/operation/file_store_scan.h
b/src/paimon/core/operation/file_store_scan.h
index 25911d0e..df8c6d61 100644
--- a/src/paimon/core/operation/file_store_scan.h
+++ b/src/paimon/core/operation/file_store_scan.h
@@ -276,9 +276,13 @@ class FileStoreScan {
const std::shared_ptr<TableSchema>&
data_schema) const;
private:
- Status ReadManifests(std::optional<Snapshot>* snapshot_ptr,
- std::vector<ManifestFileMeta>* all_manifests_ptr,
- std::vector<ManifestFileMeta>*
filtered_manifests_ptr) const;
+ Status ReadManifests(std::optional<Snapshot>* snapshot,
+ std::vector<ManifestFileMeta>* all_manifests,
+ std::vector<ManifestFileMeta>* filtered_manifests)
const;
+ Result<std::optional<Snapshot>> ReadSnapshot() const;
+ Status ReadManifestLists(const std::optional<Snapshot>& snapshot,
+ std::vector<ManifestFileMeta>* all_manifests,
+ std::vector<ManifestFileMeta>*
filtered_manifests) const;
Status ReadManifestsWithSnapshot(const Snapshot& snapshot,
std::vector<ManifestFileMeta>* manifests)
const;
@@ -289,8 +293,10 @@ class FileStoreScan {
/// Read request-filtered entries while caching complete bucket snapshots.
/// @return Number of entries considered before request-specific filtering.
Result<uint64_t> ReadManifestEntriesWithCache(
- const Snapshot& snapshot, const std::vector<ManifestFileMeta>&
bucket_manifest_metas,
- int32_t bucket, std::vector<ManifestEntry>* manifest_entries, bool*
cache_hit) const;
+ const Snapshot& snapshot, int32_t bucket,
std::vector<ManifestFileMeta>* all_manifest_metas,
+ std::vector<ManifestFileMeta>* filtered_manifest_metas,
+ std::vector<ManifestEntry>* manifest_entries, bool* cache_hit,
+ int64_t* all_data_files) const;
std::shared_ptr<CacheKey>
CreateSnapshotLiveManifestEntriesCacheKey(int32_t bucket) const;
Status StoreSnapshotLiveManifestEntries(int32_t bucket,
const SnapshotLiveManifestEntries&
entries) const;
@@ -306,8 +312,8 @@ class FileStoreScan {
std::vector<ManifestEntry>*
manifest_entries) const;
Status ReadFileEntries(const std::vector<ManifestFileMeta>& manifest_metas,
- std::vector<ManifestEntry>* manifest_entries,
- bool apply_scan_filter) const;
+ bool apply_scan_filter,
+ std::vector<ManifestEntry>* manifest_entries) const;
bool MayContainBucket(const ManifestFileMeta& manifest, int32_t bucket)
const;
diff --git a/src/paimon/core/operation/file_store_scan_test.cpp
b/src/paimon/core/operation/file_store_scan_test.cpp
index 5c53f138..eb49f354 100644
--- a/src/paimon/core/operation/file_store_scan_test.cpp
+++ b/src/paimon/core/operation/file_store_scan_test.cpp
@@ -232,8 +232,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, std::move(manifest_entries),
/*total_data_files=*/1);
+ entries.Put(/*snapshot_id=*/3, {}, /*total_data_files=*/0);
ASSERT_OK_AND_ASSIGN(auto bytes, entries.Serialize(GetDefaultPool()));
ASSERT_OK_AND_ASSIGN(auto deserialized,
@@ -273,9 +273,9 @@ TEST_F(FileStoreScanTest, SelectExactRetainedSnapshot) {
BinaryRowGenerator::GenerateRow({partition},
pool.get()), 0, 1, file);
}
SnapshotLiveManifestEntries entries(3);
- entries.Put(1, {records[0]});
- entries.Put(3, {records[1], records[2]});
- entries.Put(5, {});
+ entries.Put(1, {records[0]}, 1);
+ entries.Put(3, {records[1], records[2]}, 17);
+ entries.Put(5, {}, 0);
ASSERT_OK_AND_ASSIGN(auto bytes, entries.Serialize(pool));
const MemorySegment segment = MemorySegment::Wrap(bytes);
ASSERT_OK_AND_ASSIGN(auto first,
@@ -300,6 +300,7 @@ TEST_F(FileStoreScanTest, SelectExactRetainedSnapshot) {
pool));
ASSERT_TRUE(selected);
ASSERT_EQ(selected->candidate_entries, 2);
+ ASSERT_EQ(selected->total_data_files, 17);
ASSERT_EQ(filter_calls, 2);
ASSERT_EQ(selected->entries, std::vector<ManifestEntry>({records[2]}));
ASSERT_OK_AND_ASSIGN(auto absent,
@@ -326,7 +327,7 @@ TEST_F(FileStoreScanTest, SelectExactRetainedSnapshot) {
pool),
"injected filter error");
- entries.Put(7, {});
+ entries.Put(7, {}, 0);
ASSERT_EQ(entries.Size(), 3);
ASSERT_OK_AND_ASSIGN(auto updated, entries.Serialize(pool));
// Put must evict the oldest snapshot even when the reader permits more
snapshots.
@@ -336,6 +337,7 @@ TEST_F(FileStoreScanTest, SelectExactRetainedSnapshot) {
ASSERT_OK_AND_ASSIGN(auto retained,
SnapshotLiveManifestEntries::ReadSnapshot(
MemorySegment::Wrap(updated), 3,
4, nullptr, pool));
ASSERT_TRUE(retained);
+ ASSERT_EQ(retained->total_data_files, 17);
ASSERT_EQ(retained->entries, std::vector<ManifestEntry>({records[1],
records[2]}));
}
@@ -349,7 +351,7 @@ TEST_F(FileStoreScanTest,
SkipDecodingFilteredManifestEntries) {
ManifestEntry excluded(FileKind(static_cast<int8_t>(2)),
BinaryRowGenerator::GenerateRow({10}, pool.get()),
0, 1, file);
SnapshotLiveManifestEntries entries(1);
- entries.Put(1, {excluded});
+ entries.Put(1, {excluded}, 1);
ASSERT_OK_AND_ASSIGN(auto bytes, entries.Serialize(pool));
const MemorySegment segment = MemorySegment::Wrap(bytes);
ASSERT_OK_AND_ASSIGN(auto selected,
SnapshotLiveManifestEntries::ReadSnapshot(
@@ -373,9 +375,10 @@ TEST_F(FileStoreScanTest,
RejectUnsupportedSnapshotEntryVersionBeforeFiltering)
BinaryRow row = BinaryRowGenerator::GenerateRow({3}, pool.get());
BinaryRowSerializer serializer(1, pool);
MemorySegmentOutputStream out(64, pool);
- out.WriteValue<int32_t>(0x534d4543);
+ out.WriteValue<int32_t>(0x534d4544);
out.WriteValue<int32_t>(1);
out.WriteValue<int64_t>(1);
+ out.WriteValue<int64_t>(7);
out.WriteValue<int32_t>(1);
ASSERT_OK(serializer.Serialize(row, &out));
bool filter_called = false;
@@ -400,16 +403,17 @@ TEST_F(FileStoreScanTest,
RejectInvalidSnapshotSelectionHeader) {
out.WriteValue<int32_t>(magic);
out.WriteValue<int32_t>(snapshots);
out.WriteValue<int64_t>(1);
+ out.WriteValue<int64_t>(7);
out.WriteValue<int32_t>(entries);
return out.Segments().front();
};
auto segment = header(0, 1, 0);
ASSERT_NOK_WITH_MSG(SnapshotLiveManifestEntries::ReadSnapshot(segment, 1,
3, nullptr, pool),
"invalid snapshot live manifest entries magic");
- segment = header(0x534d4543, -1, 0);
+ segment = header(0x534d4544, -1, 0);
ASSERT_NOK_WITH_MSG(SnapshotLiveManifestEntries::ReadSnapshot(segment, 1,
3, nullptr, pool),
"snapshot count is negative");
- segment = header(0x534d4543, 1, -1);
+ segment = header(0x534d4544, 1, -1);
ASSERT_NOK_WITH_MSG(SnapshotLiveManifestEntries::ReadSnapshot(segment, 1,
3, nullptr, pool),
"entry count is negative");
auto truncated = MemorySegment::WrapView(segment.Data(), 7);
