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 3a45772c perf(manifest): prune entries by row range and cache decoded
Arrow IPC (#411)
3a45772c is described below
commit 3a45772c37f3b022fcb9da2096ade662e3eeb296
Author: gripleaf <[email protected]>
AuthorDate: Sat Oct 10 17:21:23 2026 +0800
perf(manifest): prune entries by row range and cache decoded Arrow IPC
(#411)
---
docs/source/user_guide/manifest_cache.rst | 44 +-
include/paimon/cache/cache.h | 9 +
src/paimon/CMakeLists.txt | 1 +
src/paimon/common/io/cache/cache.cpp | 12 +-
src/paimon/common/io/cache/lru_cache.cpp | 2 +-
src/paimon/common/io/cache/lru_cache.h | 2 +-
src/paimon/common/io/cache/lru_cache_test.cpp | 34 +
src/paimon/core/io/data_file_meta.h | 4 +
src/paimon/core/io/data_file_meta_serializer.cpp | 12 +-
.../manifest/index_manifest_file_handler_test.cpp | 53 +-
src/paimon/core/manifest/manifest_entry.h | 3 +
.../core/manifest/manifest_entry_serializer.cpp | 4 +-
src/paimon/core/manifest/manifest_file.cpp | 59 ++
src/paimon/core/manifest/manifest_file.h | 10 +
src/paimon/core/manifest/manifest_file_test.cpp | 6 +-
src/paimon/core/manifest/manifest_list_test.cpp | 80 ++-
.../core/manifest/manifest_row_range_test.cpp | 705 +++++++++++++++++++++
.../operation/append_only_file_store_scan_test.cpp | 45 +-
src/paimon/core/operation/file_store_scan.cpp | 9 +
src/paimon/core/utils/objects_file.h | 181 ++++--
20 files changed, 1173 insertions(+), 102 deletions(-)
diff --git a/docs/source/user_guide/manifest_cache.rst
b/docs/source/user_guide/manifest_cache.rst
index e1c095a7..f6dd034a 100644
--- a/docs/source/user_guide/manifest_cache.rst
+++ b/docs/source/user_guide/manifest_cache.rst
@@ -21,18 +21,24 @@ Manifest Cache
Overview
--------
-Paimon C++ caches raw manifest file bytes at the ``ObjectsFile<T>::Read()``
-layer. The cache uses the public ``Cache`` abstraction and is enabled through
-``ScanContextBuilder::WithCache()``. The cache covers data manifests, manifest
-lists, and index manifests because they all read through ``ObjectsFile<T>``.
+Paimon C++ caches decoded, schema-aligned manifest batches as Arrow IPC streams
+in ``ObjectsFile<T>``. The cache uses the public ``Cache`` abstraction and is
+enabled through ``ScanContextBuilder::WithCache()``. It covers data manifests,
+manifest lists, and index manifests because they all read through
``ObjectsFile<T>``.
For repeated ``get``, ``scan``, or batch ``get/scan -f`` requests in the same
process, the same snapshot often reads the same manifest files repeatedly. On a
-cache hit, the read path skips remote filesystem ``open/read``, builds an
-in-memory input stream from cached bytes, and still runs the format reader,
-Arrow decoding, and object deserialization. This design primarily reduces
-remote IO latency and bandwidth while keeping cache weight aligned with the
-actual cached bytes.
+cache hit, the read path skips remote filesystem ``open/read`` and Avro/ORC
+source-format decoding. It still reads the IPC stream, deserializes objects,
+and applies scan filters. Bucket and row-range reads can select entries before
+constructing their file metadata. Ordinary, bucket, and row-range reads share
+the same complete, query-independent cache entry.
+
+A cold cache load decodes the complete manifest and serializes its aligned
+batches into IPC. The loading reader consumes the original batches without an
+IPC round trip. Cold bucket reads may therefore decode more entries than
+uncached selective reads. Without a cache, reads retain the existing
+source-format reader path.
Configuration
-------------
@@ -86,11 +92,17 @@ Example:
Passing ``nullptr`` or omitting ``ScanContextBuilder::WithCache()`` leaves
manifest caching disabled.
-Future Optimizations
---------------------
+Cache Implementation Responsibilities
+-------------------------------------
-- Add hit, miss, bypass, and eviction metrics to read trace or metrics.
-- Add single-flight loading for high-concurrency misses on the same manifest
- path.
-- Evaluate a decoded-records second-level cache, configurable as a
- CPU-vs-memory tradeoff.
+Embedding applications can implement hit/miss and eviction statistics in their
+``Cache`` implementation. Coordination of concurrent loads for the same key
+also belongs to that implementation; ``ObjectsFile<T>`` does not deduplicate
+concurrent cache misses.
+
+Cache implementations should use ``CacheValue::GetMemoryUsage()`` for admission
+and eviction accounting. For manifest IPC, this reports the retained Arrow
+buffer capacity, including unused space from growth. ``GetSegment().Size()``
+continues to describe the valid IPC byte length. The built-in ``LruCache`` uses
+the memory usage value; existing cache values without an explicit allocation
+size continue to charge their segment length.
diff --git a/include/paimon/cache/cache.h b/include/paimon/cache/cache.h
index b282da74..e05b555e 100644
--- a/include/paimon/cache/cache.h
+++ b/include/paimon/cache/cache.h
@@ -88,12 +88,20 @@ class PAIMON_EXPORT Cache {
class PAIMON_EXPORT CacheValue {
public:
+ /// Charge the segment's logical size when no separate allocation size is
supplied.
CacheValue(const MemorySegment& segment, CacheCallback callback);
+ /// @param memory_usage Retained allocation size in bytes, including
unused capacity.
+ /// The charge is at least the segment's logical size.
+ CacheValue(const MemorySegment& segment, CacheCallback callback, int64_t
memory_usage);
+
~CacheValue();
const MemorySegment& GetSegment() const;
+ /// Bytes to charge against the cache budget. May exceed
GetSegment().Size().
+ int64_t GetMemoryUsage() const;
+
void OnEvict(const std::shared_ptr<CacheKey>& key) const;
bool operator==(const CacheValue& other) const;
@@ -101,6 +109,7 @@ class PAIMON_EXPORT CacheValue {
private:
MemorySegment segment_;
CacheCallback callback_;
+ int64_t memory_usage_;
};
} // namespace paimon
diff --git a/src/paimon/CMakeLists.txt b/src/paimon/CMakeLists.txt
index eb17538d..ca2630e4 100644
--- a/src/paimon/CMakeLists.txt
+++ b/src/paimon/CMakeLists.txt
@@ -834,6 +834,7 @@ if(PAIMON_BUILD_TESTS)
core/manifest/manifest_entry_serializer_test.cpp
core/manifest/manifest_file_meta_serializer_test.cpp
core/manifest/manifest_file_test.cpp
+ core/manifest/manifest_row_range_test.cpp
core/manifest/manifest_list_test.cpp
core/manifest/partition_entry_test.cpp
core/manifest/file_entry_test.cpp
diff --git a/src/paimon/common/io/cache/cache.cpp
b/src/paimon/common/io/cache/cache.cpp
index 37a26e91..c4c9e541 100644
--- a/src/paimon/common/io/cache/cache.cpp
+++ b/src/paimon/common/io/cache/cache.cpp
@@ -19,12 +19,18 @@
#include "paimon/cache/cache.h"
+#include <algorithm>
#include <utility>
namespace paimon {
CacheValue::CacheValue(const MemorySegment& segment, CacheCallback callback)
- : segment_(segment), callback_(std::move(callback)) {}
+ : CacheValue(segment, std::move(callback), segment.Size()) {}
+
+CacheValue::CacheValue(const MemorySegment& segment, CacheCallback callback,
int64_t memory_usage)
+ : segment_(segment),
+ callback_(std::move(callback)),
+ memory_usage_(std::max<int64_t>(segment.Size(), memory_usage)) {}
CacheValue::~CacheValue() = default;
@@ -32,6 +38,10 @@ const MemorySegment& CacheValue::GetSegment() const {
return segment_;
}
+int64_t CacheValue::GetMemoryUsage() const {
+ return memory_usage_;
+}
+
void CacheValue::OnEvict(const std::shared_ptr<CacheKey>& key) const {
if (callback_) {
callback_(key);
diff --git a/src/paimon/common/io/cache/lru_cache.cpp
b/src/paimon/common/io/cache/lru_cache.cpp
index fbfca83c..892ddb6f 100644
--- a/src/paimon/common/io/cache/lru_cache.cpp
+++ b/src/paimon/common/io/cache/lru_cache.cpp
@@ -26,7 +26,7 @@ LruCache::LruCache(int64_t max_weight)
.expire_after_access_ms = -1,
.weigh_func = [](const std::shared_ptr<CacheKey>& /*key*/,
const std::shared_ptr<CacheValue>& value) ->
int64_t {
- return value ? value->GetSegment().Size() : 0;
+ return value ? value->GetMemoryUsage() : 0;
},
.removal_callback =
[](const std::shared_ptr<CacheKey>& key, const
std::shared_ptr<CacheValue>& value,
diff --git a/src/paimon/common/io/cache/lru_cache.h
b/src/paimon/common/io/cache/lru_cache.h
index 745ca7f5..3bf2b10a 100644
--- a/src/paimon/common/io/cache/lru_cache.h
+++ b/src/paimon/common/io/cache/lru_cache.h
@@ -32,7 +32,7 @@ namespace paimon {
/// LRU Cache implementation with weight-based eviction for block cache.
///
/// Wraps GenericLruCache with CacheKey/CacheValue types. Capacity is measured
-/// in bytes (sum of MemorySegment sizes). When an entry is evicted, its
+/// in bytes (sum of CacheValue::GetMemoryUsage()). When an entry is evicted,
its
/// CacheCallback is invoked to notify the upper layer.
///
/// @note Thread-safe: all public methods are protected by the underlying
GenericLruCache lock.
diff --git a/src/paimon/common/io/cache/lru_cache_test.cpp
b/src/paimon/common/io/cache/lru_cache_test.cpp
index 29eb10fe..6a9cab5b 100644
--- a/src/paimon/common/io/cache/lru_cache_test.cpp
+++ b/src/paimon/common/io/cache/lru_cache_test.cpp
@@ -113,6 +113,40 @@ TEST_F(LruCacheTest, TestPutInsertAndUpdate) {
ASSERT_EQ(result->GetSegment().Get(0), 'B');
}
+TEST_F(LruCacheTest, TestRetainedMemoryAccounting) {
+ LruCache cache(192);
+ auto segment = MemorySegment::AllocateHeapMemory(64, pool_.get());
+ auto value = std::make_shared<CacheValue>(segment, CacheCallback(), 128);
+ ASSERT_EQ(64, value->GetSegment().Size());
+ ASSERT_EQ(128, value->GetMemoryUsage());
+ ASSERT_EQ(64, CacheValue(segment, CacheCallback()).GetMemoryUsage());
+ ASSERT_EQ(64, CacheValue(segment, CacheCallback(), 32).GetMemoryUsage());
+
+ std::vector<std::shared_ptr<CacheKey>> evicted;
+ auto first = std::make_shared<CacheValue>(
+ segment, [&](const std::shared_ptr<CacheKey>& key) {
evicted.push_back(key); }, 128);
+ auto first_key = MakeKey(0);
+ ASSERT_OK(cache.Put(first_key, first));
+ ASSERT_EQ(128, cache.GetCurrentWeight());
+ ASSERT_OK(cache.Put(MakeKey(1), value));
+ ASSERT_EQ(1, cache.Size());
+ ASSERT_EQ(128, cache.GetCurrentWeight());
+ ASSERT_EQ(std::vector<std::shared_ptr<CacheKey>>{first_key}, evicted);
+ cache.InvalidateAll();
+ ASSERT_EQ(0, cache.GetCurrentWeight());
+
+ // The payload fits, but its retained allocation exceeds the cache budget.
+ LruCache small_cache(100);
+ ASSERT_OK_AND_ASSIGN(
+ std::shared_ptr<CacheValue> loaded,
+ small_cache.Get(MakeKey(0),
+ [&](const std::shared_ptr<CacheKey>&)
+ -> Result<std::shared_ptr<CacheValue>> { return
value; }));
+ ASSERT_EQ(value, loaded);
+ ASSERT_EQ(0, small_cache.Size());
+ ASSERT_EQ(0, small_cache.GetCurrentWeight());
+}
+
/// Verifies weight-based eviction: when total weight exceeds max, LRU entries
are evicted.
TEST_F(LruCacheTest, TestWeightBasedEviction) {
// Cache can hold at most 200 bytes
diff --git a/src/paimon/core/io/data_file_meta.h
b/src/paimon/core/io/data_file_meta.h
index 436d7378..ee3d4a9f 100644
--- a/src/paimon/core/io/data_file_meta.h
+++ b/src/paimon/core/io/data_file_meta.h
@@ -43,6 +43,10 @@ class Bytes;
/// Metadata of a data file.
struct DataFileMeta {
+ /// Field positions in DataType(), shared by serialization and manifest
pruning.
+ static constexpr int32_t kRowCountFieldIndex = 2;
+ static constexpr int32_t kFirstRowIdFieldIndex = 18;
+
static const BinaryRow& EmptyMinKey();
static const BinaryRow& EmptyMaxKey();
static constexpr int32_t DUMMY_LEVEL = 0;
diff --git a/src/paimon/core/io/data_file_meta_serializer.cpp
b/src/paimon/core/io/data_file_meta_serializer.cpp
index 9f88ebec..8fa7eba0 100644
--- a/src/paimon/core/io/data_file_meta_serializer.cpp
+++ b/src/paimon/core/io/data_file_meta_serializer.cpp
@@ -46,7 +46,7 @@ Result<BinaryRow> DataFileMetaSerializer::ToRow(const
std::shared_ptr<DataFileMe
BinaryRowWriter writer(&row, 32 * 1024, pool_.get());
writer.WriteString(0, BinaryString::FromString(meta->file_name,
pool_.get()));
writer.WriteLong(1, meta->file_size);
- writer.WriteLong(2, meta->row_count);
+ writer.WriteLong(DataFileMeta::kRowCountFieldIndex, meta->row_count);
auto min_key_bytes = SerializationUtils::SerializeBinaryRow(meta->min_key,
pool_.get());
writer.WriteBinary(3, *min_key_bytes);
auto max_key_bytes = SerializationUtils::SerializeBinaryRow(meta->max_key,
pool_.get());
@@ -86,9 +86,9 @@ Result<BinaryRow> DataFileMetaSerializer::ToRow(const
std::shared_ptr<DataFileMe
writer.WriteString(17,
BinaryString::FromString(meta->external_path.value(), pool_.get()));
}
if (meta->first_row_id == std::nullopt) {
- writer.SetNullAt(18);
+ writer.SetNullAt(DataFileMeta::kFirstRowIdFieldIndex);
} else {
- writer.WriteLong(18, meta->first_row_id.value());
+ writer.WriteLong(DataFileMeta::kFirstRowIdFieldIndex,
meta->first_row_id.value());
}
if (meta->write_cols == std::nullopt) {
writer.SetNullAt(19);
@@ -110,7 +110,7 @@ Result<std::shared_ptr<DataFileMeta>>
DataFileMetaSerializer::FromRow(
const InternalRow& row) const {
auto file_name = row.GetString(0);
auto file_size = row.GetLong(1);
- auto row_count = row.GetLong(2);
+ auto row_count = row.GetLong(DataFileMeta::kRowCountFieldIndex);
auto min_key = row.GetBinary(3);
auto max_key = row.GetBinary(4);
auto key_stats_row = row.GetRow(5, 3);
@@ -155,8 +155,8 @@ Result<std::shared_ptr<DataFileMeta>>
DataFileMetaSerializer::FromRow(
external_path = row.GetString(17).ToString();
}
std::optional<int64_t> first_row_id;
- if (!row.IsNullAt(18)) {
- first_row_id = row.GetLong(18);
+ if (!row.IsNullAt(DataFileMeta::kFirstRowIdFieldIndex)) {
+ first_row_id = row.GetLong(DataFileMeta::kFirstRowIdFieldIndex);
}
std::optional<std::vector<std::string>> write_cols;
diff --git a/src/paimon/core/manifest/index_manifest_file_handler_test.cpp
b/src/paimon/core/manifest/index_manifest_file_handler_test.cpp
index 42048dc7..e0ce213a 100644
--- a/src/paimon/core/manifest/index_manifest_file_handler_test.cpp
+++ b/src/paimon/core/manifest/index_manifest_file_handler_test.cpp
@@ -25,7 +25,10 @@
#include <vector>
#include "arrow/api.h"
+#include "arrow/io/memory.h"
+#include "arrow/ipc/api.h"
#include "gtest/gtest.h"
+#include "paimon/common/utils/path_util.h"
#include "paimon/core/core_options.h"
#include "paimon/core/deletionvectors/deletion_vectors_index_file.h"
#include "paimon/core/index/index_file_meta.h"
@@ -37,6 +40,7 @@
#include "paimon/format/file_format_factory.h"
#include "paimon/memory/memory_pool.h"
#include "paimon/testing/utils/binary_row_generator.h"
+#include "paimon/testing/utils/counting_cache_test_utils.h"
#include "paimon/testing/utils/testharness.h"
namespace paimon::test {
@@ -50,7 +54,8 @@ class IndexManifestFileHandlerTest : public testing::Test {
}
Result<std::unique_ptr<IndexManifestFile>> CreateManifestFile(
- int32_t bucket_mode, const std::string& file_format_identifier =
"avro") const {
+ int32_t bucket_mode, const std::string& file_format_identifier =
"avro",
+ const std::shared_ptr<Cache>& cache = nullptr) const {
PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<FileFormat> file_format,
FileFormatFactory::Get(file_format_identifier,
{}));
auto schema = arrow::schema({arrow::field("f0", arrow::int32())});
@@ -63,6 +68,7 @@ class IndexManifestFileHandlerTest : public testing::Test {
/*global_index_external_path=*/std::nullopt,
/*index_file_in_data_file_dir=*/false, pool_));
PAIMON_ASSIGN_OR_RAISE(CoreOptions options, CoreOptions::FromMap({}));
+ options.WithCache(cache);
return IndexManifestFile::Create(dir_->GetFileSystem(), file_format,
"zstd", path_factory,
bucket_mode, pool_, options);
}
@@ -100,6 +106,51 @@ class IndexManifestFileHandlerTest : public testing::Test {
std::unique_ptr<UniqueTestDirectory> dir_;
};
+TEST_F(IndexManifestFileHandlerTest, DecodedCacheReusesIpcAcrossReaders) {
+ auto cache = std::make_shared<CountingRoutingCache>(CacheKind::MANIFEST,
1024 * 1024);
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<IndexManifestFile> writer,
+ CreateManifestFile(2, "avro", cache));
+ const std::vector<IndexManifestEntry> expected = {
+ MakeDvEntry(FileKind::Add(), BinaryRow::EmptyRow(), 0, "dv-0",
{"data-a", "data-b"}, 10),
+ MakeEntry(FileKind::Add(), BinaryRow::EmptyRow(), 1, "BTREE",
"index-1", 20)};
+ using WrittenFile = std::pair<std::string, int64_t>;
+ ASSERT_OK_AND_ASSIGN(WrittenFile written,
writer->WriteWithoutRolling(expected));
+ std::vector<IndexManifestEntry> cold;
+ ASSERT_OK(writer->Read(written.first, nullptr, written.second, &cold));
+ ASSERT_EQ(expected, cold);
+ auto key = CacheKey::ForKind(
+ PathUtil::JoinPath(FileStorePathFactory::ManifestPath(dir_->Str()),
written.first),
+ /*position=*/0, /*length=*/-1, CacheKind::MANIFEST);
+ ASSERT_OK_AND_ASSIGN(
+ std::shared_ptr<CacheValue> cached,
+ cache->Get(key,
+ [](const std::shared_ptr<CacheKey>&) ->
Result<std::shared_ptr<CacheValue>> {
+ return Status::Invalid("index manifest was not cached");
+ }));
+ ASSERT_TRUE(cached);
+ const auto& segment = cached->GetSegment();
+ auto buffer = std::make_shared<arrow::Buffer>(reinterpret_cast<const
uint8_t*>(segment.Data()),
+ segment.Size());
+ auto ipc = arrow::ipc::RecordBatchStreamReader::Open(
+ std::make_shared<arrow::io::BufferReader>(buffer));
+ ASSERT_TRUE(ipc.ok()) << ipc.status().ToString();
+ auto batch = ipc.ValueOrDie()->Next();
+ ASSERT_TRUE(batch.ok()) << batch.status().ToString();
+ ASSERT_TRUE(batch.ValueOrDie());
+ ASSERT_EQ(expected.size(), batch.ValueOrDie()->num_rows());
+ writer->DeleteQuietly(written.first);
+ writer.reset();
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<IndexManifestFile> reader,
+ CreateManifestFile(2, "avro", cache));
+ std::vector<IndexManifestEntry> warm;
+ ASSERT_OK(reader->Read(written.first, nullptr, written.second, &warm));
+ ASSERT_EQ(expected, warm);
+ ASSERT_EQ(1, cache->SupplierCallCount());
+ cache->Invalidate(key);
+ ASSERT_EQ(0, cache->Size());
+ ASSERT_NOK(reader->Read(written.first, nullptr, written.second, &warm));
+}
+
TEST_F(IndexManifestFileHandlerTest, GlobalCombinerDeletesThenAddsByFileName) {
ASSERT_OK_AND_ASSIGN(auto index_manifest_file,
CreateManifestFile(/*bucket_mode=*/4));
diff --git a/src/paimon/core/manifest/manifest_entry.h
b/src/paimon/core/manifest/manifest_entry.h
index ca6e8ffa..f7546113 100644
--- a/src/paimon/core/manifest/manifest_entry.h
+++ b/src/paimon/core/manifest/manifest_entry.h
@@ -38,6 +38,9 @@ namespace paimon {
/// Entry of a manifest file, representing an addition / deletion of a data
file.
class ManifestEntry : public FileEntry {
public:
+ /// Field position in DataType(), before the serializer prepends _VERSION.
+ static constexpr int32_t kFileFieldIndex = 4;
+
static const std::shared_ptr<arrow::DataType>& DataType();
static int64_t RecordCount(const std::vector<ManifestEntry>&
manifest_entries);
static std::optional<int64_t> NullableRecordCount(
diff --git a/src/paimon/core/manifest/manifest_entry_serializer.cpp
b/src/paimon/core/manifest/manifest_entry_serializer.cpp
index 2389cd8d..5e1a2be0 100644
--- a/src/paimon/core/manifest/manifest_entry_serializer.cpp
+++ b/src/paimon/core/manifest/manifest_entry_serializer.cpp
@@ -58,7 +58,7 @@ Result<ManifestEntry>
ManifestEntrySerializer::ConvertFrom(int32_t version,
SerializationUtils::DeserializeBinaryRow(partition_bytes));
auto bucket = row.GetInt(2);
auto total_buckets = row.GetInt(3);
- auto file = row.GetRow(4, data_file_meta_serializer_.NumFields());
+ auto file = row.GetRow(ManifestEntry::kFileFieldIndex,
data_file_meta_serializer_.NumFields());
if (!file) {
return Status::Invalid("ManifestEntry convert from row failed, with
null DataFileMeta");
}
@@ -80,7 +80,7 @@ Result<BinaryRow> ManifestEntrySerializer::ToRow(const
ManifestEntry& record) co
writer.WriteInt(4, record.TotalBuckets());
PAIMON_ASSIGN_OR_RAISE(BinaryRow data_file_meta_row,
data_file_meta_serializer_.ToRow(record.File()));
- writer.WriteRow(5, data_file_meta_row);
+ writer.WriteRow(ManifestEntry::kFileFieldIndex + 1, data_file_meta_row);
writer.Complete();
return row;
}
diff --git a/src/paimon/core/manifest/manifest_file.cpp
b/src/paimon/core/manifest/manifest_file.cpp
index d525f62a..c0a71a2f 100644
--- a/src/paimon/core/manifest/manifest_file.cpp
+++ b/src/paimon/core/manifest/manifest_file.cpp
@@ -44,6 +44,7 @@
#include "paimon/format/writer_builder.h"
#include "paimon/predicate/predicate_builder.h"
#include "paimon/status.h"
+#include "paimon/utils/row_range_index.h"
namespace arrow {
class DataType;
@@ -133,6 +134,64 @@ 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 {
+ return ReadArrowBatches(
+ file_name, file_size,
+ [this, &row_ranges, &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");
+ }
+ 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 (!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;
+ }
+ }
+ PAIMON_ASSIGN_OR_RAISE(ManifestEntry entry,
serializer_->FromRow(row));
+ if (filter) {
+ PAIMON_ASSIGN_OR_RAISE(bool keep, filter(entry));
+ if (!keep) {
+ continue;
+ }
+ }
+ entries->push_back(std::move(entry));
+ }
+ return Status::OK();
+ },
+ /*prepare_reader=*/nullptr);
+}
+
Status ManifestFile::PrepareBucketRead(int32_t bucket,
const std::optional<int32_t>&
expected_total_buckets,
std::unique_ptr<FileBatchReader>*
reader) const {
diff --git a/src/paimon/core/manifest/manifest_file.h
b/src/paimon/core/manifest/manifest_file.h
index 3840368f..c37882b5 100644
--- a/src/paimon/core/manifest/manifest_file.h
+++ b/src/paimon/core/manifest/manifest_file.h
@@ -19,6 +19,7 @@
#pragma once
#include <cstdint>
+#include <functional>
#include <memory>
#include <optional>
#include <string>
@@ -46,6 +47,7 @@ class WriterBuilder;
class PathFactory;
class ManifestFileMeta;
class ManifestEntry;
+class RowRangeIndex;
class MemoryPool;
/// This file includes several `ManifestEntry`s, representing the additional
changes since last
@@ -76,6 +78,14 @@ 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;
+
private:
Status PrepareBucketRead(int32_t bucket, const std::optional<int32_t>&
expected_total_buckets,
std::unique_ptr<FileBatchReader>* reader) const;
diff --git a/src/paimon/core/manifest/manifest_file_test.cpp
b/src/paimon/core/manifest/manifest_file_test.cpp
index df5bcbd8..b0ce4b9e 100644
--- a/src/paimon/core/manifest/manifest_file_test.cpp
+++ b/src/paimon/core/manifest/manifest_file_test.cpp
@@ -314,7 +314,7 @@ TEST_F(ManifestFileTest,
TestManifestCacheIsDisabledWithoutInjectedCache) {
ASSERT_EQ(0, counting_file_system->get_file_status_count);
}
-TEST_F(ManifestFileTest, TestManifestCacheReusesCachedBytes) {
+TEST_F(ManifestFileTest, TestManifestCacheReusesDecodedBatches) {
auto pool = GetDefaultPool();
auto counting_file_system = std::make_shared<CountingFileSystem>();
auto manifest_cache =
@@ -500,7 +500,7 @@ TEST_F(ManifestFileTest,
TestInferredBucketProbeSkipsArrowMaterialization) {
manifest_file->WriteWithoutRolling({excluded,
selected}));
std::vector<ManifestEntry> warm;
ASSERT_OK(manifest_file->Read(written.first, nullptr,
/*file_size=*/std::nullopt, &warm));
- // Cached bytes, if available, belong to a separate pool from reader
allocations.
+ // Warm decoded batches, if available, belong to a separate pool from
reader allocations.
for (bool inferred : {false, true}) {
std::shared_ptr<MemoryPool> read_pool = GetMemoryPool();
ASSERT_OK_AND_ASSIGN(
@@ -571,7 +571,7 @@ TEST_F(ManifestFileTest,
TestReadBucketEntriesSkipsDeserializingOtherBuckets) {
WrittenFile written_file,
manifest_file->WriteWithoutRolling({invalid_other_bucket,
valid_target_bucket}));
- // Exercise both a cold cache and reuse of the retained manifest bytes.
+ // Exercise both a cold cache and reuse of the decoded manifest
batches.
for (int32_t read = 0; read < 2; ++read) {
std::vector<ManifestEntry> bucket_entries;
ASSERT_OK(manifest_file->ReadBucketEntries(written_file.first,
/*bucket=*/0,
diff --git a/src/paimon/core/manifest/manifest_list_test.cpp
b/src/paimon/core/manifest/manifest_list_test.cpp
index d8d25538..fa839b3f 100644
--- a/src/paimon/core/manifest/manifest_list_test.cpp
+++ b/src/paimon/core/manifest/manifest_list_test.cpp
@@ -22,9 +22,11 @@
#include <optional>
#include <variant>
+#include "arrow/io/memory.h"
+#include "arrow/ipc/api.h"
#include "arrow/type.h"
#include "gtest/gtest.h"
-#include "paimon/core/core_options.h"
+#include "paimon/common/utils/path_util.h"
#include "paimon/core/manifest/manifest_file_meta.h"
#include "paimon/core/snapshot.h"
#include "paimon/core/stats/simple_stats.h"
@@ -34,6 +36,7 @@
#include "paimon/fs/local/local_file_system.h"
#include "paimon/memory/memory_pool.h"
#include "paimon/testing/utils/binary_row_generator.h"
+#include "paimon/testing/utils/counting_cache_test_utils.h"
#include "paimon/testing/utils/testharness.h"
namespace paimon::test {
@@ -97,7 +100,8 @@ class ManifestListTest : public testing::Test {
std::unique_ptr<ManifestList> CreateManifestList(
const std::shared_ptr<FileSystem>& file_system, const std::string&
file_format_str,
- const std::string& root_path, const std::shared_ptr<MemoryPool>& pool)
const {
+ const std::string& root_path, const std::shared_ptr<MemoryPool>& pool,
+ const std::shared_ptr<Cache>& cache = nullptr) const {
EXPECT_OK_AND_ASSIGN(std::shared_ptr<FileFormat> file_format,
FileFormatFactory::Get(file_format_str, {}));
auto unused_schema =
arrow::schema(arrow::FieldVector({arrow::field("f0", arrow::utf8())}));
@@ -109,10 +113,9 @@ class ManifestListTest : public testing::Test {
/*legacy_partition_name_enabled=*/true,
/*external_paths=*/{},
/*global_index_external_path=*/std::nullopt,
/*index_file_in_data_file_dir=*/false, pool));
- EXPECT_OK_AND_ASSIGN(CoreOptions options, CoreOptions::FromMap({}));
- EXPECT_OK_AND_ASSIGN(auto manifest_list,
- ManifestList::Create(file_system, file_format,
"zstd", path_factory,
- options.GetCache(), pool));
+ EXPECT_OK_AND_ASSIGN(
+ auto manifest_list,
+ ManifestList::Create(file_system, file_format, "zstd",
path_factory, cache, pool));
return manifest_list;
}
@@ -127,6 +130,71 @@ class ManifestListTest : public testing::Test {
}
};
+TEST_F(ManifestListTest, DecodedCacheReusesIpcAcrossReaders) {
+ auto dir = UniqueTestDirectory::Create();
+ ASSERT_TRUE(dir);
+ auto pool = GetDefaultPool();
+ auto cache = std::make_shared<CountingRoutingCache>(CacheKind::MANIFEST,
1024 * 1024);
+ auto writer = CreateManifestList(dir->GetFileSystem(), "avro", dir->Str(),
pool, cache);
+ const std::vector<ManifestFileMeta> expected = {MakeMeta("manifest-a",
100, 2, 0),
+ MakeMeta("manifest-b",
200, 0, 1)};
+ using WrittenFile = std::pair<std::string, int64_t>;
+ ASSERT_OK_AND_ASSIGN(WrittenFile written, writer->Write(expected));
+ std::vector<ManifestFileMeta> cold;
+ ASSERT_OK(writer->Read(written.first, nullptr, written.second, &cold));
+ ASSERT_EQ(expected, cold);
+ auto key = CacheKey::ForKind(
+ PathUtil::JoinPath(FileStorePathFactory::ManifestPath(dir->Str()),
written.first),
+ /*position=*/0, /*length=*/-1, CacheKind::MANIFEST);
+ ASSERT_OK_AND_ASSIGN(
+ std::shared_ptr<CacheValue> cached,
+ cache->Get(key,
+ [](const std::shared_ptr<CacheKey>&) ->
Result<std::shared_ptr<CacheValue>> {
+ return Status::Invalid("manifest list was not cached");
+ }));
+ ASSERT_TRUE(cached);
+ const auto& segment = cached->GetSegment();
+ auto buffer = std::make_shared<arrow::Buffer>(reinterpret_cast<const
uint8_t*>(segment.Data()),
+ segment.Size());
+ // The existing cache key contains IPC, not source-format bytes.
+ auto ipc = arrow::ipc::RecordBatchStreamReader::Open(
+ std::make_shared<arrow::io::BufferReader>(buffer));
+ ASSERT_TRUE(ipc.ok()) << ipc.status().ToString();
+ auto batch = ipc.ValueOrDie()->Next();
+ ASSERT_TRUE(batch.ok()) << batch.status().ToString();
+ ASSERT_TRUE(batch.ValueOrDie());
+ ASSERT_EQ(expected.size(), batch.ValueOrDie()->num_rows());
+ writer->DeleteQuietly(written.first);
+ writer.reset();
+ auto reader = CreateManifestList(dir->GetFileSystem(), "avro", dir->Str(),
pool, cache);
+ std::vector<ManifestFileMeta> warm;
+ ASSERT_OK(reader->Read(written.first, nullptr, written.second, &warm));
+ ASSERT_EQ(expected, warm);
+ ASSERT_EQ(1, cache->SupplierCallCount());
+ cache->Invalidate(key);
+ ASSERT_EQ(0, cache->Size());
+ ASSERT_NOK(reader->Read(written.first, nullptr, written.second, &warm));
+}
+
+TEST_F(ManifestListTest, OrcDecodedCachePreservesLegacyMetadata) {
+ auto pool = GetDefaultPool();
+ auto fs = std::make_shared<LocalFileSystem>();
+ const std::string path = GetDataDir() + "/orc/append_09.db/append_09";
+ const std::string file =
"manifest-list-f2d59cb8-3ec6-4860-b34b-050b1a533416-2";
+ auto uncached = CreateManifestList(fs, "orc", path, pool);
+ std::vector<ManifestFileMeta> expected;
+ ASSERT_OK(uncached->Read(file, nullptr, std::nullopt, &expected));
+ ASSERT_EQ(4, expected.size());
+ auto cache = std::make_shared<CountingRoutingCache>(CacheKind::MANIFEST,
1024 * 1024);
+ for (int32_t attempt = 0; attempt < 2; ++attempt) {
+ auto reader = CreateManifestList(fs, "orc", path, pool, cache);
+ std::vector<ManifestFileMeta> actual;
+ ASSERT_OK(reader->Read(file, nullptr, std::nullopt, &actual));
+ ASSERT_EQ(expected, actual);
+ ASSERT_EQ(1, cache->SupplierCallCount());
+ }
+}
+
TEST_F(ManifestListTest, TestSimple) {
auto pool = GetDefaultPool();
auto manifest_file_metas =
diff --git a/src/paimon/core/manifest/manifest_row_range_test.cpp
b/src/paimon/core/manifest/manifest_row_range_test.cpp
new file mode 100644
index 00000000..a4f60fea
--- /dev/null
+++ b/src/paimon/core/manifest/manifest_row_range_test.cpp
@@ -0,0 +1,705 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include <atomic>
+#include <future>
+#include <limits>
+#include <map>
+#include <memory>
+#include <optional>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "arrow/api.h"
+#include "arrow/c/bridge.h"
+#include "fmt/format.h"
+#include "gtest/gtest.h"
+#include "paimon/common/utils/checked_cast.h"
+#include "paimon/common/utils/path_util.h"
+#include "paimon/core/io/data_file_meta.h"
+#include "paimon/core/io/meta_to_arrow_array_converter.h"
+#include "paimon/core/manifest/manifest_entry_serializer.h"
+#include "paimon/core/manifest/manifest_file.h"
+#include "paimon/core/manifest/manifest_list.h"
+#include "paimon/core/operation/data_evolution_file_store_scan.h"
+#include "paimon/core/operation/file_store_scan.h"
+#include "paimon/core/utils/file_store_path_factory.h"
+#include "paimon/format/file_format.h"
+#include "paimon/format/file_format_factory.h"
+#include "paimon/format/format_writer.h"
+#include "paimon/format/writer_builder.h"
+#include "paimon/fs/local/local_file_system.h"
+#include "paimon/testing/utils/counting_cache_test_utils.h"
+#include "paimon/testing/utils/testharness.h"
+#include "paimon/utils/row_range_index.h"
+
+namespace paimon::test {
+class RowRangeManifestFileTest : public ::testing::Test {
+ protected:
+ Result<std::unique_ptr<ManifestFile>> CreateManifest(
+ const std::string& path, const std::shared_ptr<FileSystem>& fs, bool
cache_enabled,
+ const std::shared_ptr<Cache>& cache = nullptr,
+ const std::shared_ptr<MemoryPool>& pool = GetDefaultPool()) {
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<FileFormat> format,
+ FileFormatFactory::Get("avro", {}));
+ auto schema = arrow::schema(arrow::FieldVector{});
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<FileStorePathFactory>
path_factory,
+ FileStorePathFactory::Create(path, schema, {},
"", "avro", "data-",
+ true, {},
std::nullopt, false, pool));
+ PAIMON_ASSIGN_OR_RAISE(CoreOptions options,
+
CoreOptions::FromMap({{Options::READ_BATCH_SIZE, "2"}}));
+ if (cache) {
+ options.WithCache(cache);
+ } else if (cache_enabled) {
+ options.WithCache(std::make_shared<LruCache>(64 * 1024 * 1024));
+ }
+
PAIMON_RETURN_NOT_OK(fs->Mkdirs(FileStorePathFactory::ManifestPath(path)));
+ return ManifestFile::Create(fs, format, "null", path_factory, 1024,
pool, options, schema);
+ }
+
+ Result<ManifestEntry> Entry(const std::string& name,
std::optional<int64_t> first,
+ int64_t count) {
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<DataFileMeta> meta,
+ DataFileMeta::ForAppend(name, 100, count,
SimpleStats::EmptyStats(),
+ 0, 0, 0,
FileSource::Append(), std::nullopt,
+ std::nullopt, first,
std::nullopt));
+ return ManifestEntry(FileKind::Add(), BinaryRow::EmptyRow(), 0, 1,
meta);
+ }
+
+ Status WriteArray(const std::string& path, const std::string& name,
+ const std::shared_ptr<arrow::StructArray>& array) {
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<FileFormat> format,
+ FileFormatFactory::Get("avro", {}));
+ ArrowSchema schema;
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportType(*array->type(),
&schema));
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<WriterBuilder> builder,
+ format->CreateWriterBuilder(&schema, 2));
+ LocalFileSystem fs;
+ PAIMON_ASSIGN_OR_RAISE(
+ std::shared_ptr<OutputStream> output,
+ fs.Create(FileStorePathFactory::ManifestPath(path) + "/" + name,
false));
+ PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<FormatWriter> writer,
+ builder->Build(output, "null"));
+ ArrowArray batch;
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportArray(*array, &batch));
+ PAIMON_RETURN_NOT_OK(writer->AddBatch(&batch));
+ PAIMON_RETURN_NOT_OK(writer->Flush());
+ PAIMON_RETURN_NOT_OK(writer->Finish());
+ return output->Close();
+ }
+};
+
+TEST_F(RowRangeManifestFileTest, ArrowCacheReuseEvictionAndConcurrentReaders) {
+ auto dir = UniqueTestDirectory::Create();
+ ASSERT_TRUE(dir);
+ const auto cache_kind = CacheKind::MANIFEST;
+ auto cache = std::make_shared<CountingRoutingCache>(cache_kind, 1024 *
1024);
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<ManifestFile> manifest,
+ CreateManifest(dir->Str(), dir->GetFileSystem(),
true, cache));
+ ASSERT_OK_AND_ASSIGN(ManifestEntry a, Entry("a.parquet", 100, 10));
+ ASSERT_OK_AND_ASSIGN(ManifestEntry b, Entry("b.parquet", 110, 10));
+ using WrittenFile = std::pair<std::string, int64_t>;
+ ASSERT_OK_AND_ASSIGN(WrittenFile written,
manifest->WriteWithoutRolling({a, b}));
+ ASSERT_OK_AND_ASSIGN(RowRangeIndex ranges,
RowRangeIndex::Create({Range(110, 110)}));
+ std::vector<ManifestEntry> expected{b};
+ auto read = [&]() -> Status {
+ 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));
+ if (entries != expected) {
+ return Status::Invalid("cached manifest result differs from
source");
+ }
+ return Status::OK();
+ };
+ ASSERT_OK(read());
+ // The first read loads the manifest once; concurrent warm reads reuse it.
+ ASSERT_EQ(1, cache->SupplierCallCount(cache_kind));
+ // Readers own their decode state; the cache only shares immutable bytes.
+ std::vector<std::future<Status>> readers;
+ for (int32_t i = 0; i < 8; ++i) {
+ readers.push_back(std::async(std::launch::async, read));
+ }
+ for (auto& reader : readers) {
+ ASSERT_OK(reader.get());
+ }
+ ASSERT_EQ(1, cache->SupplierCallCount(cache_kind));
+ cache->InvalidateAll();
+ ASSERT_OK(read());
+ ASSERT_EQ(2, cache->SupplierCallCount(cache_kind));
+ // 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_TRUE(entries.empty());
+ ASSERT_EQ(3, cache->SupplierCallCount(cache_kind));
+
+ auto unsupported_cache =
+ 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_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_EQ(expected, entries);
+ ASSERT_LE(small_cache->GetCurrentWeight(), 1);
+ small_cache->InvalidateAll();
+ uncached.reset();
+ // Materialized entries must remain valid after both cache eviction and
reader destruction.
+ ASSERT_EQ(expected, entries);
+}
+
+TEST_F(RowRangeManifestFileTest,
ScanPlanPreservesResultsAcrossLazyDecodeAndCacheModes) {
+ auto pool = GetDefaultPool();
+ std::shared_ptr<Executor> executor = CreateDefaultExecutor();
+ auto filters = std::make_shared<ScanFilter>(
+ nullptr, std::vector<std::map<std::string, std::string>>{},
std::nullopt);
+ auto schema = arrow::schema({arrow::field("value", arrow::int32())});
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> table_schema,
+ TableSchema::Create(0, schema, {}, {}, {}));
+ ASSERT_OK_AND_ASSIGN(ManifestEntry a, Entry("a.parquet", 100, 10));
+ ASSERT_OK_AND_ASSIGN(ManifestEntry b, Entry("b.parquet", 110, 10));
+ ASSERT_OK_AND_ASSIGN(ManifestEntry c, Entry("c.parquet", 120, 10));
+ ASSERT_OK_AND_ASSIGN(ManifestEntry unknown, Entry("unknown.parquet",
std::nullopt, 10));
+ ManifestEntry deleted(FileKind::Delete(), a.Partition(), a.Bucket(),
a.TotalBuckets(),
+ a.File());
+ struct Query {
+ std::optional<std::vector<Range>> ranges;
+ std::vector<ManifestEntry> expected;
+ int32_t retained_entries;
+ };
+ const std::vector<Query> queries = {
+ {std::vector<Range>{Range(100, 100)}, {unknown}, 3},
+ {std::vector<Range>{Range(110, 110)}, {b, unknown}, 2},
+ {std::vector<Range>{Range(120, 129)}, {unknown, c}, 2},
+ {std::vector<Range>{Range(109, 110), Range(129, 129)}, {b, unknown,
c}, 5},
+ {std::vector<Range>{Range(1000, 1000)}, {unknown}, 1},
+ {std::vector<Range>{}, {unknown}, 1},
+ {std::nullopt, {b, unknown, c}, 5}};
+ for (bool cache_enabled : {false, true}) {
+ SCOPED_TRACE(cache_enabled);
+ auto dir = UniqueTestDirectory::Create();
+ ASSERT_TRUE(dir);
+ auto fs = dir->GetFileSystem();
+ auto cache = cache_enabled
+ ?
std::make_shared<CountingRoutingCache>(CacheKind::MANIFEST, 1024 * 1024)
+ : nullptr;
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<ManifestFile> manifest,
+ CreateManifest(dir->Str(), fs, cache_enabled,
cache));
+ using WrittenFile = std::pair<std::string, int64_t>;
+ ASSERT_OK_AND_ASSIGN(WrittenFile base,
manifest->WriteWithoutRolling({a, b, unknown}));
+ ASSERT_OK_AND_ASSIGN(WrittenFile delta,
manifest->WriteWithoutRolling({deleted, c}));
+ // Unknown manifest bounds force entry-level pruning even for disjoint
queries.
+ ManifestFileMeta base_meta(base.first, base.second, 3, 0,
SimpleStats::EmptyStats(), 0, 0,
+ 0, 0, 0, std::nullopt, std::nullopt);
+ ManifestFileMeta delta_meta(delta.first, delta.second, 1, 1,
SimpleStats::EmptyStats(), 0,
+ 0, 0, 0, 0, std::nullopt, std::nullopt);
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<FileFormat> format,
+ FileFormatFactory::Get("avro", {}));
+ ASSERT_OK_AND_ASSIGN(
+ std::shared_ptr<FileStorePathFactory> paths,
+ FileStorePathFactory::Create(dir->Str(), schema, {}, "", "avro",
"data-", true, {},
+ std::nullopt, false, pool));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<ManifestList> lists,
+ ManifestList::Create(fs, format, "null", paths,
cache, pool));
+ ASSERT_OK_AND_ASSIGN(WrittenFile base_list, lists->Write({base_meta}));
+ ASSERT_OK_AND_ASSIGN(WrittenFile delta_list,
lists->Write({delta_meta}));
+ Snapshot snapshot(1, 0, base_list.first, base_list.second,
delta_list.first,
+ delta_list.second, std::nullopt, std::nullopt,
std::nullopt, "test", 0,
+ Snapshot::CommitKind::Append(), 0, 30, 0,
std::nullopt, std::nullopt,
+ std::nullopt, std::nullopt, std::nullopt);
+ for (bool lazy_decode : {false, true}) {
+ SCOPED_TRACE(lazy_decode);
+ const int64_t previous_loads = cache ? cache->SupplierCallCount()
: 0;
+ if (cache) {
+ cache->InvalidateAll();
+ }
+ ASSERT_OK_AND_ASSIGN(
+ CoreOptions options,
+ CoreOptions::FromMap({{Options::DATA_EVOLUTION_ENABLED,
"true"},
+
{Options::SCAN_MANIFEST_ENTRY_LAZY_DECODE_ENABLED,
+ lazy_decode ? "true" : "false"}}));
+ options.WithCache(cache);
+ for (int32_t attempt = 0; attempt < 2; ++attempt) {
+ SCOPED_TRACE(attempt);
+ for (size_t i = 0; i < queries.size(); ++i) {
+ SCOPED_TRACE(i);
+ const auto& query = queries[i];
+
ASSERT_OK_AND_ASSIGN(std::unique_ptr<DataEvolutionFileStoreScan> scan,
+ DataEvolutionFileStoreScan::Create(
+ nullptr, nullptr, lists,
manifest, table_schema,
+ schema, filters, options,
executor, pool));
+ scan->WithSnapshot(snapshot);
+ if (query.ranges) {
+ ASSERT_OK_AND_ASSIGN(RowRangeIndex ranges,
+
RowRangeIndex::Create(query.ranges.value()));
+ scan->WithRowRangeIndex(ranges);
+ }
+ std::atomic<int32_t> filter_calls{0};
+ scan->WithLevelFilter([&](int32_t) {
+ ++filter_calls;
+ return true;
+ });
+
ASSERT_OK_AND_ASSIGN(std::shared_ptr<FileStoreScan::RawPlan> plan,
+ scan->CreatePlan());
+ ASSERT_EQ(query.expected, plan->Files());
+ // The regular filter sees only retained entries when
early pruning is on.
+ // This also verifies that both the Add and Delete of a
reach the merge.
+ ASSERT_EQ(lazy_decode ? query.retained_entries : 5,
filter_calls.load());
+ if (cache) {
+ // Two manifest lists and two manifests load once per
cache reset.
+ ASSERT_EQ(previous_loads + 4,
cache->SupplierCallCount());
+ }
+ }
+ }
+ }
+ }
+}
+
+TEST_F(RowRangeManifestFileTest, OrcCachePreservesManifestMetadata) {
+ auto pool = GetDefaultPool();
+ auto fs = std::make_shared<LocalFileSystem>();
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<FileFormat> format,
FileFormatFactory::Get("orc", {}));
+ auto schema = arrow::schema(arrow::FieldVector{});
+ const std::string path = GetDataDir() + "/orc/append_09.db/append_09";
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<FileStorePathFactory> paths,
+ FileStorePathFactory::Create(path, schema, {}, "",
"orc", "data-", true,
+ {}, std::nullopt, false,
pool));
+ ASSERT_OK_AND_ASSIGN(CoreOptions options,
+ CoreOptions::FromMap({{Options::READ_BATCH_SIZE,
"2"}}));
+ ASSERT_OK_AND_ASSIGN(
+ std::unique_ptr<ManifestFile> uncached,
+ ManifestFile::Create(fs, format, "zstd", paths, 1024, pool, options,
schema));
+ const std::string name = "manifest-3ea5ee21-d399-4f1c-a749-2fc63dbf0852-1";
+ std::vector<ManifestEntry> expected;
+ ASSERT_OK(uncached->Read(name, nullptr, std::nullopt, &expected));
+ ASSERT_EQ(5, expected.size());
+ ASSERT_EQ(1721643142456LL,
expected[0].File()->creation_time.GetMillisecond());
+ // 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_EQ(expected, without_cache);
+
+ auto cache = std::make_shared<CountingRoutingCache>(CacheKind::MANIFEST,
1024 * 1024);
+ options.WithCache(cache);
+ for (int32_t attempt = 0; attempt < 2; ++attempt) {
+ SCOPED_TRACE(attempt);
+ ASSERT_OK_AND_ASSIGN(
+ 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_EQ(expected, actual);
+ ASSERT_EQ(1, cache->SupplierCallCount(CacheKind::MANIFEST));
+ }
+}
+
+TEST_F(RowRangeManifestFileTest, EmptyManifestCacheReuse) {
+ auto dir = UniqueTestDirectory::Create();
+ ASSERT_TRUE(dir);
+ auto cache = std::make_shared<CountingRoutingCache>(CacheKind::MANIFEST,
1024 * 1024);
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<ManifestFile> manifest,
+ CreateManifest(dir->Str(), dir->GetFileSystem(),
true, cache));
+ using WrittenFile = std::pair<std::string, int64_t>;
+ ASSERT_OK_AND_ASSIGN(WrittenFile written,
manifest->WriteWithoutRolling({}));
+ 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_TRUE(entries.empty());
+ ASSERT_EQ(1, cache->SupplierCallCount(CacheKind::MANIFEST));
+ }
+}
+
+TEST_F(RowRangeManifestFileTest, AllReadPathsShareOneDecodedEntry) {
+ for (int32_t first_reader : {0, 1, 2}) {
+ SCOPED_TRACE(first_reader);
+ auto dir = UniqueTestDirectory::Create();
+ ASSERT_TRUE(dir);
+ auto cache =
std::make_shared<CountingRoutingCache>(CacheKind::MANIFEST, 1024 * 1024);
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<ManifestFile> manifest,
+ CreateManifest(dir->Str(), dir->GetFileSystem(),
true, cache));
+ ASSERT_OK_AND_ASSIGN(ManifestEntry a, Entry("a.parquet", 100, 10));
+ ASSERT_OK_AND_ASSIGN(ManifestEntry b, Entry("b.parquet", 110, 10));
+ ASSERT_OK_AND_ASSIGN(ManifestEntry c, Entry("c.parquet", 120, 10));
+ std::vector<ManifestEntry> source = {a, b, c};
+ using WrittenFile = std::pair<std::string, int64_t>;
+ ASSERT_OK_AND_ASSIGN(WrittenFile written,
manifest->WriteWithoutRolling(source));
+ ASSERT_OK_AND_ASSIGN(RowRangeIndex ranges,
RowRangeIndex::Create({Range(110, 110)}));
+ auto read = [&](int32_t mode) {
+ std::vector<ManifestEntry> actual;
+ if (mode == 0) {
+ ASSERT_OK(manifest->Read(written.first, nullptr,
written.second, &actual));
+ ASSERT_EQ(source, actual);
+ } else if (mode == 1) {
+ ASSERT_OK(manifest->ReadBucketEntries(written.first, 0,
std::nullopt,
+ written.second,
&actual));
+ ASSERT_EQ(source, actual);
+ } else {
+ ASSERT_OK(manifest->ReadRowRangeEntries(written.first, ranges,
nullptr,
+ written.second,
&actual));
+ ASSERT_EQ(std::vector<ManifestEntry>{b}, actual);
+ }
+ };
+ read(first_reader);
+ ASSERT_EQ(1, cache->Size());
+ ASSERT_EQ(1, cache->SupplierCallCount());
+ auto key = CacheKey::ForKind(
+ PathUtil::JoinPath(FileStorePathFactory::ManifestPath(dir->Str()),
written.first),
+ /*position=*/0, /*length=*/-1, CacheKind::MANIFEST);
+ ASSERT_OK_AND_ASSIGN(
+ std::shared_ptr<CacheValue> cached,
+ cache->Get(
+ key, [](const std::shared_ptr<CacheKey>&) ->
Result<std::shared_ptr<CacheValue>> {
+ return Status::Invalid("manifest is missing under the
existing cache key");
+ }));
+ ASSERT_TRUE(cached);
+ // All later paths must work entirely from the same decoded cache
entry.
+ manifest->DeleteQuietly(written.first);
+ for (int32_t mode : {0, 1, 2}) {
+ read(mode);
+ }
+ ASSERT_EQ(1, cache->Size());
+ ASSERT_EQ(1, cache->SupplierCallCount());
+ // Existing callers can still invalidate a manifest by its whole-file
key.
+ cache->Invalidate(key);
+ ASSERT_EQ(0, cache->Size());
+ }
+}
+
+TEST_F(RowRangeManifestFileTest,
ColdFilterFailureDoesNotPoisonCacheOrRepeatCallbacks) {
+ auto dir = UniqueTestDirectory::Create();
+ ASSERT_TRUE(dir);
+ auto cache = std::make_shared<CountingRoutingCache>(CacheKind::MANIFEST,
1024 * 1024);
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<ManifestFile> manifest,
+ CreateManifest(dir->Str(), dir->GetFileSystem(),
true, cache));
+ ASSERT_OK_AND_ASSIGN(ManifestEntry a, Entry("a.parquet", 100, 10));
+ ASSERT_OK_AND_ASSIGN(ManifestEntry b, Entry("b.parquet", 110, 10));
+ using WrittenFile = std::pair<std::string, int64_t>;
+ ASSERT_OK_AND_ASSIGN(WrittenFile written,
manifest->WriteWithoutRolling({a, b}));
+ int32_t calls = 0;
+ auto filter = [&](const ManifestEntry&) -> Result<bool> {
+ ++calls;
+ return Status::IOError("consumer failed");
+ };
+ std::vector<ManifestEntry> entries;
+ ASSERT_NOK_WITH_MSG(manifest->Read(written.first, filter, written.second,
&entries),
+ "consumer failed");
+ ASSERT_EQ(1, calls);
+ ASSERT_TRUE(entries.empty());
+ manifest->DeleteQuietly(written.first);
+ ASSERT_OK(manifest->Read(written.first, nullptr, written.second,
&entries));
+ ASSERT_EQ(std::vector<ManifestEntry>({a, b}), entries);
+ ASSERT_EQ(1, cache->SupplierCallCount());
+}
+
+TEST_F(RowRangeManifestFileTest, CachedBufferRetainsAllocatorUntilEviction) {
+ auto dir = UniqueTestDirectory::Create();
+ ASSERT_TRUE(dir);
+ auto cache = std::make_shared<CountingRoutingCache>(CacheKind::MANIFEST,
1024 * 1024);
+ std::weak_ptr<MemoryPool> weak_pool;
+ ASSERT_OK_AND_ASSIGN(ManifestEntry entry, Entry("a.parquet", 100, 10));
+ ASSERT_OK_AND_ASSIGN(ManifestEntry next, Entry("b.parquet", 110, 10));
+ ASSERT_OK_AND_ASSIGN(ManifestEntry last, Entry("c.parquet", 120, 10));
+ const std::vector<ManifestEntry> expected = {entry, next, last};
+ using WrittenFile = std::pair<std::string, int64_t>;
+ WrittenFile written;
+ {
+ std::shared_ptr<MemoryPool> pool = GetMemoryPool();
+ weak_pool = pool;
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<ManifestFile> manifest,
+ CreateManifest(dir->Str(), dir->GetFileSystem(),
true, cache, pool));
+ ASSERT_OK_AND_ASSIGN(written, manifest->WriteWithoutRolling(expected));
+ std::vector<ManifestEntry> entries;
+ ASSERT_OK(manifest->Read(written.first, nullptr, written.second,
&entries));
+ }
+ ASSERT_FALSE(weak_pool.expired());
+ std::vector<ManifestEntry> entries;
+ {
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<ManifestFile> reader,
+ CreateManifest(dir->Str(), dir->GetFileSystem(),
true, cache));
+ reader->DeleteQuietly(written.first);
+ auto filter = [&](const ManifestEntry&) -> Result<bool> {
+ cache->InvalidateAll();
+ EXPECT_FALSE(weak_pool.expired());
+ return true;
+ };
+ // Eviction during the first batch must not invalidate later IPC
batches.
+ ASSERT_OK(reader->Read(written.first, filter, written.second,
&entries));
+ }
+ ASSERT_TRUE(weak_pool.expired());
+ ASSERT_EQ(expected, entries);
+}
+
+TEST_F(RowRangeManifestFileTest, CacheChargesRetainedArrowCapacity) {
+ auto dir = UniqueTestDirectory::Create();
+ ASSERT_TRUE(dir);
+ auto fs = dir->GetFileSystem();
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<ManifestFile> writer,
+ CreateManifest(dir->Str(), fs, false));
+ // Exercise growth beyond the initial IPC buffer allocation.
+ ASSERT_OK_AND_ASSIGN(ManifestEntry entry, Entry(std::string(5000, 'a') +
".parquet", 100, 10));
+ using WrittenFile = std::pair<std::string, int64_t>;
+ ASSERT_OK_AND_ASSIGN(WrittenFile written,
writer->WriteWithoutRolling({entry}));
+ std::shared_ptr<MemoryPool> pool = GetMemoryPool();
+ auto cache = std::make_shared<LruCache>(1024 * 1024);
+ {
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<ManifestFile> reader,
+ CreateManifest(dir->Str(), fs, true, cache,
pool));
+ std::vector<ManifestEntry> entries;
+ ASSERT_OK(reader->Read(written.first, nullptr, written.second,
&entries));
+ ASSERT_EQ(std::vector<ManifestEntry>{entry}, entries);
+ }
+ // Only the cached IPC allocation remains in this reader's pool.
+ ASSERT_EQ(1, cache->Size());
+ ASSERT_EQ(pool->CurrentUsage(), cache->GetCurrentWeight());
+ auto key = CacheKey::ForKind(
+ PathUtil::JoinPath(FileStorePathFactory::ManifestPath(dir->Str()),
written.first),
+ /*position=*/0, /*length=*/-1, CacheKind::MANIFEST);
+ ASSERT_OK_AND_ASSIGN(
+ std::shared_ptr<CacheValue> cached,
+ cache->Get(key,
+ [](const std::shared_ptr<CacheKey>&) ->
Result<std::shared_ptr<CacheValue>> {
+ return Status::Invalid("manifest was not cached");
+ }));
+ ASSERT_TRUE(cached);
+ const int64_t logical_size = cached->GetSegment().Size();
+ ASSERT_GT(cached->GetMemoryUsage(), logical_size);
+ ASSERT_EQ(pool->CurrentUsage(), cached->GetMemoryUsage());
+ cached.reset();
+ cache->InvalidateAll();
+ ASSERT_EQ(0, pool->CurrentUsage());
+
+ // Admission must reject the allocation even though the valid IPC bytes
would fit.
+ auto small_cache = std::make_shared<LruCache>(logical_size);
+ {
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<ManifestFile> reader,
+ CreateManifest(dir->Str(), fs, true, small_cache,
pool));
+ std::vector<ManifestEntry> entries;
+ ASSERT_OK(reader->Read(written.first, nullptr, written.second,
&entries));
+ ASSERT_EQ(std::vector<ManifestEntry>{entry}, entries);
+ ASSERT_EQ(0, small_cache->Size());
+ ASSERT_EQ(0, small_cache->GetCurrentWeight());
+ }
+ ASSERT_EQ(0, pool->CurrentUsage());
+}
+
+TEST_F(RowRangeManifestFileTest,
CacheAdmissionFailureReusesAlreadyDecodedBatches) {
+ class FailingCache : public CountingRoutingCache {
+ public:
+ FailingCache() : CountingRoutingCache(CacheKind::MANIFEST, 1024 *
1024) {}
+
+ Result<std::shared_ptr<CacheValue>> Get(
+ const std::shared_ptr<CacheKey>& key,
+ std::function<Result<std::shared_ptr<CacheValue>>(const
std::shared_ptr<CacheKey>&)>
+ supplier) override {
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<CacheValue> value,
supplier(key));
+ after_load();
+ return Status::IOError("cache admission failed");
+ }
+
+ std::function<void()> after_load;
+ };
+ auto dir = UniqueTestDirectory::Create();
+ ASSERT_TRUE(dir);
+ auto cache = std::make_shared<FailingCache>();
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<ManifestFile> manifest,
+ CreateManifest(dir->Str(), dir->GetFileSystem(),
true, cache));
+ ASSERT_OK_AND_ASSIGN(ManifestEntry entry, Entry("a.parquet", 100, 10));
+ using WrittenFile = std::pair<std::string, int64_t>;
+ ASSERT_OK_AND_ASSIGN(WrittenFile written,
manifest->WriteWithoutRolling({entry}));
+ // Reopening the source after an optional cache failure would now fail.
+ cache->after_load = [&]() { manifest->DeleteQuietly(written.first); };
+ int32_t calls = 0;
+ auto filter = [&](const ManifestEntry&) -> Result<bool> {
+ ++calls;
+ return true;
+ };
+ std::vector<ManifestEntry> entries;
+ ASSERT_OK(manifest->Read(written.first, filter, written.second, &entries));
+ ASSERT_EQ(1, calls);
+ ASSERT_EQ(std::vector<ManifestEntry>{entry}, entries);
+}
+
+TEST_F(RowRangeManifestFileTest, BoundariesUnknownRangesAndDeleteMerging) {
+ 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));
+ ASSERT_OK_AND_ASSIGN(ManifestEntry a, Entry("a.parquet", 100, 10));
+ ASSERT_OK_AND_ASSIGN(ManifestEntry b, Entry("b.parquet", 110, 10));
+ ASSERT_OK_AND_ASSIGN(ManifestEntry unknown, Entry("unknown.parquet",
std::nullopt, 10));
+ ManifestEntry deleted(FileKind::Delete(), a.Partition(), a.Bucket(),
a.TotalBuckets(),
+ a.File());
+ std::vector<ManifestEntry> source = {a, b, deleted, unknown};
+ using WrittenFile = std::pair<std::string, int64_t>;
+ ASSERT_OK_AND_ASSIGN(WrittenFile written,
manifest->WriteWithoutRolling(source));
+ const std::vector<std::vector<Range>> queries = {{Range(100, 100)},
+ {Range(109, 109)},
+ {Range(110, 110)},
+ {Range(119, 119)},
+ {Range(99, 99)},
+ {Range(120, 120)},
+ {Range(109, 110)},
+ {Range(100, 100),
Range(119, 119)},
+ {}};
+ for (const auto& query : queries) {
+ ASSERT_OK_AND_ASSIGN(RowRangeIndex ranges,
RowRangeIndex::Create(query));
+ auto filter = [&ranges](const ManifestEntry& entry) ->
Result<bool> {
+ const auto& meta = entry.File();
+ return !meta->first_row_id ||
+ ranges.Intersects(*meta->first_row_id,
+ *meta->first_row_id + meta->row_count
- 1);
+ };
+ 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_EQ(ordinary, selected);
+ std::vector<ManifestEntry> ordinary_live;
+ std::vector<ManifestEntry> selected_live;
+ ASSERT_OK(FileStoreScan::MergeLiveEntries(ordinary,
&ordinary_live));
+ ASSERT_OK(FileStoreScan::MergeLiveEntries(selected,
&selected_live));
+ ASSERT_EQ(ordinary_live, selected_live);
+ for (const auto& entry : selected_live) {
+ ASSERT_NE("a.parquet", entry.File()->file_name);
+ }
+ }
+ ASSERT_OK_AND_ASSIGN(RowRangeIndex ranges,
RowRangeIndex::Create({Range(100, 100)}));
+ std::vector<ManifestEntry> entries;
+ ASSERT_NOK_WITH_MSG(manifest->ReadRowRangeEntries(
+ written.first, ranges,
+ [](const ManifestEntry&) -> Result<bool> {
+ return Status::IOError("filter error");
+ },
+ written.second, &entries),
+ "filter error");
+ ASSERT_NOK(manifest->ReadRowRangeEntries("missing-manifest", ranges,
nullptr, std::nullopt,
+ &entries));
+ }
+}
+
+TEST_F(RowRangeManifestFileTest, UncertainAndOverflowingRangesAreRetained) {
+ auto dir = UniqueTestDirectory::Create();
+ ASSERT_TRUE(dir);
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<ManifestFile> manifest,
+ CreateManifest(dir->Str(), dir->GetFileSystem(),
false));
+ std::vector<ManifestEntry> source;
+ constexpr int64_t kMax = std::numeric_limits<int64_t>::max();
+ for (const auto& range : std::vector<std::pair<int64_t, int64_t>>{
+ {kMax, 2}, {kMax, 1}, {0, 0}, {-1, 10}, {0, -1}}) {
+ ASSERT_OK_AND_ASSIGN(ManifestEntry entry,
+ Entry(fmt::format("file-{}-{}.parquet",
range.first, range.second),
+ range.first, range.second));
+ source.push_back(entry);
+ }
+ using WrittenFile = std::pair<std::string, int64_t>;
+ 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));
+ source.erase(source.begin() + 1);
+ ASSERT_EQ(source, actual);
+}
+
+TEST_F(RowRangeManifestFileTest, SchemaEvolutionAndVersionValidation) {
+ constexpr int32_t kVersionedFileFieldIndex =
ManifestEntry::kFileFieldIndex + 1;
+ auto dir = UniqueTestDirectory::Create();
+ ASSERT_TRUE(dir);
+ auto pool = GetDefaultPool();
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<ManifestFile> manifest,
+ CreateManifest(dir->Str(), dir->GetFileSystem(),
true));
+ ASSERT_OK_AND_ASSIGN(ManifestEntry entry, Entry("a.parquet", 100, 10));
+ ManifestEntrySerializer serializer(pool);
+ ASSERT_OK_AND_ASSIGN(BinaryRow row, serializer.ToRow(entry));
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<MetaToArrowArrayConverter> converter,
+
MetaToArrowArrayConverter::Create(serializer.GetDataType(), pool));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::Array> array,
converter->NextBatch({row}));
+ auto batch = checked_pointer_cast<arrow::StructArray>(array);
+ ASSERT_OK_AND_ASSIGN(RowRangeIndex ranges,
RowRangeIndex::Create({Range(1000, 1000)}));
+ for (int32_t mode : {0, 1, 2, 3}) {
+ SCOPED_TRACE(mode);
+ auto fields = serializer.GetDataType()->fields();
+ auto columns = batch->fields();
+ auto file =
checked_pointer_cast<arrow::StructArray>(columns[kVersionedFileFieldIndex]);
+ auto file_fields =
checked_pointer_cast<arrow::StructType>(file->type())->fields();
+ auto file_columns = file->fields();
+ if (mode == 0) {
+ file_fields.erase(file_fields.begin() +
DataFileMeta::kFirstRowIdFieldIndex);
+ file_columns.erase(file_columns.begin() +
DataFileMeta::kFirstRowIdFieldIndex);
+ } else if (mode == 1) {
+ std::swap(file_fields[DataFileMeta::kRowCountFieldIndex],
+ file_fields[DataFileMeta::kFirstRowIdFieldIndex]);
+ std::swap(file_columns[DataFileMeta::kRowCountFieldIndex],
+ file_columns[DataFileMeta::kFirstRowIdFieldIndex]);
+ } else {
+ arrow::Int32Builder versions;
+ ASSERT_TRUE(versions.Append(mode == 2 ? 1 : 999).ok());
+ ASSERT_TRUE(versions.Finish(&columns[0]).ok());
+ }
+ auto updated_file = arrow::StructArray::Make(file_columns,
file_fields);
+ ASSERT_TRUE(updated_file.ok()) << updated_file.status().ToString();
+ columns[kVersionedFileFieldIndex] = updated_file.ValueOrDie();
+ fields[kVersionedFileFieldIndex] =
+
fields[kVersionedFileFieldIndex]->WithType(arrow::struct_(file_fields));
+ std::swap(fields[1], fields[kVersionedFileFieldIndex]);
+ std::swap(columns[1], columns[kVersionedFileFieldIndex]);
+ auto evolved_result = arrow::StructArray::Make(columns, fields);
+ ASSERT_TRUE(evolved_result.ok()) << evolved_result.status().ToString();
+ std::shared_ptr<arrow::StructArray> evolved =
evolved_result.ValueOrDie();
+ const std::string name = fmt::format("manifest-evolved-{}", mode);
+ ASSERT_OK(WriteArray(dir->Str(), name, evolved));
+ for (int32_t attempt = 0; attempt < 2; ++attempt) {
+ SCOPED_TRACE(attempt);
+ std::vector<ManifestEntry> actual;
+ if (mode >= 2) {
+ ASSERT_NOK_WITH_MSG(
+ manifest->ReadRowRangeEntries(name, ranges, nullptr,
std::nullopt, &actual),
+ (mode == 2 ? "not compatible" : "Unsupported version:
999"));
+ } else {
+ ASSERT_OK(
+ manifest->ReadRowRangeEntries(name, ranges, nullptr,
std::nullopt, &actual));
+ ASSERT_EQ(mode == 0 ? 1 : 0, actual.size());
+ if (mode == 0) {
+ ASSERT_FALSE(actual[0].File()->first_row_id.has_value());
+ }
+ }
+ if (attempt == 0) {
+ // The second read must preserve schema evolution and
validation from IPC alone.
+ manifest->DeleteQuietly(name);
+ }
+ }
+ }
+}
+
+} // namespace paimon::test
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 908f9bbb..1bb5bda0 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
@@ -235,7 +235,6 @@ TEST_F(AppendBucketPruningTest,
ReadsSingleBucketManifestOnce) {
{}, std::nullopt, false, pool_));
ASSERT_OK(fs->Mkdirs(FileStorePathFactory::ManifestPath(dir->Str())));
ASSERT_OK_AND_ASSIGN(CoreOptions options, CoreOptions::FromMap({}));
- options.WithCache(std::make_shared<LruCache>(16 * 1024 * 1024));
ASSERT_OK_AND_ASSIGN(std::shared_ptr<ManifestFile> manifest,
ManifestFile::Create(fs, format, "null", paths, 1024
* 1024, pool_,
options, arrow::schema({})));
@@ -252,24 +251,34 @@ TEST_F(AppendBucketPruningTest,
ReadsSingleBucketManifestOnce) {
ASSERT_OK_AND_ASSIGN(BinaryRow row, serializer.ToRow(entries[0]));
ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::Array> data,
converter->NextBatch({row}));
auto reader_builder =
std::make_shared<CountingManifestReaderBuilder>(data);
- manifest->reader_builder_ = reader_builder;
options_[Options::SCAN_MANIFEST_ENTRY_LAZY_DECODE_ENABLED] = "true";
- for (bool inferred : {false, true}) {
- SCOPED_TRACE(inferred);
- ASSERT_OK_AND_ASSIGN(
- std::unique_ptr<AppendOnlyFileStoreScan> scan,
- CreateScan(KeyEquals(), inferred ? std::nullopt :
std::optional<int32_t>(bucket)));
- scan->manifest_file_ = manifest;
- for (bool known_bounds : {false, true}) {
- SCOPED_TRACE(known_bounds);
- auto meta = metas[0];
- meta.min_bucket_ = known_bounds ? std::optional<int32_t>(bucket) :
std::nullopt;
- meta.max_bucket_ = meta.min_bucket_;
- reader_builder->rows_read = 0;
- std::vector<ManifestEntry> actual;
- ASSERT_OK(scan->ReadAndMergeBucketFileEntries({meta}, bucket,
&actual));
- ASSERT_EQ(actual, entries);
- ASSERT_EQ(reader_builder->rows_read.load(), known_bounds ? 1 : 2);
+ for (bool cache_enabled : {false, true}) {
+ SCOPED_TRACE(cache_enabled);
+ options.WithCache(cache_enabled ? std::make_shared<LruCache>(16 * 1024
* 1024) : nullptr);
+ ASSERT_OK_AND_ASSIGN(manifest, ManifestFile::Create(fs, format,
"null", paths, 1024 * 1024,
+ pool_, options,
arrow::schema({})));
+ manifest->reader_builder_ = reader_builder;
+ bool cache_warm = false;
+ for (bool inferred : {false, true}) {
+ SCOPED_TRACE(inferred);
+ ASSERT_OK_AND_ASSIGN(
+ std::unique_ptr<AppendOnlyFileStoreScan> scan,
+ CreateScan(KeyEquals(), inferred ? std::nullopt :
std::optional<int32_t>(bucket)));
+ scan->manifest_file_ = manifest;
+ for (bool known_bounds : {false, true}) {
+ SCOPED_TRACE(known_bounds);
+ auto meta = metas[0];
+ meta.min_bucket_ = known_bounds ?
std::optional<int32_t>(bucket) : std::nullopt;
+ meta.max_bucket_ = meta.min_bucket_;
+ reader_builder->rows_read = 0;
+ std::vector<ManifestEntry> actual;
+ ASSERT_OK(scan->ReadAndMergeBucketFileEntries({meta}, bucket,
&actual));
+ ASSERT_EQ(actual, entries);
+ const int64_t expected_rows =
+ cache_enabled ? (cache_warm ? 0 : 1) : (known_bounds ? 1 :
2);
+ ASSERT_EQ(reader_builder->rows_read.load(), expected_rows);
+ cache_warm = cache_enabled;
+ }
}
}
}
diff --git a/src/paimon/core/operation/file_store_scan.cpp
b/src/paimon/core/operation/file_store_scan.cpp
index fa22ac56..f1c9f786 100644
--- a/src/paimon/core/operation/file_store_scan.cpp
+++ b/src/paimon/core/operation/file_store_scan.cpp
@@ -613,6 +613,15 @@ bool FileStoreScan::FilterManifestByRowRanges(const
ManifestFileMeta& manifest)
Status FileStoreScan::ReadManifestFileMeta(const ManifestFileMeta& manifest,
std::vector<ManifestEntry>*
entries) const {
+ 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(),
diff --git a/src/paimon/core/utils/objects_file.h
b/src/paimon/core/utils/objects_file.h
index e65ee75a..2a99e98d 100644
--- a/src/paimon/core/utils/objects_file.h
+++ b/src/paimon/core/utils/objects_file.h
@@ -19,6 +19,7 @@
#pragma once
#include <functional>
+#include <limits>
#include <memory>
#include <optional>
#include <string>
@@ -27,6 +28,8 @@
#include "arrow/c/bridge.h"
#include "arrow/c/helpers.h"
+#include "arrow/io/memory.h"
+#include "arrow/ipc/api.h"
#include "fmt/format.h"
#include "paimon/cache/cache.h"
#include "paimon/common/data/columnar/columnar_row.h"
@@ -44,8 +47,6 @@
#include "paimon/format/reader_builder.h"
#include "paimon/format/writer_builder.h"
#include "paimon/fs/file_system.h"
-#include "paimon/io/byte_array_input_stream.h"
-#include "paimon/memory/bytes.h"
#include "paimon/record_batch.h"
namespace paimon {
@@ -92,8 +93,7 @@ class ObjectsFile {
return Status::OK();
}
- // Optional preparation may wrap the reader and reread the file, using
either cached bytes
- // or the underlying file stream.
+ // Cached batches are query-independent. Optional preparation only applies
to uncached reads.
Status ReadArrowBatches(
const std::string& file_name, std::optional<int64_t> file_size,
const std::function<Status(const
std::shared_ptr<arrow::StructArray>&)>& consumer,
@@ -113,8 +113,15 @@ class ObjectsFile {
std::string compression_;
std::shared_ptr<Cache> cache_;
- Result<MemorySegment> ReadFileSegment(const std::string& file_path,
- const std::optional<int64_t>&
file_size) const;
+ Result<std::shared_ptr<CacheValue>> SerializeArrowBatches(
+ const std::string& file_name, std::optional<int64_t> file_size,
+ std::vector<std::shared_ptr<arrow::StructArray>>* batches,
+ std::optional<Status>* read_status) const;
+
+ Status ReadUncachedArrowBatches(
+ const std::string& file_name, std::optional<int64_t> file_size,
+ const std::function<Status(const
std::shared_ptr<arrow::StructArray>&)>& consumer,
+ const std::function<Status(std::unique_ptr<FileBatchReader>*)>&
prepare_reader) const;
/// Opens the file for reading, handing over the length when the caller
already has it.
Result<std::unique_ptr<InputStream>> OpenForRead(const std::string&
file_path,
@@ -183,37 +190,13 @@ Status ObjectsFile<T>::Read(const std::string& file_name,
}
template <typename T>
-Status ObjectsFile<T>::ReadArrowBatches(
+Status ObjectsFile<T>::ReadUncachedArrowBatches(
const std::string& file_name, std::optional<int64_t> file_size,
const std::function<Status(const std::shared_ptr<arrow::StructArray>&)>&
consumer,
const std::function<Status(std::unique_ptr<FileBatchReader>*)>&
prepare_reader) const {
std::string file_path = path_factory_->ToPath(file_name);
- std::shared_ptr<InputStream> file_input_stream;
- std::shared_ptr<Bytes> cached_bytes;
- if (cache_) {
- // Use a whole-file key so cache hits do not need a metadata lookup
just to discover file
- // length.
- auto cache_key =
- CacheKey::ForKind(file_path, /*position=*/0, /*length=*/-1,
CacheKind::MANIFEST);
- auto supplier =
- [this, &file_path,
- &file_size](const std::shared_ptr<CacheKey>&) ->
Result<std::shared_ptr<CacheValue>> {
- PAIMON_ASSIGN_OR_RAISE(MemorySegment segment,
ReadFileSegment(file_path, file_size));
- return std::make_shared<CacheValue>(segment, CacheCallback());
- };
- Result<std::shared_ptr<CacheValue>> cache_result =
cache_->Get(cache_key, supplier);
- if (cache_result.ok() && cache_result.value() &&
- cache_result.value()->GetSegment().Data() != nullptr) {
- cached_bytes =
cache_result.value()->GetSegment().GetOrCreateHeapMemory(pool_.get());
- file_input_stream =
- std::make_shared<ByteArrayInputStream>(cached_bytes->data(),
cached_bytes->size());
- }
- }
- if (!file_input_stream) {
- PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<InputStream>
unique_file_input_stream,
- OpenForRead(file_path, file_size));
- file_input_stream =
std::shared_ptr<InputStream>(std::move(unique_file_input_stream));
- }
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<InputStream> file_input_stream,
+ OpenForRead(file_path, file_size));
PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<FileBatchReader> batch_reader,
reader_builder_->Build(file_input_stream));
@@ -256,22 +239,126 @@ Result<std::unique_ptr<InputStream>>
ObjectsFile<T>::OpenForRead(
}
template <typename T>
-Result<MemorySegment> ObjectsFile<T>::ReadFileSegment(
- const std::string& file_path, const std::optional<int64_t>& file_size)
const {
- PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<InputStream> input_stream,
- OpenForRead(file_path, file_size));
- PAIMON_ASSIGN_OR_RAISE(int64_t input_length, input_stream->Length());
+Result<std::shared_ptr<CacheValue>> ObjectsFile<T>::SerializeArrowBatches(
+ const std::string& file_name, std::optional<int64_t> file_size,
+ std::vector<std::shared_ptr<arrow::StructArray>>* batches,
+ std::optional<Status>* read_status) const {
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
+ std::shared_ptr<arrow::io::BufferOutputStream> output,
+ arrow::io::BufferOutputStream::Create(4096, arrow_pool_.get()));
+ auto write_options = arrow::ipc::IpcWriteOptions::Defaults();
+ write_options.memory_pool = arrow_pool_.get();
+ write_options.use_threads = false;
+ std::shared_ptr<arrow::ipc::RecordBatchWriter> writer;
+ Status cache_status;
+ *read_status = ReadUncachedArrowBatches(
+ file_name, file_size,
+ [&](const std::shared_ptr<arrow::StructArray>& batch) -> Status {
+ // The loading caller consumes these original batches, without an
IPC round trip.
+ batches->push_back(batch);
+ if (!cache_status.ok()) {
+ return Status::OK();
+ }
+ auto write_batch = [&]() -> Status {
+ // Preserve physical types, including ORC nanosecond
timestamps.
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
+ std::shared_ptr<arrow::RecordBatch> record_batch,
+ arrow::RecordBatch::FromStructArray(batch,
arrow_pool_.get()));
+ if (!writer) {
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
+ writer, arrow::ipc::MakeStreamWriter(output,
record_batch->schema(),
+ write_options));
+ }
+
PAIMON_RETURN_NOT_OK_FROM_ARROW(writer->WriteRecordBatch(*record_batch));
+ return Status::OK();
+ };
+ // Optional serialization failures must not interrupt the source
read.
+ cache_status = write_batch();
+ return Status::OK();
+ },
+ /*prepare_reader=*/nullptr);
+ PAIMON_RETURN_NOT_OK(read_status->value());
+ PAIMON_RETURN_NOT_OK(cache_status);
+ if (!writer) {
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
+ writer,
+ arrow::ipc::MakeStreamWriter(
+ output, arrow::schema(serializer_->GetDataType()->fields()),
write_options));
+ }
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(writer->Close());
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Buffer> buffer,
output->Finish());
+ if (buffer->size() > std::numeric_limits<int32_t>::max()) {
+ return Status::Invalid("Manifest Arrow cache entry exceeds the memory
segment limit");
+ }
+ // Retain the IPC buffer and its allocator without copying the complete
stream into Bytes.
+ // The aliasing CacheValue keeps both alive, including after eviction with
active readers.
+ struct OwnedBuffer {
+ OwnedBuffer(const std::shared_ptr<arrow::MemoryPool>& owner,
+ const std::shared_ptr<arrow::Buffer>& data)
+ : pool(owner),
+ buffer(data),
+ value(MemorySegment::WrapView(reinterpret_cast<const
char*>(data->data()),
+
static_cast<int32_t>(data->size())),
+ CacheCallback(), data->capacity()) {}
+ std::shared_ptr<arrow::MemoryPool> pool;
+ std::shared_ptr<arrow::Buffer> buffer;
+ CacheValue value;
+ };
+ auto owner = std::make_shared<OwnedBuffer>(arrow_pool_, buffer);
+ auto* value = &owner->value;
+ return std::shared_ptr<CacheValue>(std::move(owner), value);
+}
- PAIMON_RETURN_NOT_OK(input_stream->Seek(0, FS_SEEK_SET));
- auto bytes = std::make_shared<Bytes>(input_length, pool_.get());
- PAIMON_ASSIGN_OR_RAISE(int64_t actual_read_size,
- input_stream->Read(bytes->data(), input_length));
- if (actual_read_size != input_length) {
- return Status::IOError(fmt::format(
- "Unexpected EOF while reading manifest file {}, expected {} bytes,
got {} bytes",
- file_path, input_length, actual_read_size));
+template <typename T>
+Status ObjectsFile<T>::ReadArrowBatches(
+ const std::string& file_name, std::optional<int64_t> file_size,
+ const std::function<Status(const std::shared_ptr<arrow::StructArray>&)>&
consumer,
+ const std::function<Status(std::unique_ptr<FileBatchReader>*)>&
prepare_reader) const {
+ if (!cache_) {
+ return ReadUncachedArrowBatches(file_name, file_size, consumer,
prepare_reader);
+ }
+ const std::string path = path_factory_->ToPath(file_name);
+ auto key = CacheKey::ForKind(path, /*position=*/0, /*length=*/-1,
CacheKind::MANIFEST);
+ std::optional<Status> read_status;
+ std::vector<std::shared_ptr<arrow::StructArray>> batches;
+ auto cached = cache_->Get(key, [&](const std::shared_ptr<CacheKey>&) {
+ return SerializeArrowBatches(file_name, file_size, &batches,
&read_status);
+ });
+ const bool usable = cached.ok() && cached.value() &&
cached.value()->GetSegment().Data();
+ if (read_status) {
+ // Publish the query-independent cache before invoking user filters. A
filter error
+ // does not invalidate the cache or affect other readers of this file.
+ PAIMON_RETURN_NOT_OK(read_status.value());
+ for (const auto& batch : batches) {
+ PAIMON_RETURN_NOT_OK(consumer(batch));
+ }
+ return Status::OK();
}
- return MemorySegment::Wrap(bytes);
+ if (!usable) {
+ return ReadUncachedArrowBatches(file_name, file_size, consumer,
prepare_reader);
+ }
+ const auto& segment = cached.value()->GetSegment();
+ auto buffer = std::make_shared<arrow::Buffer>(reinterpret_cast<const
uint8_t*>(segment.Data()),
+ segment.Size());
+ auto input = std::make_shared<arrow::io::BufferReader>(buffer);
+ auto read_options = arrow::ipc::IpcReadOptions::Defaults();
+ read_options.memory_pool = arrow_pool_.get();
+ read_options.use_threads = false;
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
+ std::shared_ptr<arrow::ipc::RecordBatchStreamReader> reader,
+ arrow::ipc::RecordBatchStreamReader::Open(input, read_options));
+ while (true) {
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::RecordBatch>
batch,
+ reader->Next());
+ if (!batch) {
+ break;
+ }
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::StructArray>
array,
+ batch->ToStructArray());
+ // Cached batches were already aligned by ManifestMetaReader before
serialization.
+ PAIMON_RETURN_NOT_OK(consumer(array));
+ }
+ return Status::OK();
}
template <typename T>
