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/Core/Settings.cpp b/src/Core/Settings.cpp index 770a99e6f63b..9f7c0b1854f8 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). 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 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..51ac1d234cc7 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 @@ -84,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; @@ -99,6 +110,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 +165,25 @@ 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, - 1 /* max_threads */)) + , 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)) { } 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( @@ -406,11 +421,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 +678,23 @@ Chunk StorageObjectStorageSource::generate() total_rows_in_file = 0; - assert(reader_future.valid()); - reader = reader_future.get(); + chassert(!reader_futures.empty()); + { + ProfileEventTimeIncrement 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 +1357,22 @@ 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) + { + 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 createReadBuffer( diff --git a/src/Storages/ObjectStorage/StorageObjectStorageSource.h b/src/Storages/ObjectStorage/StorageObjectStorageSource.h index d30329acce01..c5f23e637e42 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageSource.h +++ b/src/Storages/ObjectStorage/StorageObjectStorageSource.h @@ -1,4 +1,7 @@ #pragma once + +#include +#include #include #include #include @@ -86,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; @@ -122,6 +131,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 +149,33 @@ 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`). + /// 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> 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 +196,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(); 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"