perf(read): open data file streams and build readers concurrently

src/paimon/CMakeLists.txt

@@ -63,6 +63,7 @@ set(PAIMON_COMMON_SRCS
     common/data/variant/variant_type_utils.cpp
     common/defs.cpp
     common/executor/executor.cpp
+    common/executor/reader_build_executor.cpp
     common/factories/singleton.cpp
     common/factories/io_hook.cpp
     common/factories/factory_creator.cpp

src/paimon/common/executor/reader_build_executor.cpp

@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "paimon/common/executor/reader_build_executor.h"
+
+#include <algorithm>
+#include <mutex>
+#include <thread>
+
+#include "paimon/result.h"
+
+namespace paimon {
+
+std::shared_ptr<Executor> GetReaderBuildExecutor() {
+    static std::shared_ptr<Executor> executor;
+    static std::once_flag once;
+    std::call_once(once, []() {
+        const uint32_t hardware_threads = std::thread::hardware_concurrency();
+        const uint32_t upper_bound = hardware_threads > 0 ? hardware_threads : 1;
+        const uint32_t threads = std::clamp(kReaderBuildMaxParallelNum, 1u, upper_bound);
+        // CreateDefaultExecutor only rejects a zero thread count, which the clamp rules out.
+        executor = CreateDefaultExecutor(threads).value();
+    });
+    return executor;
+}
+
+}  // namespace paimon

src/paimon/common/executor/reader_build_executor.h

@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#pragma once
+
+#include <cstdint>
+#include <memory>
+
+#include "paimon/executor.h"
+
+namespace paimon {
+
+/// Number of data file readers built in parallel within one split, and the size of the dedicated
+/// reader build pool (clamped to hardware concurrency). Building a reader mostly waits on remote
+/// I/O, so a small fixed pool overlaps those waits without unbounded thread growth.
+inline constexpr uint32_t kReaderBuildMaxParallelNum = 4;
+
+/// Returns the process-wide thread pool dedicated to building data file readers.
+///
+/// This pool is kept separate from the executor of a read context on purpose. That executor is
+/// handed to the readers this pool builds, and they use it for async prefetch once reading
+/// starts; running the per-file builds on a dedicated pool keeps the blocking build phase from
+/// contending with it, and keeps the pool a single reused instance instead of a fresh one per
+/// split.
+///
+/// The pool is created once, on first use, sized to `kReaderBuildMaxParallelNum` clamped to
+/// hardware concurrency. Callers that build readers serially never call this and therefore never
+/// pay for any thread.
+std::shared_ptr<Executor> GetReaderBuildExecutor();
+
+}  // namespace paimon

src/paimon/common/reader/prefetch_file_batch_reader_impl.cpp

@@ -38,6 +38,7 @@
 #include "paimon/common/utils/scope_guard.h"
 #include "paimon/format/reader_builder.h"
 #include "paimon/fs/file_system.h"
+#include "paimon/memory/memory_pool.h"
 
 namespace arrow {
 class Schema;
@@ -233,10 +234,44 @@ Result<std::unique_ptr<PrefetchFileBatchReaderImpl>> PrefetchFileBatchReaderImpl
     if (enable_io_metrics) {
         io_metrics = std::make_shared<PrefetchIoMetricsState>();
     }
+    // Opening a file mostly waits on remote I/O, so the streams are all opened concurrently
+    // instead of opening the cache's stream first and paying its round trip before the readers'
+    // begin. Both waves run on the read context's `executor`, which is sized to the prefetch
+    // parallelism, so no separate pool is spun up and torn down per file and the open/build
+    // concurrency follows the configured parallelism instead of a fixed default. That reuse is
+    // safe: neither wave's tasks wait on one another or on `executor`, and Create never runs on
+    // an `executor` worker, so the CollectAll calls below only ever wait on tasks that progress.
+    // The build has to be a second wave rather than folded into the first: each build task wraps
+    // its stream in the cache, and a task waiting on the cache would block a worker while the
+    // cache's own open is still queued behind it. Wave one only opens, so its tasks never wait on
+    // one another.
+    const uint32_t open_count =
+        read_ahead_cache_enabled ? prefetch_max_parallel_num + 1 : prefetch_max_parallel_num;
+    std::vector<std::future<Result<std::unique_ptr<InputStream>>>> open_futures;
+    open_futures.reserve(open_count);
+    for (uint32_t i = 0; i < open_count; i++) {
+        open_futures.push_back(
+            Via(executor.get(),
+                [&fs, &data_file_path, data_file_size]() -> Result<std::unique_ptr<InputStream>> {
+                    return fs->Open(FileStatus(data_file_path, data_file_size));
+                }));
+    }
+    // The tasks only reference locals of this frame, which stay alive because both CollectAll
+    // calls below drain every future before returning.
+    std::vector<Result<std::unique_ptr<InputStream>>> opened_streams = CollectAll(open_futures);
+    std::vector<std::unique_ptr<InputStream>> streams;
+    streams.reserve(opened_streams.size());
+    for (auto& opened_stream : opened_streams) {
+        if (!opened_stream.ok()) {
+            return opened_stream.status();
+        }
+        streams.push_back(std::move(opened_stream).value());
+    }
+
+    size_t next_stream = 0;
     std::shared_ptr<ReadAheadCache> cache;
     if (read_ahead_cache_enabled) {
-        PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<InputStream> input_stream,
-                               fs->Open(FileStatus(data_file_path, data_file_size)));
+        std::shared_ptr<InputStream> input_stream = std::move(streams[next_stream++]);
         if (io_metrics) {
             input_stream = std::make_shared<MetricsInputStream>(input_stream, io_metrics);
         }
@@ -246,22 +281,24 @@ Result<std::unique_ptr<PrefetchFileBatchReaderImpl>> PrefetchFileBatchReaderImpl
         cache = std::make_shared<ReadAheadCache>(input_stream, cache_config,
                                                  static_cast<uint64_t>(data_file_size), pool);
     }
+    // Wave two builds the readers, which reads each file's footer and so still needs the cache
+    // above to exist. The builds are concurrent, like the opens were.
     std::vector<std::future<Result<std::unique_ptr<FileBatchReader>>>> futures;
+    futures.reserve(prefetch_max_parallel_num);
     for (uint32_t i = 0; i < prefetch_max_parallel_num; i++) {
-        futures.push_back(Via(
-            executor.get(),
-            [&fs, &data_file_path, data_file_size, &reader_builder, &cache,
-             io_metrics]() -> Result<std::unique_ptr<FileBatchReader>> {
-                PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<InputStream> input_stream,
-                                       fs->Open(FileStatus(data_file_path, data_file_size)));
-                if (io_metrics) {
-                    input_stream =
-                        std::make_unique<MetricsInputStream>(std::move(input_stream), io_metrics);
-                }
-                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(),
+                              [&reader_builder, &cache, &streams, i, next_stream,
+                               io_metrics]() -> Result<std::unique_ptr<FileBatchReader>> {
+                                  std::unique_ptr<InputStream> input_stream =
+                                      std::move(streams[next_stream + i]);
+                                  if (io_metrics) {
+                                      input_stream = std::make_unique<MetricsInputStream>(
+                                          std::move(input_stream), io_metrics);
+                                  }
+                                  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)) {

src/paimon/common/reader/prefetch_file_batch_reader_impl_test.cpp

@@ -19,7 +19,10 @@
 #include "paimon/common/reader/prefetch_file_batch_reader_impl.h"
 
 #include <atomic>
+#include <chrono>
+#include <condition_variable>
 #include <limits>
+#include <mutex>
 #include <set>
 
 #include "arrow/compute/api.h"
@@ -130,6 +133,37 @@ class FailingFileSystem : public MockFileSystem {
     }
 };
 
+/// Blocks every Open() until `opens_needed` of them are in flight at once, so a caller that
+/// opens its streams one at a time never reaches the barrier and fails instead of passing
+/// silently. The wait is bounded so that failure is reported rather than hanging the test.
+class BarrierFileSystem : public MockFileSystem {
+ public:
+    explicit BarrierFileSystem(uint32_t opens_needed) : opens_needed_(opens_needed) {}
+
+    Result<std::unique_ptr<InputStream>> Open(const std::string& path) const override {
+        const uint32_t arrived = ++inflight_opens_;
+        std::unique_lock<std::mutex> lock(mutex_);
+        if (!condition_.wait_for(lock, kBarrierTimeout,
+                                 [this] { return inflight_opens_ >= opens_needed_; })) {
+            return Status::IOError(fmt::format(
+                "only {} of {} opens were in flight at once, so they are not concurrent", arrived,
+                opens_needed_));
+        }
+        condition_.notify_all();
+        return std::make_unique<MockInputStream>();
+    }
+
+ private:
+    // Generous so a loaded machine does not turn this into a flaky test; a regression makes the
+    // opens serial, which no wait ever satisfies.
+    static constexpr std::chrono::seconds kBarrierTimeout{60};
+
+    const uint32_t opens_needed_;
+    mutable std::atomic<uint32_t> inflight_opens_{0};
+    mutable std::mutex mutex_;
+    mutable std::condition_variable condition_;
+};
+
 class IoReadingMockFileBatchReader : public MockFileBatchReader {
  public:
     IoReadingMockFileBatchReader(const std::shared_ptr<arrow::Array>& data,
@@ -1377,4 +1411,31 @@ TEST_P(PrefetchFileBatchReaderImplTest, TestWarmupLevelDecoded) {
     ASSERT_GT(produced_batches, 0);
 }
 
+// The cache's stream used to be opened before the readers' streams were dispatched, which
+// serialized one remote round trip ahead of all of them. They are opened together now, and this
+// pins that down: the barrier needs every one of them in flight at once, so opening the cache's
+// stream first would leave the others waiting on a round trip that has not been started.
+TEST_F(PrefetchFileBatchReaderImplTest, TestCreateOpensAllStreamsConcurrently) {
+    auto data_array = PrepareArray(101);
+    const int32_t batch_size = 10;
+    const int32_t prefetch_max_parallel_num = 3;
+    // One stream for the cache plus one per reader.
+    const uint32_t opens_needed = prefetch_max_parallel_num + 1;
+    auto barrier_fs = std::make_shared<BarrierFileSystem>(opens_needed);
+    MockFormatReaderBuilder reader_builder(data_array, data_type_, batch_size);
+    // Unlike executor_ above, this one has a thread per open: with fewer threads than opens the
+    // barrier could not be reached however concurrent the dispatch is.
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<Executor> executor, CreateDefaultExecutor(opens_needed));
+
+    ASSERT_OK_AND_ASSIGN(
+        auto reader,
+        PrefetchFileBatchReaderImpl::Create(
+            /*data_file_path=*/"", /*data_file_size=*/0, &reader_builder, barrier_fs,
+            prefetch_max_parallel_num, batch_size, prefetch_max_parallel_num * 2,
+            /*enable_adaptive_prefetch_strategy=*/false, executor,
+            /*initialize_read_ranges=*/true, /*read_ahead_cache_enabled=*/true, CacheConfig(),
+            /*enable_io_metrics=*/false, WarmupLevel::DECODED, pool_, GetArrowPool(pool_)));
+    ASSERT_NE(reader, nullptr);
+}
+
 }  // namespace paimon::test

src/paimon/core/operation/abstract_split_read.cpp

@@ -18,8 +18,10 @@
 
 #include "paimon/core/operation/abstract_split_read.h"
 
+#include <algorithm>
 #include <cassert>
 #include <cstddef>
+#include <future>
 #include <map>
 #include <set>
 #include <utility>
@@ -33,6 +35,8 @@
 #include "paimon/common/data/shredding/shredding_file_reader.h"
 #include "paimon/common/data/variant/variant_shredding_read_plan_factory.h"
 #include "paimon/common/data/variant/variant_type_utils.h"
+#include "paimon/common/executor/future.h"
+#include "paimon/common/executor/reader_build_executor.h"
 #include "paimon/common/reader/data_file_reader_factory.h"
 #include "paimon/common/reader/delegating_prefetch_reader.h"
 #include "paimon/common/reader/late_materializing_reader_builder.h"
@@ -93,23 +97,69 @@ Result<std::vector<std::unique_ptr<FileBatchReader>>> AbstractSplitRead::CreateR
 
     std::vector<std::unique_ptr<FileBatchReader>> raw_file_readers;
     raw_file_readers.reserve(data_files.size());
+    const uint32_t parallel_num =
+        static_cast<uint32_t>(std::min<size_t>(kReaderBuildMaxParallelNum, data_files.size()));
+    if (parallel_num <= 1) {
+        for (const auto& file : data_files) {
+            PAIMON_ASSIGN_OR_RAISE(
+                std::unique_ptr<FileBatchReader> file_reader,
+                CreateRawFileReader(partition, file, field_mapping_builder.get(), dv_factory,
+                                    row_ranges, data_file_path_factory, extra_format_options));
+            if (file_reader) {
+                raw_file_readers.push_back(std::move(file_reader));
+            }
+        }
+        return std::move(raw_file_readers);
+    }
+
+    // Opening a file, reading its footer and building the reader is mostly waiting on remote
+    // I/O, so files are built concurrently. The tasks only reference locals of this frame,
+    // which stay alive because CollectAll below drains every future before returning.
+    std::shared_ptr<Executor> build_executor = GetReaderBuildExecutor();
+    std::vector<std::future<Result<std::unique_ptr<FileBatchReader>>>> futures;
+    futures.reserve(data_files.size());
     for (const auto& file : data_files) {
-        auto data_file_path = data_file_path_factory->ToPath(file);
-        PAIMON_ASSIGN_OR_RAISE(std::string data_file_identifier, file->FileFormat());
-        PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<ReaderBuilder> reader_builder,
-                               PrepareReaderBuilder(data_file_identifier, extra_format_options));
-        PAIMON_ASSIGN_OR_RAISE(
-            std::unique_ptr<FileBatchReader> file_reader,
-            CreateFieldMappingReader(data_file_path, file, partition, std::move(reader_builder),
-                                     field_mapping_builder.get(), dv_factory, row_ranges,
-                                     data_file_path_factory));
+        futures.push_back(Via(
+            build_executor.get(), [this, file, &partition, &field_mapping_builder, &dv_factory,
+                                   &row_ranges, &data_file_path_factory, &extra_format_options]() {
+                return CreateRawFileReader(partition, file, field_mapping_builder.get(), dv_factory,
+                                           row_ranges, data_file_path_factory,
+                                           extra_format_options);
+            }));
+    }
+    // Readers keep the file order of the split, and CollectAll preserves the submit order.
+    Status first_error;
+    for (auto& built : CollectAll(futures)) {
+        if (!built.ok()) {
+            if (first_error.ok()) {
+                first_error = built.status();
+            }
+            continue;
+        }
+        std::unique_ptr<FileBatchReader> file_reader = std::move(built).value();
         if (file_reader) {
             raw_file_readers.push_back(std::move(file_reader));
         }
     }
+    PAIMON_RETURN_NOT_OK(first_error);
     return std::move(raw_file_readers);
 }
 
+Result<std::unique_ptr<FileBatchReader>> AbstractSplitRead::CreateRawFileReader(
+    const BinaryRow& partition, const std::shared_ptr<DataFileMeta>& file,
+    const FieldMappingBuilder* field_mapping_builder, DeletionVector::Factory dv_factory,
+    const std::optional<std::vector<Range>>& row_ranges,
+    const std::shared_ptr<DataFilePathFactory>& data_file_path_factory,
+    const std::map<std::string, std::string>& extra_format_options) const {
+    auto data_file_path = data_file_path_factory->ToPath(file);
+    PAIMON_ASSIGN_OR_RAISE(std::string data_file_identifier, file->FileFormat());
+    PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<ReaderBuilder> reader_builder,
+                           PrepareReaderBuilder(data_file_identifier, extra_format_options));
+    return CreateFieldMappingReader(data_file_path, file, partition, std::move(reader_builder),
+                                    field_mapping_builder, dv_factory, row_ranges,
+                                    data_file_path_factory);
+}
+
 bool AbstractSplitRead::NeedCompleteRowTrackingFields(
     bool row_tracking_enabled, const std::shared_ptr<arrow::Schema>& read_schema) {
     if (row_tracking_enabled &&

src/paimon/core/operation/abstract_split_read.h

@@ -107,6 +107,14 @@ class AbstractSplitRead : public SplitRead {
     /// that the reader builder and the file reader cannot be built from different answers.
     DataFileReadOptions BuildDataFileReadOptions() const;
 
+    // return nullptr if data file is skipped by index or dv
+    Result<std::unique_ptr<FileBatchReader>> CreateRawFileReader(
+        const BinaryRow& partition, const std::shared_ptr<DataFileMeta>& file,
+        const FieldMappingBuilder* field_mapping_builder, DeletionVector::Factory dv_factory,
+        const std::optional<std::vector<Range>>& row_ranges,
+        const std::shared_ptr<DataFilePathFactory>& data_file_path_factory,
+        const std::map<std::string, std::string>& extra_format_options) const;
+
     Result<std::unique_ptr<ReaderBuilder>> PrepareReaderBuilder(
         const std::string& format_identifier,
         const std::map<std::string, std::string>& extra_format_options) const;

src/paimon/core/schema/schema_manager.h

@@ -47,6 +47,7 @@ class SchemaManager {
                   const std::string& branch);
 
     /// Read schema for schema id. Find schema in cache first.
+    /// Safe to call concurrently.
     Result<std::shared_ptr<TableSchema>> ReadSchema(int64_t schema_id) const;
     Result<std::optional<std::shared_ptr<TableSchema>>> Latest() const;
     Result<std::unique_ptr<TableSchema>> CreateTable(