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

@@ -48,6 +48,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/Formats/FormatParserSharedResources.cpp

@@ -10,11 +10,13 @@ namespace Setting
 {
     extern const SettingsMaxThreads max_download_threads;
     extern const SettingsMaxThreads max_parsing_threads;
+    extern const SettingsUInt64 object_storage_max_files_to_prefetch;
 }
 
 FormatParserSharedResources::FormatParserSharedResources(const Settings & settings, size_t num_streams_)
     : max_parsing_threads(settings[Setting::max_parsing_threads])
     , max_io_threads(settings[Setting::max_download_threads])
+    , max_concurrent_readers_per_stream(std::max<size_t>(settings[Setting::object_storage_max_files_to_prefetch], 1))
     , num_streams(num_streams_)
 {
 }

src/Formats/FormatParserSharedResources.h

@@ -22,6 +22,15 @@ struct FormatParserSharedResources
     const size_t max_parsing_threads = 0;
     const size_t max_io_threads = 0;
 
+    /// How many readers a single stream may keep concurrently active (e.g. via
+    /// `object_storage_max_files_to_prefetch` priming several files' readers ahead of the one
+    /// currently being consumed). Per-reader budgets (see Parquet's `getLimitsPerReader`) divide by
+    /// `num_streams * max_concurrent_readers_per_stream`, not just `num_streams`, so that priming
+    /// more files ahead doesn't let each of them claim a whole stream's share of memory/threads as
+    /// if it were the only reader alive for that stream. Defaults to 1 (today's behaviour: exactly
+    /// one active reader per stream) so this is a no-op unless a caller opts into more.
+    const size_t max_concurrent_readers_per_stream = 1;
+
     std::atomic<size_t> num_streams{0};
     ThreadPoolCallbackRunnerFast parsing_runner;
     ThreadPoolCallbackRunnerFast io_runner;

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/Parquet/ReadCommon.cpp

@@ -10,6 +10,11 @@ SharedResourcesExt::Limits SharedResourcesExt::getLimitsPerReader(const FormatPa
 {
     const SharedResourcesExt & ext = *static_cast<const SharedResourcesExt *>(parser_shared_resources.opaque.get());
     size_t n = parser_shared_resources.num_streams.load(std::memory_order_relaxed);
+    /// Up to `max_concurrent_readers_per_stream` readers can be alive per stream at once (e.g.
+    /// `object_storage_max_files_to_prefetch` priming several files ahead). Without this factor,
+    /// each of them would compute its budget as if it alone owned its stream's whole share,
+    /// oversubscribing memory/parsing-thread budget by that factor once more than one is primed.
+    n *= std::max<size_t>(parser_shared_resources.max_concurrent_readers_per_stream, 1);
     fraction /= static_cast<double>(std::max(n, size_t(1)));
     return Limits {
         .memory_low_watermark = size_t(ext.total_memory_low_watermark * fraction),

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(