perf(fs): avoid redundant object-store metadata requests

Reuse file sizes already obtained during read planning, so
object-store-backed reads do not need an extra metadata request.

Clarify known-size opening semantics and keep both file-system opening
overloads available on concrete implementations.

Record a follow-up to reuse known deletion-vector index sizes.

Co-authored-by: GPT-5.6 Terra <codex@users.noreply.github.com>

Author: mrdrivingduck

include/paimon/fs/file_system.h

@@ -193,6 +193,25 @@ 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 a known file size.
+    /// @param path The file path to open.
+    /// @param file_size The exact size of the file in bytes. It 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_size` to skip metadata requests. The caller must
+    ///       provide the size of an existing regular file and must not expect this method to
+    ///       validate the path, file type, or size. A stale or incorrect size, 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 std::string& path,
+                                                      int64_t file_size) const {
+        if (file_size < 0) {
+            return Status::Invalid("file size must be non-negative");
+        }
+        return Open(path);
+    }
+
     /// 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.

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

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));
+
+    ASSERT_OK_AND_ASSIGN(auto input_stream,
+                         fs_->Open(file_path, static_cast<int64_t>(content.size())));
+    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(file_path, /*file_size=*/-1).status().IsInvalid());
+}
+
 TEST_P(FileSystemTest, TestWriteMultipleTimes) {
     std::vector<std::string> content_vec = {"abc", "defg", "hi", "j", "k"};
     std::string content = "abcdefghijk";

src/paimon/common/fs/object_store_file_system.cpp

@@ -388,6 +388,17 @@ Result<std::unique_ptr<InputStream>> ObjectStoreFileSystem::Open(const std::stri
                                                     ToUri(object_path), metadata.value().size);
 }
 
+Result<std::unique_ptr<InputStream>> ObjectStoreFileSystem::Open(const std::string& path,
+                                                                 int64_t file_size) const {
+    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));

src/paimon/common/fs/object_store_file_system.h

@@ -85,6 +85,8 @@ 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 std::string& path,
+                                              int64_t file_size) 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;

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

src/paimon/common/fs/resolving_file_system.cpp

@@ -79,6 +79,12 @@ Result<std::unique_ptr<InputStream>> ResolvingFileSystem::Open(const std::string
     return fs->Open(path);
 }
 
+Result<std::unique_ptr<InputStream>> ResolvingFileSystem::Open(const std::string& path,
+                                                               int64_t file_size) const {
+    PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<FileSystem> fs, GetRealFileSystem(path));
+    return fs->Open(path, file_size);
+}
+
 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));

src/paimon/common/fs/resolving_file_system.h

@@ -41,6 +41,8 @@ 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 std::string& path,
+                                              int64_t file_size) const override;
     Result<std::unique_ptr<OutputStream>> Create(const std::string& path,
                                                  bool overwrite) const override;
 

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,16 +83,17 @@ 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(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,
+                              [&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(data_file_path));
+                                                         fs->Open(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);

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;