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 b0439d93 perf(scan): prune manifest partitions before metadata
decoding (#430)
b0439d93 is described below
commit b0439d9318b51b0c5f422a86d7573cbc182ff415
Author: gripleaf <[email protected]>
AuthorDate: Mon Oct 19 08:52:02 2026 +0800
perf(scan): prune manifest partitions before metadata decoding (#430)
---
docs/source/user_guide/manifest_entry_cache.rst | 41 ++-
include/paimon/defs.h | 3 +-
.../data/serializer/binary_row_serializer.cpp | 12 +
.../common/data/serializer/binary_row_serializer.h | 3 +
.../data/serializer/binary_row_serializer_test.cpp | 34 +++
src/paimon/common/utils/serialization_utils.h | 50 +++-
.../common/utils/serialization_utils_test.cpp | 46 +++
.../core/manifest/manifest_entry_serializer.cpp | 4 +
.../core/manifest/manifest_entry_serializer.h | 5 +
.../manifest/manifest_entry_serializer_test.cpp | 6 +
src/paimon/core/manifest/manifest_file.cpp | 82 ++---
src/paimon/core/manifest/manifest_file.h | 17 +-
.../core/manifest/manifest_row_range_test.cpp | 124 ++++++--
.../manifest/snapshot_live_manifest_entries.cpp | 73 ++++-
.../core/manifest/snapshot_live_manifest_entries.h | 19 +-
.../operation/append_only_file_store_scan_test.cpp | 330 ++++++++++++++++++++-
src/paimon/core/operation/file_store_scan.cpp | 177 ++++++-----
src/paimon/core/operation/file_store_scan.h | 18 +-
src/paimon/core/operation/file_store_scan_test.cpp | 231 +++++++++++----
19 files changed, 1034 insertions(+), 241 deletions(-)
diff --git a/docs/source/user_guide/manifest_entry_cache.rst
b/docs/source/user_guide/manifest_entry_cache.rst
index 97b33cde..0f37138c 100644
--- a/docs/source/user_guide/manifest_entry_cache.rst
+++ b/docs/source/user_guide/manifest_entry_cache.rst
@@ -36,6 +36,37 @@ Request-specific filters are not stored in the cache.
Partition, level, and
predicate filters are still evaluated for each scan, so cached entries can be
reused safely across different scan predicates for the same bucket.
+Selective manifest decoding
+---------------------------
+
+With ``Options::SCAN_MANIFEST_ENTRY_LAZY_DECODE_ENABLED`` enabled (the
default),
+scans evaluate partition filters before constructing ``ManifestEntry`` and
+``DataFileMeta`` objects or decoding file statistics. The same row filter runs
+on ordinary manifest reads, cached manifest Arrow batches, and exact snapshot
+cache hits. This avoids full metadata decoding for excluded partitions even
+when no cache is configured. Row-range reads apply both range and partition
+selection before decoding; ADD and DELETE entries receive the same selection.
+Level and data predicates still run through the normal scan filters.
+
+The snapshot cache keeps the complete merged entries for each bucket and
+snapshot in serialized form. A hit selects the exact retained snapshot.
+Rows belonging to other snapshots are skipped by their serialized lengths,
+without allocating or copying their payloads. When there is no partition
filter,
+all entries of the selected snapshot are decoded. Disabling lazy decoding also
+restores full entry decoding before partition filtering.
+
+For example, if a bucket contains 604 entries across 84 partitions and a query
+needs six entries from one partition, the scan inspects 604 partition fields
+but fully decodes only six entries. This reduces metadata allocation and
+statistics decoding without introducing another cache or changing the
+serialized cache format.
+
+The filtered result is request-local. Manifest Arrow caches retain complete
+batches, and snapshot cache misses rebuild the complete bucket before storing
+it. A later scan with another partition filter or snapshot cannot reuse an
+incomplete result.
+The existing snapshot retention limit also applies to selective reads.
+
Configuration
-------------
@@ -67,11 +98,11 @@ the cache path. Set it to a positive value to enable the
cache when
is still controlled by the configured ``Cache`` implementation, for example the
capacity of ``LruCache``.
-If no cache is provided through ``ScanContextBuilder::WithCache()``, this
-optimization is skipped. The snapshot manifest entry cache shares the same
-``Cache`` interface with raw manifest and data-file footer caches, but it uses
a
-dedicated ``CacheKind`` and a table/branch/bucket key instead of file byte
-ranges.
+If no cache is provided through ``ScanContextBuilder::WithCache()``, snapshot
+caching is skipped; selective manifest decoding remains available. The snapshot
+manifest entry cache shares the same ``Cache`` interface with raw manifest and
+data-file footer caches, but it uses a dedicated ``CacheKind`` and a
+table/branch/bucket key instead of file byte ranges.
Limitations
-----------
diff --git a/include/paimon/defs.h b/include/paimon/defs.h
index 8c88c3de..88cb78bd 100644
--- a/include/paimon/defs.h
+++ b/include/paimon/defs.h
@@ -224,7 +224,8 @@ struct PAIMON_EXPORT Options {
static const char SCAN_MANIFEST_ENTRY_CACHE_MAX_SNAPSHOTS[];
/// "scan.manifest-entry.lazy-decode.enabled" - Whether to deserialize
only manifest entries
- /// for the target bucket when rebuilding the cache. Default value is true.
+ /// selected by bucket, partition, and row-range pruning before
constructing file metadata.
+ /// Applies to supported manifest read paths with or without caching.
Default value is true.
static const char SCAN_MANIFEST_ENTRY_LAZY_DECODE_ENABLED[];
/// "prefetch.io-metrics.enabled" - Whether to collect per-I/O metrics for
prefetch reads.
diff --git a/src/paimon/common/data/serializer/binary_row_serializer.cpp
b/src/paimon/common/data/serializer/binary_row_serializer.cpp
index 56075ac0..5c406a4e 100644
--- a/src/paimon/common/data/serializer/binary_row_serializer.cpp
+++ b/src/paimon/common/data/serializer/binary_row_serializer.cpp
@@ -53,4 +53,16 @@ Result<BinaryRow>
BinaryRowSerializer::Deserialize(DataInputStream* source) cons
return row;
}
+Status BinaryRowSerializer::Skip(DataInputStream* source) {
+ PAIMON_ASSIGN_OR_RAISE(int32_t read_length, source->ReadValue<int32_t>());
+ PAIMON_ASSIGN_OR_RAISE(int64_t pos, source->GetPos());
+ PAIMON_ASSIGN_OR_RAISE(int64_t length, source->Length());
+ if (read_length < 0 || pos < 0 || pos > length || read_length > length -
pos) {
+ return Status::Invalid(
+ fmt::format("invalid binary row length {} at position {}, stream
length {}",
+ read_length, pos, length));
+ }
+ return source->Seek(pos + read_length);
+}
+
} // namespace paimon
diff --git a/src/paimon/common/data/serializer/binary_row_serializer.h
b/src/paimon/common/data/serializer/binary_row_serializer.h
index b1f0a05a..1b8b90f3 100644
--- a/src/paimon/common/data/serializer/binary_row_serializer.h
+++ b/src/paimon/common/data/serializer/binary_row_serializer.h
@@ -57,6 +57,9 @@ class BinaryRowSerializer {
Result<BinaryRow> Deserialize(DataInputStream* source) const;
+ /// Skip a length-prefixed row without allocating or copying its payload.
+ static Status Skip(DataInputStream* source);
+
Status Serialize(const BinaryRow& row, MemorySegmentOutputStream* target)
const;
private:
diff --git a/src/paimon/common/data/serializer/binary_row_serializer_test.cpp
b/src/paimon/common/data/serializer/binary_row_serializer_test.cpp
index 51a6d3aa..298d77fa 100644
--- a/src/paimon/common/data/serializer/binary_row_serializer_test.cpp
+++ b/src/paimon/common/data/serializer/binary_row_serializer_test.cpp
@@ -70,4 +70,38 @@ TEST_F(BinaryRowSerializerTest, SerializeAndDeserialize) {
EXPECT_EQ(deserialized_row.GetDouble(2), 3.14);
}
+TEST_F(BinaryRowSerializerTest, SkipPreservesNextRow) {
+ BinaryRowSerializer serializer(1, pool_);
+ MemorySegmentOutputStream out(128, pool_);
+ for (int32_t value : {10, 20}) {
+ BinaryRow row(1);
+ BinaryRowWriter writer(&row, 0, pool_.get());
+ writer.WriteInt(0, value);
+ writer.Complete();
+ ASSERT_OK(serializer.Serialize(row, &out));
+ }
+ auto stream =
+ std::make_shared<ByteArrayInputStream>(out.Segments().front().Data(),
out.CurrentSize());
+ DataInputStream in(stream);
+ ASSERT_OK(BinaryRowSerializer::Skip(&in));
+ ASSERT_OK_AND_ASSIGN(auto row, serializer.Deserialize(&in));
+ ASSERT_EQ(row.GetInt(0), 20);
+ ASSERT_OK_AND_ASSIGN(int64_t pos, in.GetPos());
+ ASSERT_EQ(pos, out.CurrentSize());
+ ASSERT_NOK(BinaryRowSerializer::Skip(&in));
+}
+
+TEST_F(BinaryRowSerializerTest, SkipRejectsInvalidLength) {
+ for (int32_t length : {-1, 5}) {
+ SCOPED_TRACE(length);
+ MemorySegmentOutputStream out(64, pool_);
+ out.WriteValue<int32_t>(length);
+ out.WriteValue<int32_t>(0);
+ auto stream =
std::make_shared<ByteArrayInputStream>(out.Segments().front().Data(),
+
out.CurrentSize());
+ DataInputStream in(stream);
+ ASSERT_NOK_WITH_MSG(BinaryRowSerializer::Skip(&in), "invalid binary
row length");
+ }
+}
+
} // namespace paimon::test
diff --git a/src/paimon/common/utils/serialization_utils.h
b/src/paimon/common/utils/serialization_utils.h
index 449766ca..8c5d7a73 100644
--- a/src/paimon/common/utils/serialization_utils.h
+++ b/src/paimon/common/utils/serialization_utils.h
@@ -21,7 +21,9 @@
#include <cstdint>
#include <cstring>
+#include <limits>
#include <memory>
+#include <string_view>
#include <utility>
#include "fmt/format.h"
@@ -75,31 +77,53 @@ class SerializationUtils {
/// Schemaless deserialization for `BinaryRow`.
static Result<BinaryRow> DeserializeBinaryRow(const
std::shared_ptr<Bytes>& bytes) {
- if (PAIMON_UNLIKELY(bytes->size() < 4)) {
- return Status::Invalid(fmt::format("bytes size {} is less than 4",
bytes->size()));
+ return DeserializeBinaryRow(MemorySegment::Wrap(bytes));
+ }
+
+ /// Borrow a schemaless row. The caller must keep the backing bytes alive
while using it.
+ static Result<BinaryRow> DeserializeBinaryRowView(std::string_view bytes) {
+ if (PAIMON_UNLIKELY(bytes.size() >
std::numeric_limits<int32_t>::max())) {
+ return Status::Invalid("binary row exceeds maximum segment size");
+ }
+ return DeserializeBinaryRow(
+ MemorySegment::WrapView(bytes.data(),
static_cast<int32_t>(bytes.size())));
+ }
+
+ /// Schemaless deserialization for `BinaryRow` from a `DataInputStream`.
+ static Result<BinaryRow> DeserializeBinaryRow(DataInputStream* input,
MemoryPool* pool) {
+ int32_t read_length = -1;
+ PAIMON_ASSIGN_OR_RAISE(read_length, input->ReadValue<int32_t>());
+ std::shared_ptr<Bytes> bytes = Bytes::AllocateBytes(read_length, pool);
+ PAIMON_RETURN_NOT_OK(input->ReadBytes(bytes.get()));
+ return DeserializeBinaryRow(bytes);
+ }
+
+ private:
+ static Result<BinaryRow> DeserializeBinaryRow(const MemorySegment&
segment) {
+ if (PAIMON_UNLIKELY(segment.Size() < 4)) {
+ return Status::Invalid(fmt::format("bytes size {} is less than 4",
segment.Size()));
}
// The buffer is byte-filled, so memcpy avoids the strict-aliasing UB
of reinterpret_cast.
int32_t arity;
- memcpy(&arity, bytes->data(), sizeof(int32_t));
+ memcpy(&arity, segment.Data(), sizeof(int32_t));
if (SystemByteOrder() == ByteOrder::PAIMON_LITTLE_ENDIAN) {
arity = EndianSwapValue(arity);
}
if (PAIMON_UNLIKELY(arity < 0)) {
return Status::Invalid("arity is less than 0");
}
+ const int32_t payload_size = segment.Size() - 4;
+ // Bound the arity before the bitset helper, then compute the total
size without overflow.
+ if (PAIMON_UNLIKELY(arity > payload_size / 8 ||
+ BinaryRow::CalculateBitSetWidthInBytes(arity) +
+ static_cast<int64_t>(arity) * 8 >
+ payload_size)) {
+ return Status::Invalid("binary row fixed fields exceed serialized
length");
+ }
BinaryRow row(arity);
- row.PointTo(MemorySegment::Wrap(bytes), 4, bytes->size() - 4);
+ row.PointTo(segment, 4, segment.Size() - 4);
return row;
}
-
- /// Schemaless deserialization for `BinaryRow` from a `DataInputStream`.
- static Result<BinaryRow> DeserializeBinaryRow(DataInputStream* input,
MemoryPool* pool) {
- int32_t read_length = -1;
- PAIMON_ASSIGN_OR_RAISE(read_length, input->ReadValue<int32_t>());
- std::shared_ptr<Bytes> bytes = Bytes::AllocateBytes(read_length, pool);
- PAIMON_RETURN_NOT_OK(input->ReadBytes(bytes.get()));
- return DeserializeBinaryRow(bytes);
- }
};
} // namespace paimon
diff --git a/src/paimon/common/utils/serialization_utils_test.cpp
b/src/paimon/common/utils/serialization_utils_test.cpp
index 56a39a29..3df436a9 100644
--- a/src/paimon/common/utils/serialization_utils_test.cpp
+++ b/src/paimon/common/utils/serialization_utils_test.cpp
@@ -85,4 +85,50 @@ TEST_F(SerializationUtilsTest,
TestDeserializeBinaryRowFromStream) {
ASSERT_EQ(de_row.GetString(2).ToString(), "hello paimon!");
}
+TEST_F(SerializationUtilsTest, DeserializeBorrowedBinaryRow) {
+ auto pool = GetDefaultPool();
+ BinaryRow row(3);
+ BinaryRowWriter writer(&row, 0, pool.get());
+ writer.WriteInt(0, -42);
+ writer.SetNullAt(1);
+ writer.WriteString(2, BinaryString::FromString("partition", pool.get()));
+ writer.Complete();
+ auto bytes = SerializationUtils::SerializeBinaryRow(row, pool.get());
+ // Use an unaligned backing address, as Arrow binary values need not be
aligned.
+ std::string backing = "x" + std::string(bytes->data(), bytes->size());
+ std::string_view data(backing.data() + 1, bytes->size());
+ ASSERT_OK_AND_ASSIGN(auto view,
SerializationUtils::DeserializeBinaryRowView(data));
+ ASSERT_EQ(view.GetSegment().Data(), data.data());
+ ASSERT_EQ(view.GetInt(0), -42);
+ ASSERT_TRUE(view.IsNullAt(1));
+ ASSERT_EQ(view.GetString(2).ToString(), "partition");
+ ASSERT_EQ(view, row);
+
+ ASSERT_OK_AND_ASSIGN(auto owned,
SerializationUtils::DeserializeBinaryRow(bytes));
+ bytes.reset();
+ backing.clear();
+ ASSERT_EQ(owned, row);
+ auto empty_bytes =
SerializationUtils::SerializeBinaryRow(BinaryRow::EmptyRow(), pool.get());
+ ASSERT_OK_AND_ASSIGN(auto empty,
SerializationUtils::DeserializeBinaryRowView(std::string_view(
+ empty_bytes->data(),
empty_bytes->size())));
+ ASSERT_EQ(empty, BinaryRow::EmptyRow());
+}
+
+TEST_F(SerializationUtilsTest, RejectInvalidBorrowedBinaryRow) {
+ const std::string negative_arity(4, static_cast<char>(0xff));
+
ASSERT_NOK_WITH_MSG(SerializationUtils::DeserializeBinaryRowView(negative_arity),
+ "arity is less than 0");
+ const std::string huge_arity = {0x7f, static_cast<char>(0xff),
static_cast<char>(0xff),
+ static_cast<char>(0xff)};
+
ASSERT_NOK_WITH_MSG(SerializationUtils::DeserializeBinaryRowView(huge_arity),
+ "fixed fields exceed serialized length");
+ auto bytes =
+ SerializationUtils::SerializeBinaryRow(BinaryRow::EmptyRow(),
GetDefaultPool().get());
+ for (size_t size = 0; size < bytes->size(); ++size) {
+ SCOPED_TRACE(size);
+ ASSERT_NOK(
+
SerializationUtils::DeserializeBinaryRowView(std::string_view(bytes->data(),
size)));
+ }
+}
+
} // namespace paimon::test
diff --git a/src/paimon/core/manifest/manifest_entry_serializer.cpp
b/src/paimon/core/manifest/manifest_entry_serializer.cpp
index 5e1a2be0..fec2b6a5 100644
--- a/src/paimon/core/manifest/manifest_entry_serializer.cpp
+++ b/src/paimon/core/manifest/manifest_entry_serializer.cpp
@@ -48,6 +48,10 @@ int32_t ManifestEntrySerializer::GetBucket(const
InternalRow& row) {
return row.GetInt(3);
}
+Result<BinaryRow> ManifestEntrySerializer::GetPartitionView(const InternalRow&
row) {
+ return SerializationUtils::DeserializeBinaryRowView(row.GetStringView(2));
+}
+
Result<ManifestEntry> ManifestEntrySerializer::ConvertFrom(int32_t version,
const InternalRow&
row) const {
PAIMON_RETURN_NOT_OK(ValidateVersion(version));
diff --git a/src/paimon/core/manifest/manifest_entry_serializer.h
b/src/paimon/core/manifest/manifest_entry_serializer.h
index 9d4ddcfa..5f334aad 100644
--- a/src/paimon/core/manifest/manifest_entry_serializer.h
+++ b/src/paimon/core/manifest/manifest_entry_serializer.h
@@ -56,6 +56,11 @@ class ManifestEntrySerializer : public
VersionedObjectSerializer<ManifestEntry>
/// Get the bucket from a versioned manifest entry row without fully
deserializing it.
static int32_t GetBucket(const InternalRow& row);
+ /// Borrow the partition without copying or constructing file metadata.
+ /// The caller must validate the entry version before accessing this field.
+ /// The row backing memory must remain valid while the returned view is
used.
+ static Result<BinaryRow> GetPartitionView(const InternalRow& row);
+
Result<ManifestEntry> ConvertFrom(int32_t version, const InternalRow& row)
const override;
Result<BinaryRow> ToRow(const ManifestEntry& record) const override;
diff --git a/src/paimon/core/manifest/manifest_entry_serializer_test.cpp
b/src/paimon/core/manifest/manifest_entry_serializer_test.cpp
index a31f4bf7..2838829e 100644
--- a/src/paimon/core/manifest/manifest_entry_serializer_test.cpp
+++ b/src/paimon/core/manifest/manifest_entry_serializer_test.cpp
@@ -57,9 +57,15 @@ TEST_F(ManifestEntrySerializerTest, TestToFromRow) {
for (const auto& entry : entries) {
ASSERT_OK_AND_ASSIGN(auto row, serializer.ToRow(entry));
ASSERT_EQ(entry.Bucket(), ManifestEntrySerializer::GetBucket(row));
+ ASSERT_OK_AND_ASSIGN(auto partition,
ManifestEntrySerializer::GetPartitionView(row));
+ ASSERT_EQ(entry.Partition(), partition);
+ ASSERT_EQ(partition.GetSegment().Data(), row.GetStringView(2).data());
ASSERT_OK_AND_ASSIGN(auto result_entry, serializer.FromRow(row));
ASSERT_EQ(entry, result_entry);
ASSERT_EQ(entry.ToString(), result_entry.ToString());
+ // The final entry owns its partition even after the source row is
released.
+ row = BinaryRow::EmptyRow();
+ ASSERT_EQ(entry.Partition(), result_entry.Partition());
}
}
diff --git a/src/paimon/core/manifest/manifest_file.cpp
b/src/paimon/core/manifest/manifest_file.cpp
index c0a71a2f..f47a63a9 100644
--- a/src/paimon/core/manifest/manifest_file.cpp
+++ b/src/paimon/core/manifest/manifest_file.cpp
@@ -134,53 +134,69 @@ Status ManifestFile::ReadBucketEntries(const std::string&
file_name, int32_t buc
});
}
-Status ManifestFile::ReadRowRangeEntries(
- const std::string& file_name, const RowRangeIndex& row_ranges,
- const std::function<Result<bool>(const ManifestEntry&)>& filter,
- std::optional<int64_t> file_size, std::vector<ManifestEntry>* entries)
const {
+Status ManifestFile::ReadSelectedEntries(
+ const std::string& file_name, std::optional<int64_t> file_size, const
RowRangeIndex* row_ranges,
+ const std::function<Result<bool>(const InternalRow&)>& row_filter,
+ const std::function<Result<bool>(const ManifestEntry&)>& entry_filter,
+ std::vector<ManifestEntry>* entries) const {
+ if (!row_ranges && !row_filter) {
+ return Read(file_name, entry_filter, file_size, entries);
+ }
return ReadArrowBatches(
file_name, file_size,
- [this, &row_ranges, &filter,
+ [this, row_ranges, &row_filter, &entry_filter,
entries](const std::shared_ptr<arrow::StructArray>& batch) -> Status {
- // ManifestMetaReader has aligned both the entry and its nested
file schema. Probe
- // the two range columns without allocating DataFileMeta, stats or
binary keys.
- // The serialized entry has a leading _VERSION field.
- const auto& file_column =
batch->field(ManifestEntry::kFileFieldIndex + 1);
- if (file_column->type_id() != arrow::Type::STRUCT) {
- return Status::Invalid("Manifest entry file metadata must be a
struct");
- }
- auto files = checked_pointer_cast<arrow::StructArray>(file_column);
- const auto& count_column =
files->field(DataFileMeta::kRowCountFieldIndex);
- const auto& first_column =
files->field(DataFileMeta::kFirstRowIdFieldIndex);
- if (count_column->type_id() != arrow::Type::INT64 ||
- first_column->type_id() != arrow::Type::INT64) {
- return Status::Invalid("Manifest entry row range must contain
int64 fields");
+ // ManifestMetaReader has aligned both the entry and its nested
file schema.
+ // Only row-range scans need to probe the file columns before
entry decoding.
+ std::shared_ptr<arrow::StructArray> files;
+ std::shared_ptr<arrow::Int64Array> counts;
+ std::shared_ptr<arrow::Int64Array> first_ids;
+ if (row_ranges) {
+ const auto& file_column =
batch->field(ManifestEntry::kFileFieldIndex + 1);
+ if (file_column->type_id() != arrow::Type::STRUCT) {
+ return Status::Invalid("Manifest entry file metadata must
be a struct");
+ }
+ files = checked_pointer_cast<arrow::StructArray>(file_column);
+ const auto& count_column =
files->field(DataFileMeta::kRowCountFieldIndex);
+ const auto& first_column =
files->field(DataFileMeta::kFirstRowIdFieldIndex);
+ if (count_column->type_id() != arrow::Type::INT64 ||
+ first_column->type_id() != arrow::Type::INT64) {
+ return Status::Invalid("Manifest entry row range must
contain int64 fields");
+ }
+ counts = checked_pointer_cast<arrow::Int64Array>(count_column);
+ first_ids =
checked_pointer_cast<arrow::Int64Array>(first_column);
}
- auto counts =
checked_pointer_cast<arrow::Int64Array>(count_column);
- auto first_ids =
checked_pointer_cast<arrow::Int64Array>(first_column);
ColumnarRow row(batch->fields(), pool_, /*row_id=*/0);
for (int64_t i = 0; i < batch->length(); ++i) {
row.SetRowId(i);
PAIMON_RETURN_NOT_OK(
ManifestEntrySerializer::ValidateVersion(row.GetInt(kVersionFieldIndex)));
- if (files->IsNull(i)) {
- return Status::Invalid(
- "ManifestEntry convert from row failed, with null
DataFileMeta");
+ if (row_ranges) {
+ if (files->IsNull(i)) {
+ return Status::Invalid(
+ "ManifestEntry convert from row failed, with null
DataFileMeta");
+ }
+ if (!first_ids->IsNull(i) && !counts->IsNull(i)) {
+ const int64_t first = first_ids->Value(i);
+ const int64_t count = counts->Value(i);
+ // Missing or invalid ranges cannot prove an entry is
irrelevant. In
+ // particular, do not overflow when computing the
inclusive upper bound.
+ if (first >= 0 && count > 0 &&
+ first <= std::numeric_limits<int64_t>::max() -
(count - 1) &&
+ !row_ranges->Intersects(first, first + (count -
1))) {
+ continue;
+ }
+ }
}
- if (!first_ids->IsNull(i) && !counts->IsNull(i)) {
- const int64_t first = first_ids->Value(i);
- const int64_t count = counts->Value(i);
- // Missing or invalid ranges cannot prove an entry is
irrelevant. In
- // particular, do not overflow when computing the
inclusive upper bound.
- if (first >= 0 && count > 0 &&
- first <= std::numeric_limits<int64_t>::max() - (count
- 1) &&
- !row_ranges.Intersects(first, first + (count - 1))) {
+ if (row_filter) {
+ PAIMON_ASSIGN_OR_RAISE(bool keep, row_filter(row));
+ if (!keep) {
continue;
}
}
PAIMON_ASSIGN_OR_RAISE(ManifestEntry entry,
serializer_->FromRow(row));
- if (filter) {
- PAIMON_ASSIGN_OR_RAISE(bool keep, filter(entry));
+ if (entry_filter) {
+ PAIMON_ASSIGN_OR_RAISE(bool keep, entry_filter(entry));
if (!keep) {
continue;
}
diff --git a/src/paimon/core/manifest/manifest_file.h
b/src/paimon/core/manifest/manifest_file.h
index c37882b5..22ffba49 100644
--- a/src/paimon/core/manifest/manifest_file.h
+++ b/src/paimon/core/manifest/manifest_file.h
@@ -78,13 +78,16 @@ class ManifestFile : public ObjectsFile<ManifestEntry> {
std::optional<int64_t> file_size,
std::vector<ManifestEntry>* entries) const;
- /// Read entries intersecting row ID ranges before constructing their file
metadata.
- /// Unknown row ranges are retained. Add and Delete entries use the same
selection, and
- /// the ordinary entry filter still runs on every retained entry.
- Status ReadRowRangeEntries(const std::string& file_name, const
RowRangeIndex& row_ranges,
- const std::function<Result<bool>(const
ManifestEntry&)>& filter,
- std::optional<int64_t> file_size,
- std::vector<ManifestEntry>* entries) const;
+ /// Select entries by row ranges and a row filter before constructing file
metadata.
+ /// A null row_ranges or row_filter disables that selection. Unknown
ranges are retained.
+ /// Add and Delete entries use the same selection. The entry filter runs
after decoding.
+ /// The reader validates the entry version before inspecting row ranges or
invoking the filter.
+ Status ReadSelectedEntries(
+ const std::string& file_name, std::optional<int64_t> file_size,
+ const RowRangeIndex* row_ranges,
+ const std::function<Result<bool>(const InternalRow&)>& row_filter,
+ const std::function<Result<bool>(const ManifestEntry&)>& entry_filter,
+ std::vector<ManifestEntry>* entries) const;
private:
Status PrepareBucketRead(int32_t bucket, const std::optional<int32_t>&
expected_total_buckets,
diff --git a/src/paimon/core/manifest/manifest_row_range_test.cpp
b/src/paimon/core/manifest/manifest_row_range_test.cpp
index a4f60fea..db2dd75c 100644
--- a/src/paimon/core/manifest/manifest_row_range_test.cpp
+++ b/src/paimon/core/manifest/manifest_row_range_test.cpp
@@ -46,6 +46,7 @@
#include "paimon/format/format_writer.h"
#include "paimon/format/writer_builder.h"
#include "paimon/fs/local/local_file_system.h"
+#include "paimon/testing/utils/binary_row_generator.h"
#include "paimon/testing/utils/counting_cache_test_utils.h"
#include "paimon/testing/utils/testharness.h"
#include "paimon/utils/row_range_index.h"
@@ -106,6 +107,65 @@ class RowRangeManifestFileTest : public ::testing::Test {
}
};
+TEST_F(RowRangeManifestFileTest, PartitionFilterComposesWithRowRanges) {
+ for (bool cache_enabled : {false, true}) {
+ SCOPED_TRACE(cache_enabled);
+ auto dir = UniqueTestDirectory::Create();
+ ASSERT_TRUE(dir);
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<ManifestFile> manifest,
+ CreateManifest(dir->Str(), dir->GetFileSystem(),
cache_enabled));
+ auto pool = GetDefaultPool();
+ auto partition10 = BinaryRowGenerator::GenerateRow({10}, pool.get());
+ auto partition20 = BinaryRowGenerator::GenerateRow({20}, pool.get());
+ ASSERT_OK_AND_ASSIGN(ManifestEntry first, Entry("a.parquet", 100, 10));
+ ASSERT_OK_AND_ASSIGN(ManifestEntry second, Entry("b.parquet", 100,
10));
+ ASSERT_OK_AND_ASSIGN(ManifestEntry third, Entry("c.parquet", 200, 10));
+ ManifestEntry a(FileKind::Add(), partition10, 0, 1, first.File());
+ ManifestEntry deleted(FileKind::Delete(), partition10, 0, 1,
first.File());
+ // Decoding the excluded partition would reject its invalid file kind.
+ ManifestEntry b(FileKind(static_cast<int8_t>(2)), partition20, 0, 1,
second.File());
+ ManifestEntry c(FileKind::Add(), partition10, 0, 1, third.File());
+ std::vector<ManifestEntry> source = {a, deleted, b, c};
+ using WrittenFile = std::pair<std::string, int64_t>;
+ ASSERT_OK_AND_ASSIGN(WrittenFile written,
manifest->WriteWithoutRolling(source));
+ auto row_filter = [](const InternalRow& row) -> Result<bool> {
+ PAIMON_ASSIGN_OR_RAISE(BinaryRow partition,
+
ManifestEntrySerializer::GetPartitionView(row));
+ return partition.GetInt(0) == 10;
+ };
+ std::vector<ManifestEntry> selected;
+ ASSERT_OK(manifest->ReadSelectedEntries(
+ written.first, written.second, /*row_ranges=*/nullptr, row_filter,
nullptr, &selected));
+ ASSERT_EQ(std::vector<ManifestEntry>({a, deleted, c}), selected);
+ std::vector<ManifestEntry> live;
+ ASSERT_OK(FileStoreScan::MergeLiveEntries(selected, &live));
+ ASSERT_EQ(std::vector<ManifestEntry>({c}), live);
+
+ ASSERT_OK_AND_ASSIGN(RowRangeIndex ranges,
RowRangeIndex::Create({Range(100, 100)}));
+ selected.clear();
+ ASSERT_OK(manifest->ReadSelectedEntries(written.first, written.second,
&ranges, row_filter,
+ nullptr, &selected));
+ ASSERT_EQ(std::vector<ManifestEntry>({a, deleted}), selected);
+ live.clear();
+ ASSERT_OK(FileStoreScan::MergeLiveEntries(selected, &live));
+ ASSERT_TRUE(live.empty());
+
+ auto error_filter = [](const InternalRow&) -> Result<bool> {
+ return Status::IOError("injected row filter error");
+ };
+ selected.clear();
+ ASSERT_NOK_WITH_MSG(
+ manifest->ReadSelectedEntries(written.first, written.second,
/*row_ranges=*/nullptr,
+ error_filter, nullptr, &selected),
+ "injected row filter error");
+ // The unfiltered path still decodes the excluded entry from the
complete batches.
+ selected.clear();
+ ASSERT_NOK_WITH_MSG(manifest->ReadSelectedEntries(written.first,
written.second, nullptr,
+ nullptr, nullptr,
&selected),
+ "Unsupported byte value 2 for file kind.");
+ }
+}
+
TEST_F(RowRangeManifestFileTest, ArrowCacheReuseEvictionAndConcurrentReaders) {
auto dir = UniqueTestDirectory::Create();
ASSERT_TRUE(dir);
@@ -123,8 +183,8 @@ TEST_F(RowRangeManifestFileTest,
ArrowCacheReuseEvictionAndConcurrentReaders) {
PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<ManifestFile> reader,
CreateManifest(dir->Str(),
dir->GetFileSystem(), true, cache));
std::vector<ManifestEntry> entries;
- PAIMON_RETURN_NOT_OK(
- reader->ReadRowRangeEntries(written.first, ranges, nullptr,
written.second, &entries));
+ PAIMON_RETURN_NOT_OK(reader->ReadSelectedEntries(
+ written.first, written.second, &ranges, /*row_filter=*/nullptr,
nullptr, &entries));
if (entries != expected) {
return Status::Invalid("cached manifest result differs from
source");
}
@@ -148,7 +208,8 @@ TEST_F(RowRangeManifestFileTest,
ArrowCacheReuseEvictionAndConcurrentReaders) {
// A different immutable manifest must not reuse the previous file's
cached content.
ASSERT_OK_AND_ASSIGN(WrittenFile next, manifest->WriteWithoutRolling({a}));
std::vector<ManifestEntry> entries;
- ASSERT_OK(manifest->ReadRowRangeEntries(next.first, ranges, nullptr,
next.second, &entries));
+ ASSERT_OK(manifest->ReadSelectedEntries(next.first, next.second, &ranges,
+ /*row_filter=*/nullptr, nullptr,
&entries));
ASSERT_TRUE(entries.empty());
ASSERT_EQ(3, cache->SupplierCallCount(cache_kind));
@@ -156,15 +217,15 @@ TEST_F(RowRangeManifestFileTest,
ArrowCacheReuseEvictionAndConcurrentReaders) {
std::make_shared<CountingRoutingCache>(CacheKind::DATA_FILE_FOOTER,
1024 * 1024);
ASSERT_OK_AND_ASSIGN(std::unique_ptr<ManifestFile> unsupported,
CreateManifest(dir->Str(), dir->GetFileSystem(),
true, unsupported_cache));
- ASSERT_OK(
- unsupported->ReadRowRangeEntries(written.first, ranges, nullptr,
written.second, &entries));
+ ASSERT_OK(unsupported->ReadSelectedEntries(written.first, written.second,
&ranges,
+ /*row_filter=*/nullptr,
nullptr, &entries));
ASSERT_EQ(expected, entries);
entries.clear();
auto small_cache = std::make_shared<LruCache>(1);
ASSERT_OK_AND_ASSIGN(std::unique_ptr<ManifestFile> uncached,
CreateManifest(dir->Str(), dir->GetFileSystem(),
true, small_cache));
- ASSERT_OK(
- uncached->ReadRowRangeEntries(written.first, ranges, nullptr,
written.second, &entries));
+ ASSERT_OK(uncached->ReadSelectedEntries(written.first, written.second,
&ranges,
+ /*row_filter=*/nullptr, nullptr,
&entries));
ASSERT_EQ(expected, entries);
ASSERT_LE(small_cache->GetCurrentWeight(), 1);
small_cache->InvalidateAll();
@@ -302,7 +363,8 @@ TEST_F(RowRangeManifestFileTest,
OrcCachePreservesManifestMetadata) {
// This legacy manifest has unknown row IDs, so every entry must be
retained.
ASSERT_OK_AND_ASSIGN(RowRangeIndex ranges, RowRangeIndex::Create({Range(0,
0)}));
std::vector<ManifestEntry> without_cache;
- ASSERT_OK(uncached->ReadRowRangeEntries(name, ranges, nullptr,
std::nullopt, &without_cache));
+ ASSERT_OK(uncached->ReadSelectedEntries(name, std::nullopt, &ranges,
/*row_filter=*/nullptr,
+ nullptr, &without_cache));
ASSERT_EQ(expected, without_cache);
auto cache = std::make_shared<CountingRoutingCache>(CacheKind::MANIFEST,
1024 * 1024);
@@ -313,7 +375,8 @@ TEST_F(RowRangeManifestFileTest,
OrcCachePreservesManifestMetadata) {
std::unique_ptr<ManifestFile> reader,
ManifestFile::Create(fs, format, "zstd", paths, 1024, pool,
options, schema));
std::vector<ManifestEntry> actual;
- ASSERT_OK(reader->ReadRowRangeEntries(name, ranges, nullptr,
std::nullopt, &actual));
+ ASSERT_OK(reader->ReadSelectedEntries(name, std::nullopt, &ranges,
/*row_filter=*/nullptr,
+ nullptr, &actual));
ASSERT_EQ(expected, actual);
ASSERT_EQ(1, cache->SupplierCallCount(CacheKind::MANIFEST));
}
@@ -330,8 +393,8 @@ TEST_F(RowRangeManifestFileTest, EmptyManifestCacheReuse) {
ASSERT_OK_AND_ASSIGN(RowRangeIndex ranges, RowRangeIndex::Create({Range(0,
0)}));
for (int32_t attempt = 0; attempt < 2; ++attempt) {
std::vector<ManifestEntry> entries;
- ASSERT_OK(manifest->ReadRowRangeEntries(written.first, ranges,
nullptr, written.second,
- &entries));
+ ASSERT_OK(manifest->ReadSelectedEntries(written.first, written.second,
&ranges,
+ /*row_filter=*/nullptr,
nullptr, &entries));
ASSERT_TRUE(entries.empty());
ASSERT_EQ(1, cache->SupplierCallCount(CacheKind::MANIFEST));
}
@@ -362,8 +425,8 @@ TEST_F(RowRangeManifestFileTest,
AllReadPathsShareOneDecodedEntry) {
written.second,
&actual));
ASSERT_EQ(source, actual);
} else {
- ASSERT_OK(manifest->ReadRowRangeEntries(written.first, ranges,
nullptr,
- written.second,
&actual));
+ ASSERT_OK(manifest->ReadSelectedEntries(written.first,
written.second, &ranges,
+
/*row_filter=*/nullptr, nullptr, &actual));
ASSERT_EQ(std::vector<ManifestEntry>{b}, actual);
}
};
@@ -582,8 +645,8 @@ TEST_F(RowRangeManifestFileTest,
BoundariesUnknownRangesAndDeleteMerging) {
std::vector<ManifestEntry> ordinary;
ASSERT_OK(manifest->Read(written.first, filter, written.second,
&ordinary));
std::vector<ManifestEntry> selected;
- ASSERT_OK(manifest->ReadRowRangeEntries(written.first, ranges,
filter, written.second,
- &selected));
+ ASSERT_OK(manifest->ReadSelectedEntries(written.first,
written.second, &ranges,
+ /*row_filter=*/nullptr,
filter, &selected));
ASSERT_EQ(ordinary, selected);
std::vector<ManifestEntry> ordinary_live;
std::vector<ManifestEntry> selected_live;
@@ -596,15 +659,15 @@ TEST_F(RowRangeManifestFileTest,
BoundariesUnknownRangesAndDeleteMerging) {
}
ASSERT_OK_AND_ASSIGN(RowRangeIndex ranges,
RowRangeIndex::Create({Range(100, 100)}));
std::vector<ManifestEntry> entries;
- ASSERT_NOK_WITH_MSG(manifest->ReadRowRangeEntries(
- written.first, ranges,
+ ASSERT_NOK_WITH_MSG(manifest->ReadSelectedEntries(
+ written.first, written.second, &ranges,
/*row_filter=*/nullptr,
[](const ManifestEntry&) -> Result<bool> {
return Status::IOError("filter error");
},
- written.second, &entries),
+ &entries),
"filter error");
- ASSERT_NOK(manifest->ReadRowRangeEntries("missing-manifest", ranges,
nullptr, std::nullopt,
- &entries));
+ ASSERT_NOK(manifest->ReadSelectedEntries("missing-manifest",
std::nullopt, &ranges,
+ /*row_filter=*/nullptr,
nullptr, &entries));
}
}
@@ -626,8 +689,8 @@ TEST_F(RowRangeManifestFileTest,
UncertainAndOverflowingRangesAreRetained) {
ASSERT_OK_AND_ASSIGN(WrittenFile written,
manifest->WriteWithoutRolling(source));
ASSERT_OK_AND_ASSIGN(RowRangeIndex ranges, RowRangeIndex::Create({Range(0,
0)}));
std::vector<ManifestEntry> actual;
- ASSERT_OK(
- manifest->ReadRowRangeEntries(written.first, ranges, nullptr,
written.second, &actual));
+ ASSERT_OK(manifest->ReadSelectedEntries(written.first, written.second,
&ranges,
+ /*row_filter=*/nullptr, nullptr,
&actual));
source.erase(source.begin() + 1);
ASSERT_EQ(source, actual);
}
@@ -684,11 +747,22 @@ TEST_F(RowRangeManifestFileTest,
SchemaEvolutionAndVersionValidation) {
std::vector<ManifestEntry> actual;
if (mode >= 2) {
ASSERT_NOK_WITH_MSG(
- manifest->ReadRowRangeEntries(name, ranges, nullptr,
std::nullopt, &actual),
+ manifest->ReadSelectedEntries(name, std::nullopt, &ranges,
+ /*row_filter=*/nullptr,
nullptr, &actual),
(mode == 2 ? "not compatible" : "Unsupported version:
999"));
+ bool filter_called = false;
+ ASSERT_NOK_WITH_MSG(manifest->ReadSelectedEntries(
+ name, std::nullopt,
/*row_ranges=*/nullptr,
+ [&filter_called](const InternalRow&)
-> Result<bool> {
+ filter_called = true;
+ return false;
+ },
+ nullptr, &actual),
+ (mode == 2 ? "not compatible" :
"Unsupported version: 999"));
+ ASSERT_FALSE(filter_called);
} else {
- ASSERT_OK(
- manifest->ReadRowRangeEntries(name, ranges, nullptr,
std::nullopt, &actual));
+ ASSERT_OK(manifest->ReadSelectedEntries(name, std::nullopt,
&ranges,
+
/*row_filter=*/nullptr, nullptr, &actual));
ASSERT_EQ(mode == 0 ? 1 : 0, actual.size());
if (mode == 0) {
ASSERT_FALSE(actual[0].File()->first_row_id.has_value());
diff --git a/src/paimon/core/manifest/snapshot_live_manifest_entries.cpp
b/src/paimon/core/manifest/snapshot_live_manifest_entries.cpp
index 8982209a..c6d981e1 100644
--- a/src/paimon/core/manifest/snapshot_live_manifest_entries.cpp
+++ b/src/paimon/core/manifest/snapshot_live_manifest_entries.cpp
@@ -61,16 +61,6 @@ std::shared_ptr<Bytes> ToBytes(const
MemorySegmentOutputStream& out,
SnapshotLiveManifestEntries::SnapshotLiveManifestEntries(int32_t max_snapshots)
: max_snapshots_(max_snapshots) {}
-std::optional<SnapshotLiveManifestEntries::Entry>
SnapshotLiveManifestEntries::LatestBeforeOrEqual(
- int64_t snapshot_id) const {
- auto iter = entries_by_snapshot_.upper_bound(snapshot_id);
- if (iter == entries_by_snapshot_.begin()) {
- return std::optional<Entry>();
- }
- --iter;
- return Entry{iter->first, iter->second};
-}
-
void SnapshotLiveManifestEntries::Put(int64_t snapshot_id,
std::vector<ManifestEntry>&& entries) {
if (NormalizeMaxSnapshots(max_snapshots_) == 0) {
return;
@@ -136,4 +126,67 @@ void SnapshotLiveManifestEntries::EvictIfNeeded() {
}
}
+Result<std::optional<SnapshotLiveManifestEntries::Selection>>
+SnapshotLiveManifestEntries::ReadSnapshot(
+ const MemorySegment& segment, int64_t snapshot_id, int32_t max_snapshots,
+ const std::function<Result<bool>(const InternalRow&)>& row_filter,
+ const std::shared_ptr<MemoryPool>& pool) {
+ std::optional<Selection> selected;
+ if (segment.Data() == nullptr || segment.Size() == 0 || max_snapshots <=
0) {
+ return selected;
+ }
+ auto bytes = segment.GetOrCreateHeapMemory(pool.get());
+ auto stream = std::make_shared<ByteArrayInputStream>(bytes->data(),
bytes->size());
+ DataInputStream in(stream);
+ PAIMON_ASSIGN_OR_RAISE(int32_t magic, in.ReadValue<int32_t>());
+ if (magic != kMagic) {
+ return Status::Invalid("invalid snapshot live manifest entries magic");
+ }
+ PAIMON_ASSIGN_OR_RAISE(int32_t snapshot_count, in.ReadValue<int32_t>());
+ if (snapshot_count < 0) {
+ return Status::Invalid("snapshot live manifest entries snapshot count
is negative");
+ }
+ ManifestEntrySerializer serializer(pool);
+ BinaryRowSerializer row_serializer(serializer.NumFields(), pool);
+ 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>());
+ // 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()) {
+ return Status::Invalid("snapshot live manifest entries are not
ordered");
+ }
+ previous_id = id;
+ const bool keep = id == snapshot_id && i >= snapshot_count -
max_snapshots;
+ PAIMON_ASSIGN_OR_RAISE(int32_t entry_count, in.ReadValue<int32_t>());
+ if (entry_count < 0) {
+ return Status::Invalid("snapshot live manifest entries entry count
is negative");
+ }
+ if (keep) {
+ selected.emplace();
+ selected->candidate_entries = entry_count;
+ if (!row_filter) {
+ selected->entries.reserve(entry_count);
+ }
+ }
+ for (int32_t j = 0; j < entry_count; ++j) {
+ if (!keep) {
+ PAIMON_RETURN_NOT_OK(BinaryRowSerializer::Skip(&in));
+ continue;
+ }
+ PAIMON_ASSIGN_OR_RAISE(BinaryRow row,
row_serializer.Deserialize(&in));
+ if (row_filter) {
+
PAIMON_RETURN_NOT_OK(ManifestEntrySerializer::ValidateVersion(row.GetInt(0)));
+ PAIMON_ASSIGN_OR_RAISE(bool matches, row_filter(row));
+ if (!matches) {
+ continue;
+ }
+ }
+ PAIMON_ASSIGN_OR_RAISE(ManifestEntry entry,
serializer.FromRow(row));
+ selected->entries.emplace_back(std::move(entry));
+ }
+ }
+ return selected;
+}
+
} // namespace paimon
diff --git a/src/paimon/core/manifest/snapshot_live_manifest_entries.h
b/src/paimon/core/manifest/snapshot_live_manifest_entries.h
index acf1d8e7..784d31b2 100644
--- a/src/paimon/core/manifest/snapshot_live_manifest_entries.h
+++ b/src/paimon/core/manifest/snapshot_live_manifest_entries.h
@@ -21,6 +21,7 @@
#include <cstddef>
#include <cstdint>
+#include <functional>
#include <map>
#include <memory>
#include <optional>
@@ -31,6 +32,7 @@
namespace paimon {
class Bytes;
+class InternalRow;
class MemoryPool;
class MemorySegment;
@@ -40,14 +42,13 @@ class MemorySegment;
/// cache; callers are responsible for storing the serialized bytes in the
cache layer.
class SnapshotLiveManifestEntries {
public:
- struct Entry {
- int64_t snapshot_id;
- std::shared_ptr<const std::vector<ManifestEntry>> entries;
+ struct Selection {
+ std::vector<ManifestEntry> entries;
+ uint64_t candidate_entries = 0;
};
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);
size_t Size() const;
@@ -56,6 +57,16 @@ class SnapshotLiveManifestEntries {
int32_t
max_snapshots,
const
std::shared_ptr<MemoryPool>& pool);
+ /// 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
+ /// deliberately not a cache value: query-filtered entries must never
replace the complete
+ /// bucket's cached entries.
+ static Result<std::optional<Selection>> ReadSnapshot(
+ const MemorySegment& segment, int64_t snapshot_id, int32_t
max_snapshots,
+ const std::function<Result<bool>(const InternalRow&)>& row_filter,
+ const std::shared_ptr<MemoryPool>& pool);
+
private:
void EvictIfNeeded();
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 1bb5bda0..29d19d7f 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
@@ -29,9 +29,12 @@
#include "arrow/type.h"
#include "gtest/gtest.h"
+#include "paimon/cache/cache.h"
#include "paimon/common/data/binary_row.h"
#include "paimon/common/data/binary_row_writer.h"
+#include "paimon/common/data/serializer/binary_row_serializer.h"
#include "paimon/common/io/cache/lru_cache.h"
+#include "paimon/common/io/memory_segment_output_stream.h"
#include "paimon/common/utils/math.h"
#include "paimon/core/bucket/default_bucket_function.h"
#include "paimon/core/io/meta_to_arrow_array_converter.h"
@@ -50,6 +53,7 @@
#include "paimon/executor.h"
#include "paimon/format/file_format_factory.h"
#include "paimon/fs/local/local_file_system.h"
+#include "paimon/memory/bytes.h"
#include "paimon/memory/memory_pool.h"
#include "paimon/metrics.h"
#include "paimon/predicate/literal.h"
@@ -61,6 +65,7 @@
#include "paimon/testing/mock/mock_file_batch_reader.h"
#include "paimon/testing/mock/mock_format_reader_builder.h"
#include "paimon/testing/utils/binary_row_generator.h"
+#include "paimon/testing/utils/counting_cache_test_utils.h"
#include "paimon/testing/utils/test_helper.h"
#include "paimon/testing/utils/testharness.h"
#include "paimon/testing/utils/timezone_guard.h"
@@ -689,14 +694,61 @@ TEST(AppendOnlyFileStoreScanTest, TestScanDurationMetric)
{
namespace {
+class CountingPartitionPredicate : public PredicateFilter {
+ public:
+ explicit CountingPartitionPredicate(const
std::shared_ptr<PredicateFilter>& predicate,
+ const Status& first_row_error =
Status::OK())
+ : predicate_(predicate), first_row_error_(first_row_error) {}
+
+ bool operator==(const Predicate& other) const override {
+ return *predicate_ == other;
+ }
+ const Function& GetFunction() const override {
+ return predicate_->GetFunction();
+ }
+ std::shared_ptr<Predicate> Negate() const override {
+ return predicate_->Negate();
+ }
+ std::string ToString() const override {
+ return predicate_->ToString();
+ }
+ Result<std::vector<char>> Test(const arrow::Array& array,
+ arrow::MemoryPool* pool) const override {
+ return predicate_->Test(array, pool);
+ }
+ Result<bool> Test(const std::shared_ptr<arrow::Schema>& schema,
+ const InternalRow& row) const override {
+ if (++row_calls_ == 1 && !first_row_error_.ok()) {
+ return first_row_error_;
+ }
+ return predicate_->Test(schema, row);
+ }
+ Result<bool> Test(const std::shared_ptr<arrow::Schema>& schema, int64_t
row_count,
+ const InternalRow& min_values, const InternalRow&
max_values,
+ const InternalArray& null_counts) const override {
+ ++stats_calls_;
+ // Retain every manifest so even an empty result exercises entry
filtering.
+ return true;
+ }
+
+ mutable std::atomic<int32_t> row_calls_{0};
+ mutable std::atomic<int32_t> stats_calls_{0};
+
+ private:
+ std::shared_ptr<PredicateFilter> predicate_;
+ Status first_row_error_;
+};
+
std::shared_ptr<FileStoreScan> BuildScan(const std::string& table_path,
const std::shared_ptr<Cache>& cache,
const std::optional<int32_t>& bucket
= std::nullopt,
const std::shared_ptr<Predicate>&
predicate = nullptr,
- bool
manifest_entry_lazy_decode_enabled = true) {
+ bool
manifest_entry_lazy_decode_enabled = true,
+ int32_t max_snapshots = 8) {
ScanContextBuilder context_builder(table_path);
context_builder.AddOption(Options::FILE_FORMAT, "orc")
- .AddOption(Options::SCAN_MANIFEST_ENTRY_CACHE_MAX_SNAPSHOTS, "8")
+ .AddOption(Options::SCAN_MANIFEST_ENTRY_CACHE_MAX_SNAPSHOTS,
+ fmt::format("{}", max_snapshots))
.AddOption(Options::SCAN_MANIFEST_ENTRY_LAZY_DECODE_ENABLED,
manifest_entry_lazy_decode_enabled ? "true" : "false")
.WithCache(cache);
@@ -811,7 +863,8 @@ TEST(AppendOnlyFileStoreScanTest,
TestSnapshotLiveManifestCachePath) {
TEST(AppendOnlyFileStoreScanTest, TestSnapshotLiveManifestCacheRebuildOnMiss) {
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>(CacheKind::SNAPSHOT_LIVE_MANIFEST, 16 *
1024 * 1024);
// Seed the cache with an earlier snapshot.
auto scan_base = BuildScan(table_path, cache, /*bucket=*/0);
@@ -819,7 +872,7 @@ TEST(AppendOnlyFileStoreScanTest,
TestSnapshotLiveManifestCacheRebuildOnMiss) {
scan_base->GetSnapshotManager()->LoadSnapshot(/*snapshot_id=*/3));
scan_base->WithSnapshot(snapshot_3);
ASSERT_OK_AND_ASSIGN(auto plan_base, scan_base->CreatePlan());
- (void)plan_base;
+ ASSERT_EQ(cache->GetCount(CacheKind::SNAPSHOT_LIVE_MANIFEST), 1);
// Advance to a newer snapshot: cache miss rebuilds the target snapshot
bucket.
auto scan_next = BuildScan(table_path, cache, /*bucket=*/0);
@@ -827,11 +880,30 @@ TEST(AppendOnlyFileStoreScanTest,
TestSnapshotLiveManifestCacheRebuildOnMiss) {
scan_next->GetSnapshotManager()->LoadSnapshot(/*snapshot_id=*/5));
scan_next->WithSnapshot(snapshot_5);
ASSERT_OK_AND_ASSIGN(auto plan_next, scan_next->CreatePlan());
+ ASSERT_EQ(cache->GetCount(CacheKind::SNAPSHOT_LIVE_MANIFEST), 2);
auto scan_expected = BuildScan(table_path, /*cache=*/nullptr,
/*bucket=*/0);
scan_expected->WithSnapshot(snapshot_5);
ASSERT_OK_AND_ASSIGN(auto plan_expected, scan_expected->CreatePlan());
ASSERT_EQ(SortedFileNames(plan_expected->Files()),
SortedFileNames(plan_next->Files()));
+
+ auto scan_retained = BuildScan(table_path, cache, /*bucket=*/0);
+ scan_retained->WithSnapshot(snapshot_3);
+ ASSERT_OK_AND_ASSIGN(auto plan_retained, scan_retained->CreatePlan());
+ ASSERT_EQ(SortedFileNames(plan_base->Files()),
SortedFileNames(plan_retained->Files()));
+ ASSERT_EQ(cache->GetCount(CacheKind::SNAPSHOT_LIVE_MANIFEST), 3);
+ ASSERT_OK_AND_ASSIGN(uint64_t hit,
scan_retained->GetScanMetrics()->GetCounter(
+
ScanMetrics::LAST_SNAPSHOT_CACHE_HIT));
+ ASSERT_EQ(hit, 1);
+ auto scan_evicted = BuildScan(table_path, cache, /*bucket=*/0,
/*predicate=*/nullptr,
+ /*manifest_entry_lazy_decode_enabled=*/true,
/*max_snapshots=*/1);
+ scan_evicted->WithSnapshot(snapshot_3);
+ ASSERT_OK_AND_ASSIGN(auto plan_evicted, scan_evicted->CreatePlan());
+ ASSERT_EQ(SortedFileNames(plan_base->Files()),
SortedFileNames(plan_evicted->Files()));
+ ASSERT_EQ(cache->GetCount(CacheKind::SNAPSHOT_LIVE_MANIFEST), 4);
+ ASSERT_OK_AND_ASSIGN(
+ hit,
scan_evicted->GetScanMetrics()->GetCounter(ScanMetrics::LAST_SNAPSHOT_CACHE_HIT));
+ ASSERT_EQ(hit, 0);
}
TEST(AppendOnlyFileStoreScanTest,
TestSnapshotLiveManifestCacheFallbackWithoutLazyDecode) {
@@ -851,4 +923,254 @@ TEST(AppendOnlyFileStoreScanTest,
TestSnapshotLiveManifestCacheFallbackWithoutLa
ASSERT_OK_AND_ASSIGN(auto plan_expected, scan_expected->CreatePlan());
ASSERT_EQ(SortedFileNames(plan_expected->Files()),
SortedFileNames(plan_fallback->Files()));
}
+
+TEST(AppendOnlyFileStoreScanTest,
SnapshotCachePropagatesPartitionPredicateError) {
+ TimezoneGuard guard("Asia/Shanghai");
+ const std::string table_path = GetDataDir() +
"/orc/append_09.db/append_09/";
+ auto cache = std::make_shared<LruCache>(16 * 1024 * 1024);
+ auto warm = BuildScan(table_path, cache, /*bucket=*/0);
+ ASSERT_OK_AND_ASSIGN(Snapshot snapshot,
warm->GetSnapshotManager()->LoadSnapshot(5));
+ warm->WithSnapshot(snapshot);
+ ASSERT_OK_AND_ASSIGN(auto warm_plan, warm->CreatePlan());
+ auto files = warm_plan->Files();
+ ASSERT_FALSE(files.empty());
+ const int32_t partition = files.front().Partition().GetInt(0);
+ auto predicate = PredicateBuilder::Equal(1, "f1", FieldType::INT,
Literal(partition));
+ for (bool lazy_decode : {false, true}) {
+ SCOPED_TRACE(lazy_decode);
+ for (const Status& error : {Status::IOError("injected partition
predicate error"),
+ Status::Invalid("injected partition
predicate error")}) {
+ SCOPED_TRACE(error.ToString());
+ auto scan = BuildScan(table_path, cache, /*bucket=*/0, predicate,
lazy_decode);
+ scan->WithSnapshot(snapshot);
+ auto failing =
+
std::make_shared<CountingPartitionPredicate>(scan->GetPartitionPredicate(),
error);
+ scan->partition_filter_ = failing;
+ // A transient predicate failure must reach the caller, without an
internal retry.
+ auto result = scan->CreatePlan();
+ ASSERT_FALSE(result.ok());
+ ASSERT_EQ(result.status().code(), error.code());
+ ASSERT_NE(result.status().message().find(error.message()),
std::string::npos);
+ ASSERT_EQ(failing->row_calls_.load(), 1);
+ // The failure must leave the complete cached snapshot available
for the next request.
+ ASSERT_OK_AND_ASSIGN(auto plan, scan->CreatePlan());
+ ASSERT_FALSE(plan->Files().empty());
+ for (const auto& entry : plan->Files()) {
+ ASSERT_EQ(entry.Partition().GetInt(0), partition);
+ }
+ ASSERT_OK_AND_ASSIGN(uint64_t hit,
scan->GetScanMetrics()->GetCounter(
+
ScanMetrics::LAST_SNAPSHOT_CACHE_HIT));
+ ASSERT_EQ(hit, 1);
+ }
+ }
+}
+
+TEST(AppendOnlyFileStoreScanTest, SnapshotCacheRebuildsCorruptValues) {
+ TimezoneGuard guard("Asia/Shanghai");
+ const std::string table_path = GetDataDir() +
"/orc/append_09.db/append_09/";
+ auto source = BuildScan(table_path, /*cache=*/nullptr, /*bucket=*/0);
+ ASSERT_OK_AND_ASSIGN(Snapshot snapshot,
source->GetSnapshotManager()->LoadSnapshot(5));
+ source->WithSnapshot(snapshot);
+ ASSERT_OK_AND_ASSIGN(auto source_plan, source->CreatePlan());
+ auto files = source_plan->Files();
+ ASSERT_FALSE(files.empty());
+ const ManifestEntry first = files.front();
+ const auto all_names = SortedFileNames(std::move(files));
+ auto predicate =
+ PredicateBuilder::Equal(1, "f1", FieldType::INT,
Literal(first.Partition().GetInt(0)));
+ auto expected = BuildScan(table_path, /*cache=*/nullptr, /*bucket=*/0,
predicate);
+ expected->WithSnapshot(snapshot);
+ ASSERT_OK_AND_ASSIGN(auto expected_plan, expected->CreatePlan());
+ const auto expected_names = SortedFileNames(expected_plan->Files());
+ auto pool = GetDefaultPool();
+ ManifestEntrySerializer serializer(pool);
+ BinaryRowSerializer row_serializer(serializer.NumFields(), pool);
+ for (const std::string corruption : {"magic", "version", "truncated",
"partition"}) {
+ SCOPED_TRACE(corruption);
+ ASSERT_OK_AND_ASSIGN(BinaryRow row, serializer.ToRow(first));
+ if (corruption == "version") {
+ row.SetInt(0, 999);
+ } else if (corruption == "partition") {
+ // A negative partition arity fails before predicate evaluation,
regardless of mode.
+ auto segment = row.GetSegment();
+ const int32_t offset =
+ static_cast<int32_t>(row.GetStringView(2).data() -
segment.Data());
+ 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>(1);
+ out.WriteValue<int64_t>(snapshot.Id());
+ out.WriteValue<int32_t>(1);
+ ASSERT_OK(row_serializer.Serialize(row, &out));
+ ASSERT_EQ(out.Segments().size(), 1);
+ const int64_t size = out.CurrentSize() - (corruption == "truncated" ?
1 : 0);
+ std::shared_ptr<Bytes> bytes =
+ Bytes::AllocateBytes(std::string(out.Segments().front().Data(),
size), pool.get());
+ const auto segment = MemorySegment::Wrap(bytes);
+ ASSERT_NOK(SnapshotLiveManifestEntries::ReadSnapshot(segment,
snapshot.Id(), 8,
+
/*row_filter=*/nullptr, pool));
+ for (bool lazy_decode : {false, true}) {
+ SCOPED_TRACE(lazy_decode);
+ auto cache =
std::make_shared<CountingRoutingCache>(CacheKind::SNAPSHOT_LIVE_MANIFEST,
+ 16 * 1024 *
1024);
+ auto scan = BuildScan(table_path, cache, /*bucket=*/0, predicate,
lazy_decode);
+ scan->WithSnapshot(snapshot);
+
ASSERT_OK(cache->Put(scan->CreateSnapshotLiveManifestEntriesCacheKey(0),
+ std::make_shared<CacheValue>(segment,
CacheCallback())));
+ for (int32_t attempt = 0; attempt < 2; ++attempt) {
+ ASSERT_OK_AND_ASSIGN(auto plan, scan->CreatePlan());
+ ASSERT_EQ(SortedFileNames(plan->Files()), expected_names);
+ ASSERT_OK_AND_ASSIGN(uint64_t hit,
scan->GetScanMetrics()->GetCounter(
+
ScanMetrics::LAST_SNAPSHOT_CACHE_HIT));
+ ASSERT_EQ(hit, attempt);
+ ASSERT_EQ(cache->GetCount(CacheKind::SNAPSHOT_LIVE_MANIFEST),
attempt + 1);
+ }
+ auto unfiltered = BuildScan(table_path, cache, /*bucket=*/0,
nullptr, lazy_decode);
+ unfiltered->WithSnapshot(snapshot);
+ ASSERT_OK_AND_ASSIGN(auto all, unfiltered->CreatePlan());
+ ASSERT_EQ(SortedFileNames(all->Files()), all_names);
+ }
+ }
+}
+
+TEST(AppendOnlyFileStoreScanTest, PartitionPruningAcrossCacheModes) {
+ auto dir = UniqueTestDirectory::Create();
+ ASSERT_TRUE(dir);
+ auto fields =
+ arrow::FieldVector{arrow::field("f0", arrow::int32()),
arrow::field("f1", arrow::int32())};
+ std::map<std::string, std::string> options = {
+ {Options::BUCKET, "1"}, {Options::BUCKET_KEY, "f0"},
{Options::FILE_FORMAT, "orc"}};
+ ASSERT_OK_AND_ASSIGN(auto helper,
+ TestHelper::Create(dir->Str(), arrow::schema(fields),
{"f1"}, {}, options,
+ /*is_streaming_mode=*/false));
+ std::vector<std::unique_ptr<RecordBatch>> batches;
+ for (int32_t partition : {10, 20}) {
+ ASSERT_OK_AND_ASSIGN(
+ auto batch,
+ TestHelper::MakeRecordBatch(arrow::struct_(fields),
fmt::format("[[1, {}]]", partition),
+ {{"f1", fmt::format("{}",
partition)}}, /*bucket=*/0,
+ {RecordBatch::RowKind::INSERT}));
+ batches.push_back(std::move(batch));
+ }
+ ASSERT_OK(helper->WriteAndCommit(std::move(batches), 0, std::nullopt));
+ const std::string table_path = dir->Str() + "/foo.db/bar";
+ for (bool lazy_decode : {false, true}) {
+ SCOPED_TRACE(lazy_decode);
+ // Exercise uncached reads, manifest Arrow caching, and bucket
snapshot caching.
+ for (int32_t mode : {0, 1, 2}) {
+ SCOPED_TRACE(mode);
+ auto cache =
std::make_shared<CountingRoutingCache>(std::map<CacheKind, int64_t>{
+ {CacheKind::MANIFEST, 16 * 1024 * 1024},
+ {CacheKind::SNAPSHOT_LIVE_MANIFEST, 16 * 1024 * 1024}});
+ std::shared_ptr<Cache> scan_cache = mode == 0 ? nullptr : cache;
+ std::optional<int32_t> bucket = mode == 2 ?
std::optional<int32_t>(0) : std::nullopt;
+ int32_t scans = 0;
+ int64_t manifest_loads = 0;
+ // A filtered cold read must retain other partitions for
subsequent cached reads.
+ for (int32_t partition : {10, 20, 99, 10}) {
+ SCOPED_TRACE(partition);
+ auto predicate =
+ PredicateBuilder::Equal(1, "f1", FieldType::INT,
Literal(partition));
+ auto scan = BuildScan(table_path, scan_cache, bucket,
predicate, lazy_decode);
+ auto counting =
+
std::make_shared<CountingPartitionPredicate>(scan->GetPartitionPredicate());
+ 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);
+ ASSERT_EQ(plan->Files().size(), partition == 99 ? 0 : 1);
+ if (partition != 99) {
+ ASSERT_EQ(plan->Files()[0].Partition().GetInt(0),
partition);
+ }
+ auto metrics = scan->GetScanMetrics();
+ ASSERT_OK_AND_ASSIGN(uint64_t hit,
+
metrics->GetCounter(ScanMetrics::LAST_SNAPSHOT_CACHE_HIT));
+ ASSERT_EQ(hit, mode == 2 && scans > 0 ? 1 : 0);
+ if (mode == 2) {
+
ASSERT_EQ(cache->GetCount(CacheKind::SNAPSHOT_LIVE_MANIFEST), scans + 1);
+ ASSERT_OK_AND_ASSIGN(
+ uint64_t scanned,
+
metrics->GetCounter(ScanMetrics::LAST_LAZY_DECODE_SCANNED_ROWS));
+ ASSERT_EQ(scanned, 2);
+ }
+ if (mode == 1) {
+ if (scans == 0) {
+ manifest_loads =
cache->SupplierCallCount(CacheKind::MANIFEST);
+ ASSERT_GT(manifest_loads, 0);
+ }
+ ASSERT_EQ(cache->SupplierCallCount(CacheKind::MANIFEST),
manifest_loads);
+ ASSERT_EQ(cache->GetCount(CacheKind::MANIFEST), (scans +
1) * manifest_loads);
+ }
+ ++scans;
+ }
+ auto unfiltered = BuildScan(table_path, scan_cache, bucket,
nullptr, lazy_decode);
+ ASSERT_OK_AND_ASSIGN(auto all, unfiltered->CreatePlan());
+ ASSERT_EQ(all->Files().size(), 2);
+ }
+ }
+}
+
+TEST(AppendOnlyFileStoreScanTest,
PartitionPruningSkipsExcludedMetadataInCreatePlan) {
+ auto dir = UniqueTestDirectory::Create();
+ ASSERT_TRUE(dir);
+ auto fields =
+ arrow::FieldVector{arrow::field("f0", arrow::int32()),
arrow::field("f1", arrow::int32())};
+ ASSERT_OK_AND_ASSIGN(
+ auto helper,
+ TestHelper::Create(
+ dir->Str(), arrow::schema(fields), {"f1"}, {},
+ {{Options::BUCKET, "1"}, {Options::BUCKET_KEY, "f0"},
{Options::FILE_FORMAT, "orc"}},
+ /*is_streaming_mode=*/false));
+ const std::string table_path = dir->Str() + "/foo.db/bar";
+ auto pool = GetDefaultPool();
+ ASSERT_OK_AND_ASSIGN(
+ auto file,
+ DataFileMeta::ForAppend("file.orc", 10, 1, SimpleStats::EmptyStats(),
0, 0, 0, std::nullopt,
+ std::nullopt, std::nullopt, std::nullopt,
std::nullopt));
+ ManifestEntry kept(FileKind::Add(), BinaryRowGenerator::GenerateRow({10},
pool.get()), 0, 1,
+ file);
+ ManifestEntry excluded(FileKind(static_cast<int8_t>(2)),
+ BinaryRowGenerator::GenerateRow({20}, pool.get()),
0, 1, file);
+ auto writer = BuildScan(table_path, nullptr);
+ using WrittenFile = std::pair<std::string, int64_t>;
+ ASSERT_OK_AND_ASSIGN(WrittenFile written,
+ writer->manifest_file_->WriteWithoutRolling({kept,
excluded}));
+ // Both partitions share one manifest, whose statistics cannot prune the
requested partition.
+ ManifestFileMeta meta(written.first, written.second, 2, 0,
+ BinaryRowGenerator::GenerateStats({10}, {20}, {0},
pool.get()), 0, 0, 0,
+ 0, 0, std::nullopt, std::nullopt);
+ ASSERT_OK_AND_ASSIGN(WrittenFile base, writer->manifest_list_->Write({}));
+ ASSERT_OK_AND_ASSIGN(WrittenFile delta,
writer->manifest_list_->Write({meta}));
+ Snapshot snapshot(1, 0, base.first, base.second, delta.first,
delta.second, std::nullopt,
+ std::nullopt, std::nullopt, "test", 0,
Snapshot::CommitKind::Append(), 0, 2,
+ 2, std::nullopt, std::nullopt, std::nullopt,
std::nullopt, std::nullopt);
+ auto predicate = PredicateBuilder::Equal(1, "f1", FieldType::INT,
Literal(10));
+ for (int32_t mode : {0, 1, 2}) {
+ SCOPED_TRACE(mode);
+ std::shared_ptr<Cache> cache =
+ mode == 0 ? nullptr : std::make_shared<LruCache>(16 * 1024 * 1024);
+ std::optional<int32_t> bucket = mode == 2 ? std::optional<int32_t>(0)
: std::nullopt;
+ auto scan = BuildScan(table_path, cache, bucket, predicate);
+ scan->WithSnapshot(snapshot);
+ if (mode == 2) {
+ // Seed the complete snapshot directly: rebuilding must decode
every partition.
+ SnapshotLiveManifestEntries entries(8);
+ entries.Put(1, {kept, excluded});
+ ASSERT_OK(scan->StoreSnapshotLiveManifestEntries(0, entries));
+ }
+ for (int32_t attempt = 0; attempt < 2; ++attempt) {
+ ASSERT_OK_AND_ASSIGN(auto plan, scan->CreatePlan());
+ ASSERT_EQ(plan->Files(), std::vector<ManifestEntry>{kept});
+ ASSERT_OK_AND_ASSIGN(uint64_t hit,
scan->GetScanMetrics()->GetCounter(
+
ScanMetrics::LAST_SNAPSHOT_CACHE_HIT));
+ ASSERT_EQ(hit, mode == 2 ? 1 : 0);
+ }
+ auto full_decode = BuildScan(table_path, cache, bucket, predicate,
+
/*manifest_entry_lazy_decode_enabled=*/false);
+ full_decode->WithSnapshot(snapshot);
+ ASSERT_NOK_WITH_MSG(full_decode->CreatePlan(), "Unsupported byte value
2 for file kind.");
+ }
+}
+
} // namespace paimon::test
diff --git a/src/paimon/core/operation/file_store_scan.cpp
b/src/paimon/core/operation/file_store_scan.cpp
index e7475d17..e5f13233 100644
--- a/src/paimon/core/operation/file_store_scan.cpp
+++ b/src/paimon/core/operation/file_store_scan.cpp
@@ -42,6 +42,7 @@
#include "paimon/core/io/file_index_options.h"
#include "paimon/core/manifest/file_entry.h"
#include "paimon/core/manifest/file_kind.h"
+#include "paimon/core/manifest/manifest_entry_serializer.h"
#include "paimon/core/manifest/manifest_file.h"
#include "paimon/core/manifest/manifest_file_meta.h"
#include "paimon/core/manifest/manifest_list.h"
@@ -196,19 +197,10 @@ 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_RETURN_NOT_OK(ReadManifestEntriesWithCache(snapshot.value(),
all_manifest_file_metas,
-
cache_bucket.value(), &manifest_entries,
-
&snapshot_cache_hit));
- lazy_decode_scanned_rows = manifest_entries.size();
- std::vector<ManifestEntry> filtered_entries;
- filtered_entries.reserve(manifest_entries.size());
- for (auto& entry : manifest_entries) {
- PAIMON_ASSIGN_OR_RAISE(bool keep, FilterManifestEntry(entry));
- if (keep) {
- filtered_entries.emplace_back(std::move(entry));
- }
- }
- manifest_entries = std::move(filtered_entries);
+ PAIMON_ASSIGN_OR_RAISE(lazy_decode_scanned_rows,
+ ReadManifestEntriesWithCache(
+ snapshot.value(), all_manifest_file_metas,
cache_bucket.value(),
+ &manifest_entries, &snapshot_cache_hit));
} else {
lazy_decode_scanned_rows = std::accumulate(
filtered_manifest_file_metas.begin(),
filtered_manifest_file_metas.end(), uint64_t{0},
@@ -396,44 +388,81 @@ Status FileStoreScan::ReadManifestEntries(const
std::vector<ManifestFileMeta>& m
// bucket counts, with the current count and schema in the cache key. Exact
hits
// can be returned directly; cache misses rebuild the target snapshot bucket
from the target
// snapshot's data manifests.
-Status FileStoreScan::ReadManifestEntriesWithCache(
+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 {
Duration cache_load_duration;
- PAIMON_ASSIGN_OR_RAISE(SnapshotLiveManifestEntries cached_entries,
- LoadSnapshotLiveManifestEntries(bucket));
+ const int32_t max_snapshots =
core_options_.GetScanManifestEntryCacheMaxSnapshots();
+ SnapshotLiveManifestEntries cached_entries(max_snapshots);
+ bool predicate_failed = false;
+ auto row_filter = CreateManifestEntryRowFilter(&predicate_failed);
+ bool partition_filtered = false;
+ uint64_t candidate_entries = 0;
+ *cache_hit = false;
+ Result<std::shared_ptr<CacheValue>> value = core_options_.GetCache()->Get(
+ CreateSnapshotLiveManifestEntriesCacheKey(bucket),
+ [](const std::shared_ptr<CacheKey>&) ->
Result<std::shared_ptr<CacheValue>> {
+ return std::shared_ptr<CacheValue>();
+ });
+ if (value.ok() && value.value()) {
+ Result<std::optional<SnapshotLiveManifestEntries::Selection>> selected
=
+
SnapshotLiveManifestEntries::ReadSnapshot(value.value()->GetSegment(),
snapshot.Id(),
+ max_snapshots,
row_filter, pool_);
+ if (predicate_failed) {
+ return selected.status();
+ }
+ if (selected.ok() && selected.value()) {
+ *cache_hit = true;
+ partition_filtered = static_cast<bool>(row_filter);
+ candidate_entries = selected.value()->candidate_entries;
+ *manifest_entries = std::move(selected.value()->entries);
+ } else if (selected.ok()) {
+ // Reuse the same cache value to retain complete historical
snapshots on a miss.
+ // Query-filtered entries must never be written back into the
cache.
+ Result<SnapshotLiveManifestEntries> deserialized =
+
SnapshotLiveManifestEntries::Deserialize(value.value()->GetSegment(),
max_snapshots,
+ pool_);
+ if (deserialized.ok()) {
+ cached_entries = std::move(deserialized.value());
+ }
+ }
+ }
const uint64_t cache_load_duration_ms = cache_load_duration.Get();
metrics_->SetCounter(ScanMetrics::LAST_SNAPSHOT_CACHE_LOAD_DURATION,
cache_load_duration_ms);
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()) {
- *cache_hit = true;
- *manifest_entries = *cached->entries;
- return Status::OK();
+ if (!*cache_hit) {
+ std::vector<ManifestFileMeta> bucket_manifest_metas;
+ for (const auto& meta : all_manifest_metas) {
+ if ((!bucket_filter_ && bucket_selector_) ||
MayContainBucket(meta, bucket)) {
+ bucket_manifest_metas.push_back(meta);
+ }
+ }
+ PAIMON_RETURN_NOT_OK(
+ 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));
+ Duration cache_store_duration;
+ PAIMON_RETURN_NOT_OK(StoreSnapshotLiveManifestEntries(bucket,
cached_entries));
+ const uint64_t cache_store_duration_ms = cache_store_duration.Get();
+ metrics_->SetCounter(ScanMetrics::LAST_SNAPSHOT_CACHE_STORE_DURATION,
+ cache_store_duration_ms);
+ metrics_->ObserveHistogram(ScanMetrics::SNAPSHOT_CACHE_STORE_DURATION,
+
static_cast<double>(cache_store_duration_ms));
}
- *cache_hit = false;
-
- // Rebuild the target snapshot bucket from all manifests and write the
live entries back to the
- // cache.
- std::vector<ManifestFileMeta> bucket_manifest_metas;
- for (const auto& meta : all_manifest_metas) {
- if ((!bucket_filter_ && bucket_selector_) || MayContainBucket(meta,
bucket)) {
- bucket_manifest_metas.push_back(meta);
+ std::vector<ManifestEntry> filtered_entries;
+ filtered_entries.reserve(manifest_entries->size());
+ for (auto& entry : *manifest_entries) {
+ PAIMON_ASSIGN_OR_RAISE(bool keep, partition_filtered
+ ?
FilterNonPartitionManifestEntry(entry)
+ : FilterManifestEntry(entry));
+ if (keep) {
+ filtered_entries.emplace_back(std::move(entry));
}
}
- 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));
- Duration cache_store_duration;
- PAIMON_RETURN_NOT_OK(StoreSnapshotLiveManifestEntries(bucket,
cached_entries));
- const uint64_t cache_store_duration_ms = cache_store_duration.Get();
- metrics_->SetCounter(ScanMetrics::LAST_SNAPSHOT_CACHE_STORE_DURATION,
cache_store_duration_ms);
- metrics_->ObserveHistogram(ScanMetrics::SNAPSHOT_CACHE_STORE_DURATION,
- static_cast<double>(cache_store_duration_ms));
- return Status::OK();
+ *manifest_entries = std::move(filtered_entries);
+ return candidate_entries;
}
std::shared_ptr<CacheKey>
FileStoreScan::CreateSnapshotLiveManifestEntriesCacheKey(
@@ -447,26 +476,6 @@ std::shared_ptr<CacheKey>
FileStoreScan::CreateSnapshotLiveManifestEntriesCacheK
table_path_,
BranchManager::NormalizeBranch(core_options_.GetBranch()), bucket);
}
-Result<SnapshotLiveManifestEntries>
FileStoreScan::LoadSnapshotLiveManifestEntries(
- int32_t bucket) const {
- auto supplier = [](const std::shared_ptr<CacheKey>&) ->
Result<std::shared_ptr<CacheValue>> {
- return std::shared_ptr<CacheValue>();
- };
- std::shared_ptr<CacheKey> cache_key =
CreateSnapshotLiveManifestEntriesCacheKey(bucket);
- const auto max_snapshots =
core_options_.GetScanManifestEntryCacheMaxSnapshots();
- Result<std::shared_ptr<CacheValue>> cache_result =
- core_options_.GetCache()->Get(cache_key, supplier);
- if (!cache_result.ok() || !cache_result.value()) {
- return SnapshotLiveManifestEntries(max_snapshots);
- }
- Result<SnapshotLiveManifestEntries> deserialized =
SnapshotLiveManifestEntries::Deserialize(
- cache_result.value()->GetSegment(), max_snapshots, pool_);
- if (!deserialized.ok()) {
- return SnapshotLiveManifestEntries(max_snapshots);
- }
- return std::move(deserialized.value());
-}
-
Status FileStoreScan::StoreSnapshotLiveManifestEntries(
int32_t bucket, const SnapshotLiveManifestEntries& entries) const {
Result<std::shared_ptr<Bytes>> bytes_result = entries.Serialize(pool_);
@@ -619,25 +628,35 @@ bool FileStoreScan::FilterManifestByRowRanges(const
ManifestFileMeta& manifest)
Status FileStoreScan::ReadManifestFileMeta(const ManifestFileMeta& manifest,
std::vector<ManifestEntry>*
entries) const {
+ const RowRangeIndex* row_ranges = nullptr;
if (row_range_index_ && core_options_.DataEvolutionEnabled() &&
core_options_.ScanManifestEntryLazyDecodeEnabled()) {
- return manifest_file_->ReadRowRangeEntries(
- manifest.FileName(), row_range_index_.value(),
- [this](const ManifestEntry& entry) -> Result<bool> {
- return FilterManifestEntry(entry);
- },
- manifest.FileSize(), entries);
- }
- std::vector<ManifestEntry> unfiltered_entries;
- PAIMON_RETURN_NOT_OK(manifest_file_->Read(
- manifest.FileName(),
- [this](const ManifestEntry& entry) -> Result<bool> { return
FilterManifestEntry(entry); },
- manifest.FileSize(), &unfiltered_entries));
- entries->reserve(entries->size() + unfiltered_entries.size());
- for (auto& entry : unfiltered_entries) {
- entries->emplace_back(std::move(entry));
+ row_ranges = &row_range_index_.value();
+ }
+ auto row_filter =
CreateManifestEntryRowFilter(/*predicate_failed=*/nullptr);
+ return manifest_file_->ReadSelectedEntries(
+ manifest.FileName(), manifest.FileSize(), row_ranges, row_filter,
+ [this, partition_filtered =
+ static_cast<bool>(row_filter)](const ManifestEntry& entry)
-> Result<bool> {
+ return partition_filtered ? FilterNonPartitionManifestEntry(entry)
+ : FilterManifestEntry(entry);
+ },
+ entries);
+}
+
+std::function<Result<bool>(const InternalRow&)>
FileStoreScan::CreateManifestEntryRowFilter(
+ bool* predicate_failed) const {
+ if (!partition_filter_ ||
!core_options_.ScanManifestEntryLazyDecodeEnabled()) {
+ return nullptr;
}
- return Status::OK();
+ return [this, predicate_failed](const InternalRow& row) -> Result<bool> {
+ PAIMON_ASSIGN_OR_RAISE(BinaryRow partition,
ManifestEntrySerializer::GetPartitionView(row));
+ Result<bool> result = partition_filter_->Test(partition_schema_,
partition);
+ if (!result.ok() && predicate_failed) {
+ *predicate_failed = true;
+ }
+ return result;
+ };
}
Result<bool> FileStoreScan::FilterManifestEntry(const ManifestEntry& entry)
const {
@@ -648,6 +667,10 @@ Result<bool> FileStoreScan::FilterManifestEntry(const
ManifestEntry& entry) cons
return false;
}
}
+ return FilterNonPartitionManifestEntry(entry);
+}
+
+Result<bool> FileStoreScan::FilterNonPartitionManifestEntry(const
ManifestEntry& entry) const {
if (only_read_real_buckets_ && entry.Bucket() < 0) {
return false;
}
diff --git a/src/paimon/core/operation/file_store_scan.h
b/src/paimon/core/operation/file_store_scan.h
index 5fe128da..b73ad095 100644
--- a/src/paimon/core/operation/file_store_scan.h
+++ b/src/paimon/core/operation/file_store_scan.h
@@ -63,6 +63,7 @@ namespace paimon {
class CacheKey;
class Executor;
class FileKind;
+class InternalRow;
class ManifestFile;
class ManifestFileMeta;
class ManifestList;
@@ -276,13 +277,12 @@ class FileStoreScan {
Status ReadManifestEntries(const std::vector<ManifestFileMeta>&
manifest_metas,
std::vector<ManifestEntry>* manifest_entries)
const;
- Status ReadManifestEntriesWithCache(const Snapshot& snapshot,
- const std::vector<ManifestFileMeta>&
bucket_manifest_metas,
- int32_t bucket,
- std::vector<ManifestEntry>*
manifest_entries,
- bool* cache_hit) const;
+ /// 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;
std::shared_ptr<CacheKey>
CreateSnapshotLiveManifestEntriesCacheKey(int32_t bucket) const;
- Result<SnapshotLiveManifestEntries>
LoadSnapshotLiveManifestEntries(int32_t bucket) const;
Status StoreSnapshotLiveManifestEntries(int32_t bucket,
const SnapshotLiveManifestEntries&
entries) const;
@@ -309,7 +309,13 @@ class FileStoreScan {
Status ReadManifestFileMeta(const ManifestFileMeta& manifest,
std::vector<ManifestEntry>* entries) const;
+ /// If provided, predicate_failed must initially be false and outlive the
synchronous callback.
+ /// Only predicate evaluation errors set it; partition decoding errors
leave it unchanged.
+ std::function<Result<bool>(const InternalRow&)>
CreateManifestEntryRowFilter(
+ bool* predicate_failed) const;
+
Result<bool> FilterManifestEntry(const ManifestEntry& entry) const;
+ Result<bool> FilterNonPartitionManifestEntry(const ManifestEntry& entry)
const;
protected:
std::shared_ptr<MemoryPool> pool_;
diff --git a/src/paimon/core/operation/file_store_scan_test.cpp
b/src/paimon/core/operation/file_store_scan_test.cpp
index c4f0289e..ef784f76 100644
--- a/src/paimon/core/operation/file_store_scan_test.cpp
+++ b/src/paimon/core/operation/file_store_scan_test.cpp
@@ -21,14 +21,19 @@
#include <string>
#include "arrow/type.h"
+#include "fmt/format.h"
#include "gtest/gtest.h"
+#include "paimon/common/io/memory_segment_output_stream.h"
#include "paimon/core/io/data_file_meta.h"
#include "paimon/core/manifest/file_kind.h"
#include "paimon/core/manifest/file_source.h"
+#include "paimon/core/manifest/manifest_entry_serializer.h"
#include "paimon/core/manifest/snapshot_live_manifest_entries.h"
#include "paimon/core/stats/simple_stats.h"
#include "paimon/data/timestamp.h"
+#include "paimon/memory/bytes.h"
#include "paimon/memory/memory_segment.h"
+#include "paimon/testing/utils/binary_row_generator.h"
#include "paimon/testing/utils/testharness.h"
namespace paimon::test {
@@ -190,54 +195,6 @@ TEST_F(FileStoreScanTest, TestFilterManifestByRowRanges) {
ASSERT_TRUE(file_store_scan->FilterManifestByRowRanges(manifest2));
}
-TEST_F(FileStoreScanTest, TestSnapshotLiveManifestEntries) {
- std::vector<ManifestEntry> snapshot1;
- ASSERT_OK_AND_ASSIGN(
- auto file1,
- DataFileMeta::ForAppend("file-1", /*file_size=*/10, /*row_count=*/1,
- SimpleStats::EmptyStats(),
/*min_sequence_number=*/0,
- /*max_sequence_number=*/0, /*schema_id=*/0,
- /*file_source=*/std::nullopt,
/*value_stats_cols=*/std::nullopt,
- /*external_path=*/std::nullopt,
/*first_row_id=*/std::nullopt,
- /*write_cols=*/std::nullopt));
- 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));
- ASSERT_EQ(entries.Size(), 1);
- auto hit = entries.LatestBeforeOrEqual(/*snapshot_id=*/1);
- ASSERT_TRUE(hit);
- ASSERT_EQ(hit->snapshot_id, 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);
-
- std::vector<ManifestEntry> snapshot3;
- ASSERT_OK_AND_ASSIGN(
- auto file3,
- DataFileMeta::ForAppend("file-3", /*file_size=*/10, /*row_count=*/1,
- SimpleStats::EmptyStats(),
/*min_sequence_number=*/0,
- /*max_sequence_number=*/0, /*schema_id=*/0,
- /*file_source=*/std::nullopt,
/*value_stats_cols=*/std::nullopt,
- /*external_path=*/std::nullopt,
/*first_row_id=*/std::nullopt,
- /*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));
-
- 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, {});
- ASSERT_EQ(entries.Size(), 2);
- ASSERT_FALSE(entries.LatestBeforeOrEqual(/*snapshot_id=*/1));
- ASSERT_TRUE(entries.LatestBeforeOrEqual(/*snapshot_id=*/3));
- ASSERT_TRUE(entries.LatestBeforeOrEqual(/*snapshot_id=*/5));
-}
-
TEST_F(FileStoreScanTest, TestSnapshotLiveManifestEntriesSerialization) {
std::vector<ManifestEntry> manifest_entries;
ASSERT_OK_AND_ASSIGN(
@@ -259,18 +216,180 @@ TEST_F(FileStoreScanTest,
TestSnapshotLiveManifestEntriesSerialization) {
SnapshotLiveManifestEntries::Deserialize(
MemorySegment::Wrap(bytes), /*max_snapshots=*/2,
GetDefaultPool()));
ASSERT_EQ(deserialized.Size(), 2);
- auto hit = deserialized.LatestBeforeOrEqual(/*snapshot_id=*/2);
- ASSERT_TRUE(hit);
- ASSERT_EQ(hit->snapshot_id, 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_OK_AND_ASSIGN(auto round_trip,
deserialized.Serialize(GetDefaultPool()));
+ ASSERT_EQ(*bytes, *round_trip);
ASSERT_OK_AND_ASSIGN(auto evicted_deserialized,
SnapshotLiveManifestEntries::Deserialize(
MemorySegment::Wrap(bytes),
/*max_snapshots=*/1,
GetDefaultPool()));
ASSERT_EQ(evicted_deserialized.Size(), 1);
- ASSERT_FALSE(evicted_deserialized.LatestBeforeOrEqual(/*snapshot_id=*/1));
-
ASSERT_EQ(evicted_deserialized.LatestBeforeOrEqual(/*snapshot_id=*/3)->snapshot_id,
3);
+ ASSERT_OK_AND_ASSIGN(auto retained,
evicted_deserialized.Serialize(GetDefaultPool()));
+ // Use a larger read limit to verify that Deserialize itself evicted the
older snapshot.
+ ASSERT_OK_AND_ASSIGN(
+ auto evicted,
SnapshotLiveManifestEntries::ReadSnapshot(MemorySegment::Wrap(retained), 1, 2,
+ nullptr,
GetDefaultPool()));
+ ASSERT_FALSE(evicted);
+ ASSERT_OK_AND_ASSIGN(
+ auto empty,
SnapshotLiveManifestEntries::ReadSnapshot(MemorySegment::Wrap(retained), 3, 2,
+ nullptr,
GetDefaultPool()));
+ ASSERT_TRUE(empty);
+ ASSERT_TRUE(empty->entries.empty());
}
+
+TEST_F(FileStoreScanTest, SelectExactRetainedSnapshot) {
+ auto pool = GetDefaultPool();
+ std::vector<ManifestEntry> records;
+ for (int32_t partition = 0; partition < 3; ++partition) {
+ ASSERT_OK_AND_ASSIGN(
+ auto file,
+ DataFileMeta::ForAppend(fmt::format("file-{}", partition), 10,
partition + 1,
+ SimpleStats::EmptyStats(), 0, 0, 0,
std::nullopt, std::nullopt,
+ std::nullopt, std::nullopt, std::nullopt));
+ records.emplace_back(FileKind::Add(),
+ 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, {});
+ ASSERT_OK_AND_ASSIGN(auto bytes, entries.Serialize(pool));
+ const MemorySegment segment = MemorySegment::Wrap(bytes);
+ ASSERT_OK_AND_ASSIGN(auto first,
+ SnapshotLiveManifestEntries::ReadSnapshot(segment, 1,
3, nullptr, pool));
+ ASSERT_TRUE(first);
+ ASSERT_EQ(first->entries, std::vector<ManifestEntry>({records[0]}));
+ ASSERT_OK_AND_ASSIGN(auto third,
+ SnapshotLiveManifestEntries::ReadSnapshot(segment, 3,
3, nullptr, pool));
+ ASSERT_TRUE(third);
+ ASSERT_EQ(third->entries, std::vector<ManifestEntry>({records[1],
records[2]}));
+ int32_t filter_calls = 0;
+ ASSERT_OK_AND_ASSIGN(auto selected,
+ SnapshotLiveManifestEntries::ReadSnapshot(
+ segment, 3, 2,
+ [&filter_calls](const InternalRow& row) ->
Result<bool> {
+ ++filter_calls;
+ PAIMON_ASSIGN_OR_RAISE(
+ BinaryRow partition,
+
ManifestEntrySerializer::GetPartitionView(row));
+ return partition.GetInt(0) == 2;
+ },
+ pool));
+ ASSERT_TRUE(selected);
+ ASSERT_EQ(selected->candidate_entries, 2);
+ ASSERT_EQ(filter_calls, 2);
+ ASSERT_EQ(selected->entries, std::vector<ManifestEntry>({records[2]}));
+ ASSERT_OK_AND_ASSIGN(auto absent,
+ SnapshotLiveManifestEntries::ReadSnapshot(segment, 2,
3, nullptr, pool));
+ ASSERT_FALSE(absent);
+ ASSERT_OK_AND_ASSIGN(auto evicted,
+ SnapshotLiveManifestEntries::ReadSnapshot(segment, 3,
1, nullptr, pool));
+ ASSERT_FALSE(evicted);
+ ASSERT_OK_AND_ASSIGN(auto disabled,
+ SnapshotLiveManifestEntries::ReadSnapshot(segment, 3,
0, nullptr, pool));
+ ASSERT_FALSE(disabled);
+ // Skipping other snapshots must not allocate row payloads in the reader's
pool.
+ std::shared_ptr<MemoryPool> read_pool = GetMemoryPool();
+ ASSERT_OK_AND_ASSIGN(
+ auto empty, SnapshotLiveManifestEntries::ReadSnapshot(segment, 5, 1,
nullptr, read_pool));
+ ASSERT_TRUE(empty);
+ ASSERT_TRUE(empty->entries.empty());
+ ASSERT_EQ(read_pool->MaxMemoryUsage(), 0);
+ ASSERT_NOK_WITH_MSG(SnapshotLiveManifestEntries::ReadSnapshot(
+ segment, 3, 3,
+ [](const InternalRow&) -> Result<bool> {
+ return Status::IOError("injected filter
error");
+ },
+ pool),
+ "injected filter error");
+
+ entries.Put(7, {});
+ 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.
+ ASSERT_OK_AND_ASSIGN(auto oldest,
SnapshotLiveManifestEntries::ReadSnapshot(
+ MemorySegment::Wrap(updated), 1, 4,
nullptr, pool));
+ ASSERT_FALSE(oldest);
+ ASSERT_OK_AND_ASSIGN(auto retained,
SnapshotLiveManifestEntries::ReadSnapshot(
+ MemorySegment::Wrap(updated), 3,
4, nullptr, pool));
+ ASSERT_TRUE(retained);
+ ASSERT_EQ(retained->entries, std::vector<ManifestEntry>({records[1],
records[2]}));
+}
+
+TEST_F(FileStoreScanTest, SkipDecodingFilteredManifestEntries) {
+ auto pool = GetDefaultPool();
+ ASSERT_OK_AND_ASSIGN(
+ auto file,
+ DataFileMeta::ForAppend("file", 10, 1, SimpleStats::EmptyStats(), 0,
0, 0, std::nullopt,
+ std::nullopt, std::nullopt, std::nullopt,
std::nullopt));
+ // Like the bucket-pruning tests, use an invalid kind to detect full entry
decoding.
+ ManifestEntry excluded(FileKind(static_cast<int8_t>(2)),
+ BinaryRowGenerator::GenerateRow({10}, pool.get()),
0, 1, file);
+ SnapshotLiveManifestEntries entries(1);
+ entries.Put(1, {excluded});
+ ASSERT_OK_AND_ASSIGN(auto bytes, entries.Serialize(pool));
+ const MemorySegment segment = MemorySegment::Wrap(bytes);
+ ASSERT_OK_AND_ASSIGN(auto selected,
SnapshotLiveManifestEntries::ReadSnapshot(
+ segment, 1, 1,
+ [](const InternalRow& row) ->
Result<bool> {
+ PAIMON_ASSIGN_OR_RAISE(
+ BinaryRow partition,
+
ManifestEntrySerializer::GetPartitionView(row));
+ return partition.GetInt(0) ==
20;
+ },
+ pool));
+ ASSERT_TRUE(selected);
+ ASSERT_TRUE(selected->entries.empty());
+ ASSERT_NOK_WITH_MSG(SnapshotLiveManifestEntries::ReadSnapshot(segment, 1,
1, nullptr, pool),
+ "Unsupported byte value 2 for file kind.");
+}
+
+TEST_F(FileStoreScanTest,
RejectUnsupportedSnapshotEntryVersionBeforeFiltering) {
+ auto pool = GetDefaultPool();
+ // Only the version is present: the reader must reject it before the
filter accesses fields.
+ BinaryRow row = BinaryRowGenerator::GenerateRow({3}, pool.get());
+ BinaryRowSerializer serializer(1, pool);
+ MemorySegmentOutputStream out(64, pool);
+ out.WriteValue<int32_t>(0x534d4543);
+ out.WriteValue<int32_t>(1);
+ out.WriteValue<int64_t>(1);
+ out.WriteValue<int32_t>(1);
+ ASSERT_OK(serializer.Serialize(row, &out));
+ bool filter_called = false;
+ ASSERT_NOK_WITH_MSG(SnapshotLiveManifestEntries::ReadSnapshot(
+ out.Segments().front(), 1, 1,
+ [&filter_called](const InternalRow&) ->
Result<bool> {
+ filter_called = true;
+ return false;
+ },
+ pool),
+ "Unsupported version: 3");
+ ASSERT_FALSE(filter_called);
+}
+
+TEST_F(FileStoreScanTest, RejectInvalidSnapshotSelectionHeader) {
+ auto pool = GetDefaultPool();
+ ASSERT_OK_AND_ASSIGN(auto absent,
SnapshotLiveManifestEntries::ReadSnapshot(MemorySegment(), 1,
+
3, nullptr, pool));
+ ASSERT_FALSE(absent);
+ auto header = [&pool](int32_t magic, int32_t snapshots, int32_t entries) {
+ MemorySegmentOutputStream out(64, pool);
+ out.WriteValue<int32_t>(magic);
+ out.WriteValue<int32_t>(snapshots);
+ out.WriteValue<int64_t>(1);
+ 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);
+ ASSERT_NOK_WITH_MSG(SnapshotLiveManifestEntries::ReadSnapshot(segment, 1,
3, nullptr, pool),
+ "snapshot count is negative");
+ segment = header(0x534d4543, 1, -1);
+ ASSERT_NOK_WITH_MSG(SnapshotLiveManifestEntries::ReadSnapshot(segment, 1,
3, nullptr, pool),
+ "entry count is negative");
+ auto truncated = MemorySegment::WrapView(segment.Data(), 7);
+ ASSERT_NOK(SnapshotLiveManifestEntries::ReadSnapshot(truncated, 1, 3,
nullptr, pool));
+}
+
} // namespace paimon::test
