perf(scan): lazily decode manifest bucket entries

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[];

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";

src/paimon/core/core_options.cpp

@@ -425,6 +425,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;
@@ -827,6 +828,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"
@@ -1166,6 +1169,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;
 }

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;

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());
@@ -217,6 +218,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"},
@@ -353,6 +355,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());

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);

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;

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());

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()) {

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,

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