diff --git a/src/Common/FailPoint.cpp b/src/Common/FailPoint.cpp index b8fee32bdf13..2421d5413c3e 100644 --- a/src/Common/FailPoint.cpp +++ b/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) \ 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..af0c8e6adaac 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,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( 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(*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`. 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 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::createReaderAsync() +std::future 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 createReadBuffer( diff --git a/src/Storages/ObjectStorage/StorageObjectStorageSource.h b/src/Storages/ObjectStorage/StorageObjectStorageSource.h index d30329acce01..af71585054db 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,19 @@ class StorageObjectStorageSource : public ISource FormatFilterInfoPtr format_filter_info; ReadFromFormatInfo read_from_format_info; - const std::shared_ptr 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 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 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(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 +155,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 +202,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" diff --git a/tests/queries/0_stateless/04627_object_storage_prefetch_pool_selection.reference b/tests/queries/0_stateless/04627_object_storage_prefetch_pool_selection.reference new file mode 100644 index 000000000000..d86bac9de59a --- /dev/null +++ b/tests/queries/0_stateless/04627_object_storage_prefetch_pool_selection.reference @@ -0,0 +1 @@ +OK diff --git a/tests/queries/0_stateless/04627_object_storage_prefetch_pool_selection.sh b/tests/queries/0_stateless/04627_object_storage_prefetch_pool_selection.sh new file mode 100755 index 000000000000..557bf8235075 --- /dev/null +++ b/tests/queries/0_stateless/04627_object_storage_prefetch_pool_selection.sh @@ -0,0 +1,80 @@ +#!/usr/bin/env bash +# Tags: no-fasttest, no-parallel +# no-fasttest: depends on MinIO via s3_conn. +# no-parallel: `object_storage_reader_pool_pause` is a process-wide PAUSEABLE failpoint. While it is +# enabled it would also pause background reader creation in any other object storage query +# running concurrently in a parallel test session. +# +# Regression test: at the default `object_storage_max_files_to_prefetch` (1), background reader +# creation must run on a dedicated single-thread pool (`StorageObjectStorageThreads*`), not on the +# shared, server-wide prefetch pool (`IOPrefetchThreads*`) that `MergeTreePrefetchedReadPool` also +# submits to - otherwise this source's prefetches contend with unrelated queries on that pool. +# Above 1, a private pool sized from the setting would make thread count unbounded +# (streams * max_files_to_prefetch), so the shared pool is used instead; that direction is checked +# too, so a change that always uses one pool regardless of the setting cannot pass either half. +# +# The pool is exercised even at the default of 1 (see `refillReaderFutures`): a lookahead future for +# the next file is always queued via `createReaderAsync`, just left unprimed. A pauseable failpoint +# at the top of that scheduled task holds it on whichever pool it was submitted to for long enough +# to read `system.metrics` and see which pool went active. + +CURDIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd) +# shellcheck source=../shell_config.sh +. "$CURDIR"/../shell_config.sh + +path="04627_prefetch_pool_${CLICKHOUSE_DATABASE}" + +function cleanup() +{ + $CLICKHOUSE_CLIENT -q "SYSTEM DISABLE FAILPOINT object_storage_reader_pool_pause" 2>/dev/null ||: +} +trap cleanup EXIT + +$CLICKHOUSE_CLIENT --query "DROP TABLE IF EXISTS test_04627_write" +$CLICKHOUSE_CLIENT --query " + CREATE TABLE test_04627_write (id UInt64) + ENGINE = S3(s3_conn, filename = '$path/file.parquet', format = Parquet)" +$CLICKHOUSE_CLIENT --query "INSERT INTO test_04627_write SELECT number FROM numbers(10)" +$CLICKHOUSE_CLIENT --query "DROP TABLE test_04627_write" + +function metric_value() +{ + $CLICKHOUSE_CLIENT --query "SELECT value FROM system.metrics WHERE metric = '$1'" +} + +# $1: object_storage_max_files_to_prefetch value to test +# $2: metric expected to go active (the pool this setting should route through) +# $3: metric expected to stay at its pre-query baseline (the pool it should NOT use) +function check_pool() +{ + local depth=$1 expect_active=$2 expect_idle=$3 + local idle_baseline active idle + idle_baseline=$(metric_value "$expect_idle") + + $CLICKHOUSE_CLIENT -q "SYSTEM ENABLE FAILPOINT object_storage_reader_pool_pause" + + $CLICKHOUSE_CLIENT --query " + SELECT count() FROM s3(s3_conn, filename = '$path/file.parquet', format = Parquet) + SETTINGS max_threads = 1, object_storage_max_files_to_prefetch = $depth" & + local q_pid=$! + + $CLICKHOUSE_CLIENT -q "SYSTEM WAIT FAILPOINT object_storage_reader_pool_pause PAUSE" + + active=$(metric_value "$expect_active") + idle=$(metric_value "$expect_idle") + + if [[ "$active" -lt 1 ]]; then + echo "FAIL depth=$depth: expected $expect_active active, got $active" + fi + if [[ "$idle" -ne "$idle_baseline" ]]; then + echo "FAIL depth=$depth: expected $expect_idle to stay at $idle_baseline, got $idle" + fi + + $CLICKHOUSE_CLIENT -q "SYSTEM DISABLE FAILPOINT object_storage_reader_pool_pause" + wait "$q_pid" +} + +check_pool 1 StorageObjectStorageThreadsActive IOPrefetchThreadsActive +check_pool 4 IOPrefetchThreadsActive StorageObjectStorageThreadsActive + +echo "OK"