perf(manifest): selectively decode bucket entries with bounded Avro indexes

Probe manifest versions and buckets before materializing selected entries
when the input bytes are retained in memory. Keep single-pass reads for
uncached inputs and manifests known to contain only the requested bucket.

Support precise Avro bitmap selection and preserve physical row IDs across
schema resets. Reuse block positions from complete scans with a bounded
index, falling back to sequential skipping when the bound is exceeded.

Cover cached schema evolution, invalid versions, cache fallback, sparse
selection, compressed blocks, empty inputs and index overflow.

Generated-by: OpenAI Codex (GPT-6)

Author: zhongzheyun

src/paimon/core/manifest/manifest_file.cpp

@@ -22,10 +22,12 @@
 #include <limits>
 #include <utility>
 
+#include "arrow/array/array_primitive.h"
 #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/common/utils/checked_cast.h"
 #include "paimon/core/io/rolling_file_writer.h"
 #include "paimon/core/manifest/manifest_entry.h"
 #include "paimon/core/manifest/manifest_entry_serializer.h"
@@ -106,7 +108,81 @@ Status ManifestFile::ReadBucketEntries(const std::string& file_name, int32_t buc
                 entries->push_back(std::move(entry));
             }
             return Status::OK();
-        });
+        },
+        [this, bucket](FileBatchReader* reader) { return PrepareBucketRead(reader, bucket); });
+}
+
+Status ManifestFile::PrepareBucketRead(FileBatchReader* reader, int32_t bucket) const {
+    if (!reader->SupportPreciseBitmapSelection()) {
+        return Status::OK();
+    }
+    PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<ArrowSchema> c_schema, reader->GetFileSchema());
+    PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Schema> file_schema,
+                                      arrow::ImportSchema(c_schema.get()));
+    const auto& target_type = serializer_->GetDataType();
+    const std::string& version_name = target_type->field(0)->name();
+    const std::string& bucket_name = target_type->field(3)->name();
+    auto version_field = file_schema->GetFieldByName(version_name);
+    auto bucket_field = file_schema->GetFieldByName(bucket_name);
+    if (!version_field || !bucket_field || version_field->type()->id() != arrow::Type::INT32 ||
+        bucket_field->type()->id() != arrow::Type::INT32) {
+        return Status::OK();
+    }
+    arrow::FieldVector probe_fields;
+    for (const auto& field : file_schema->fields()) {
+        if (field->name() == version_name || field->name() == bucket_name) {
+            probe_fields.push_back(field);
+        }
+    }
+    auto probe_schema = arrow::schema(probe_fields);
+    ArrowSchema probe_c_schema;
+    PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportSchema(*probe_schema, &probe_c_schema));
+    PAIMON_RETURN_NOT_OK(
+        reader->SetReadSchema(&probe_c_schema, /*predicate=*/nullptr, std::nullopt));
+    RoaringBitmap32 selected;
+    bool row_ids_fit = true;
+    while (true) {
+        PAIMON_ASSIGN_OR_RAISE(BatchReader::ReadBatch batch, reader->NextBatch());
+        if (BatchReader::IsEofBatch(batch)) {
+            break;
+        }
+        PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
+            std::shared_ptr<arrow::Array> array,
+            arrow::ImportArray(batch.first.get(), batch.second.get()));
+        if (!array || array->type_id() != arrow::Type::STRUCT) {
+            return Status::Invalid("Manifest bucket probe must return a struct array");
+        }
+        auto rows = checked_pointer_cast<arrow::StructArray>(array);
+        auto version_array = rows->GetFieldByName(version_name);
+        auto bucket_array = rows->GetFieldByName(bucket_name);
+        if (!version_array || !bucket_array || version_array->type_id() != arrow::Type::INT32 ||
+            bucket_array->type_id() != arrow::Type::INT32) {
+            return Status::Invalid("Manifest bucket probe must return int32 version and bucket");
+        }
+        auto versions = checked_pointer_cast<arrow::Int32Array>(version_array);
+        auto buckets = checked_pointer_cast<arrow::Int32Array>(bucket_array);
+        for (int64_t i = 0; i < rows->length(); ++i) {
+            if (rows->IsNull(i) || versions->IsNull(i) || buckets->IsNull(i)) {
+                return Status::Invalid("Manifest version and bucket must not be null");
+            }
+            // Validate every entry, including buckets not selected by this scan.
+            PAIMON_RETURN_NOT_OK(ManifestEntrySerializer::ValidateVersion(versions->Value(i)));
+            if (buckets->Value(i) == bucket) {
+                PAIMON_ASSIGN_OR_RAISE(uint64_t file_row, reader->GetPreviousBatchFileRowId(i));
+                if (file_row > std::numeric_limits<uint32_t>::max()) {
+                    row_ids_fit = false;
+                } else {
+                    selected.Add(static_cast<uint32_t>(file_row));
+                }
+            }
+        }
+    }
+    // Keep the on-disk schema; ManifestMetaReader still performs schema evolution afterwards.
+    ArrowSchema full_c_schema;
+    PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportSchema(*file_schema, &full_c_schema));
+    return reader->SetReadSchema(
+        &full_c_schema, /*predicate=*/nullptr,
+        row_ids_fit ? std::optional<RoaringBitmap32>(std::move(selected)) : std::nullopt);
 }
 
 Result<std::vector<ManifestFileMeta>> ManifestFile::Write(

src/paimon/core/manifest/manifest_file.h

@@ -67,6 +67,8 @@ class ManifestFile : public ObjectsFile<ManifestEntry> {
                              std::vector<ManifestEntry>* entries) const;
 
  private:
+    Status PrepareBucketRead(FileBatchReader* reader, int32_t bucket) const;
+
     ManifestFile(const std::shared_ptr<FileSystem>& file_system,
                  const std::shared_ptr<ReaderBuilder>& reader_builder,
                  const std::shared_ptr<WriterBuilder>& writer_builder,

src/paimon/core/manifest/manifest_file_test.cpp

@@ -25,13 +25,15 @@
 #include <utility>
 
 #include "arrow/api.h"
+#include "arrow/ipc/json_simple.h"
 #include "gtest/gtest.h"
 #include "paimon/common/data/binary_row.h"
 #include "paimon/common/data/data_define.h"
 #include "paimon/core/io/data_file_meta.h"
 #include "paimon/core/manifest/file_kind.h"
 #include "paimon/core/manifest/file_source.h"
 #include "paimon/core/manifest/manifest_entry.h"
+#include "paimon/core/manifest/manifest_entry_serializer.h"
 #include "paimon/core/manifest/manifest_file_meta.h"
 #include "paimon/core/stats/simple_stats.h"
 #include "paimon/core/utils/file_store_path_factory.h"
@@ -40,6 +42,8 @@
 #include "paimon/defs.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/memory/memory_pool.h"
 #include "paimon/testing/utils/binary_row_generator.h"
@@ -104,30 +108,45 @@ class ManifestFileTest : public testing::Test {
         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 {
+        EXPECT_OK_AND_ASSIGN(
+            std::vector<ManifestEntry> entries,
+            TryReadManifestEntry(file_format_str, root_path, file_name, pool, bucket));
+        return entries;
+    }
+
+    Result<std::vector<ManifestEntry>> TryReadManifestEntry(
+        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, bool cache_enabled = false) 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, {}));
+        PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<FileFormat> file_format,
+                               FileFormatFactory::Get(file_format_str, {}));
         auto unused_schema = arrow::schema(arrow::FieldVector({arrow::field("f0", arrow::utf8())}));
