Object storage: prefetch upcoming files ahead of the current one

src/Common/FailPoint.cpp

@@ -83,6 +83,8 @@ static struct InitFiu
     REGULAR(check_table_query_delay_for_part) \
     REGULAR(dummy_failpoint) \
     REGULAR(prefetched_reader_pool_failpoint) \
+    REGULAR(object_storage_file_prefetch_failpoint) \
+    PAUSEABLE(object_storage_reader_pool_pause) \
     REGULAR(taskstats_counters_reset_throw) \
     REGULAR(shared_set_sleep_during_update) \
     REGULAR(smt_outdated_parts_exception_response) \

src/Common/ProfileEvents.cpp

@@ -871,6 +871,7 @@ The server successfully detected this situation and will download merged part fr
     M(RemoteFSBuffers, "Number of buffers created for asynchronous reading from remote filesystem", ValueType::Number) \
     M(MergeTreePrefetchedReadPoolInit, "Time spent preparing tasks in MergeTreePrefetchedReadPool", ValueType::Microseconds) \
     M(WaitPrefetchTaskMicroseconds, "Time spend waiting for prefetched reader", ValueType::Microseconds) \
+    M(ObjectStorageWaitPrefetchedReaderMicroseconds, "Time spent waiting for a primed object storage file reader to become available (see object_storage_max_files_to_prefetch)", ValueType::Microseconds) \
     \
     M(ThreadpoolReaderTaskMicroseconds, "Time spent getting the data in asynchronous reading", ValueType::Microseconds) \
     M(ThreadpoolReaderPrepareMicroseconds, "Time spent on preparation (e.g. call to reader seek() method)", ValueType::Microseconds) \

src/Core/Settings.cpp

@@ -4388,6 +4388,9 @@ Possible values:
 Enables caching of rows number during count from files in table functions `file`/`s3`/`url`/`hdfs`/`azureBlobStorage`.
 
 Enabled by default.
