diff --git a/include/paimon/fs/file_system.h b/include/paimon/fs/file_system.h
index bee3ecc82..99081b151 100644
--- a/include/paimon/fs/file_system.h
+++ b/include/paimon/fs/file_system.h
@@ -22,6 +22,7 @@
 #include <functional>
 #include <memory>
 #include <string>
+#include <utility>
 #include <vector>
 
 #include "paimon/result.h"
@@ -151,7 +152,7 @@ class PAIMON_EXPORT BasicFileStatus {
     virtual std::string GetPath() const = 0;
 };
 
-/// Extended file status information interface.
+/// Extended file status information.
 ///
 /// This class extends BasicFileStatus to provide comprehensive file system metadata including file
 /// size, modification time, and other attributes. It's used for operations that require detailed
@@ -161,21 +162,45 @@ class PAIMON_EXPORT FileStatus {
     FileStatus() = default;
     virtual ~FileStatus() = default;
 
+    /// Sentinel returned by `GetModificationTime()` when the modification time is not known.
+    static constexpr int64_t kUnknownModificationTime = -1;
+
+    /// Create a file status from caller-supplied metadata.
+    /// @param path The path of the file or directory.
+    /// @param length The size of the file in bytes. It may be negative only when the size is
+    ///               unknown.
+    /// @param is_dir Whether the path represents a directory. Defaults to false.
+    FileStatus(std::string path, int64_t length, bool is_dir = false)
+        : path_(std::move(path)), length_(length), is_dir_(is_dir) {}
+
     /// Get the size of the file in bytes.
     /// @note For directories, this method is undefined behavior.
-    virtual int64_t GetLen() const = 0;
+    virtual int64_t GetLen() const {
+        return length_;
+    }
 
     /// Check if this entry represents a directory.
-    virtual bool IsDir() const = 0;
+    virtual bool IsDir() const {
+        return is_dir_;
+    }
 
     /// Get the path of this file or directory.
-    virtual std::string GetPath() const = 0;
+    virtual std::string GetPath() const {
+        return path_;
+    }
 
     /// Get the last modification time of the file.
     ///
     /// @return A long value representing the time the file was last modified, measured in
     /// milliseconds since the epoch (UTC January 1, 1970).
-    virtual int64_t GetModificationTime() const = 0;
+    virtual int64_t GetModificationTime() const {
+        return kUnknownModificationTime;
+    }
+
+ private:
+    std::string path_;
+    int64_t length_ = -1;
+    bool is_dir_ = false;
 };
 
 /// Abstract file system interface.