-        EXPECT_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));
-        EXPECT_OK_AND_ASSIGN(CoreOptions options,
-                             CoreOptions::FromMap({{Options::FILE_FORMAT, "orc"}}));
-        EXPECT_OK_AND_ASSIGN(
+        PAIMON_ASSIGN_OR_RAISE(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));
+        PAIMON_ASSIGN_OR_RAISE(CoreOptions options,
+                               CoreOptions::FromMap({{Options::FILE_FORMAT, "orc"}}));
+        if (cache_enabled) {
+            options.WithCache(
+                std::make_shared<CountingRoutingCache>(CacheKind::MANIFEST, 64 * 1024 * 1024));
+        }
+        PAIMON_ASSIGN_OR_RAISE(
             std::unique_ptr<ManifestFile> manifest_file,
             ManifestFile::Create(file_system, file_format, "zstd", path_factory,
                                  /*target_file_size=*/1024, pool, options, unused_schema));
         std::vector<ManifestEntry> manifest_entries;
         if (bucket) {
-            EXPECT_OK(
+            PAIMON_RETURN_NOT_OK(
                 manifest_file->ReadBucketEntries(file_name, bucket.value(), &manifest_entries));
         } else {
-            EXPECT_OK(manifest_file->Read(file_name, /*filter=*/nullptr, &manifest_entries));
+            PAIMON_RETURN_NOT_OK(
+                manifest_file->Read(file_name, /*filter=*/nullptr, &manifest_entries));
         }
 
         return manifest_entries;
@@ -373,45 +392,104 @@ TEST_F(ManifestFileTest, TestReadBucketEntriesSkipsDeserializingOtherBuckets) {
                           "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("avro", {}));
-    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"}}));
-    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}));
+    // A cache that rejects manifest keys must fall back to the single-pass reader.
+    for (int32_t cache_mode : {0, 1, 2}) {
+        SCOPED_TRACE(cache_mode);
+        auto test_dir = UniqueTestDirectory::Create();
+        ASSERT_TRUE(test_dir);
+        auto file_system = std::make_shared<CountingFileSystem>();
+        ASSERT_OK(file_system->Mkdirs(FileStorePathFactory::ManifestPath(test_dir->Str())));
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<FileFormat> file_format,
+                             FileFormatFactory::Get("avro", {}));
+        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"}}));
+        if (cache_mode != 0) {
+            options.WithCache(std::make_shared<CountingRoutingCache>(
+                cache_mode == 1 ? CacheKind::MANIFEST : CacheKind::DEFAULT, 64 * 1024 * 1024));
+        }
+        ASSERT_OK_AND_ASSIGN(
+            std::unique_ptr<ManifestFile> manifest_file,
+            ManifestFile::Create(file_system, file_format, "null", path_factory,
+                                 /*target_file_size=*/1024, pool, options, unused_schema));
 
