perf(fs): avoid redundant object-store metadata requests for known-size files

include/paimon/fs/file_system.h

@@ -193,6 +193,15 @@ 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 file for reading with a known file size.
+    ///
+    /// File systems that can use the size to avoid metadata requests may override this method.
+    /// The default implementation ignores the size and opens the file normally.
+    virtual Result<std::unique_ptr<InputStream>> Open(const std::string& path,
+                                                      int64_t /*file_size*/) const {
+        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/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

@@ -62,6 +62,19 @@ Result<std::unique_ptr<PrefetchFileBatchReaderImpl>> PrefetchFileBatchReaderImpl
     const std::shared_ptr<Executor>& executor, bool initialize_read_ranges,
     PrefetchCacheMode prefetch_cache_mode, const CacheConfig& cache_config,
     const std::shared_ptr<MemoryPool>& pool) {
+    return Create(data_file_path, /*data_file_size=*/-1, reader_builder, fs,
+                  prefetch_max_parallel_num, batch_size, prefetch_batch_count,
+                  enable_adaptive_prefetch_strategy, executor, initialize_read_ranges,
+                  prefetch_cache_mode, cache_config, pool);
+}
+
+Result<std::unique_ptr<PrefetchFileBatchReaderImpl>> PrefetchFileBatchReaderImpl::Create(
+    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) {
     if (prefetch_max_parallel_num == 0) {
         return Status::Invalid("prefetch max parallel num should be greater than 0.");
     }
@@ -83,16 +96,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

@@ -55,6 +55,14 @@ class Metrics;
 
 class PrefetchFileBatchReaderImpl : public PrefetchFileBatchReader {
  public:
+    static Result<std::unique_ptr<PrefetchFileBatchReaderImpl>> Create(
+        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);
+
     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,

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_,
@@ -159,7 +159,7 @@ Result<std::unique_ptr<FileBatchReader>> AbstractSplitRead::CreateFileBatchReade
         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));
+                               options_.GetFileSystem()->Open(data_file_path, data_file_size));
         return reader_builder->Build(input_stream);
     }
 }
@@ -204,9 +204,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;

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(