@@ -193,6 +218,24 @@ class PAIMON_EXPORT FileSystem {
     ///         failure (e.g., file not found, permission denied).
     virtual Result<std::unique_ptr<InputStream>> Open(const std::string& path) const = 0;
 
+    /// Open an existing regular file for reading with known file metadata.
+    /// @param file_status The trusted status of the file to open. Its path and length must
+    ///                    identify an existing regular file. Its length must be non-negative;
+    ///                    zero is valid for an empty file.
+    /// @return Result containing a unique pointer to `InputStream` on success, or error status on
+    ///         failure (e.g., invalid file size, file not found, permission denied).
+    /// @note File systems may rely on `file_status` to skip metadata requests. The caller must
+    ///       not expect this method to validate the path, file type, or size. A stale or
+    ///       incorrect status, or a file removed after planning, can cause reads to end early or
+    ///       fail when read instead of failing at open time. Wrapping file systems should forward
+    ///       both `Open` overloads.
+    virtual Result<std::unique_ptr<InputStream>> Open(const FileStatus& file_status) const {
+        if (file_status.GetLen() < 0) {
+            return Status::Invalid("file size must be non-negative");
+        }
+        return Open(file_status.GetPath());
+    }
+
     /// Create a new file for writing.
     /// @param path The file path to create.
     /// @param overwrite If true, overwrite existing file; if false, fail if file exists.
diff --git a/src/paimon/common/file_index/bitmap/apply_bitmap_index_batch_reader_test.cpp b/src/paimon/common/file_index/bitmap/apply_bitmap_index_batch_reader_test.cpp
index 7b1ebd3ff..d068d3c2b 100644
--- a/src/paimon/common/file_index/bitmap/apply_bitmap_index_batch_reader_test.cpp
+++ b/src/paimon/common/file_index/bitmap/apply_bitmap_index_batch_reader_test.cpp
@@ -93,8 +93,8 @@ class ApplyBitmapIndexBatchReaderTest : public ::testing::Test,
                 ASSERT_OK_AND_ASSIGN(
                     file_batch_reader,
                     PrefetchFileBatchReaderImpl::Create(
-                        /*data_file_path=*/"DUMMY", &reader_builder, fs_, prefetch_batch_count,
-                        batch_size, prefetch_batch_count * 2,
+                        /*data_file_path=*/"DUMMY", /*data_file_size=*/0, &reader_builder, fs_,
+                        prefetch_batch_count, batch_size, prefetch_batch_count * 2,
                         /*enable_adaptive_prefetch_strategy=*/false, executor_,
                         /*initialize_read_ranges=*/true,
                         /*prefetch_cache_mode=*/PrefetchCacheMode::ALWAYS, CacheConfig(), pool_));
diff --git a/src/paimon/common/fs/file_system_test.cpp b/src/paimon/common/fs/file_system_test.cpp
index 494823811..22cbf3f03 100644
--- a/src/paimon/common/fs/file_system_test.cpp
+++ b/src/paimon/common/fs/file_system_test.cpp
@@ -283,6 +283,20 @@ TEST_P(FileSystemTest, TestSimpleWriteAndRead) {
     ASSERT_OK(in_stream->Close());
 }
 
+TEST_P(FileSystemTest, TestOpenWithKnownFileSize) {
+    const std::string content = "abcdefghijk";
+    const std::string file_path = test_root_ + "/file.data";
+    ASSERT_OK(fs_->WriteFile(file_path, content, /*overwrite=*/true));
+
+    FileStatus file_status(file_path, static_cast<int64_t>(content.size()));
+    ASSERT_OK_AND_ASSIGN(auto input_stream, fs_->Open(file_status));
+    ASSERT_OK_AND_ASSIGN(int64_t file_size, input_stream->Length());
+    ASSERT_EQ(file_size, content.size());
+    ASSERT_OK(input_stream->Close());
+
+    ASSERT_TRUE(fs_->Open(FileStatus(file_path, /*length=*/-1)).status().IsInvalid());
+}
+
 TEST_P(FileSystemTest, TestWriteMultipleTimes) {
     std::vector<std::string> content_vec = {"abc", "defg", "hi", "j", "k"};
     std::string content = "abcdefghijk";
diff --git a/src/paimon/common/fs/object_store_file_system.cpp b/src/paimon/common/fs/object_store_file_system.cpp
index 2eb2b9be0..77193d9bb 100644
--- a/src/paimon/common/fs/object_store_file_system.cpp
+++ b/src/paimon/common/fs/object_store_file_system.cpp
@@ -388,6 +388,19 @@ Result<std::unique_ptr<InputStream>> ObjectStoreFileSystem::Open(const std::stri
                                                     ToUri(object_path), metadata.value().size);
 }
 
+Result<std::unique_ptr<InputStream>> ObjectStoreFileSystem::Open(
+    const FileStatus& file_status) const {
+    const std::string path = file_status.GetPath();
+    const int64_t file_size = file_status.GetLen();
+    PAIMON_RETURN_NOT_OK(ValidateValueNonNegative(file_size, "file size"));
+    PAIMON_ASSIGN_OR_RAISE(ObjectStorePath object_path, ParsePath(path));
+    if (object_path.key.empty()) {
+        return Status::Invalid(fmt::format("{} is a directory", path));
+    }
+    return std::make_unique<ObjectStoreInputStream>(client_, read_ahead_limiter_, object_path,
+                                                    ToUri(object_path), file_size);
+}
+
 Result<std::unique_ptr<FileStatus>> ObjectStoreFileSystem::GetFileStatus(
     const std::string& path) const {
     PAIMON_ASSIGN_OR_RAISE(ObjectStorePath object_path, ParsePath(path));
diff --git a/src/paimon/common/fs/object_store_file_system.h b/src/paimon/common/fs/object_store_file_system.h
index 9ef305284..5bc54a43f 100644
--- a/src/paimon/common/fs/object_store_file_system.h
+++ b/src/paimon/common/fs/object_store_file_system.h
@@ -85,6 +85,7 @@ class PAIMON_EXPORT ObjectStoreFileSystem : public FileSystem {
     ~ObjectStoreFileSystem() override = default;
 
     Result<std::unique_ptr<InputStream>> Open(const std::string& path) const override;
+    Result<std::unique_ptr<InputStream>> Open(const FileStatus& file_status) const override;
     Result<std::unique_ptr<FileStatus>> GetFileStatus(const std::string& path) const override;
     Status ListDir(const std::string& directory,
                    std::vector<std::unique_ptr<BasicFileStatus>>* file_status_list) const override;
diff --git a/src/paimon/common/fs/object_store_file_system_test.cpp b/src/paimon/common/fs/object_store_file_system_test.cpp
index 1b34917ab..131bfd17a 100644
--- a/src/paimon/common/fs/object_store_file_system_test.cpp
+++ b/src/paimon/common/fs/object_store_file_system_test.cpp
@@ -159,6 +159,15 @@ TEST(ObjectStoreFileSystemTest, TestOpenBucketRootIsDirectory) {
     ASSERT_EQ(client->list_calls_, 0);
 }
 
+TEST(ObjectStoreFileSystemTest, TestOpenWithKnownLengthSkipsHead) {
+    auto client = std::make_shared<MockObjectStoreClient>();
+    client->objects_["file"] = "data";
+    ObjectStoreFileSystem fs("s3", client);
+    ASSERT_OK_AND_ASSIGN(auto stream, fs.Open(FileStatus("s3://bucket/file", 4)));
+    ASSERT_EQ(stream->Length().value(), 4);
+    ASSERT_EQ(client->head_calls_, 0);
+}
+
 TEST(ObjectStoreFileSystemTest, TestPathWithLeadingSlashes) {
     auto client = std::make_shared<MockObjectStoreClient>();
     client->objects_["file"] = "data";
diff --git a/src/paimon/common/fs/resolving_file_system.cpp b/src/paimon/common/fs/resolving_file_system.cpp
index 69fc28228..a9d6aec53 100644
--- a/src/paimon/common/fs/resolving_file_system.cpp
+++ b/src/paimon/common/fs/resolving_file_system.cpp
@@ -79,6 +79,13 @@ Result<std::unique_ptr<InputStream>> ResolvingFileSystem::Open(const std::string
     return fs->Open(path);
 }
 
+Result<std::unique_ptr<InputStream>> ResolvingFileSystem::Open(
+    const FileStatus& file_status) const {
+    PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<FileSystem> fs,
+                           GetRealFileSystem(file_status.GetPath()));
+    return fs->Open(file_status);
+}
+
 Result<std::unique_ptr<OutputStream>> ResolvingFileSystem::Create(const std::string& path,
                                                                   bool overwrite) const {
     PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<FileSystem> fs, GetRealFileSystem(path));
diff --git a/src/paimon/common/fs/resolving_file_system.h b/src/paimon/common/fs/resolving_file_system.h
index 2c5c64349..c3625caa3 100644
--- a/src/paimon/common/fs/resolving_file_system.h
+++ b/src/paimon/common/fs/resolving_file_system.h
@@ -41,6 +41,7 @@ class ResolvingFileSystem : public FileSystem {
     ~ResolvingFileSystem() override = default;
 
     Result<std::unique_ptr<InputStream>> Open(const std::string& path) const override;
+    Result<std::unique_ptr<InputStream>> Open(const FileStatus& file_status) const override;
     Result<std::unique_ptr<OutputStream>> Create(const std::string& path,
                                                  bool overwrite) const override;
 
diff --git a/src/paimon/common/reader/prefetch_file_batch_reader_impl.cpp b/src/paimon/common/reader/prefetch_file_batch_reader_impl.cpp
index defdd660f..c44651790 100644
--- a/src/paimon/common/reader/prefetch_file_batch_reader_impl.cpp
+++ b/src/paimon/common/reader/prefetch_file_batch_reader_impl.cpp
@@ -56,7 +56,7 @@ std::pair<int64_t, int64_t> ComputeBatchSliceByReadRange(
 }  // namespace
 
 Result<std::unique_ptr<PrefetchFileBatchReaderImpl>> PrefetchFileBatchReaderImpl::Create(
-    const std::string& data_file_path, const ReaderBuilder* reader_builder,
+    const std::string& data_file_path, int64_t data_file_size, const ReaderBuilder* reader_builder,
     const std::shared_ptr<FileSystem>& fs, uint32_t prefetch_max_parallel_num, int32_t batch_size,
     uint32_t prefetch_batch_count, bool enable_adaptive_prefetch_strategy,
     const std::shared_ptr<Executor>& executor, bool initialize_read_ranges,
@@ -83,20 +83,22 @@ Result<std::unique_ptr<PrefetchFileBatchReaderImpl>> PrefetchFileBatchReaderImpl
 
     std::shared_ptr<ReadAheadCache> cache;
     if (prefetch_cache_mode != PrefetchCacheMode::NEVER) {
-        PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<InputStream> input_stream, fs->Open(data_file_path));
+        PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<InputStream> input_stream,
+                               fs->Open(FileStatus(data_file_path, data_file_size)));
         cache = std::make_shared<ReadAheadCache>(input_stream, cache_config, pool);
     }
     std::vector<std::future<Result<std::unique_ptr<FileBatchReader>>>> futures;
     for (uint32_t i = 0; i < prefetch_max_parallel_num; i++) {
-        futures.push_back(Via(executor.get(),
-                              [&fs, &data_file_path, &reader_builder,
-                               &cache]() -> Result<std::unique_ptr<FileBatchReader>> {
-                                  PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<InputStream> input_stream,
-                                                         fs->Open(data_file_path));
-                                  auto cache_input_stream = std::make_shared<CacheInputStream>(
-                                      std::move(input_stream), cache);
-                                  return reader_builder->Build(cache_input_stream);
-                              }));
+        futures.push_back(
+            Via(executor.get(),
+                [&fs, &data_file_path, data_file_size, &reader_builder,
+                 &cache]() -> Result<std::unique_ptr<FileBatchReader>> {
+                    PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<InputStream> input_stream,
+                                           fs->Open(FileStatus(data_file_path, data_file_size)));
+                    auto cache_input_stream =
+                        std::make_shared<CacheInputStream>(std::move(input_stream), cache);
+                    return reader_builder->Build(cache_input_stream);
+                }));
     }
     std::vector<std::shared_ptr<PrefetchFileBatchReader>> readers;
     for (auto& file_batch_reader : CollectAll(futures)) {
diff --git a/src/paimon/common/reader/prefetch_file_batch_reader_impl.h b/src/paimon/common/reader/prefetch_file_batch_reader_impl.h
index f0c302e21..78cfbb5f9 100644
--- a/src/paimon/common/reader/prefetch_file_batch_reader_impl.h
+++ b/src/paimon/common/reader/prefetch_file_batch_reader_impl.h
@@ -56,12 +56,12 @@ class Metrics;
 class PrefetchFileBatchReaderImpl : public PrefetchFileBatchReader {
  public:
     static Result<std::unique_ptr<PrefetchFileBatchReaderImpl>> Create(
-        const std::string& data_file_path, const ReaderBuilder* reader_builder,
-        const std::shared_ptr<FileSystem>& fs, uint32_t prefetch_max_parallel_num,
-        int32_t batch_size, uint32_t prefetch_batch_count, bool enable_adaptive_prefetch_strategy,
-        const std::shared_ptr<Executor>& executor, bool initialize_read_ranges,
-        PrefetchCacheMode prefetch_cache_mode, const CacheConfig& cache_config,
-        const std::shared_ptr<MemoryPool>& pool);
+        const std::string& data_file_path, int64_t data_file_size,
+        const ReaderBuilder* reader_builder, const std::shared_ptr<FileSystem>& fs,
+        uint32_t prefetch_max_parallel_num, int32_t batch_size, uint32_t prefetch_batch_count,
+        bool enable_adaptive_prefetch_strategy, const std::shared_ptr<Executor>& executor,
+        bool initialize_read_ranges, PrefetchCacheMode prefetch_cache_mode,
+        const CacheConfig& cache_config, const std::shared_ptr<MemoryPool>& pool);
 
     ~PrefetchFileBatchReaderImpl() override;
 
diff --git a/src/paimon/common/reader/prefetch_file_batch_reader_impl_test.cpp b/src/paimon/common/reader/prefetch_file_batch_reader_impl_test.cpp
index 30ecdf65e..e1514adab 100644
--- a/src/paimon/common/reader/prefetch_file_batch_reader_impl_test.cpp
+++ b/src/paimon/common/reader/prefetch_file_batch_reader_impl_test.cpp
@@ -199,14 +199,16 @@ class PrefetchFileBatchReaderImplTest : public ::testing::Test,
         EXPECT_OK_AND_ASSIGN(auto reader_builder, file_format->CreateReaderBuilder(batch_size));
         EXPECT_OK_AND_ASSIGN(std::shared_ptr<Executor> executor,
                              CreateDefaultExecutor(prefetch_max_parallel_num - 1));
+        const std::string data_file_path =
+            PathUtil::JoinPath(dir_->Str(), "file." + file_format->Identifier());
+        EXPECT_OK_AND_ASSIGN(auto data_file_status, local_fs_->GetFileStatus(data_file_path));
         EXPECT_OK_AND_ASSIGN(
             std::unique_ptr<PrefetchFileBatchReaderImpl> reader,
             PrefetchFileBatchReaderImpl::Create(
-                PathUtil::JoinPath(dir_->Str(), "file." + file_format->Identifier()),
-                reader_builder.get(), local_fs_, prefetch_max_parallel_num, batch_size,
-                prefetch_max_parallel_num * 2, /*enable_adaptive_prefetch_strategy=*/false,
-                executor, /*initialize_read_ranges=*/false, cache_mode, CacheConfig(),
-                GetDefaultPool()));
+                data_file_path, data_file_status->GetLen(), reader_builder.get(), local_fs_,
+                prefetch_max_parallel_num, batch_size, prefetch_max_parallel_num * 2,
+                /*enable_adaptive_prefetch_strategy=*/false, executor,
+                /*initialize_read_ranges=*/false, cache_mode, CacheConfig(), GetDefaultPool()));
         std::unique_ptr<ArrowSchema> c_schema = std::make_unique<ArrowSchema>();
         auto arrow_status = arrow::ExportSchema(*read_schema, c_schema.get());
         EXPECT_TRUE(arrow_status.ok());