-    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.");
+        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}));
+
+        // Exercise both a cold cache and reuse of the retained manifest bytes.
+        for (int32_t read = 0; read < 2; ++read) {
+            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);
+            ASSERT_EQ(cache_mode == 1 ? 1 : read + 1, file_system->open_count);
+        }
+        std::vector<ManifestEntry> missing_bucket_entries;
+        ASSERT_OK(manifest_file->ReadBucketEntries(written_file.first, /*bucket=*/2,
+                                                   &missing_bucket_entries));
+        ASSERT_TRUE(missing_bucket_entries.empty());
+        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, TestBucketProbeValidatesVersionsOutsideSelectedBucket) {
+    auto pool = GetDefaultPool();
+    ManifestEntrySerializer serializer(pool);
+    auto type =
+        arrow::struct_({arrow::field(serializer.GetDataType()->field(0)->name(), arrow::int32()),
+                        arrow::field(serializer.GetDataType()->field(3)->name(), arrow::int32())});
+    const std::vector<std::pair<std::string, std::string>> cases = {
+        {R"([[999,1],[2,0]])", "Unsupported version: 999"},
+        {R"([[1,1],[2,0]])", "not compatible"},
+        {R"([[null,1],[2,0]])", "must not be null"},
+        {R"([[2,null],[2,0]])", "must not be null"}};
+    for (const auto& item : cases) {
+        SCOPED_TRACE(item.first);
+        auto dir = UniqueTestDirectory::Create();
+        ASSERT_TRUE(dir);
+        auto fs = dir->GetFileSystem();
+        const std::string manifest_dir = FileStorePathFactory::ManifestPath(dir->Str());
+        ASSERT_OK(fs->Mkdirs(manifest_dir));
+        const std::string file_name = "manifest-invalid-version";
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<FileFormat> format,
+                             FileFormatFactory::Get("avro", {}));
+        ArrowSchema c_schema;
+        ASSERT_TRUE(arrow::ExportSchema(*arrow::schema(type->fields()), &c_schema).ok());
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<WriterBuilder> builder,
+                             format->CreateWriterBuilder(&c_schema, 2));
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<OutputStream> output,
+                             fs->Create(manifest_dir + "/" + file_name, false));
+        ASSERT_OK_AND_ASSIGN(std::unique_ptr<FormatWriter> writer, builder->Build(output, "null"));
+        auto array = arrow::ipc::internal::json::ArrayFromJSON(type, item.first).ValueOrDie();
+        ArrowArray c_array;
+        ASSERT_TRUE(arrow::ExportArray(*array, &c_array).ok());
+        ASSERT_OK(writer->AddBatch(&c_array));
+        ASSERT_OK(writer->Flush());
+        ASSERT_OK(writer->Finish());
+        ASSERT_OK(output->Close());
+        ASSERT_NOK_WITH_MSG(TryReadManifestEntry("avro", dir->Str(), file_name, pool, /*bucket=*/0,
+                                                 /*cache_enabled=*/true),
+                            item.second);
+    }
 }
 
 TEST_F(ManifestFileTest, TestLegacyManifestFormatIsReadOnly) {
@@ -539,6 +617,11 @@ TEST_F(ManifestFileTest, TestManifestFileCompatibleWithJavaPaimon09) {
     ASSERT_EQ(expected_manifest_entries,
               ReadManifestEntry("avro", paimon::test::GetDataDir() + "/avro", "avro_manifest_09",
                                 pool, /*bucket=*/0));
+    ASSERT_OK_AND_ASSIGN(std::vector<ManifestEntry> cached_entries,
+                         TryReadManifestEntry("avro", paimon::test::GetDataDir() + "/avro",
+                                              "avro_manifest_09", pool, /*bucket=*/0,
+                                              /*cache_enabled=*/true));
+    ASSERT_EQ(expected_manifest_entries, cached_entries);
 }
 
 TEST_F(ManifestFileTest, TestManifestFileCompatibleWithJavaPaimon11) {
@@ -578,6 +661,11 @@ TEST_F(ManifestFileTest, TestManifestFileCompatibleWithJavaPaimon11) {
     ASSERT_EQ(expected_manifest_entries,
               ReadManifestEntry("avro", paimon::test::GetDataDir() + "/avro", "avro_manifest_11",
                                 pool, /*bucket=*/0));
+    ASSERT_OK_AND_ASSIGN(std::vector<ManifestEntry> cached_entries,
+                         TryReadManifestEntry("avro", paimon::test::GetDataDir() + "/avro",
+                                              "avro_manifest_11", pool, /*bucket=*/0,
+                                              /*cache_enabled=*/true));
+    ASSERT_EQ(expected_manifest_entries, cached_entries);
 }
 
 }  // namespace paimon::test