This is an automated email from the ASF dual-hosted git repository.
SteNicholas 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 aa60634a perf(scan): lazily decode manifest bucket entries (#212)
aa60634a is described below
commit aa60634a318420c819a5357ea13ef3991fb0685f
Author: gripleaf <[email protected]>
AuthorDate: Thu Aug 20 22:48:44 2026 +0800
perf(scan): lazily decode manifest bucket entries (#212)
---
include/paimon/defs.h | 4 +
src/paimon/common/defs.cpp | 2 +
src/paimon/core/core_options.cpp | 7 ++
src/paimon/core/core_options.h | 1 +
src/paimon/core/core_options_test.cpp | 3 +
.../core/manifest/manifest_entry_serializer.cpp | 27 +++--
.../core/manifest/manifest_entry_serializer.h | 6 ++
.../manifest/manifest_entry_serializer_test.cpp | 10 ++
src/paimon/core/manifest/manifest_file.cpp | 19 ++++
src/paimon/core/manifest/manifest_file.h | 4 +
src/paimon/core/manifest/manifest_file_test.cpp | 120 +++++++++++++++++++--
.../operation/append_only_file_store_scan_test.cpp | 40 ++++++-
src/paimon/core/operation/file_store_scan.cpp | 27 +++++
src/paimon/core/utils/objects_file.h | 47 +++++---
14 files changed, 282 insertions(+), 35 deletions(-)
diff --git a/include/paimon/defs.h b/include/paimon/defs.h
index e944587f..338eda30 100644
--- a/include/paimon/defs.h
+++ b/include/paimon/defs.h
@@ -200,6 +200,10 @@ struct PAIMON_EXPORT Options {
/// cache. Default value is 0.
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.
+ static const char SCAN_MANIFEST_ENTRY_LAZY_DECODE_ENABLED[];
+
/// "read.batch-size" - Read batch size for any file format if it supports.
/// The default value is 1024.
static const char READ_BATCH_SIZE[];
diff --git a/src/paimon/common/defs.cpp b/src/paimon/common/defs.cpp
index ef35940e..8c5336e1 100644
--- a/src/paimon/common/defs.cpp
+++ b/src/paimon/common/defs.cpp
@@ -59,6 +59,8 @@ const char Options::SCAN_SNAPSHOT_ID[] = "scan.snapshot-id";
const char Options::SCAN_MODE[] = "scan.mode";
const char Options::SCAN_MANIFEST_ENTRY_CACHE_MAX_SNAPSHOTS[] =
"scan.manifest-entry-cache.max-snapshots";
+const char Options::SCAN_MANIFEST_ENTRY_LAZY_DECODE_ENABLED[] =
+ "scan.manifest-entry.lazy-decode.enabled";
const char Options::READ_BATCH_SIZE[] = "read.batch-size";
const char Options::WRITE_BATCH_SIZE[] = "write.batch-size";
const char Options::WRITE_BUFFER_SIZE[] = "write-buffer-size";
diff --git a/src/paimon/core/core_options.cpp b/src/paimon/core/core_options.cpp
index 1c2e164b..1e8203a5 100644
--- a/src/paimon/core/core_options.cpp
+++ b/src/paimon/core/core_options.cpp
@@ -426,6 +426,7 @@ struct CoreOptions::Impl {
int32_t manifest_merge_min_count = 30;
int32_t scan_manifest_entry_cache_max_snapshots = 0;
+ bool scan_manifest_entry_lazy_decode_enabled = true;
int32_t read_batch_size = 1024;
int32_t write_batch_size = 1024;
int32_t local_sort_max_num_file_handles = 128;
@@ -828,6 +829,8 @@ struct CoreOptions::Impl {
return Status::Invalid(fmt::format("{} must be non-negative",
Options::SCAN_MANIFEST_ENTRY_CACHE_MAX_SNAPSHOTS));
}
+
PAIMON_RETURN_NOT_OK(parser.Parse<bool>(Options::SCAN_MANIFEST_ENTRY_LAZY_DECODE_ENABLED,
+
&scan_manifest_entry_lazy_decode_enabled));
// Parse scan.fallback-branch - fallback branch when partition not
found
PAIMON_RETURN_NOT_OK(parser.Parse(Options::SCAN_FALLBACK_BRANCH,
&scan_fallback_branch));
// Parse branch - branch name, default "main"
@@ -1170,6 +1173,10 @@ int32_t
CoreOptions::GetScanManifestEntryCacheMaxSnapshots() const {
return impl_->scan_manifest_entry_cache_max_snapshots;
}
+bool CoreOptions::ScanManifestEntryLazyDecodeEnabled() const {
+ return impl_->scan_manifest_entry_lazy_decode_enabled;
+}
+
int64_t CoreOptions::GetManifestTargetFileSize() const {
return impl_->manifest_target_file_size;
}
diff --git a/src/paimon/core/core_options.h b/src/paimon/core/core_options.h
index 53ef4ad0..3bb17d6f 100644
--- a/src/paimon/core/core_options.h
+++ b/src/paimon/core/core_options.h
@@ -108,6 +108,7 @@ class PAIMON_EXPORT CoreOptions {
std::optional<int64_t> GetScanTimestampMillis() const;
int64_t GetRealtimeReadViewTtlMillis() const;
int32_t GetScanManifestEntryCacheMaxSnapshots() const;
+ bool ScanManifestEntryLazyDecodeEnabled() const;
int64_t GetManifestTargetFileSize() const;
std::shared_ptr<Cache> GetCache() const;
diff --git a/src/paimon/core/core_options_test.cpp
b/src/paimon/core/core_options_test.cpp
index 0054a5b5..f057206d 100644
--- a/src/paimon/core/core_options_test.cpp
+++ b/src/paimon/core/core_options_test.cpp
@@ -65,6 +65,7 @@ TEST(CoreOptionsTest, TestDefaultValue) {
ASSERT_EQ(30, core_options.GetManifestMergeMinCount());
ASSERT_FALSE(core_options.ManifestDeleteFileDropStats());
ASSERT_EQ(0, core_options.GetScanManifestEntryCacheMaxSnapshots());
+ ASSERT_TRUE(core_options.ScanManifestEntryLazyDecodeEnabled());
ASSERT_EQ(nullptr, core_options.GetCache());
ASSERT_EQ(128 * 1024 * 1024L, core_options.GetSourceSplitTargetSize());
ASSERT_EQ(4 * 1024 * 1024L, core_options.GetSourceSplitOpenFileCost());
@@ -218,6 +219,7 @@ TEST(CoreOptionsTest, TestFromMap) {
{Options::SCAN_SNAPSHOT_ID, "5"},
{Options::SCAN_MODE, "from-snapshot-full"},
{Options::SCAN_MANIFEST_ENTRY_CACHE_MAX_SNAPSHOTS, "7"},
+ {Options::SCAN_MANIFEST_ENTRY_LAZY_DECODE_ENABLED, "false"},
{Options::SNAPSHOT_NUM_RETAINED_MIN, "15"},
{Options::SNAPSHOT_NUM_RETAINED_MAX, "30"},
{Options::SNAPSHOT_EXPIRE_LIMIT, "20"},
@@ -355,6 +357,7 @@ TEST(CoreOptionsTest, TestFromMap) {
ASSERT_TRUE(core_options.CommitDiscardDuplicateFiles());
ASSERT_EQ(5, core_options.GetScanSnapshotId().value_or(-1));
ASSERT_EQ(7, core_options.GetScanManifestEntryCacheMaxSnapshots());
+ ASSERT_FALSE(core_options.ScanManifestEntryLazyDecodeEnabled());
ExpireConfig expire_config = core_options.GetExpireConfig();
ASSERT_EQ(15, expire_config.GetSnapshotRetainMin());
ASSERT_EQ(30, expire_config.GetSnapshotRetainMax());
diff --git a/src/paimon/core/manifest/manifest_entry_serializer.cpp
b/src/paimon/core/manifest/manifest_entry_serializer.cpp
index 053405b8..2389cd8d 100644
--- a/src/paimon/core/manifest/manifest_entry_serializer.cpp
+++ b/src/paimon/core/manifest/manifest_entry_serializer.cpp
@@ -31,17 +31,26 @@ namespace paimon {
class MemoryPool;
struct DataFileMeta;
+Status ManifestEntrySerializer::ValidateVersion(int32_t version) {
+ if (version == VERSION_2) {
+ return Status::OK();
+ }
+ if (version == VERSION_1) {
+ return Status::Invalid(
+ fmt::format("The current version {} is not compatible with the
version {}, "
+ "please recreate the table.",
+ VERSION_2, version));
+ }
+ return Status::Invalid(fmt::format("Unsupported version: {}", version));
+}
+
+int32_t ManifestEntrySerializer::GetBucket(const InternalRow& row) {
+ return row.GetInt(3);
+}
+
Result<ManifestEntry> ManifestEntrySerializer::ConvertFrom(int32_t version,
const InternalRow&
row) const {
- if (version != VERSION_2) {
- if (version == VERSION_1) {
- return Status::Invalid(
- fmt::format("The current version {} is not compatible with the
version {}, "
- "please recreate the table.",
- GetVersion(), version));
- }
- return Status::Invalid("Unsupported version", std::to_string(version));
- }
+ PAIMON_RETURN_NOT_OK(ValidateVersion(version));
auto kind = row.GetByte(0);
PAIMON_ASSIGN_OR_RAISE(FileKind file_kind, FileKind::FromByteValue(kind));
auto partition_bytes = row.GetBinary(1);
diff --git a/src/paimon/core/manifest/manifest_entry_serializer.h
b/src/paimon/core/manifest/manifest_entry_serializer.h
index 7438895f..4a71a1b6 100644
--- a/src/paimon/core/manifest/manifest_entry_serializer.h
+++ b/src/paimon/core/manifest/manifest_entry_serializer.h
@@ -50,6 +50,12 @@ class ManifestEntrySerializer : public
VersionedObjectSerializer<ManifestEntry>
return VERSION_2;
}
+ /// Validate the serialization version before reading fields that may vary
by version.
+ static Status ValidateVersion(int32_t version);
+
+ /// Get the bucket from a versioned manifest entry row without fully
deserializing it.
+ static int32_t GetBucket(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 2aa2db52..2d8cffc3 100644
--- a/src/paimon/core/manifest/manifest_entry_serializer_test.cpp
+++ b/src/paimon/core/manifest/manifest_entry_serializer_test.cpp
@@ -55,12 +55,22 @@ TEST_F(ManifestEntrySerializerTest, TestToFromRow) {
ManifestEntrySerializer serializer(pool);
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 result_entry, serializer.FromRow(row));
ASSERT_EQ(entry, result_entry);
ASSERT_EQ(entry.ToString(), result_entry.ToString());
}
}
+TEST_F(ManifestEntrySerializerTest, TestValidateVersion) {
+ ASSERT_OK(ManifestEntrySerializer::ValidateVersion(/*version=*/2));
+
ASSERT_NOK_WITH_MSG(ManifestEntrySerializer::ValidateVersion(/*version=*/1),
+ "The current version 2 is not compatible with the
version 1, please "
+ "recreate the table.");
+
ASSERT_NOK_WITH_MSG(ManifestEntrySerializer::ValidateVersion(/*version=*/3),
+ "Unsupported version: 3");
+}
+
TEST_F(ManifestEntrySerializerTest, TestNullableRecordCount) {
std::vector<ManifestEntry> empty_entries;
ASSERT_FALSE(ManifestEntry::NullableRecordCount(empty_entries).has_value());
diff --git a/src/paimon/core/manifest/manifest_file.cpp
b/src/paimon/core/manifest/manifest_file.cpp
index 22f2681f..1be49d0b 100644
--- a/src/paimon/core/manifest/manifest_file.cpp
+++ b/src/paimon/core/manifest/manifest_file.cpp
@@ -24,6 +24,7 @@
#include "arrow/c/abi.h"
#include "arrow/c/bridge.h"
+#include "paimon/common/data/columnar/columnar_row.h"
#include "paimon/common/utils/arrow/status_utils.h"
#include "paimon/core/io/rolling_file_writer.h"
#include "paimon/core/manifest/manifest_entry.h"
@@ -86,6 +87,24 @@ Result<std::unique_ptr<ManifestFile>> ManifestFile::Create(
manifest_file_factory, target_file_size, pool,
options, partition_type));
}
+Status ManifestFile::ReadBucketEntries(const std::string& file_name, int32_t
bucket,
+ std::vector<ManifestEntry>* entries)
const {
+ return ReadArrowBatches(
+ file_name,
+ [this, bucket, entries](const std::shared_ptr<arrow::StructArray>&
batch) -> Status {
+ for (int64_t i = 0; i < batch->length(); i++) {
+ ColumnarRow row(batch->fields(), pool_, i);
+
PAIMON_RETURN_NOT_OK(ManifestEntrySerializer::ValidateVersion(row.GetInt(0)));
+ if (ManifestEntrySerializer::GetBucket(row) != bucket) {
+ continue;
+ }
+ PAIMON_ASSIGN_OR_RAISE(ManifestEntry entry,
serializer_->FromRow(row));
+ entries->push_back(std::move(entry));
+ }
+ return Status::OK();
+ });
+}
+
Result<std::vector<ManifestFileMeta>> ManifestFile::Write(
const std::vector<ManifestEntry>& entries) {
if (entries.empty()) {
diff --git a/src/paimon/core/manifest/manifest_file.h
b/src/paimon/core/manifest/manifest_file.h
index d34764b5..0211e14d 100644
--- a/src/paimon/core/manifest/manifest_file.h
+++ b/src/paimon/core/manifest/manifest_file.h
@@ -62,6 +62,10 @@ class ManifestFile : public ObjectsFile<ManifestEntry> {
/// @note This method is atomic.
Result<std::vector<ManifestFileMeta>> Write(const
std::vector<ManifestEntry>& entries);
+ /// Read a manifest file and deserialize only entries for the specified
bucket.
+ Status ReadBucketEntries(const std::string& file_name, int32_t bucket,
+ std::vector<ManifestEntry>* entries) const;
+
private:
ManifestFile(const std::shared_ptr<FileSystem>& file_system,
const std::shared_ptr<ReaderBuilder>& reader_builder,
diff --git a/src/paimon/core/manifest/manifest_file_test.cpp
b/src/paimon/core/manifest/manifest_file_test.cpp
index 8f6e0b2e..a34f4152 100644
--- a/src/paimon/core/manifest/manifest_file_test.cpp
+++ b/src/paimon/core/manifest/manifest_file_test.cpp
@@ -23,7 +23,6 @@
#include <optional>
#include <string>
#include <utility>
-#include <variant>
#include "arrow/api.h"
#include "gtest/gtest.h"
@@ -100,10 +99,10 @@ class CountingFileSystem : public FileSystem {
class ManifestFileTest : public testing::Test {
public:
- std::vector<ManifestEntry> ReadManifestEntry(const std::string&
file_format_str,
- const std::string& root_path,
- const std::string& file_name,
- const
std::shared_ptr<MemoryPool>& pool) const {
+ std::vector<ManifestEntry> ReadManifestEntry(
+ const std::string& file_format_str, const std::string& root_path,
+ const std::string& file_name, const std::shared_ptr<MemoryPool>& pool,
+ const std::optional<int32_t>& bucket = std::nullopt) const {
std::shared_ptr<FileSystem> file_system =
std::make_shared<LocalFileSystem>();
EXPECT_OK_AND_ASSIGN(std::shared_ptr<FileFormat> file_format,
FileFormatFactory::Get(file_format_str, {}));
@@ -124,7 +123,12 @@ class ManifestFileTest : public testing::Test {
ManifestFile::Create(file_system, file_format, "zstd",
path_factory,
/*target_file_size=*/1024, pool, options,
unused_schema));
std::vector<ManifestEntry> manifest_entries;
- EXPECT_OK(manifest_file->Read(file_name, /*filter=*/nullptr,
&manifest_entries));
+ if (bucket) {
+ EXPECT_OK(
+ manifest_file->ReadBucketEntries(file_name, bucket.value(),
&manifest_entries));
+ } else {
+ EXPECT_OK(manifest_file->Read(file_name, /*filter=*/nullptr,
&manifest_entries));
+ }
return manifest_entries;
}
@@ -316,6 +320,104 @@ TEST_F(ManifestFileTest,
TestManifestCacheReusesCachedBytes) {
ASSERT_EQ(1, manifest_cache->Size());
}
+TEST_F(ManifestFileTest, TestReadBucketEntriesMaterializesOnlySelectedBucket) {
+ auto pool = GetDefaultPool();
+ auto counting_file_system = std::make_shared<CountingFileSystem>();
+ auto manifest_cache =
+ std::make_shared<CountingRoutingCache>(CacheKind::MANIFEST, 64 * 1024
* 1024);
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<FileFormat> file_format,
+ FileFormatFactory::Get("orc", {}));
+ std::string root_path = paimon::test::GetDataDir() +
"/orc/append_09.db/append_09";
+ auto unused_schema = arrow::schema(arrow::FieldVector({arrow::field("f0",
arrow::utf8())}));
+ ASSERT_OK_AND_ASSIGN(
+ std::shared_ptr<FileStorePathFactory> path_factory,
+ FileStorePathFactory::Create(root_path, unused_schema,
/*partition_keys=*/{},
+ /*default_part_value=*/"",
file_format->Identifier(),
+ /*data_file_prefix=*/"data-",
+ /*legacy_partition_name_enabled=*/true,
/*external_paths=*/{},
+
/*global_index_external_path=*/std::nullopt,
+ /*index_file_in_data_file_dir=*/false,
pool));
+ ASSERT_OK_AND_ASSIGN(
+ CoreOptions options,
+ CoreOptions::FromMap({{Options::FILE_FORMAT, "orc"},
{Options::MANIFEST_FORMAT, "orc"}}));
+ options.WithCache(manifest_cache);
+ ASSERT_OK_AND_ASSIGN(
+ std::unique_ptr<ManifestFile> manifest_file,
+ ManifestFile::Create(counting_file_system, file_format, "zstd",
path_factory,
+ /*target_file_size=*/1024, pool, options,
unused_schema));
+
+ const std::string manifest_name =
"manifest-3a44a0da-1008-463c-914e-28d271375e24-0";
+ std::vector<ManifestEntry> all_entries;
+ ASSERT_OK(manifest_file->Read(manifest_name, /*filter=*/nullptr,
&all_entries));
+ ASSERT_EQ(2, all_entries.size());
+
+ std::vector<ManifestEntry> bucket_one_entries;
+ ASSERT_OK(manifest_file->ReadBucketEntries(manifest_name, /*bucket=*/1,
&bucket_one_entries));
+ ASSERT_EQ(std::vector<ManifestEntry>({all_entries[0]}),
bucket_one_entries);
+
+ std::vector<ManifestEntry> bucket_zero_entries;
+ ASSERT_OK(manifest_file->ReadBucketEntries(manifest_name, /*bucket=*/0,
&bucket_zero_entries));
+ ASSERT_EQ(std::vector<ManifestEntry>({all_entries[1]}),
bucket_zero_entries);
+
+ std::vector<ManifestEntry> missing_bucket_entries;
+ ASSERT_OK(
+ manifest_file->ReadBucketEntries(manifest_name, /*bucket=*/2,
&missing_bucket_entries));
+ ASSERT_TRUE(missing_bucket_entries.empty());
+
+ ASSERT_EQ(1, counting_file_system->open_count);
+ ASSERT_EQ(4, manifest_cache->GetCount());
+ ASSERT_EQ(1, manifest_cache->SupplierCallCount());
+}
+
+TEST_F(ManifestFileTest, TestReadBucketEntriesSkipsDeserializingOtherBuckets) {
+ auto pool = GetDefaultPool();
+ std::vector<ManifestEntry> source_entries =
+ ReadManifestEntry("orc", paimon::test::GetDataDir() +
"/orc/append_09.db/append_09",
+ "manifest-3a44a0da-1008-463c-914e-28d271375e24-0",
pool);
+ ASSERT_EQ(2, source_entries.size());
+
+ auto test_dir = UniqueTestDirectory::Create();
+ ASSERT_TRUE(test_dir);
+ std::shared_ptr<FileSystem> file_system = test_dir->GetFileSystem();
+
ASSERT_OK(file_system->Mkdirs(FileStorePathFactory::ManifestPath(test_dir->Str())));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<FileFormat> file_format,
+ FileFormatFactory::Get("orc", {}));
+ auto unused_schema = arrow::schema(arrow::FieldVector({arrow::field("f0",
arrow::utf8())}));
+ ASSERT_OK_AND_ASSIGN(
+ std::shared_ptr<FileStorePathFactory> path_factory,
+ FileStorePathFactory::Create(test_dir->Str(), unused_schema,
/*partition_keys=*/{},
+ /*default_part_value=*/"",
file_format->Identifier(),
+ /*data_file_prefix=*/"data-",
+ /*legacy_partition_name_enabled=*/true,
/*external_paths=*/{},
+
/*global_index_external_path=*/std::nullopt,
+ /*index_file_in_data_file_dir=*/false,
pool));
+ ASSERT_OK_AND_ASSIGN(
+ CoreOptions options,
+ CoreOptions::FromMap({{Options::FILE_FORMAT, "orc"},
{Options::MANIFEST_FORMAT, "orc"}}));
+ ASSERT_OK_AND_ASSIGN(
+ std::unique_ptr<ManifestFile> manifest_file,
+ ManifestFile::Create(file_system, file_format, "zstd", path_factory,
+ /*target_file_size=*/1024, pool, options,
unused_schema));
+
+ ManifestEntry invalid_other_bucket(FileKind(static_cast<int8_t>(2)),
+ source_entries[0].Partition(),
/*bucket=*/1,
+ /*total_buckets=*/2,
source_entries[0].File());
+ ManifestEntry valid_target_bucket(FileKind::Add(),
source_entries[1].Partition(), /*bucket=*/0,
+ /*total_buckets=*/2,
source_entries[1].File());
+ using WrittenFile = std::pair<std::string, int64_t>;
+ ASSERT_OK_AND_ASSIGN(
+ WrittenFile written_file,
+ manifest_file->WriteWithoutRolling({invalid_other_bucket,
valid_target_bucket}));
+
+ std::vector<ManifestEntry> all_entries;
+ ASSERT_NOK_WITH_MSG(manifest_file->Read(written_file.first,
/*filter=*/nullptr, &all_entries),
+ "Unsupported byte value 2 for file kind.");
+
+ std::vector<ManifestEntry> bucket_entries;
+ ASSERT_OK(manifest_file->ReadBucketEntries(written_file.first,
/*bucket=*/0, &bucket_entries));
+ ASSERT_EQ(std::vector<ManifestEntry>({valid_target_bucket}),
bucket_entries);
+}
+
TEST_F(ManifestFileTest, TestWithNullCount) {
auto pool = GetDefaultPool();
auto manifest_entries =
@@ -406,6 +508,9 @@ TEST_F(ManifestFileTest,
TestManifestFileCompatibleWithJavaPaimon09) {
std::vector<ManifestEntry> expected_manifest_entries;
expected_manifest_entries.emplace_back(manifest_entry);
ASSERT_EQ(expected_manifest_entries, manifest_entries);
+ ASSERT_EQ(expected_manifest_entries,
+ ReadManifestEntry("avro", paimon::test::GetDataDir() + "/avro",
"avro_manifest_09",
+ pool, /*bucket=*/0));
}
TEST_F(ManifestFileTest, TestManifestFileCompatibleWithJavaPaimon11) {
@@ -442,6 +547,9 @@ TEST_F(ManifestFileTest,
TestManifestFileCompatibleWithJavaPaimon11) {
std::vector<ManifestEntry> expected_manifest_entries;
expected_manifest_entries.emplace_back(manifest_entry);
ASSERT_EQ(expected_manifest_entries, manifest_entries);
+ ASSERT_EQ(expected_manifest_entries,
+ ReadManifestEntry("avro", paimon::test::GetDataDir() + "/avro",
"avro_manifest_11",
+ pool, /*bucket=*/0));
}
} // 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 f319498a..e1fb5a43 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,6 +29,7 @@
#include "paimon/common/data/binary_row.h"
#include "paimon/common/data/binary_row_writer.h"
#include "paimon/common/io/cache/lru_cache.h"
+#include "paimon/core/manifest/manifest_entry.h"
#include "paimon/core/manifest/partition_entry.h"
#include "paimon/core/operation/metrics/scan_metrics.h"
#include "paimon/core/schema/schema_manager.h"
@@ -186,11 +187,14 @@ namespace {
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) {
+ const std::shared_ptr<Predicate>&
predicate = nullptr,
+ bool
manifest_entry_lazy_decode_enabled = true) {
ScanContextBuilder context_builder(table_path);
context_builder.AddOption(Options::FILE_FORMAT, "orc")
.AddOption(Options::MANIFEST_FORMAT, "orc")
.AddOption(Options::SCAN_MANIFEST_ENTRY_CACHE_MAX_SNAPSHOTS, "8")
+ .AddOption(Options::SCAN_MANIFEST_ENTRY_LAZY_DECODE_ENABLED,
+ manifest_entry_lazy_decode_enabled ? "true" : "false")
.WithCache(cache);
if (bucket) {
context_builder.SetBucketFilter(bucket.value());
@@ -205,6 +209,16 @@ std::shared_ptr<FileStoreScan> BuildScan(const
std::string& table_path,
return typed_table_scan->snapshot_reader_->scan_;
}
+std::vector<std::string> SortedFileNames(std::vector<ManifestEntry>&& entries)
{
+ std::vector<std::string> file_names;
+ file_names.reserve(entries.size());
+ for (const auto& entry : entries) {
+ file_names.push_back(entry.FileName());
+ }
+ std::sort(file_names.begin(), file_names.end());
+ return file_names;
+}
+
} // namespace
TEST(AppendOnlyFileStoreScanTest, TestDropStatsAfterFiltering) {
@@ -253,13 +267,13 @@ TEST(AppendOnlyFileStoreScanTest,
TestSnapshotLiveManifestCachePath) {
scan_first->GetSnapshotManager()->LoadSnapshot(/*snapshot_id=*/5));
scan_first->WithSnapshot(snapshot_5);
ASSERT_OK_AND_ASSIGN(auto plan_first, scan_first->CreatePlan());
- size_t first_size = plan_first->Files().size();
+ std::vector<std::string> first_file_names =
SortedFileNames(plan_first->Files());
// Second scan on the same snapshot should read the same bucket live
entries from cache.
auto scan_second = BuildScan(table_path, cache, /*bucket=*/0);
scan_second->WithSnapshot(snapshot_5);
ASSERT_OK_AND_ASSIGN(auto plan_second, scan_second->CreatePlan());
- ASSERT_EQ(first_size, plan_second->Files().size());
+ ASSERT_EQ(first_file_names, SortedFileNames(plan_second->Files()));
}
TEST(AppendOnlyFileStoreScanTest, TestSnapshotLiveManifestCacheRebuildOnMiss) {
@@ -285,6 +299,24 @@ TEST(AppendOnlyFileStoreScanTest,
TestSnapshotLiveManifestCacheRebuildOnMiss) {
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(plan_expected->Files().size(), plan_next->Files().size());
+ ASSERT_EQ(SortedFileNames(plan_expected->Files()),
SortedFileNames(plan_next->Files()));
+}
+
+TEST(AppendOnlyFileStoreScanTest,
TestSnapshotLiveManifestCacheFallbackWithoutLazyDecode) {
+ 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 scan_fallback = BuildScan(table_path, cache, /*bucket=*/0,
/*predicate=*/nullptr,
+
/*manifest_entry_lazy_decode_enabled=*/false);
+ ASSERT_OK_AND_ASSIGN(Snapshot snapshot_5,
+
scan_fallback->GetSnapshotManager()->LoadSnapshot(/*snapshot_id=*/5));
+ scan_fallback->WithSnapshot(snapshot_5);
+ ASSERT_OK_AND_ASSIGN(auto plan_fallback, scan_fallback->CreatePlan());
+
+ 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_fallback->Files()));
}
} // namespace paimon::test
diff --git a/src/paimon/core/operation/file_store_scan.cpp
b/src/paimon/core/operation/file_store_scan.cpp
index 865e006f..f21b0bb7 100644
--- a/src/paimon/core/operation/file_store_scan.cpp
+++ b/src/paimon/core/operation/file_store_scan.cpp
@@ -365,6 +365,33 @@ Status FileStoreScan::StoreSnapshotLiveManifestEntries(
Status FileStoreScan::ReadAndMergeBucketFileEntries(
const std::vector<ManifestFileMeta>& manifest_metas, int32_t bucket,
std::vector<ManifestEntry>* merged_entries) const {
+ if (core_options_.ScanManifestEntryLazyDecodeEnabled()) {
+ std::vector<std::future<Result<std::vector<ManifestEntry>>>> futures;
+ futures.reserve(manifest_metas.size());
+ for (const auto& meta : manifest_metas) {
+ auto read_meta_task = [this, meta, bucket]() ->
Result<std::vector<ManifestEntry>> {
+ std::vector<ManifestEntry> bucket_entries;
+ PAIMON_RETURN_NOT_OK(
+ manifest_file_->ReadBucketEntries(meta.FileName(), bucket,
&bucket_entries));
+ return bucket_entries;
+ };
+ futures.push_back(Via(executor_.get(), read_meta_task));
+ }
+
+ std::vector<ManifestEntry> bucket_entries;
+ std::vector<Result<std::vector<ManifestEntry>>> entry_lists =
CollectAll(futures);
+ for (auto& entry_list : entry_lists) {
+ if (!entry_list.ok()) {
+ return entry_list.status();
+ }
+ bucket_entries.reserve(bucket_entries.size() +
entry_list.value().size());
+ for (auto& entry : entry_list.value()) {
+ bucket_entries.emplace_back(std::move(entry));
+ }
+ }
+ return MergeLiveEntries(bucket_entries, merged_entries);
+ }
+
std::vector<ManifestEntry> unmerged_entries;
std::vector<ManifestEntry> entries;
PAIMON_RETURN_NOT_OK(ReadFileEntries(manifest_metas, &entries,
/*apply_scan_filter=*/false));
diff --git a/src/paimon/core/utils/objects_file.h
b/src/paimon/core/utils/objects_file.h
index f8509fe2..a56952ae 100644
--- a/src/paimon/core/utils/objects_file.h
+++ b/src/paimon/core/utils/objects_file.h
@@ -19,7 +19,6 @@
#pragma once
#include <functional>
-#include <limits>
#include <memory>
#include <string>
#include <utility>
@@ -78,6 +77,10 @@ class ObjectsFile {
Result<std::pair<std::string, int64_t>> WriteWithoutRolling(const
std::vector<T>& records);
protected:
+ Status ReadArrowBatches(
+ const std::string& file_name,
+ const std::function<Status(const
std::shared_ptr<arrow::StructArray>&)>& consumer) const;
+
std::shared_ptr<PathFactory> path_factory_;
std::shared_ptr<MemoryPool> pool_;
std::unique_ptr<ObjectSerializer<T>> serializer_;
@@ -127,6 +130,30 @@ template <typename T>
Status ObjectsFile<T>::Read(const std::string& file_name,
const std::function<Result<bool>(const T&)>&
filter,
std::vector<T>* result) const {
+ return ReadArrowBatches(
+ file_name,
+ [this, &filter, result](const std::shared_ptr<arrow::StructArray>&
struct_array) -> Status {
+ result->reserve(result->size() + struct_array->length());
+ for (int64_t i = 0; i < struct_array->length(); i++) {
+ ColumnarRow row(struct_array->fields(), pool_, i);
+ PAIMON_ASSIGN_OR_RAISE(T obj, serializer_->FromRow(row));
+ if (filter) {
+ PAIMON_ASSIGN_OR_RAISE(bool filter_res, filter(obj));
+ if (filter_res) {
+ result->push_back(std::move(obj));
+ }
+ } else {
+ result->push_back(std::move(obj));
+ }
+ }
+ return Status::OK();
+ });
+}
+
+template <typename T>
+Status ObjectsFile<T>::ReadArrowBatches(
+ const std::string& file_name,
+ const std::function<Status(const std::shared_ptr<arrow::StructArray>&)>&
consumer) const {
std::string file_path = path_factory_->ToPath(file_name);
std::shared_ptr<InputStream> file_input_stream;
std::shared_ptr<Bytes> cached_bytes;
@@ -171,22 +198,10 @@ Status ObjectsFile<T>::Read(const std::string& file_name,
if (!typed_array || typed_array->type_id() != arrow::Type::STRUCT) {
return Status::Invalid(fmt::format("file {}, cannot cast to struct
array", file_name));
}
- auto* struct_array =
checked_cast<arrow::StructArray*>(typed_array.get());
- result->reserve(struct_array->length());
- for (int64_t i = 0; i < struct_array->length(); i++) {
- ColumnarRow row(struct_array->fields(), pool_, i);
- PAIMON_ASSIGN_OR_RAISE(T obj, serializer_->FromRow(row));
- if (filter) {
- PAIMON_ASSIGN_OR_RAISE(bool filter_res, filter(obj));
- if (filter_res) {
- result->push_back(std::move(obj));
- }
- } else {
- result->push_back(std::move(obj));
- }
- }
+ std::shared_ptr<arrow::StructArray> struct_array =
+ checked_pointer_cast<arrow::StructArray>(typed_array);
+ PAIMON_RETURN_NOT_OK(consumer(struct_array));
}
- reader->Close();
return Status::OK();
}
