From 62bd030c2c299f8ebacac253269ec778deb0512b Mon Sep 17 00:00:00 2001 From: VighneshPath Date: Thu, 23 Jul 2026 11:50:01 +0530 Subject: [PATCH 1/6] Object storage: prime upcoming files' reads ahead of the current one StorageObjectStorageSource read files one at a time, opening the next file's reader only after the current one is fully consumed; the number of concurrently open files was therefore tied to num_streams (~max_threads), even though object storage reads are IO-wait-bound rather than CPU-bound and idle cores could be overlapping many more files' fetch latency instead of decode threads' worth. A one-file-ahead reader was already built asynchronously (createReaderAsync via create_reader_pool), but only the pipeline objects were constructed in advance - IInputFormat::read() (where Parquet's ReadManager/Prefetcher actually start fetching) was never called until the reader became current, so no bytes were fetched early. Add a new setting, object_storage_max_files_to_prefetch (default 1, matching prior behaviour exactly), that widens the single lookahead future into a queue of futures and, for slots beyond the first, calls a new IInputFormat::prefetch() hook (no-op by default; overridden by ParquetV3BlockInputFormat to eagerly run initializeIfNeeded()) so their background reads are already in flight by the time generate() reaches them. Per-stream row order is unaffected, since files within a stream are still consumed strictly front-to-back from the queue. Locally, against a latency-injected MinIO/Iceberg REST catalog harness (max_threads=2, cacheless), raising this setting from 1 to 4 took wall time from ~42s to ~15s on a 70-file scan, with identical result checksums throughout. Signed-off-by: VighneshPath --- src/Core/Settings.cpp | 3 ++ src/Core/SettingsChangesHistory.cpp | 1 + src/Processors/Formats/IInputFormat.h | 6 +++ .../Impl/ParquetV3BlockInputFormat.cpp | 7 +++ .../Formats/Impl/ParquetV3BlockInputFormat.h | 2 + .../StorageObjectStorageSource.cpp | 50 +++++++++++++++---- .../StorageObjectStorageSource.h | 28 ++++++++++- 7 files changed, 85 insertions(+), 12 deletions(-) diff --git a/src/Core/Settings.cpp b/src/Core/Settings.cpp index 770a99e6f63b..8d83f82a2bfe 100644 --- a/src/Core/Settings.cpp +++ b/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). )", 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 diff --git a/src/Core/SettingsChangesHistory.cpp b/src/Core/SettingsChangesHistory.cpp index 2615ded19715..2fc3109b9abd 100644 --- a/src/Core/SettingsChangesHistory.cpp +++ b/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", { diff --git a/src/Processors/Formats/IInputFormat.h b/src/Processors/Formats/IInputFormat.h index 768d1fea7163..91c9e5ee7138 100644 --- a/src/Processors/Formats/IInputFormat.h +++ b/src/Processors/Formats/IInputFormat.h @@ -136,6 +136,12 @@ class IInputFormat : public ISource virtual std::optional, 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; } diff --git a/src/Processors/Formats/Impl/ParquetV3BlockInputFormat.cpp b/src/Processors/Formats/Impl/ParquetV3BlockInputFormat.cpp index 3195e1195890..420c993e682c 100644 --- a/src/Processors/Formats/Impl/ParquetV3BlockInputFormat.cpp +++ b/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) diff --git a/src/Processors/Formats/Impl/ParquetV3BlockInputFormat.h b/src/Processors/Formats/Impl/ParquetV3BlockInputFormat.h index c93b5c3b3af0..ccbc434f89ef 100644 --- a/src/Processors/Formats/Impl/ParquetV3BlockInputFormat.h +++ b/src/Processors/Formats/Impl/ParquetV3BlockInputFormat.h @@ -40,6 +40,8 @@ class ParquetV3BlockInputFormat : public IInputFormat std::optional, size_t>> getMatchedBuckets() const override; + void prefetch() override; + private: Chunk read() override; diff --git a/src/Storages/ObjectStorage/StorageObjectStorageSource.cpp b/src/Storages/ObjectStorage/StorageObjectStorageSource.cpp index b07a0abb85ae..e43de4c71bf1 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageSource.cpp +++ b/src/Storages/ObjectStorage/StorageObjectStorageSource.cpp @@ -99,6 +99,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 @@ -158,10 +159,11 @@ StorageObjectStorageSource::StorageObjectStorageSource( CurrentMetrics::StorageObjectStorageThreads, CurrentMetrics::StorageObjectStorageThreadsActive, CurrentMetrics::StorageObjectStorageThreadsScheduled, - 1 /* max_threads */)) + std::max(context_->getSettingsRef()[Setting::object_storage_max_files_to_prefetch], 1))) , file_iterator(file_iterator_) , schema_cache(StorageObjectStorage::getSchemaCache(context_, configuration->getTypeName())) , create_reader_scheduler(threadPoolCallbackRunnerUnsafe(*create_reader_pool, ThreadName::READER_POOL)) + , max_files_to_prefetch(std::max(context_->getSettingsRef()[Setting::object_storage_max_files_to_prefetch], 1)) { } @@ -406,11 +408,23 @@ 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)); + } +} + Chunk StorageObjectStorageSource::generate() { lazyInitialize(); @@ -634,18 +648,25 @@ Chunk StorageObjectStorageSource::generate() total_rows_in_file = 0; - assert(reader_future.valid()); - reader = reader_future.get(); + chassert(!reader_futures.empty()); + reader = reader_futures.front().get(); + reader_futures.pop_front(); 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(); + if (max_files_to_prefetch <= 1) + { + /// 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(); + } + /// With max_files_to_prefetch > 1 the pool has multiple threads and multiple readers are + /// meant to be building/priming concurrently, so waiting for the whole pool here would + /// serialize exactly the concurrency this setting is meant to provide. + refillReaderFutures(); } return {}; @@ -1308,9 +1329,18 @@ StorageObjectStorageSource::ReaderHolder StorageObjectStorageSource::createReade std::move(constant_columns_with_values)); } -std::future StorageObjectStorageSource::createReaderAsync() +std::future StorageObjectStorageSource::createReaderAsync(bool prime) { - return create_reader_scheduler([=, this] { return createReader(); }, Priority{}); + return create_reader_scheduler([=, this] + { + auto reader_holder = createReader(); + if (prime && reader_holder) + { + if (auto * input_format = reader_holder.getInputFormat()) + input_format->prefetch(); + } + return reader_holder; + }, Priority{}); } std::unique_ptr createReadBuffer( diff --git a/src/Storages/ObjectStorage/StorageObjectStorageSource.h b/src/Storages/ObjectStorage/StorageObjectStorageSource.h index d30329acce01..b56d4ae6c80c 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageSource.h +++ b/src/Storages/ObjectStorage/StorageObjectStorageSource.h @@ -1,4 +1,7 @@ #pragma once + +#include +#include #include #include #include @@ -122,6 +125,10 @@ class StorageObjectStorageSource : public ISource ObjectInfoPtr getObjectInfo() const { return object_info; } const IInputFormat * getInputFormat() const { return dynamic_cast(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(source.get()); } + ReadBuffer * readBuffer() const { return read_buf.get(); } private: ObjectInfoPtr object_info; @@ -136,7 +143,22 @@ class StorageObjectStorageSource : public ISource ReaderHolder reader; ThreadPoolCallbackRunnerUnsafe create_reader_scheduler; - std::future reader_future; + + /// Depth of the file-lookahead queue below (>= 1; from `object_storage_max_files_to_prefetch`). + /// `reader` plus these futures together keep up to this many files' readers alive/priming at once. + const size_t max_files_to_prefetch; + /// Upcoming files' readers, in the same order the file_iterator handed them out. Built on + /// create_reader_pool and primed (createReaderAsync -> prefetch()) so their background reads are + /// already in flight by the time generate() reaches them; consumed strictly front-to-back, so + /// per-stream row order is unaffected by however many are queued. + std::deque> 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(); /// Recreate ReadBuffer and Pipeline for each file. static ReaderHolder createReader( @@ -157,7 +179,9 @@ class StorageObjectStorageSource : public ISource ReaderHolder createReader(); - std::future 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 createReaderAsync(bool prime); void addNumRowsToCache(const ObjectInfo & object_info, size_t num_rows); void lazyInitialize(); From 8d4c03fdf90c43de0bd0e9a375b53246f9755802 Mon Sep 17 00:00:00 2001 From: VighneshPath Date: Thu, 23 Jul 2026 15:20:13 +0530 Subject: [PATCH 2/6] Object storage prefetch: add ProfileEvent and FailPoint for parity Mirrors MergeTreePrefetchedReadPool's observability/testability conventions for its own background prefetch mechanism: - ObjectStorageWaitPrefetchedReaderMicroseconds measures time spent blocked in reader_futures.front().get(), the same way WaitPrefetchTaskMicroseconds does for MergeTree's prefetched reads. - object_storage_file_prefetch_failpoint lets tests deterministically exercise a failed prefetch, the same way prefetched_reader_pool_failpoint does for MergeTree. Signed-off-by: VighneshPath --- src/Common/FailPoint.cpp | 1 + src/Common/ProfileEvents.cpp | 1 + .../ObjectStorage/StorageObjectStorageSource.cpp | 13 ++++++++++++- 3 files changed, 14 insertions(+), 1 deletion(-) diff --git a/src/Common/FailPoint.cpp b/src/Common/FailPoint.cpp index b8fee32bdf13..8b753423afac 100644 --- a/src/Common/FailPoint.cpp +++ b/src/Common/FailPoint.cpp @@ -83,6 +83,7 @@ static struct InitFiu REGULAR(check_table_query_delay_for_part) \ REGULAR(dummy_failpoint) \ REGULAR(prefetched_reader_pool_failpoint) \ + REGULAR(object_storage_file_prefetch_failpoint) \ REGULAR(taskstats_counters_reset_throw) \ REGULAR(shared_set_sleep_during_update) \ REGULAR(smt_outdated_parts_exception_response) \ diff --git a/src/Common/ProfileEvents.cpp b/src/Common/ProfileEvents.cpp index a1ec3e9f8612..4f8785734cb5 100644 --- a/src/Common/ProfileEvents.cpp +++ b/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) \ diff --git a/src/Storages/ObjectStorage/StorageObjectStorageSource.cpp b/src/Storages/ObjectStorage/StorageObjectStorageSource.cpp index e43de4c71bf1..be2095e6f7be 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageSource.cpp +++ b/src/Storages/ObjectStorage/StorageObjectStorageSource.cpp @@ -44,6 +44,7 @@ #include #include #include +#include #include #include #include @@ -56,6 +57,8 @@ #endif #include +#include +#include #include #include #include @@ -73,6 +76,7 @@ namespace ProfileEvents extern const Event ObjectStorageReadObjects; extern const Event ObjectStorageClusterProcessedTasks; extern const Event ObjectStorageClusterWaitingMicroseconds; + extern const Event ObjectStorageWaitPrefetchedReaderMicroseconds; } namespace CurrentMetrics @@ -649,7 +653,10 @@ Chunk StorageObjectStorageSource::generate() total_rows_in_file = 0; chassert(!reader_futures.empty()); - reader = reader_futures.front().get(); + { + ProfileEventTimeIncrement watch(ProfileEvents::ObjectStorageWaitPrefetchedReaderMicroseconds); + reader = reader_futures.front().get(); + } reader_futures.pop_front(); if (!reader) @@ -1336,6 +1343,10 @@ std::future StorageObjectStorageSource 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(); } From 1ee5500a707bcc3e77b266cc474404ddfae04d71 Mon Sep 17 00:00:00 2001 From: VighneshPath Date: Thu, 23 Jul 2026 20:40:39 +0530 Subject: [PATCH 3/6] Object storage prefetch: fix missing FailPoints extern declaration Each .cpp using a FailPoints:: symbol needs its own local extern declaration (matching MergeTreePrefetchedReadPool.cpp's pattern) - this was missing, causing "use of undeclared identifier 'FailPoints'". Signed-off-by: VighneshPath --- src/Storages/ObjectStorage/StorageObjectStorageSource.cpp | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/src/Storages/ObjectStorage/StorageObjectStorageSource.cpp b/src/Storages/ObjectStorage/StorageObjectStorageSource.cpp index be2095e6f7be..8cce0371060c 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageSource.cpp +++ b/src/Storages/ObjectStorage/StorageObjectStorageSource.cpp @@ -79,6 +79,11 @@ namespace ProfileEvents extern const Event ObjectStorageWaitPrefetchedReaderMicroseconds; } +namespace FailPoints +{ + extern const char object_storage_file_prefetch_failpoint[]; +} + namespace CurrentMetrics { extern const Metric StorageObjectStorageThreads; From f04d3cbeb421d6d84422c534be8f3ca0681c4c8f Mon Sep 17 00:00:00 2001 From: VighneshPath Date: Thu, 30 Jul 2026 17:15:52 +0530 Subject: [PATCH 4/6] Object storage: do not drop prefetched files when a lookahead slot is empty `object_storage_max_files_to_prefetch` above 2 could make a scan silently return fewer rows than the files contain. The queued tasks each call `file_iterator->next` themselves, so a task queued earlier can lose the race and resolve to no file while tasks queued later already hold real files. `generate` treated the first empty slot as proof the file list was exhausted and stopped, abandoning those readers. With a single future in flight, as before this setting existed, the two were the same fact. `takeNextQueuedReader` now skips empty slots and reports exhaustion only once every queued slot has been drained. Also move the `FailPoints` extern declaration inside `namespace DB`. At global scope it names a different symbol than the one `FailPoint.cpp` defines, so the build did not link. Note this means row order within a stream is not deterministic above 1, which contradicts the claim in 2e3df41adc8; only which rows are returned is fixed. Co-Authored-By: Claude Opus 5 Signed-off-by: VighneshPath --- .../StorageObjectStorageSource.cpp | 32 +++++++++--- .../StorageObjectStorageSource.h | 21 ++++++-- ...ct_storage_max_files_to_prefetch.reference | 31 +++++++++++ ...26_object_storage_max_files_to_prefetch.sh | 52 +++++++++++++++++++ 4 files changed, 124 insertions(+), 12 deletions(-) create mode 100644 tests/queries/0_stateless/04626_object_storage_max_files_to_prefetch.reference create mode 100755 tests/queries/0_stateless/04626_object_storage_max_files_to_prefetch.sh diff --git a/src/Storages/ObjectStorage/StorageObjectStorageSource.cpp b/src/Storages/ObjectStorage/StorageObjectStorageSource.cpp index 8cce0371060c..48ebe1d0f927 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageSource.cpp +++ b/src/Storages/ObjectStorage/StorageObjectStorageSource.cpp @@ -79,11 +79,6 @@ namespace ProfileEvents extern const Event ObjectStorageWaitPrefetchedReaderMicroseconds; } -namespace FailPoints -{ - extern const char object_storage_file_prefetch_failpoint[]; -} - namespace CurrentMetrics { extern const Metric StorageObjectStorageThreads; @@ -93,6 +88,13 @@ 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[]; +} + namespace Setting { extern const SettingsUInt64 max_download_buffer_size; @@ -434,6 +436,23 @@ void StorageObjectStorageSource::refillReaderFutures() } } +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(); @@ -660,9 +679,8 @@ Chunk StorageObjectStorageSource::generate() chassert(!reader_futures.empty()); { ProfileEventTimeIncrement watch(ProfileEvents::ObjectStorageWaitPrefetchedReaderMicroseconds); - reader = reader_futures.front().get(); + reader = takeNextQueuedReader(); } - reader_futures.pop_front(); if (!reader) break; diff --git a/src/Storages/ObjectStorage/StorageObjectStorageSource.h b/src/Storages/ObjectStorage/StorageObjectStorageSource.h index b56d4ae6c80c..5fa61b857002 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageSource.h +++ b/src/Storages/ObjectStorage/StorageObjectStorageSource.h @@ -145,12 +145,16 @@ class StorageObjectStorageSource : public ISource ThreadPoolCallbackRunnerUnsafe create_reader_scheduler; /// Depth of the file-lookahead queue below (>= 1; from `object_storage_max_files_to_prefetch`). - /// `reader` plus these futures together keep up to this many files' readers alive/priming at once. + /// 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, in the same order the file_iterator handed them out. Built on - /// create_reader_pool and primed (createReaderAsync -> prefetch()) so their background reads are - /// already in flight by the time generate() reaches them; consumed strictly front-to-back, so - /// per-stream row order is unaffected by however many are queued. + /// 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> reader_futures; /// Fills reader_futures back up to max_files_to_prefetch entries. Only futures beyond the first @@ -160,6 +164,13 @@ class StorageObjectStorageSource : public ISource /// 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( size_t processor, diff --git a/tests/queries/0_stateless/04626_object_storage_max_files_to_prefetch.reference b/tests/queries/0_stateless/04626_object_storage_max_files_to_prefetch.reference new file mode 100644 index 000000000000..69bd3bcde955 --- /dev/null +++ b/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 diff --git a/tests/queries/0_stateless/04626_object_storage_max_files_to_prefetch.sh b/tests/queries/0_stateless/04626_object_storage_max_files_to_prefetch.sh new file mode 100755 index 000000000000..05902a113491 --- /dev/null +++ b/tests/queries/0_stateless/04626_object_storage_max_files_to_prefetch.sh @@ -0,0 +1,52 @@ +#!/usr/bin/env bash +# Tags: no-fasttest +# Tag no-fasttest: depends on MinIO via s3_conn. + +# `object_storage_max_files_to_prefetch` must never change which rows a scan returns, only when +# their background reads start. Raising it lets one reading stream have several tasks in flight +# racing for the shared file iterator, so a lookahead slot queued earlier can resolve to no file +# while later-queued slots already hold files; the stream must still read every one of them. +# +# The failure this guards against is a silently short result, and it is racy: a broken build +# returns the right answer a good fraction of the time. Each configuration is therefore checked +# repeatedly - a single run per setting would let a regression through roughly half the time. + +CURDIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd) +# shellcheck source=../shell_config.sh +. "$CURDIR"/../shell_config.sh + +path="04626_prefetch_${CLICKHOUSE_DATABASE}" + +$CLICKHOUSE_CLIENT --query "DROP TABLE IF EXISTS test_04626_write" +$CLICKHOUSE_CLIENT --query " + CREATE TABLE test_04626_write (id UInt64, payload String) + ENGINE = S3(s3_conn, filename = '$path/file_{_partition_id}.parquet', format = Parquet, partition_strategy = 'wildcard') + PARTITION BY id % 6" + +# 600 rows spread over 6 files, 100 rows each, so a lost file shows up as a short count. +$CLICKHOUSE_CLIENT --s3_truncate_on_insert 1 --query " + INSERT INTO test_04626_write SELECT number AS id, repeat('x', 16) AS payload FROM numbers(600)" + +$CLICKHOUSE_CLIENT --query "DROP TABLE test_04626_write" + +# Depth 1 is the default (nothing is primed ahead), 3 and 8 put several tasks in flight at once. +# 8 is deliberately larger than the 6 files that exist, so the lookahead queue can never fill and +# has to drain correctly once the file iterator runs out mid-refill. +for depth in 1 3 8; do + for _ in {1..10}; do + $CLICKHOUSE_CLIENT --query " + SELECT sum(id), count(), sum(sipHash64(payload)) + FROM s3(s3_conn, filename = '$path/file_*.parquet', format = Parquet) + SETTINGS max_threads = 2, object_storage_max_files_to_prefetch = $depth, enable_filesystem_cache = 0" + done +done + +# Row *order* is deliberately not asserted: with several tasks racing for the file iterator, which +# file lands in which lookahead slot is not deterministic, so only the set of rows is guaranteed. +$CLICKHOUSE_CLIENT --query " + SELECT arraySort(groupArray(id)) = ( + SELECT arraySort(groupArray(id)) + FROM s3(s3_conn, filename = '$path/file_*.parquet', format = Parquet) + SETTINGS max_threads = 1, object_storage_max_files_to_prefetch = 1) + FROM s3(s3_conn, filename = '$path/file_*.parquet', format = Parquet) + SETTINGS max_threads = 1, object_storage_max_files_to_prefetch = 4, enable_filesystem_cache = 0" From 77027d63f979168ca29a6885d5e61e1d349131ba Mon Sep 17 00:00:00 2001 From: VighneshPath Date: Thu, 30 Jul 2026 17:26:05 +0530 Subject: [PATCH 5/6] Object storage: prefetch readers on the shared pool, not a pool per source Each source built its own `ThreadPool` sized from `object_storage_max_files_to_prefetch`, so a query's prefetch threads came to `streams * max_files_to_prefetch` with no global ceiling - 1024 threads at 64 streams and depth 16. Nothing bounded that across concurrent queries either. Submit to `Context::getPrefetchThreadpool` instead, the server-wide pool that `MergeTreePrefetchedReadPool` already uses for remote reads and that is capped by `prefetch_threadpool_pool_size`. Reader construction stays off the format parsing pool, so the nested shared-pool deadlock described in `threadPoolCallbackRunner.h` is still avoided. The pool now outlives the source while the scheduled work still captures `this`, so the destructor waits for this source's own outstanding futures. Waiting on the whole pool, as before, would now block on unrelated queries. Drop the `create_reader_pool->wait` that ran when the lookahead was 1: it let a private one-thread pool release its thread before the next task was scheduled, and there is no private thread any more. Taking a reader from the queue already implies its callback finished, which is what that wait gave us. Teardown can still wait for a queued task to reach the front of the shared pool. That is memory-safe but not prompt; source-local cancellation would be a separate change. Co-Authored-By: Claude Opus 5 Signed-off-by: VighneshPath --- .../StorageObjectStorageSource.cpp | 32 +++++++++---------- .../StorageObjectStorageSource.h | 8 ++++- 2 files changed, 22 insertions(+), 18 deletions(-) diff --git a/src/Storages/ObjectStorage/StorageObjectStorageSource.cpp b/src/Storages/ObjectStorage/StorageObjectStorageSource.cpp index 48ebe1d0f927..51ac1d234cc7 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageSource.cpp +++ b/src/Storages/ObjectStorage/StorageObjectStorageSource.cpp @@ -165,15 +165,10 @@ 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( - std::make_shared( - CurrentMetrics::StorageObjectStorageThreads, - CurrentMetrics::StorageObjectStorageThreadsActive, - CurrentMetrics::StorageObjectStorageThreadsScheduled, - std::max(context_->getSettingsRef()[Setting::object_storage_max_files_to_prefetch], 1))) + , create_reader_pool(context_->getPrefetchThreadpool()) , file_iterator(file_iterator_) , schema_cache(StorageObjectStorage::getSchemaCache(context_, configuration->getTypeName())) - , create_reader_scheduler(threadPoolCallbackRunnerUnsafe(*create_reader_pool, ThreadName::READER_POOL)) + , create_reader_scheduler(threadPoolCallbackRunnerUnsafe(create_reader_pool, ThreadName::READER_POOL)) , max_files_to_prefetch(std::max(context_->getSettingsRef()[Setting::object_storage_max_files_to_prefetch], 1)) { } @@ -181,7 +176,14 @@ StorageObjectStorageSource::StorageObjectStorageSource( StorageObjectStorageSource::~StorageObjectStorageSource() { LOG_DEBUG(log, "Source finished: files_read={}", total_files_read); - create_reader_pool->wait(); + /// The scheduled work captures `this`, and create_reader_pool is shared with the rest of the + /// server, so waiting for the whole pool is both wrong (it would wait for unrelated queries) + /// and insufficient to express what we need. Wait for exactly this source's own futures. + for (auto & reader_future : reader_futures) + { + if (reader_future.valid()) + reader_future.wait(); + } } std::string StorageObjectStorageSource::getUniqueStoragePathIdentifier( @@ -687,15 +689,11 @@ Chunk StorageObjectStorageSource::generate() ++total_files_read; - if (max_files_to_prefetch <= 1) - { - /// 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(); - } - /// With max_files_to_prefetch > 1 the pool has multiple threads and multiple readers are - /// meant to be building/priming concurrently, so waiting for the whole pool here would - /// serialize exactly the concurrency this setting is meant to provide. + /// 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(); } diff --git a/src/Storages/ObjectStorage/StorageObjectStorageSource.h b/src/Storages/ObjectStorage/StorageObjectStorageSource.h index 5fa61b857002..c5f23e637e42 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageSource.h +++ b/src/Storages/ObjectStorage/StorageObjectStorageSource.h @@ -89,7 +89,13 @@ class StorageObjectStorageSource : public ISource FormatFilterInfoPtr format_filter_info; ReadFromFormatInfo read_from_format_info; - const std::shared_ptr create_reader_pool; + /// Server-wide pool for remote object storage prefetches (sized by the `prefetch_threadpool_pool_size` + /// server setting), the same one `MergeTreePrefetchedReadPool` submits to. It is shared on + /// purpose: a private pool per source sized from `object_storage_max_files_to_prefetch` would + /// make the thread count `streams * max_files_to_prefetch`, unbounded by anything global. + /// Because the 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 file_iterator; SchemaCache & schema_cache; From fbb6fb5ad48c592b54b19e09b2228d43e7f131e0 Mon Sep 17 00:00:00 2001 From: VighneshPath Date: Thu, 30 Jul 2026 17:43:53 +0530 Subject: [PATCH 6/6] Object storage: say in the setting description that prefetching reorders files Above 1, a stream prepares several files at once and they are claimed in whatever order the background tasks reach the shared file list, so the order files are read in is not deterministic. Which rows come back is unaffected, but rows arrive in a different order between runs when there is no `ORDER BY`. That was worth stating where a user actually looks before raising the setting. `2e3df41adc8` claimed the opposite, and the code comments and the test were corrected earlier; this is the same correction for the user-facing text. Co-Authored-By: Claude Opus 5 Signed-off-by: VighneshPath --- src/Core/Settings.cpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/Core/Settings.cpp b/src/Core/Settings.cpp index 8d83f82a2bfe..9f7c0b1854f8 100644 --- a/src/Core/Settings.cpp +++ b/src/Core/Settings.cpp @@ -4390,7 +4390,7 @@ Enables caching of rows number during count from files in table functions `file` 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). +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