diff --git a/include/paimon/defs.h b/include/paimon/defs.h
index d1ebf507a..17cac9177 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 bac4f16f7..6e16ed33d 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 578950fbe..22e97f1f7 100644
--- a/src/paimon/core/core_options.cpp
+++ b/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;
 }
diff --git a/src/paimon/core/core_options.h b/src/paimon/core/core_options.h
index 9eb289883..600e43f01 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 ebc127edb..9ec596cab 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());
@@ -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());
diff --git a/src/paimon/core/manifest/manifest_entry_serializer.cpp b/src/paimon/core/manifest/manifest_entry_serializer.cpp
index 053405b89..2389cd8d7 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 7438895f5..4a71a1b68 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 2aa2db524..2d8cffc37 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 22f2681f6..1be49d0b5 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 d34764b5f..0211e14d1 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 8f6e0b2ea..a34f41524 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 f319498a4..e1fb5a43a 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 865e006ff..f21b0bb7f 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 f8509fe23..a56952ae2 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();
 }