@@ -299,8 +301,8 @@ TEST_F(PrefetchFileBatchReaderImplTest, TestSimple) {
         ASSERT_OK_AND_ASSIGN(
             auto reader,
             PrefetchFileBatchReaderImpl::Create(
-                /*data_file_path=*/"", &reader_builder, mock_fs_, prefetch_max_parallel_num,
-                batch_size, prefetch_max_parallel_num * 2,
+                /*data_file_path=*/"", /*data_file_size=*/0, &reader_builder, mock_fs_,
+                prefetch_max_parallel_num, batch_size, prefetch_max_parallel_num * 2,
                 /*enable_adaptive_prefetch_strategy=*/false, executor_,
                 /*initialize_read_ranges=*/true, /*prefetch_cache_mode=*/PrefetchCacheMode::ALWAYS,
                 CacheConfig(), GetDefaultPool()));
@@ -323,8 +325,9 @@ TEST_F(PrefetchFileBatchReaderImplTest, TestReadWithLimits) {
     ASSERT_OK_AND_ASSIGN(
         auto reader,
         PrefetchFileBatchReaderImpl::Create(
-            /*data_file_path=*/"", &reader_builder, mock_fs_, prefetch_max_parallel_num, batch_size,
-            prefetch_max_parallel_num * 2, /*enable_adaptive_prefetch_strategy=*/false, executor_,
+            /*data_file_path=*/"", /*data_file_size=*/0, &reader_builder, mock_fs_,
+            prefetch_max_parallel_num, batch_size, prefetch_max_parallel_num * 2,
+            /*enable_adaptive_prefetch_strategy=*/false, executor_,
             /*initialize_read_ranges=*/true, /*prefetch_cache_mode=*/PrefetchCacheMode::ALWAYS,
             CacheConfig(), GetDefaultPool()));
     // simulate read limits, only read 8 batches
@@ -352,8 +355,9 @@ TEST_F(PrefetchFileBatchReaderImplTest, TestReadWithoutInitializeReadRanges) {
     ASSERT_OK_AND_ASSIGN(
         auto reader,
         PrefetchFileBatchReaderImpl::Create(
-            /*data_file_path=*/"", &reader_builder, mock_fs_, prefetch_max_parallel_num, batch_size,
-            prefetch_max_parallel_num * 2, /*enable_adaptive_prefetch_strategy=*/false, executor_,
+            /*data_file_path=*/"", /*data_file_size=*/0, &reader_builder, mock_fs_,
+            prefetch_max_parallel_num, batch_size, prefetch_max_parallel_num * 2,
+            /*enable_adaptive_prefetch_strategy=*/false, executor_,
             /*initialize_read_ranges=*/false, /*prefetch_cache_mode=*/PrefetchCacheMode::ALWAYS,
             CacheConfig(), GetDefaultPool()));
     // simulate read limits, only read 8 batches
@@ -428,8 +432,9 @@ TEST_F(PrefetchFileBatchReaderImplTest, RefreshReadRanges) {
     ASSERT_OK_AND_ASSIGN(
         auto reader,
         PrefetchFileBatchReaderImpl::Create(
-            /*data_file_path=*/"", &reader_builder, mock_fs_, prefetch_max_parallel_num, batch_size,
-            prefetch_max_parallel_num * 2, /*enable_adaptive_prefetch_strategy=*/false, executor_,
+            /*data_file_path=*/"", /*data_file_size=*/0, &reader_builder, mock_fs_,
+            prefetch_max_parallel_num, batch_size, prefetch_max_parallel_num * 2,
+            /*enable_adaptive_prefetch_strategy=*/false, executor_,
             /*initialize_read_ranges=*/false, /*prefetch_cache_mode=*/PrefetchCacheMode::ALWAYS,
             CacheConfig(), GetDefaultPool()));
     auto prefetch_reader = dynamic_cast<PrefetchFileBatchReaderImpl*>(reader.get());
@@ -457,7 +462,8 @@ TEST_F(PrefetchFileBatchReaderImplTest, RefreshReadRangesDisablePrefetchByAdapti
     ASSERT_OK_AND_ASSIGN(
         auto reader,
         PrefetchFileBatchReaderImpl::Create(
-            /*data_file_path=*/"", &reader_builder, mock_fs_, prefetch_max_parallel_num, batch_size,
+            /*data_file_path=*/"", /*data_file_size=*/0, &reader_builder, mock_fs_,
+            prefetch_max_parallel_num, batch_size,
             /*prefetch_batch_count=*/2,
             /*enable_adaptive_prefetch_strategy=*/true, executor_,
             /*initialize_read_ranges=*/true, /*prefetch_cache_mode=*/PrefetchCacheMode::ALWAYS,
@@ -474,8 +480,9 @@ TEST_F(PrefetchFileBatchReaderImplTest, SetReadRanges) {
     ASSERT_OK_AND_ASSIGN(
         auto reader,
         PrefetchFileBatchReaderImpl::Create(
-            /*data_file_path=*/"", &reader_builder, mock_fs_, prefetch_max_parallel_num, batch_size,
-            prefetch_max_parallel_num * 2, /*enable_adaptive_prefetch_strategy=*/false, executor_,
+            /*data_file_path=*/"", /*data_file_size=*/0, &reader_builder, mock_fs_,
+            prefetch_max_parallel_num, batch_size, prefetch_max_parallel_num * 2,
+            /*enable_adaptive_prefetch_strategy=*/false, executor_,
             /*initialize_read_ranges=*/false, /*prefetch_cache_mode=*/PrefetchCacheMode::ALWAYS,
             CacheConfig(), GetDefaultPool()));
     auto prefetch_reader = dynamic_cast<PrefetchFileBatchReaderImpl*>(reader.get());
@@ -517,8 +524,8 @@ TEST_F(PrefetchFileBatchReaderImplTest, SetReadRangesReturnErrorWhenPushDownFail
     ASSERT_OK_AND_ASSIGN(
         auto reader,
         PrefetchFileBatchReaderImpl::Create(
-            /*data_file_path=*/"", &reader_builder, mock_fs_, prefetch_max_parallel_num, batch_size,
-            prefetch_max_parallel_num * 2,
+            /*data_file_path=*/"", /*data_file_size=*/0, &reader_builder, mock_fs_,
+            prefetch_max_parallel_num, batch_size, prefetch_max_parallel_num * 2,
             /*enable_adaptive_prefetch_strategy=*/false, executor_,
             /*initialize_read_ranges=*/false, /*prefetch_cache_mode=*/PrefetchCacheMode::ALWAYS,
             CacheConfig(), GetDefaultPool()));
@@ -539,8 +546,8 @@ TEST_F(PrefetchFileBatchReaderImplTest, NeedInitCacheNeverMode) {
     ASSERT_OK_AND_ASSIGN(
         auto reader,
         PrefetchFileBatchReaderImpl::Create(
-            /*data_file_path=*/"", &reader_builder, mock_fs_, prefetch_max_parallel_num, batch_size,
-            prefetch_max_parallel_num * 2,
+            /*data_file_path=*/"", /*data_file_size=*/0, &reader_builder, mock_fs_,
+            prefetch_max_parallel_num, batch_size, prefetch_max_parallel_num * 2,
             /*enable_adaptive_prefetch_strategy=*/false, executor_,
             /*initialize_read_ranges=*/false, /*prefetch_cache_mode=*/PrefetchCacheMode::NEVER,
             CacheConfig(), GetDefaultPool()));
@@ -563,8 +570,8 @@ TEST_F(PrefetchFileBatchReaderImplTest, WorkloopSetReadStatusWhenCacheInitFailed
     ASSERT_OK_AND_ASSIGN(
         auto reader,
         PrefetchFileBatchReaderImpl::Create(
-            /*data_file_path=*/"", &reader_builder, mock_fs_, prefetch_max_parallel_num, batch_size,
-            prefetch_max_parallel_num * 2,
+            /*data_file_path=*/"", /*data_file_size=*/0, &reader_builder, mock_fs_,
+            prefetch_max_parallel_num, batch_size, prefetch_max_parallel_num * 2,
             /*enable_adaptive_prefetch_strategy=*/false, executor_,
             /*initialize_read_ranges=*/false, /*prefetch_cache_mode=*/PrefetchCacheMode::ALWAYS,
             invalid_cache_config, GetDefaultPool()));
@@ -584,8 +591,8 @@ TEST_F(PrefetchFileBatchReaderImplTest, DoReadBatchReturnOkWhenShutdown) {
     ASSERT_OK_AND_ASSIGN(
         auto reader,
         PrefetchFileBatchReaderImpl::Create(
-            /*data_file_path=*/"", &reader_builder, mock_fs_, prefetch_max_parallel_num, batch_size,
-            prefetch_max_parallel_num * 2,
+            /*data_file_path=*/"", /*data_file_size=*/0, &reader_builder, mock_fs_,
+            prefetch_max_parallel_num, batch_size, prefetch_max_parallel_num * 2,
             /*enable_adaptive_prefetch_strategy=*/false, executor_,
             /*initialize_read_ranges=*/false, /*prefetch_cache_mode=*/PrefetchCacheMode::ALWAYS,
             CacheConfig(), GetDefaultPool()));
@@ -603,8 +610,8 @@ TEST_F(PrefetchFileBatchReaderImplTest, DoReadBatchReturnOkWhenNoCurrentReadRang
     ASSERT_OK_AND_ASSIGN(
         auto reader,
         PrefetchFileBatchReaderImpl::Create(
-            /*data_file_path=*/"", &reader_builder, mock_fs_, prefetch_max_parallel_num, batch_size,
-            prefetch_max_parallel_num * 2,
+            /*data_file_path=*/"", /*data_file_size=*/0, &reader_builder, mock_fs_,
+            prefetch_max_parallel_num, batch_size, prefetch_max_parallel_num * 2,
             /*enable_adaptive_prefetch_strategy=*/false, executor_,
             /*initialize_read_ranges=*/false, /*prefetch_cache_mode=*/PrefetchCacheMode::ALWAYS,
             CacheConfig(), GetDefaultPool()));
@@ -622,8 +629,9 @@ TEST_F(PrefetchFileBatchReaderImplTest, TestReadWithLargeBatchSize) {
     ASSERT_OK_AND_ASSIGN(
         auto reader,
         PrefetchFileBatchReaderImpl::Create(
-            /*data_file_path=*/"", &reader_builder, mock_fs_, prefetch_max_parallel_num, batch_size,
-            prefetch_max_parallel_num * 2, /*enable_adaptive_prefetch_strategy=*/false, executor_,
+            /*data_file_path=*/"", /*data_file_size=*/0, &reader_builder, mock_fs_,
+            prefetch_max_parallel_num, batch_size, prefetch_max_parallel_num * 2,
+            /*enable_adaptive_prefetch_strategy=*/false, executor_,
             /*initialize_read_ranges=*/true, /*prefetch_cache_mode=*/PrefetchCacheMode::ALWAYS,
             CacheConfig(), GetDefaultPool()));
     ASSERT_NOK(reader->GetPreviousBatchFileRowId(0));
@@ -642,8 +650,9 @@ TEST_F(PrefetchFileBatchReaderImplTest, TestPartialReaderSuccessRead) {
     ASSERT_OK_AND_ASSIGN(
         auto reader,
         PrefetchFileBatchReaderImpl::Create(
-            /*data_file_path=*/"", &reader_builder, mock_fs_, prefetch_max_parallel_num, batch_size,
-            prefetch_max_parallel_num, /*enable_adaptive_prefetch_strategy=*/false, executor_,
+            /*data_file_path=*/"", /*data_file_size=*/0, &reader_builder, mock_fs_,
+            prefetch_max_parallel_num, batch_size, prefetch_max_parallel_num,
+            /*enable_adaptive_prefetch_strategy=*/false, executor_,
             /*initialize_read_ranges=*/true, /*prefetch_cache_mode=*/PrefetchCacheMode::ALWAYS,
             CacheConfig(), GetDefaultPool()));
     auto prefetch_reader = dynamic_cast<PrefetchFileBatchReaderImpl*>(reader.get());
@@ -687,8 +696,9 @@ TEST_F(PrefetchFileBatchReaderImplTest, TestAllReaderFailedWithIOError) {
     ASSERT_OK_AND_ASSIGN(
         auto reader,
         PrefetchFileBatchReaderImpl::Create(
-            /*data_file_path=*/"", &reader_builder, mock_fs_, prefetch_max_parallel_num, batch_size,
-            prefetch_max_parallel_num * 2, /*enable_adaptive_prefetch_strategy=*/false, executor_,
+            /*data_file_path=*/"", /*data_file_size=*/0, &reader_builder, mock_fs_,
+            prefetch_max_parallel_num, batch_size, prefetch_max_parallel_num * 2,
+            /*enable_adaptive_prefetch_strategy=*/false, executor_,
             /*initialize_read_ranges=*/true, /*prefetch_cache_mode=*/PrefetchCacheMode::ALWAYS,
             CacheConfig(), GetDefaultPool()));
 
@@ -722,8 +732,9 @@ TEST_F(PrefetchFileBatchReaderImplTest, TestPrefetchWithEmptyData) {
     ASSERT_OK_AND_ASSIGN(
         auto reader,
         PrefetchFileBatchReaderImpl::Create(
-            /*data_file_path=*/"", &reader_builder, mock_fs_, prefetch_max_parallel_num, batch_size,
-            prefetch_max_parallel_num * 2, /*enable_adaptive_prefetch_strategy=*/false, executor_,
+            /*data_file_path=*/"", /*data_file_size=*/0, &reader_builder, mock_fs_,
+            prefetch_max_parallel_num, batch_size, prefetch_max_parallel_num * 2,
+            /*enable_adaptive_prefetch_strategy=*/false, executor_,
             /*initialize_read_ranges=*/true, /*prefetch_cache_mode=*/PrefetchCacheMode::ALWAYS,
             CacheConfig(), GetDefaultPool()));
     ASSERT_NOK(reader->GetPreviousBatchFileRowId(0));
@@ -741,8 +752,9 @@ TEST_F(PrefetchFileBatchReaderImplTest, TestCallNextBatchAfterReadingEof) {
     ASSERT_OK_AND_ASSIGN(
         auto reader,
         PrefetchFileBatchReaderImpl::Create(
-            /*data_file_path=*/"", &reader_builder, mock_fs_, prefetch_max_parallel_num, batch_size,
-            prefetch_max_parallel_num * 2, /*enable_adaptive_prefetch_strategy=*/false, executor_,
+            /*data_file_path=*/"", /*data_file_size=*/0, &reader_builder, mock_fs_,
+            prefetch_max_parallel_num, batch_size, prefetch_max_parallel_num * 2,
+            /*enable_adaptive_prefetch_strategy=*/false, executor_,
             /*initialize_read_ranges=*/true, /*prefetch_cache_mode=*/PrefetchCacheMode::ALWAYS,
             CacheConfig(), GetDefaultPool()));
     ASSERT_NOK(reader->GetPreviousBatchFileRowId(0));
@@ -766,8 +778,9 @@ TEST_F(PrefetchFileBatchReaderImplTest, TestCreateReaderWithoutNextBatch) {
     ASSERT_OK_AND_ASSIGN(
         auto reader,
         PrefetchFileBatchReaderImpl::Create(
-            /*data_file_path=*/"", &reader_builder, mock_fs_, prefetch_max_parallel_num, batch_size,
-            prefetch_max_parallel_num * 2, /*enable_adaptive_prefetch_strategy=*/false, executor_,
+            /*data_file_path=*/"", /*data_file_size=*/0, &reader_builder, mock_fs_,
+            prefetch_max_parallel_num, batch_size, prefetch_max_parallel_num * 2,
+            /*enable_adaptive_prefetch_strategy=*/false, executor_,
             /*initialize_read_ranges=*/true, /*prefetch_cache_mode=*/PrefetchCacheMode::ALWAYS,
             CacheConfig(), GetDefaultPool()));
 }
@@ -780,7 +793,7 @@ TEST_F(PrefetchFileBatchReaderImplTest, TestInvalidCase) {
     MockFormatReaderBuilder reader_builder(data_array, data_type_, batch_size);
     {
         ASSERT_NOK(PrefetchFileBatchReaderImpl::Create(
-            data_file_path, &reader_builder, mock_fs_,
+            data_file_path, /*data_file_size=*/0, &reader_builder, mock_fs_,
             /*prefetch_max_parallel_num=*/0, batch_size, 2,
             /*enable_adaptive_prefetch_strategy=*/false, executor_,
             /*initialize_read_ranges=*/true, /*prefetch_cache_mode=*/PrefetchCacheMode::ALWAYS,
@@ -788,30 +801,31 @@ TEST_F(PrefetchFileBatchReaderImplTest, TestInvalidCase) {
     }
     {
         ASSERT_NOK(PrefetchFileBatchReaderImpl::Create(
-            data_file_path, &reader_builder, mock_fs_, prefetch_max_parallel_num, /*batch_size=*/-1,
-            prefetch_max_parallel_num * 2, /*enable_adaptive_prefetch_strategy=*/false, executor_,
+            data_file_path, /*data_file_size=*/0, &reader_builder, mock_fs_,
+            prefetch_max_parallel_num, /*batch_size=*/-1, prefetch_max_parallel_num * 2,
+            /*enable_adaptive_prefetch_strategy=*/false, executor_,
             /*initialize_read_ranges=*/true, /*prefetch_cache_mode=*/PrefetchCacheMode::ALWAYS,
             CacheConfig(), GetDefaultPool()));
     }
     {
         ASSERT_NOK(PrefetchFileBatchReaderImpl::Create(
-            data_file_path, &reader_builder, mock_fs_, prefetch_max_parallel_num, batch_size,
-            prefetch_max_parallel_num * 2,
+            data_file_path, /*data_file_size=*/0, &reader_builder, mock_fs_,
+            prefetch_max_parallel_num, batch_size, prefetch_max_parallel_num * 2,
             /*enable_adaptive_prefetch_strategy=*/false,
             /*executor=*/nullptr, /*initialize_read_ranges=*/true,
             /*prefetch_cache_mode=*/PrefetchCacheMode::ALWAYS, CacheConfig(), GetDefaultPool()));
     }
     {
         ASSERT_NOK(PrefetchFileBatchReaderImpl::Create(
-            data_file_path, /*reader_builder=*/nullptr, mock_fs_, prefetch_max_parallel_num,
-            batch_size, prefetch_max_parallel_num * 2,
+            data_file_path, /*data_file_size=*/0, /*reader_builder=*/nullptr, mock_fs_,
+            prefetch_max_parallel_num, batch_size, prefetch_max_parallel_num * 2,
             /*enable_adaptive_prefetch_strategy=*/false, executor_,
             /*initialize_read_ranges=*/true, /*prefetch_cache_mode=*/PrefetchCacheMode::ALWAYS,
             CacheConfig(), GetDefaultPool()));
     }
     {
         ASSERT_NOK(PrefetchFileBatchReaderImpl::Create(
-            data_file_path, &reader_builder,
+            data_file_path, /*data_file_size=*/0, &reader_builder,
             /*fs=*/nullptr, prefetch_max_parallel_num, batch_size, prefetch_max_parallel_num * 2,
             /*enable_adaptive_prefetch_strategy=*/false, executor_,
             /*initialize_read_ranges=*/true, /*prefetch_cache_mode=*/PrefetchCacheMode::ALWAYS,
@@ -821,8 +835,8 @@ TEST_F(PrefetchFileBatchReaderImplTest, TestInvalidCase) {
         ASSERT_OK_AND_ASSIGN(
             auto reader,
             PrefetchFileBatchReaderImpl::Create(
-                data_file_path, &reader_builder, mock_fs_, prefetch_max_parallel_num, batch_size,
-                prefetch_max_parallel_num * 2,
+                data_file_path, /*data_file_size=*/0, &reader_builder, mock_fs_,
+                prefetch_max_parallel_num, batch_size, prefetch_max_parallel_num * 2,
                 /*enable_adaptive_prefetch_strategy=*/false, executor_,
                 /*initialize_read_ranges=*/true, /*prefetch_cache_mode=*/PrefetchCacheMode::ALWAYS,
                 CacheConfig(), GetDefaultPool()));
@@ -910,14 +924,14 @@ TEST_F(PrefetchFileBatchReaderImplTest, TestPrefetchWithBitmap) {
     MockFormatReaderBuilder reader_builder(data_array, data_type_, bitmap,
                                            /*read_batch_size=*/100);
     int32_t prefetch_max_parallel_num = 3;
-    ASSERT_OK_AND_ASSIGN(
-        auto reader,
-        PrefetchFileBatchReaderImpl::Create(
-            /*data_file_path=*/"", &reader_builder, mock_fs_, prefetch_max_parallel_num,
-            /*batch_size=*/100, prefetch_max_parallel_num * 2,
-            /*enable_adaptive_prefetch_strategy=*/false, executor_,
-            /*initialize_read_ranges=*/true, /*prefetch_cache_mode=*/PrefetchCacheMode::ALWAYS,
-            CacheConfig(), GetDefaultPool()));
+    ASSERT_OK_AND_ASSIGN(auto reader, PrefetchFileBatchReaderImpl::Create(
+                                          /*data_file_path=*/"", /*data_file_size=*/0,
+                                          &reader_builder, mock_fs_, prefetch_max_parallel_num,
+                                          /*batch_size=*/100, prefetch_max_parallel_num * 2,
+                                          /*enable_adaptive_prefetch_strategy=*/false, executor_,
+                                          /*initialize_read_ranges=*/true,
+                                          /*prefetch_cache_mode=*/PrefetchCacheMode::ALWAYS,
+                                          CacheConfig(), GetDefaultPool()));
     ASSERT_OK_AND_ASSIGN(auto result_chunk_array, ReadResultCollector::CollectResult(reader.get()));
 
     ASSERT_OK_AND_ASSIGN(auto data_batch, ReadResultCollector::GetReadBatch(data_array));
diff --git a/src/paimon/core/deletionvectors/apply_deletion_vector_batch_reader_test.cpp b/src/paimon/core/deletionvectors/apply_deletion_vector_batch_reader_test.cpp
index daa42d37c..1fca5f374 100644
--- a/src/paimon/core/deletionvectors/apply_deletion_vector_batch_reader_test.cpp
+++ b/src/paimon/core/deletionvectors/apply_deletion_vector_batch_reader_test.cpp
@@ -83,8 +83,8 @@ class ApplyDeletionVectorBatchReaderTest : public ::testing::Test,
                 ASSERT_OK_AND_ASSIGN(
                     file_batch_reader,
                     PrefetchFileBatchReaderImpl::Create(
-                        /*data_file_path=*/"DUMMY", &reader_builder, fs_, prefetch_batch_count,
-                        batch_size, prefetch_batch_count * 2,
+                        /*data_file_path=*/"DUMMY", /*data_file_size=*/0, &reader_builder, fs_,
+                        prefetch_batch_count, batch_size, prefetch_batch_count * 2,
                         /*enable_adaptive_prefetch_strategy=*/false, executor_,
                         /*initialize_read_ranges=*/true,
                         /*prefetch_cache_mode=*/PrefetchCacheMode::ALWAYS, CacheConfig(), pool_));
diff --git a/src/paimon/core/deletionvectors/deletion_vectors_index_file.cpp b/src/paimon/core/deletionvectors/deletion_vectors_index_file.cpp
index 7124ee14f..8743b43ab 100644
--- a/src/paimon/core/deletionvectors/deletion_vectors_index_file.cpp
+++ b/src/paimon/core/deletionvectors/deletion_vectors_index_file.cpp
@@ -50,6 +50,7 @@ DeletionVectorsIndexFile::ReadAllDeletionVectors(
 
     std::map<std::string, std::shared_ptr<DeletionVector>> deletion_vectors;
     std::string file_path = path_factory_->ToPath(file_meta);
+    // TODO(mrdrivingduck): Use file_meta->FileSize() to avoid an object-store metadata request.
     PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<InputStream> input_stream, fs_->Open(file_path));
     auto data_input_stream = std::make_shared<DataInputStream>(input_stream);
     PAIMON_RETURN_NOT_OK(CheckVersion(data_input_stream));
diff --git a/src/paimon/core/operation/abstract_split_read.cpp b/src/paimon/core/operation/abstract_split_read.cpp
index 0850d943f..3a42f7da9 100644
--- a/src/paimon/core/operation/abstract_split_read.cpp
+++ b/src/paimon/core/operation/abstract_split_read.cpp
@@ -144,13 +144,13 @@ Result<std::unique_ptr<ReaderBuilder>> AbstractSplitRead::PrepareReaderBuilder(
 
 Result<std::unique_ptr<FileBatchReader>> AbstractSplitRead::CreateFileBatchReader(
     const std::string& file_format_identifier, const std::string& data_file_path,
-    const ReaderBuilder* reader_builder) const {
+    int64_t data_file_size, const ReaderBuilder* reader_builder) const {
     if (context_->EnablePrefetch() && file_format_identifier != "blob" &&
         file_format_identifier != "avro") {
         PAIMON_ASSIGN_OR_RAISE(
             std::unique_ptr<PrefetchFileBatchReaderImpl> prefetch_reader,
             PrefetchFileBatchReaderImpl::Create(
-                data_file_path, reader_builder, options_.GetFileSystem(),
+                data_file_path, data_file_size, reader_builder, options_.GetFileSystem(),
                 context_->GetPrefetchMaxParallelNum(), options_.GetReadBatchSize(),
                 context_->GetPrefetchBatchCount(), options_.EnableAdaptivePrefetchStrategy(),
                 executor_,
@@ -158,8 +158,9 @@ Result<std::unique_ptr<FileBatchReader>> AbstractSplitRead::CreateFileBatchReade
                 context_->GetCacheConfig(), pool_));
         return std::make_unique<DelegatingPrefetchReader>(std::move(prefetch_reader));
     } else {
-        PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<InputStream> input_stream,
-                               options_.GetFileSystem()->Open(data_file_path));
+        PAIMON_ASSIGN_OR_RAISE(
+            std::shared_ptr<InputStream> input_stream,
+            options_.GetFileSystem()->Open(FileStatus(data_file_path, data_file_size)));
         return reader_builder->Build(input_stream);
     }
 }
@@ -204,9 +205,9 @@ Result<std::unique_ptr<FileBatchReader>> AbstractSplitRead::CreateFieldMappingRe
         field_mapping->non_partition_info.non_partition_data_schema);
 
     PAIMON_ASSIGN_OR_RAISE(std::string file_format_identifier, file_meta->FileFormat());
-    PAIMON_ASSIGN_OR_RAISE(
-        std::unique_ptr<FileBatchReader> file_reader,
-        CreateFileBatchReader(file_format_identifier, data_file_path, reader_builder));
+    PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<FileBatchReader> file_reader,
+                           CreateFileBatchReader(file_format_identifier, data_file_path,
+                                                 file_meta->file_size, reader_builder));
     std::set<int32_t> skip_map_selected_keys_filter_field_ids;
     if (file_format_identifier != "blob") {
         std::pair<std::unique_ptr<FileBatchReader>, std::set<int32_t>> shared_shredding_result;
diff --git a/src/paimon/core/operation/abstract_split_read.h b/src/paimon/core/operation/abstract_split_read.h
index ea3f9070b..27349fec1 100644
--- a/src/paimon/core/operation/abstract_split_read.h
+++ b/src/paimon/core/operation/abstract_split_read.h
@@ -107,7 +107,7 @@ class AbstractSplitRead : public SplitRead {
 
     Result<std::unique_ptr<FileBatchReader>> CreateFileBatchReader(
         const std::string& file_format_identifier, const std::string& data_file_path,
-        const ReaderBuilder* reader_builder) const;
+        int64_t data_file_size, const ReaderBuilder* reader_builder) const;
 
     // return nullptr if data file is skipped by index or dv
     Result<std::unique_ptr<FileBatchReader>> CreateFieldMappingReader(
diff --git a/src/paimon/fs/jindo/jindo_file_system.h b/src/paimon/fs/jindo/jindo_file_system.h
index 0ba8990ba..89b3081e2 100644
--- a/src/paimon/fs/jindo/jindo_file_system.h
+++ b/src/paimon/fs/jindo/jindo_file_system.h
@@ -38,6 +38,8 @@ class JindoFileSystem : public FileSystem {
     explicit JindoFileSystem(std::unique_ptr<JdoFileSystem>&& fs);
     ~JindoFileSystem() override = default;
 
+    using FileSystem::Open;
+
     Result<std::unique_ptr<InputStream>> Open(const std::string& path) const override;
     Result<std::unique_ptr<OutputStream>> Create(const std::string& path,
                                                  bool overwrite) const override;
diff --git a/src/paimon/fs/local/local_file_system.h b/src/paimon/fs/local/local_file_system.h
index 748f24635..12ee5646a 100644
--- a/src/paimon/fs/local/local_file_system.h
+++ b/src/paimon/fs/local/local_file_system.h
@@ -39,6 +39,8 @@ class LocalFileSystem : public FileSystem {
     LocalFileSystem() = default;
     ~LocalFileSystem() override = default;
 
+    using FileSystem::Open;
+
     Result<std::unique_ptr<InputStream>> Open(const std::string& path) const override;
     Result<std::unique_ptr<OutputStream>> Create(const std::string& path,
                                                  bool overwrite) const override;
diff --git a/src/paimon/testing/mock/mock_file_system.h b/src/paimon/testing/mock/mock_file_system.h
index 484ed1b39..a691c3de1 100644
--- a/src/paimon/testing/mock/mock_file_system.h
+++ b/src/paimon/testing/mock/mock_file_system.h
@@ -103,6 +103,8 @@ class MockFileSystem : public FileSystem {
     MockFileSystem() = default;
     ~MockFileSystem() override = default;
 
+    using FileSystem::Open;
+
     Result<std::unique_ptr<InputStream>> Open(const std::string& path) const override {
         return std::make_unique<MockInputStream>();
     }