+)", 0) \
+    DECLARE(UInt64, object_storage_max_files_to_prefetch, 1, R"(
+Number of files that each reading stream of `s3`/`azureBlobStorage`/`hdfs`/data lake table engines and table functions keeps open and priming (starting their background reads) ahead of the file currently being consumed. Since object storage reads are IO-wait-bound rather than CPU-bound, priming more files in advance overlaps their fetch latency with decoding the current file, instead of only starting the next file's fetch once its stream slot is scheduled. 1 keeps today's behaviour (only the current file is being read; the next file's reader is constructed ahead of time but its data is not fetched until it becomes current). Higher values increase read concurrency at the cost of memory (each additionally primed file holds its own transient buffers). Any value above 1 lets a stream prepare several files at once, which are claimed in whatever order those background tasks reach the shared file list, so the order files are read in - and therefore the order rows come back in without an `ORDER BY` - is no longer deterministic. Which rows are returned is unaffected.
 )", 0) \
     DECLARE(Bool, optimize_respect_aliases, true, R"(
 If it is set to true, it will respect aliases in WHERE/GROUP BY/ORDER BY, that will help with partition pruning/secondary indexes/optimize_aggregation_in_order/optimize_read_in_order/optimize_trivial_count

src/Core/SettingsChangesHistory.cpp

@@ -47,6 +47,7 @@ const VersionToSettingsChangesMap & getSettingsChangesHistory()
             {"export_merge_tree_partition_retry_initial_backoff_seconds", 5, 5, "New setting for exponential back-off between failed part export retries in an export partition task"},
             {"export_merge_tree_partition_retry_max_backoff_seconds", 300, 300, "New setting capping the exponential back-off between failed part export retries in an export partition task"},
             {"export_merge_tree_partition_max_retries", 3, 3, "Obsolete and ignored: export partition tasks now retry retryable failures until the task timeout and fail immediately on non-retryable errors, instead of using a fixed retry budget"},
+            {"object_storage_max_files_to_prefetch", 1, 1, "New setting to prime upcoming files' reads ahead of the file currently being consumed by an object storage reading stream, decoupling file-level read concurrency from the number of processing threads."},
         });
         addSettingsChanges(settings_changes_history, "26.3",
         {

src/Processors/Formats/IInputFormat.h

@@ -136,6 +136,12 @@ class IInputFormat : public ISource
 
     virtual std::optional<std::pair<std::vector<size_t>, size_t>> getMatchedBuckets() const { return std::nullopt; }
 
+    /// Called (from a background thread, before this format becomes the one being pulled from) to
+    /// eagerly start background reads for formats that support it, so their data is already in
+    /// flight by the time regular reading reaches this file. No-op by default; only overridden by
+    /// formats whose reads are otherwise lazy until the first read() call.
+    virtual void prefetch() {}
+
 protected:
     ReadBuffer & getReadBuffer() const { chassert(in); return *in; }
 

src/Processors/Formats/Impl/ParquetV3BlockInputFormat.cpp

@@ -114,6 +114,13 @@ parquet::format::FileMetaData ParquetV3BlockInputFormat::getFileMetadata(Parquet
     }
 }
 
+void ParquetV3BlockInputFormat::prefetch()
+{
+    if (need_only_count)
+        return;
+    initializeIfNeeded();
+}
+
 Chunk ParquetV3BlockInputFormat::read()
 {
     if (need_only_count)

src/Processors/Formats/Impl/ParquetV3BlockInputFormat.h

@@ -40,6 +40,8 @@ class ParquetV3BlockInputFormat : public IInputFormat
 
     std::optional<std::pair<std::vector<size_t>, size_t>> getMatchedBuckets() const override;
 
+    void prefetch() override;
+
 private:
     Chunk read() override;
 

src/Storages/ObjectStorage/StorageObjectStorageSource.cpp

@@ -44,6 +44,7 @@
 #include <boost/operators.hpp>
 #include <Poco/String.h>
 #include <Common/Exception.h>
+#include <Common/FailPoint.h>
 #include <Common/SipHash.h>
 #include <Common/parseGlobs.h>
 #include <Storages/ObjectStorage/IObjectIterator.h>
@@ -56,6 +57,8 @@
 #endif
 
 #include <fmt/ranges.h>
+#include <base/sleep.h>
+#include <Common/ElapsedTimeProfileEventIncrement.h>
 #include <Common/ProfileEvents.h>
 #include <Core/SettingsEnums.h>
 #include <Poco/String.h>
@@ -73,6 +76,7 @@ namespace ProfileEvents
     extern const Event ObjectStorageReadObjects;
     extern const Event ObjectStorageClusterProcessedTasks;
     extern const Event ObjectStorageClusterWaitingMicroseconds;
+    extern const Event ObjectStorageWaitPrefetchedReaderMicroseconds;
 }
 
 namespace CurrentMetrics
@@ -84,6 +88,14 @@ namespace CurrentMetrics
 
 namespace DB
 {
+/// The failpoints are defined in `DB::FailPoints`, so this has to be declared inside `namespace DB`
+/// - at global scope it declares a different symbol and the build fails to link.
+namespace FailPoints
+{
+    extern const char object_storage_file_prefetch_failpoint[];
+    extern const char object_storage_reader_pool_pause[];
+}
+
 namespace Setting
 {
     extern const SettingsUInt64 max_download_buffer_size;
@@ -99,6 +111,7 @@ namespace Setting
     extern const SettingsBool input_format_parquet_use_native_reader_v3;
     extern const SettingsBool allow_experimental_iceberg_read_optimization;
     extern const SettingsBool use_object_storage_list_objects_cache;
+    extern const SettingsUInt64 object_storage_max_files_to_prefetch;
 }
 
 namespace ErrorCodes
@@ -153,22 +166,35 @@ StorageObjectStorageSource::StorageObjectStorageSource(
     , parser_shared_resources(std::move(parser_shared_resources_))
     , format_filter_info(std::move(format_filter_info_))
     , read_from_format_info(info)
-    , create_reader_pool(
+    , own_reader_pool(
           std::make_shared<ThreadPool>(
               CurrentMetrics::StorageObjectStorageThreads,
               CurrentMetrics::StorageObjectStorageThreadsActive,
               CurrentMetrics::StorageObjectStorageThreadsScheduled,
               1 /* max_threads */))
+    , create_reader_pool(
+          context_->getSettingsRef()[Setting::object_storage_max_files_to_prefetch] <= 1
+          ? *own_reader_pool
+          : context_->getPrefetchThreadpool())
     , file_iterator(file_iterator_)
     , schema_cache(StorageObjectStorage::getSchemaCache(context_, configuration->getTypeName()))
-    , create_reader_scheduler(threadPoolCallbackRunnerUnsafe<ReaderHolder>(*create_reader_pool, ThreadName::READER_POOL))
+    , create_reader_scheduler(threadPoolCallbackRunnerUnsafe<ReaderHolder>(create_reader_pool, ThreadName::READER_POOL))
+    , max_files_to_prefetch(std::max<size_t>(context_->getSettingsRef()[Setting::object_storage_max_files_to_prefetch], 1))
 {
 }
 
 StorageObjectStorageSource::~StorageObjectStorageSource()
 {
     LOG_DEBUG(log, "Source finished: files_read={}", total_files_read);
-    create_reader_pool->wait();
+    /// The scheduled work captures `this`. When create_reader_pool is the shared server-wide pool
+    /// (max_files_to_prefetch > 1), waiting for the whole pool would also be wrong (it would wait
+    /// for unrelated queries) and insufficient to express what we need. Wait for exactly this
+    /// source's own futures instead, which also covers the dedicated-pool case below.
+    for (auto & reader_future : reader_futures)
+    {
+        if (reader_future.valid())
+            reader_future.wait();
+    }
 }
 
 std::string StorageObjectStorageSource::getUniqueStoragePathIdentifier(
@@ -406,11 +432,40 @@ void StorageObjectStorageSource::lazyInitialize()
     if (reader)
     {
         ++total_files_read;
-        reader_future = createReaderAsync();
+        refillReaderFutures();
     }
     initialized = true;
 }
 
+void StorageObjectStorageSource::refillReaderFutures()
+{
+    while (reader_futures.size() < max_files_to_prefetch)
+    {
+        /// The first queued future preserves the pre-existing pipeline-object-only lookahead
+        /// (unprimed), so max_files_to_prefetch=1 never primes anything - identical to the
+        /// behaviour before this setting existed. Every subsequent slot is primed.
+        const bool prime = !reader_futures.empty();
+        reader_futures.push_back(createReaderAsync(prime));
+    }
+}
+
+StorageObjectStorageSource::ReaderHolder StorageObjectStorageSource::takeNextQueuedReader()
+{
+    /// Stopping at the first empty slot would silently drop files: the queued tasks race each other
+    /// for file_iterator->next(), so a slot queued earlier can resolve to nothing while later-queued
+    /// slots already hold files that have been fetched (and, when primed, read ahead) and still have
+    /// to be returned. Only when every queued slot has been drained and none held a file is this
+    /// stream really out of work.
+    while (!reader_futures.empty())
+    {
+        auto next_reader = reader_futures.front().get();
+        reader_futures.pop_front();
+        if (next_reader)
+            return next_reader;
+    }
+    return {};
+}
+
 Chunk StorageObjectStorageSource::generate()
 {
     lazyInitialize();
@@ -634,18 +689,23 @@ Chunk StorageObjectStorageSource::generate()
 
         total_rows_in_file = 0;
 
-        assert(reader_future.valid());
-        reader = reader_future.get();
+        chassert(!reader_futures.empty());
+        {
+            ProfileEventTimeIncrement<Microseconds> watch(ProfileEvents::ObjectStorageWaitPrefetchedReaderMicroseconds);
+            reader = takeNextQueuedReader();
+        }
 
         if (!reader)
             break;
 
         ++total_files_read;
 
-        /// Even if task is finished the thread may be not freed in pool.
-        /// So wait until it will be freed before scheduling a new task.
-        create_reader_pool->wait();
-        reader_future = createReaderAsync();
+        /// There used to be a `create_reader_pool->wait` here for the single-lookahead case, to let
+        /// the private one-thread pool free its thread before the next task was scheduled. The pool
+        /// is now the shared server-wide one, so there is no private thread to wait for, and waiting
+        /// on it would block on unrelated queries' prefetches. Taking a reader from the queue
+        /// already means its callback has finished, which is all that wait actually guaranteed.
+        refillReaderFutures();
     }
 
     return {};
@@ -1308,9 +1368,25 @@ StorageObjectStorageSource::ReaderHolder StorageObjectStorageSource::createReade
         std::move(constant_columns_with_values));
 }
 
-std::future<StorageObjectStorageSource::ReaderHolder> StorageObjectStorageSource::createReaderAsync()
+std::future<StorageObjectStorageSource::ReaderHolder> StorageObjectStorageSource::createReaderAsync(bool prime)
 {
-    return create_reader_scheduler([=, this] { return createReader(); }, Priority{});
+    return create_reader_scheduler([=, this]
+    {
+        /// Lets a test observe, via `system.metrics`, which pool (own_reader_pool or the shared
+        /// context_->getPrefetchThreadpool()) this task actually runs on while it is held here.
+        FailPointInjection::pauseFailPoint(FailPoints::object_storage_reader_pool_pause);
+        auto reader_holder = createReader();
+        if (prime && reader_holder)
+        {
+            fiu_do_on(FailPoints::object_storage_file_prefetch_failpoint,
+            {
+                throw Exception(ErrorCodes::BAD_ARGUMENTS, "Failpoint for object storage file prefetch enabled");
+            });
+            if (auto * input_format = reader_holder.getInputFormat())
+                input_format->prefetch();
+        }
+        return reader_holder;
+    }, Priority{});
 }
 
 std::unique_ptr<ReadBufferFromFileBase> createReadBuffer(

src/Storages/ObjectStorage/StorageObjectStorageSource.h

@@ -1,4 +1,7 @@
 #pragma once
+
+#include <deque>
+#include <future>
 #include <optional>
 #include <Common/re2.h>
 #include <Interpreters/Context_fwd.h>
@@ -86,7 +89,19 @@ class StorageObjectStorageSource : public ISource
     FormatFilterInfoPtr format_filter_info;
 
     ReadFromFormatInfo read_from_format_info;
-    const std::shared_ptr<ThreadPool> create_reader_pool;
+    /// Dedicated single-thread pool, used by create_reader_pool below at the default
+    /// `object_storage_max_files_to_prefetch` of 1 - the same private per-source pool `master` (i.e.
+    /// before this setting existed) uses unconditionally - so prefetches cannot contend with
+    /// unrelated queries on the server-wide pool. Constructing it is cheap (threads are spawned
+    /// lazily on schedule), so it's unconditional even though it goes unused above 1, where a pool
+    /// sized from the setting would make the thread count `streams * max_files_to_prefetch`,
+    /// unbounded by anything global - create_reader_pool falls back to the shared
+    /// `context_->getPrefetchThreadpool()` (the same one `MergeTreePrefetchedReadPool` submits to)
+    /// instead.
+    std::shared_ptr<ThreadPool> own_reader_pool;
+    /// Because the shared pool outlives this source, outstanding futures must be waited for in the
+    /// destructor - the scheduled work captures `this`.
+    ThreadPool & create_reader_pool;
 
     std::shared_ptr<IObjectIterator> file_iterator;
     SchemaCache & schema_cache;
@@ -122,6 +137,10 @@ class StorageObjectStorageSource : public ISource
 
         ObjectInfoPtr getObjectInfo() const { return object_info; }
         const IInputFormat * getInputFormat() const { return dynamic_cast<const IInputFormat *>(source.get()); }
+        /// Non-const overload so a freshly-built, not-yet-current reader can be primed (see
+        /// `createReaderAsync`) without exposing mutable access more broadly.
+        IInputFormat * getInputFormat() { return dynamic_cast<IInputFormat *>(source.get()); }
+        ReadBuffer * readBuffer() const { return read_buf.get(); }
 
     private:
         ObjectInfoPtr object_info;
@@ -136,7 +155,33 @@ class StorageObjectStorageSource : public ISource
 
     ReaderHolder reader;
     ThreadPoolCallbackRunnerUnsafe<ReaderHolder> create_reader_scheduler;
-    std::future<ReaderHolder> reader_future;
+
+    /// Depth of the file-lookahead queue below (>= 1; from `object_storage_max_files_to_prefetch`).
+    /// This bounds `reader_futures` only, so a stream holds up to this many queued readers *plus*
+    /// the one in `reader` it is currently being pulled from. Of the queued ones only those beyond
+    /// the first are primed (see refillReaderFutures), so at the default of 1 nothing is primed
+    /// ahead at all and the behaviour matches what it was before this setting existed.
+    const size_t max_files_to_prefetch;
+    /// Upcoming files' readers. Built on create_reader_pool and primed (createReaderAsync ->
+    /// prefetch()) so their background reads are already in flight by the time generate() reaches
+    /// them. Queued in submission order, but *which* file a slot ends up holding is not: the queued
+    /// tasks race each other for file_iterator->next(), so a slot queued earlier can resolve to
+    /// nothing while a slot queued later already holds a file. Consumed via takeNextQueuedReader.
+    std::deque<std::future<ReaderHolder>> reader_futures;
+
+    /// Fills reader_futures back up to max_files_to_prefetch entries. Only futures beyond the first
+    /// are primed (see createReaderAsync): the first preserves the pre-existing pipeline-object-only
+    /// lookahead, so max_files_to_prefetch=1 (the default) primes nothing and behaves exactly as
+    /// before this setting existed. No-op once file_iterator is exhausted (createReaderAsync's
+    /// result will resolve to an empty/false ReaderHolder, which callers already handle).
+    void refillReaderFutures();
+
+    /// Pops the oldest queued reader, skipping slots that resolved to no file. Returns an empty
+    /// holder only once every queued slot has been drained and none of them held a file.
+    /// An empty slot on its own does not mean the file list is exhausted - queued tasks race for
+    /// file_iterator->next(), so an earlier-queued slot can come up empty while later-queued slots
+    /// already hold files that have been fetched and must still be read.
+    ReaderHolder takeNextQueuedReader();
 
     /// Recreate ReadBuffer and Pipeline for each file.
     static ReaderHolder createReader(
@@ -157,7 +202,9 @@ class StorageObjectStorageSource : public ISource
 
     ReaderHolder createReader();
 
-    std::future<ReaderHolder> createReaderAsync();
+    /// If `prime` is set, calls IInputFormat::prefetch() on the built reader (see FormatFactory
+    /// formats' prefetch() override) to start its background reads before it becomes `reader`.
+    std::future<ReaderHolder> createReaderAsync(bool prime);
 
     void addNumRowsToCache(const ObjectInfo & object_info, size_t num_rows);
     void lazyInitialize();

tests/queries/0_stateless/04626_object_storage_max_files_to_prefetch.reference

@@ -0,0 +1,31 @@
+179700	600	2138863090785102664
+179700	600	2138863090785102664
+179700	600	2138863090785102664
+179700	600	2138863090785102664
+179700	600	2138863090785102664
+179700	600	2138863090785102664
+179700	600	2138863090785102664
+179700	600	2138863090785102664
+179700	600	2138863090785102664
+179700	600	2138863090785102664
+179700	600	2138863090785102664
+179700	600	2138863090785102664
+179700	600	2138863090785102664
+179700	600	2138863090785102664
+179700	600	2138863090785102664
+179700	600	2138863090785102664
+179700	600	2138863090785102664
+179700	600	2138863090785102664
+179700	600	2138863090785102664
+179700	600	2138863090785102664
+179700	600	2138863090785102664
+179700	600	2138863090785102664
+179700	600	2138863090785102664
+179700	600	2138863090785102664
+179700	600	2138863090785102664
+179700	600	2138863090785102664
+179700	600	2138863090785102664
+179700	600	2138863090785102664
+179700	600	2138863090785102664
+179700	600	2138863090785102664
+1