From 79d3658690cbd2f798a02da5cbf87cbe28c8cf60 Mon Sep 17 00:00:00 2001 From: UnamedRus Date: Tue, 18 Aug 2026 16:45:42 +0300 Subject: [PATCH 1/4] Parquet v3: rebalance read-stage budgets for deeper prefetch concurrency Budget memory and threads separately per read stage instead of one shared fraction, and add a ColumnDataPrefetch stage that issues the compressed data-page reads (charged to its own memory budget) while ColumnData only decodes. The old single 0.2 fraction capped the data stage at 0.2 of both memory and threads, so only ~2 row groups were read/decoded ahead and the S3 link sat idle on latency-bound, high-RTT reads. Now compressed reads run deep (cheap per row group) while decoded row groups stay bounded, hiding per-GET latency. Also reconcile the decoded-memory charge up to the actual footprint inside decodePrimitiveColumn, before formOutputColumn moves the column, so the honest cap actually bounds decode-ahead. Squashed extraction of 34816a35b40 + 114640eeaf6 + f26050692be from the parquet-v3 feature branch onto antalya-26.6. Co-Authored-By: Claude Opus 4.8 (1M context) Signed-off-by: UnamedRus --- .../Formats/Impl/Parquet/ReadCommon.cpp | 12 ++- .../Formats/Impl/Parquet/ReadCommon.h | 8 +- .../Formats/Impl/Parquet/ReadManager.cpp | 96 +++++++++++++++---- .../Formats/Impl/Parquet/ReadManager.h | 4 + .../Formats/Impl/Parquet/Reader.cpp | 15 ++- src/Processors/Formats/Impl/Parquet/Reader.h | 2 +- 6 files changed, 109 insertions(+), 28 deletions(-) diff --git a/src/Processors/Formats/Impl/Parquet/ReadCommon.cpp b/src/Processors/Formats/Impl/Parquet/ReadCommon.cpp index 953562e818b4..736f8ec0ed9b 100644 --- a/src/Processors/Formats/Impl/Parquet/ReadCommon.cpp +++ b/src/Processors/Formats/Impl/Parquet/ReadCommon.cpp @@ -6,15 +6,17 @@ namespace DB::Parquet { -SharedResourcesExt::Limits SharedResourcesExt::getLimitsPerReader(const FormatParserSharedResources & parser_shared_resources, double fraction) +SharedResourcesExt::Limits SharedResourcesExt::getLimitsPerReader(const FormatParserSharedResources & parser_shared_resources, double memory_fraction, double thread_fraction) { const SharedResourcesExt & ext = *static_cast(parser_shared_resources.opaque.get()); size_t n = parser_shared_resources.num_streams.load(std::memory_order_relaxed); - fraction /= static_cast(std::max(n, size_t(1))); + /// Split each budget across the files read in parallel. + memory_fraction /= static_cast(std::max(n, size_t(1))); + thread_fraction /= static_cast(std::max(n, size_t(1))); return Limits { - .memory_low_watermark = size_t(ext.total_memory_low_watermark * fraction), - .memory_high_watermark = size_t(ext.total_memory_high_watermark * fraction), - .parsing_threads = size_t(std::max(std::lround(parser_shared_resources.parsing_runner.getMaxThreads() * fraction + .5), 1l))}; + .memory_low_watermark = size_t(ext.total_memory_low_watermark * memory_fraction), + .memory_high_watermark = size_t(ext.total_memory_high_watermark * memory_fraction), + .parsing_threads = size_t(std::max(std::lround(parser_shared_resources.parsing_runner.getMaxThreads() * thread_fraction + .5), 1l))}; } #ifdef OS_LINUX diff --git a/src/Processors/Formats/Impl/Parquet/ReadCommon.h b/src/Processors/Formats/Impl/Parquet/ReadCommon.h index 76b8a0fbccd5..cbaf7f095ec6 100644 --- a/src/Processors/Formats/Impl/Parquet/ReadCommon.h +++ b/src/Processors/Formats/Impl/Parquet/ReadCommon.h @@ -50,7 +50,7 @@ struct SharedResourcesExt size_t parsing_threads; }; - static Limits getLimitsPerReader(const FormatParserSharedResources & parser_shared_resources, double fraction); + static Limits getLimitsPerReader(const FormatParserSharedResources & parser_shared_resources, double memory_fraction, double thread_fraction); }; @@ -88,6 +88,9 @@ enum class ReadStage ColumnIndexAndOffsetIndex, OffsetIndex, + /// Issues the compressed data-page reads (startPrefetch), no decode. Own memory budget, so many + /// row groups prefetch ahead while only a few decode at once. Decouples fetch from decode depth. + ColumnDataPrefetch, ColumnData, Deliver, @@ -186,6 +189,9 @@ class MemoryUsageToken val += amount; } + /// How much memory this token currently charges. + size_t charged() const { return val; } + private: ReadStage alloc_stage = ReadStage::Deallocated; size_t val = 0; diff --git a/src/Processors/Formats/Impl/Parquet/ReadManager.cpp b/src/Processors/Formats/Impl/Parquet/ReadManager.cpp index 3408cd99c032..15fce3ce890f 100644 --- a/src/Processors/Formats/Impl/Parquet/ReadManager.cpp +++ b/src/Processors/Formats/Impl/Parquet/ReadManager.cpp @@ -74,17 +74,45 @@ void ReadManager::init(FormatParserSharedResourcesPtr parser_shared_resources_, stages[i].row_group_tasks_to_schedule.resize(num_row_groups); } - /// Distribute memory budget among stages. - /// The distribution is static to make sure no stage gets starved if others eat all the memory. - /// E.g. if the budget was shared among all stages, maybe PrewhereData could run far ahead and - /// The distribution is static to make sure no stage gets starved if others eat all the memory. - double sum = 0; - stages[size_t(ReadStage::NotStarted)].memory_target_fraction = 0; - stages[size_t(ReadStage::Deliver)].memory_target_fraction = 0; + /// Distribute the memory and thread budgets among stages. + /// The distribution is static to make sure no stage gets starved if others eat all the resources. + /// E.g. if the budget was shared among all stages, maybe ColumnData could run far ahead and eat + /// all the memory, starving the small index reads that other row groups need to make progress. + /// + /// Budget memory and threads separately: a single 0.2 fraction capped ColumnData at 0.2 of both, + /// so only ~2 row groups were read/decoded ahead. Give ColumnData most of the memory and threads + /// (deep, decode-bound); give the small latency-bound index/bloom reads threads for parallelism. + using S = ReadStage; + auto set_fractions = [&](S s, double memory_fraction, double thread_fraction) + { + stages[size_t(s)].memory_target_fraction = memory_fraction; + stages[size_t(s)].thread_target_fraction = thread_fraction; + }; + set_fractions(S::NotStarted, 0, 0); + set_fractions(S::BloomFilterHeader, 0.05, 1); + set_fractions(S::BloomFilterBlocksOrDictionary, 0.10, 1); + set_fractions(S::ColumnIndexAndOffsetIndex, 0.05, 1); + set_fractions(S::OffsetIndex, 0.05, 1); + /// Compressed prefetch: large memory (cheap per row group) so many reads run ahead; 1 thread + /// (startPrefetch is cheap, reads run in the Prefetcher's own io pool). + set_fractions(S::ColumnDataPrefetch, 0.45, 1); + /// Decode: bounded memory (decoded row groups are large) but most threads. Caps resident decoded + /// row groups independently of prefetch depth. + set_fractions(S::ColumnData, 0.30, 3); + set_fractions(S::Deliver, 0, 0); + + double memory_sum = 0; + double thread_sum = 0; for (const Stage & stage : stages) - sum += stage.memory_target_fraction; + { + memory_sum += stage.memory_target_fraction; + thread_sum += stage.thread_target_fraction; + } for (Stage & stage : stages) - stage.memory_target_fraction /= sum; + { + stage.memory_target_fraction /= memory_sum; + stage.thread_target_fraction /= thread_sum; + } /// The NotStarted stage completed for all row groups, transition to next stage. MemoryUsageDiff diff(ReadStage::NotStarted); @@ -139,6 +167,7 @@ void ReadManager::finishRowGroupStage(size_t row_group_idx, ReadStage stage, Mem switch (stage) { case ReadStage::NotStarted: + case ReadStage::ColumnDataPrefetch: case ReadStage::ColumnData: case ReadStage::Deliver: chassert(false); @@ -295,9 +324,10 @@ void ReadManager::addTasksToReadColumns(size_t row_group_idx, size_t row_subgrou } else { - LOG_TEST(getLogger("ParquetReadManager"), "addTasksToReadColumns: added ColumnData: i={} step_idx={} row_group_idx={} row_subgroup_idx={}", i, step_idx, row_group_idx, row_subgroup_idx); + /// `stage` is ColumnDataPrefetch (issue reads) or ColumnData (decode). + LOG_TEST(getLogger("ParquetReadManager"), "addTasksToReadColumns: added {}: i={} step_idx={} row_group_idx={} row_subgroup_idx={}", magic_enum::enum_name(stage), i, step_idx, row_group_idx, row_subgroup_idx); add_tasks.push_back(Task { - .stage = ReadStage::ColumnData, + .stage = stage, .step_idx = step_idx, .row_group_idx = row_group_idx, .row_subgroup_idx = row_subgroup_idx, @@ -307,8 +337,8 @@ void ReadManager::addTasksToReadColumns(size_t row_group_idx, size_t row_subgrou if (add_tasks.empty() && is_offset_index) { - /// Don't need to read offset index, move on to next stage (ColumnData). - stage = ReadStage::ColumnData; + /// Don't need to read offset index, move on to the next stage (ColumnDataPrefetch). + stage = ReadStage::ColumnDataPrefetch; continue; } @@ -319,7 +349,7 @@ void ReadManager::addTasksToReadColumns(size_t row_group_idx, size_t row_subgrou /// (RowSubgroup.filter.memory) work correctly when PREWHERE expression doesn't use any /// columns (note: the expression may still be nontrivial, e.g. `rand()%2=0`).) add_tasks.push_back(Task { - .stage = ReadStage::ColumnData, + .stage = stage, .step_idx = step_idx, .row_group_idx = row_group_idx, .row_subgroup_idx = row_subgroup_idx, @@ -420,6 +450,13 @@ void ReadManager::finishRowSubgroupStage(size_t row_group_idx, size_t row_subgro case ReadStage::ColumnIndexAndOffsetIndex: case ReadStage::OffsetIndex: { + /// Prerequisites read; issue the compressed data-page reads (but don't decode yet). + addTasksToReadColumns(row_group_idx, row_subgroup_idx, ReadStage::ColumnDataPrefetch, step_idx, diff); + return; + } + case ReadStage::ColumnDataPrefetch: + { + /// Data-page reads issued (in flight in the Prefetcher's io pool); now decode. addTasksToReadColumns(row_group_idx, row_subgroup_idx, ReadStage::ColumnData, step_idx, diff); return; } @@ -544,7 +581,7 @@ void ReadManager::flushMemoryUsageDiff(MemoryUsageDiff && diff) if (!should_schedule && d < 0) { const auto & stage = stages[i]; - auto limits = SharedResourcesExt::getLimitsPerReader(*parser_shared_resources, stage.memory_target_fraction); + auto limits = SharedResourcesExt::getLimitsPerReader(*parser_shared_resources, stage.memory_target_fraction, stage.thread_target_fraction); should_schedule = checkTaskSchedulingLimits( stage.memory_usage.load(std::memory_order_relaxed), 0, stage.batches_in_progress.load(std::memory_order_relaxed), 0, limits); @@ -562,7 +599,7 @@ void ReadManager::scheduleTasksIfNeeded(ReadStage stage_idx) MemoryUsageDiff diff(stage_idx); std::vector tasks; - auto limits = SharedResourcesExt::getLimitsPerReader(*parser_shared_resources, stage.memory_target_fraction); + auto limits = SharedResourcesExt::getLimitsPerReader(*parser_shared_resources, stage.memory_target_fraction, stage.thread_target_fraction); size_t memory_usage = stage.memory_usage.load(std::memory_order_relaxed); size_t batches_in_progress = stage.batches_in_progress.load(std::memory_order_relaxed); @@ -715,12 +752,16 @@ void ReadManager::scheduleTask(Task task, bool is_first_in_group, MemoryUsageDif case ReadStage::OffsetIndex: prefetches.push_back(&column.offset_index_prefetch); break; - case ReadStage::ColumnData: + case ReadStage::ColumnDataPrefetch: { RowSubgroup & row_subgroup = row_group.subgroups.at(task.row_subgroup_idx); - ColumnSubchunk & subchunk = row_subgroup.columns.at(task.column_idx); if (row_subgroup.filter.rows_pass == 0) break; + /// Determine which data pages this subgroup needs and queue their reads. The + /// startPrefetch at the end of this function issues them against the Prefetcher's io + /// pool and charges the compressed bytes to the ColumnDataPrefetch stage budget - + /// separate from the decoded-output budget (ColumnData) - so many row groups can have + /// their reads in flight (deep prefetch) while only a few are decoded at once. reader.determinePagesToPrefetch(column, row_subgroup, row_group, prefetches); /// Side note: would be nice to avoid reading the dictionary if all dictionary-encoded @@ -737,7 +778,17 @@ void ReadManager::scheduleTask(Task task, bool is_first_in_group, MemoryUsageDif { prefetches.push_back(&column.data_pages_prefetch); } - + break; + } + case ReadStage::ColumnData: + { + RowSubgroup & row_subgroup = row_group.subgroups.at(task.row_subgroup_idx); + ColumnSubchunk & subchunk = row_subgroup.columns.at(task.column_idx); + if (row_subgroup.filter.rows_pass == 0) + break; + /// The data-page reads were already issued in ColumnDataPrefetch (and are in flight or + /// done in the Prefetcher). Here we only reserve the estimated decoded-output memory + /// against the ColumnData budget; runTask then decodes from those buffers. double bytes_per_row = reader.estimateColumnMemoryBytesPerRow(column, row_group, reader.primitive_columns.at(task.column_idx)); size_t column_memory = static_cast(bytes_per_row * static_cast(row_subgroup.filter.rows_pass)); subchunk.column_and_offsets_memory = MemoryUsageToken(column_memory, &diff); @@ -842,6 +893,11 @@ void ReadManager::runTask(Task task, bool last_in_batch, MemoryUsageDiff & diff) reader.decodeOffsetIndex(column, row_group); column.offset_index_prefetch.reset(&diff); break; + case ReadStage::ColumnDataPrefetch: + /// The compressed data-page reads were already issued in scheduleTask (startPrefetch) + /// and proceed asynchronously in the Prefetcher's io pool. Nothing to do here; the + /// subgroup advances to ColumnData, which decodes from those buffers. + break; case ReadStage::ColumnData: { RowSubgroup & row_subgroup = row_group.subgroups.at(task.row_subgroup_idx); @@ -859,7 +915,7 @@ void ReadManager::runTask(Task task, bool last_in_batch, MemoryUsageDiff & diff) chassert(task.row_subgroup_idx != UINT64_MAX); reader.decodePrimitiveColumn( column, column_info, row_subgroup.columns.at(task.column_idx), - row_group, row_subgroup); + row_group, row_subgroup, diff); for (size_t i = prev_page_idx; i < column.data_pages_idx; ++i) { diff --git a/src/Processors/Formats/Impl/Parquet/ReadManager.h b/src/Processors/Formats/Impl/Parquet/ReadManager.h index 49ac1b2942c8..7073492d2174 100644 --- a/src/Processors/Formats/Impl/Parquet/ReadManager.h +++ b/src/Processors/Formats/Impl/Parquet/ReadManager.h @@ -88,7 +88,11 @@ class ReadManager /// Tasks that are either in thread pool's queue or executing. std::atomic batches_in_progress {0}; + /// Share of the query-global memory budget for this stage, kept separate from the thread + /// share so a stage needing parallelism but little memory isn't forced to trade one off. double memory_target_fraction = 1; + /// Share of the parsing thread pool for this stage, independent of the memory share. + double thread_target_fraction = 1; /// We take advantage of the fact that each pair can have at most one group /// of tasks in flight at a time. E.g. we create tasks to read columns in subgroup n, then diff --git a/src/Processors/Formats/Impl/Parquet/Reader.cpp b/src/Processors/Formats/Impl/Parquet/Reader.cpp index e0422d65ce2d..f32a7dcc8665 100644 --- a/src/Processors/Formats/Impl/Parquet/Reader.cpp +++ b/src/Processors/Formats/Impl/Parquet/Reader.cpp @@ -1328,7 +1328,7 @@ double Reader::estimateColumnMemoryBytesPerRow(const ColumnChunk & column, const return res; } -void Reader::decodePrimitiveColumn(ColumnChunk & column, const PrimitiveColumnInfo & column_info, ColumnSubchunk & subchunk, const RowGroup & row_group, RowSubgroup & row_subgroup) +void Reader::decodePrimitiveColumn(ColumnChunk & column, const PrimitiveColumnInfo & column_info, ColumnSubchunk & subchunk, const RowGroup & row_group, RowSubgroup & row_subgroup, MemoryUsageDiff & diff) { /// Allocate columns for values, null map, and array offsets. @@ -1503,6 +1503,19 @@ void Reader::decodePrimitiveColumn(ColumnChunk & column, const PrimitiveColumnIn chassert(subchunk.column->getDataType() == column_info.output_type->getColumnType()); + /// The scheduleTask charge was an estimate; reconcile up to the actual decoded footprint here, + /// before formOutputColumn (below) moves `subchunk.column`, so the scheduler stops decoding ahead + /// before real RAM exceeds the cap. Grow-only (fail-closed). + size_t actual_bytes = subchunk.column->allocatedBytes(); + for (const auto & offsets : subchunk.arrays_offsets) + if (offsets) + actual_bytes += offsets->allocatedBytes(); + if (subchunk.group_null_map) + actual_bytes += subchunk.group_null_map->allocatedBytes(); + size_t already_charged = subchunk.column_and_offsets_memory.charged(); + if (actual_bytes > already_charged) + subchunk.column_and_offsets_memory.add(actual_bytes - already_charged, &diff); + OutputColumnState & state = row_subgroup.output.at(column_info.idx_in_output_block); chassert(!state.column); size_t prev_count = state.primitive_columns_remaining.fetch_sub(1); diff --git a/src/Processors/Formats/Impl/Parquet/Reader.h b/src/Processors/Formats/Impl/Parquet/Reader.h index 0ac46ac11f31..049bb11efe27 100644 --- a/src/Processors/Formats/Impl/Parquet/Reader.h +++ b/src/Processors/Formats/Impl/Parquet/Reader.h @@ -532,7 +532,7 @@ struct Reader /// Guess how much memory ColumnSubchunk::{column, arrays_offsets} will use, per row. double estimateColumnMemoryBytesPerRow(const ColumnChunk & column, const RowGroup & row_group, const PrimitiveColumnInfo & column_info) const; - void decodePrimitiveColumn(ColumnChunk & column, const PrimitiveColumnInfo & column_info, ColumnSubchunk & subchunk, const RowGroup & row_group, RowSubgroup & row_subgroup); + void decodePrimitiveColumn(ColumnChunk & column, const PrimitiveColumnInfo & column_info, ColumnSubchunk & subchunk, const RowGroup & row_group, RowSubgroup & row_subgroup, MemoryUsageDiff & diff); /// Returns mutable column because some of the recursive calls require it, /// e.g. ColumnArray::create does assumeMutable() on the nested columns. From 91a8a069c227ae617aedddf02c5a105b99eee5d2 Mon Sep 17 00:00:00 2001 From: UnamedRus Date: Fri, 21 Aug 2026 16:11:46 +0300 Subject: [PATCH 2/4] Parquet: expose read-stage budget weights as a setting The reader scheduler splits a static memory and thread budget across the read stages (bloom/index reads, compressed prefetch, decode) using fixed relative weights. Those weights are the main lever for latency-bound object-storage reads - the compressed prefetch weight sets the read-ahead depth, the decode thread weight sets decode concurrency - but they were hardcoded, so tuning them required a rebuild. Add `input_format_parquet_read_stage_weights`, an experimental `Map` setting. Keys are `.` (resource is `memory` or `threads`); values are relative weights, normalized per resource across stages exactly like the built-in defaults. Only the listed keys override; unspecified stages keep their defaults, so an empty map (the default) reproduces the previous behavior bit for bit. Unknown stage or resource names, negative weights, and a configuration that zeroes out the whole memory or thread budget all raise `BAD_ARGUMENTS` rather than silently falling back. Co-Authored-By: Claude Opus 4.8 (1M context) Signed-off-by: UnamedRus --- src/Core/FormatFactorySettings.h | 3 + src/Core/SettingsChangesHistory.cpp | 1 + src/Formats/FormatFactory.cpp | 25 +++++ src/Formats/FormatSettings.h | 6 + .../Formats/Impl/Parquet/ReadCommon.h | 4 + .../Formats/Impl/Parquet/ReadManager.cpp | 103 ++++++++++++++---- .../Impl/ParquetV3BlockInputFormat.cpp | 1 + 7 files changed, 124 insertions(+), 19 deletions(-) diff --git a/src/Core/FormatFactorySettings.h b/src/Core/FormatFactorySettings.h index 67396c955eac..38cb6d71cf50 100644 --- a/src/Core/FormatFactorySettings.h +++ b/src/Core/FormatFactorySettings.h @@ -201,6 +201,9 @@ Schedule prefetches more aggressively if memory usage is below than threshold. P DECLARE(UInt64, input_format_parquet_memory_high_watermark, 4ul << 30, R"( Approximate memory limit for Parquet reader v3. Limits how many row groups or columns can be read in parallel. When reading multiple files in one query, the limit is on total memory usage across those files. )", 0) \ + DECLARE(Map, input_format_parquet_read_stage_weights, "", R"( +Expert tuning knob for the Parquet reader scheduler. Overrides the relative memory and thread budget weights assigned to each read stage. Keys have the form `.`, where `` is `memory` or `threads` and `` is one of `bloom_filter_header`, `bloom_filter_blocks_or_dictionary`, `column_index_and_offset_index`, `offset_index`, `column_data_prefetch`, `column_data`. Values are relative weights (they need not sum to 1; each is normalized against the sum of that resource across stages). Only the specified keys override the built-in defaults; unspecified stages keep their defaults. Example: `{'column_data_prefetch.memory': 12, 'column_data.threads': 4}`. Advanced and experimental: the internal stage model may change between versions. +)", EXPERIMENTAL) \ DECLARE(Bool, input_format_parquet_page_filter_push_down, true, R"( Skip pages using min/max values from column index. )", 0) \ diff --git a/src/Core/SettingsChangesHistory.cpp b/src/Core/SettingsChangesHistory.cpp index 37d38b0ef104..5f1299f5b374 100644 --- a/src/Core/SettingsChangesHistory.cpp +++ b/src/Core/SettingsChangesHistory.cpp @@ -44,6 +44,7 @@ const VersionToSettingsChangesMap & getSettingsChangesHistory() {"analyzer_compatibility_apply_final_to_all_joined_tables", true, false, "Fixed a bug in the analyzer where FINAL on the left-most table of a JOIN was incorrectly applied to the other joined tables as well. previous_value=true so `compatibility` with versions before 26.6 restores the old behavior."}, {"analyzer_compatibility_allow_non_aggregate_in_having", false, false, "New compatibility setting. When enabled, the new analyzer mimics the legacy `HAVING`-to-`WHERE` rewrite for non-aggregate AND-conjuncts instead of raising `NOT_AN_AGGREGATE`."}, {"reserve_memory", 0, 0, "New setting to reserve memory for specific workload before starting a query."}, + {"input_format_parquet_read_stage_weights", "", "", "New experimental setting to override the per-stage memory and thread budget weights of the Parquet reader scheduler."}, {"output_format_image_width", 1024, 1024, "New setting controlling the width of the output image for image output formats such as PNG."}, {"output_format_image_height", 1024, 1024, "New setting controlling the height of the output image for image output formats such as PNG."}, {"output_format_image_terminal_mode", "", "", "New setting controlling whether image output formats such as PNG are rendered directly to the terminal using an inline image protocol."}, diff --git a/src/Formats/FormatFactory.cpp b/src/Formats/FormatFactory.cpp index b40fb23ca262..02c67cd6aa4a 100644 --- a/src/Formats/FormatFactory.cpp +++ b/src/Formats/FormatFactory.cpp @@ -223,6 +223,31 @@ FormatSettings getFormatSettings(const ContextPtr & context, const Settings & se format_settings.parquet.enable_json_parsing = settings[Setting::input_format_parquet_enable_json_parsing]; format_settings.parquet.memory_low_watermark = settings[Setting::input_format_parquet_memory_low_watermark]; format_settings.parquet.memory_high_watermark = settings[Setting::input_format_parquet_memory_high_watermark]; + { + const Map & stage_weights = settings[Setting::input_format_parquet_read_stage_weights].value; + for (const auto & key_value : stage_weights) + { + if (key_value.getType() != Field::Types::Tuple || key_value.safeGet().size() != 2) + throw Exception(ErrorCodes::BAD_ARGUMENTS, + "The value of the `input_format_parquet_read_stage_weights` setting must be a Map(String, Float64)"); + const Tuple & pair = key_value.safeGet(); + if (pair.at(0).getType() != Field::Types::String) + throw Exception(ErrorCodes::BAD_ARGUMENTS, + "The keys of the `input_format_parquet_read_stage_weights` setting must be Strings"); + const Field & weight = pair.at(1); + double value; + switch (weight.getType()) + { + case Field::Types::UInt64: value = static_cast(weight.safeGet()); break; + case Field::Types::Int64: value = static_cast(weight.safeGet()); break; + case Field::Types::Float64: value = weight.safeGet(); break; + default: + throw Exception(ErrorCodes::BAD_ARGUMENTS, + "The values of the `input_format_parquet_read_stage_weights` setting must be numbers"); + } + format_settings.parquet.read_stage_weights[pair.at(0).safeGet()] = value; + } + } format_settings.parquet.allow_missing_columns = settings[Setting::input_format_parquet_allow_missing_columns]; format_settings.parquet.skip_columns_with_unsupported_types_in_schema_inference = settings[Setting::input_format_parquet_skip_columns_with_unsupported_types_in_schema_inference]; format_settings.parquet.output_string_as_string = settings[Setting::output_format_parquet_string_as_string]; diff --git a/src/Formats/FormatSettings.h b/src/Formats/FormatSettings.h index 728a1faba670..1620e4603451 100644 --- a/src/Formats/FormatSettings.h +++ b/src/Formats/FormatSettings.h @@ -5,6 +5,8 @@ #include #include +#include + namespace DB { @@ -361,6 +363,10 @@ struct FormatSettings size_t local_read_min_bytes_for_seek = 8192; size_t memory_low_watermark = 2ul << 20; size_t memory_high_watermark = 4ul << 30; + /// Optional per-stage budget weight overrides for the reader scheduler, keyed by + /// "." (resource = "memory" | "threads"). Empty means use the built-in + /// defaults. Parsed from the input_format_parquet_read_stage_weights setting. + std::map read_stage_weights; /// Write. UInt64 row_group_rows = 1000000; diff --git a/src/Processors/Formats/Impl/Parquet/ReadCommon.h b/src/Processors/Formats/Impl/Parquet/ReadCommon.h index cbaf7f095ec6..b44723e5bf00 100644 --- a/src/Processors/Formats/Impl/Parquet/ReadCommon.h +++ b/src/Processors/Formats/Impl/Parquet/ReadCommon.h @@ -42,6 +42,10 @@ struct SharedResourcesExt { size_t total_memory_low_watermark = 0; size_t total_memory_high_watermark = 0; + /// Optional per-stage budget weight overrides, keyed by "." + /// (resource = "memory" | "threads"). Empty means use the built-in defaults. + /// Applied in ReadManager::init. See input_format_parquet_read_stage_weights. + std::map read_stage_weights; struct Limits { diff --git a/src/Processors/Formats/Impl/Parquet/ReadManager.cpp b/src/Processors/Formats/Impl/Parquet/ReadManager.cpp index 15fce3ce890f..92d6888695f3 100644 --- a/src/Processors/Formats/Impl/Parquet/ReadManager.cpp +++ b/src/Processors/Formats/Impl/Parquet/ReadManager.cpp @@ -13,10 +13,12 @@ #include #include #include +#include #include namespace DB::ErrorCodes { + extern const int BAD_ARGUMENTS; extern const int LOGICAL_ERROR; extern const int QUERY_WAS_CANCELLED; } @@ -79,27 +81,86 @@ void ReadManager::init(FormatParserSharedResourcesPtr parser_shared_resources_, /// E.g. if the budget was shared among all stages, maybe ColumnData could run far ahead and eat /// all the memory, starving the small index reads that other row groups need to make progress. /// - /// Budget memory and threads separately: a single 0.2 fraction capped ColumnData at 0.2 of both, - /// so only ~2 row groups were read/decoded ahead. Give ColumnData most of the memory and threads - /// (deep, decode-bound); give the small latency-bound index/bloom reads threads for parallelism. - using S = ReadStage; - auto set_fractions = [&](S s, double memory_fraction, double thread_fraction) + /// The values below are relative weights, not fractions: they need not sum to 1 (the memory weights + /// sum to 20, the thread weights to 8). The normalization loop right after divides each by the + /// per-resource sum to turn them into the actual budget fractions stored in the Stage. A stage then + /// gets `weight/sum` of the query-global memory watermark and of the parsing thread pool (further + /// divided across files read in parallel, see `getLimitsPerReader`). + /// + /// Memory and threads are budgeted separately (an earlier single 0.2 fraction capped ColumnData at + /// 0.2 of *both*, so only ~2 row groups were read/decoded ahead) because they answer two different + /// questions, and the weights follow from those: + /// + /// Threads = "does a task in this stage occupy a CPU?" Index/bloom/prefetch tasks only *issue* an + /// async read - they return immediately and the actual IO runs in the Prefetcher's own io pool - so + /// 1 slot each is enough to keep issuing and extra slots buy nothing. Decode (ColumnData) is real + /// CPU work (decompress + decode), so it gets the majority (3 of 8) to overlap several row groups + /// without monopolizing the pool and starving the cheap read-issuing stages. + /// + /// Memory = "how much outstanding work does a byte here buy?" ColumnDataPrefetch holds *compressed* + /// pages in flight, i.e. the read-ahead depth (~bandwidth-delay product): high-RTT S3 needs many + /// GETs outstanding to hide latency, and compressed pages are small, so memory here buys the most + /// concurrency per byte - hence the largest share (9). ColumnData holds *decoded* columns, which are + /// far larger per row group, so it is bounded (6) to keep resident decoded groups from blowing the + /// budget. Index/bloom footprints are tiny (a few KB per row group), so they get just a floor to + /// keep a few row groups' indexes resident for pipelining; BloomFilterBlocksOrDictionary gets 2 vs + /// 1 because dictionary/bloom blocks are larger than headers and offset indexes. + /// + /// The exact integers are relative priorities tuned empirically on high-RTT S3, not derived constants. + auto set_weights = [&](ReadStage s, double memory_weight, double thread_weight) { - stages[size_t(s)].memory_target_fraction = memory_fraction; - stages[size_t(s)].thread_target_fraction = thread_fraction; + stages[size_t(s)].memory_target_fraction = memory_weight; + stages[size_t(s)].thread_target_fraction = thread_weight; }; - set_fractions(S::NotStarted, 0, 0); - set_fractions(S::BloomFilterHeader, 0.05, 1); - set_fractions(S::BloomFilterBlocksOrDictionary, 0.10, 1); - set_fractions(S::ColumnIndexAndOffsetIndex, 0.05, 1); - set_fractions(S::OffsetIndex, 0.05, 1); - /// Compressed prefetch: large memory (cheap per row group) so many reads run ahead; 1 thread - /// (startPrefetch is cheap, reads run in the Prefetcher's own io pool). - set_fractions(S::ColumnDataPrefetch, 0.45, 1); - /// Decode: bounded memory (decoded row groups are large) but most threads. Caps resident decoded - /// row groups independently of prefetch depth. - set_fractions(S::ColumnData, 0.30, 3); - set_fractions(S::Deliver, 0, 0); + set_weights(ReadStage::NotStarted, 0, 0); + set_weights(ReadStage::BloomFilterHeader, 1, 1); + set_weights(ReadStage::BloomFilterBlocksOrDictionary, 2, 1); + set_weights(ReadStage::ColumnIndexAndOffsetIndex, 1, 1); + set_weights(ReadStage::OffsetIndex, 1, 1); + set_weights(ReadStage::ColumnDataPrefetch, 9, 1); + set_weights(ReadStage::ColumnData, 6, 3); + set_weights(ReadStage::Deliver, 0, 0); + + /// Expert override of the weights above, from input_format_parquet_read_stage_weights. + /// Keys are "."; only listed keys override, the rest keep the defaults. + const auto & ext = *static_cast(parser_shared_resources->opaque.get()); + if (!ext.read_stage_weights.empty()) + { + static const std::unordered_map stage_by_name = + { + {"bloom_filter_header", ReadStage::BloomFilterHeader}, + {"bloom_filter_blocks_or_dictionary", ReadStage::BloomFilterBlocksOrDictionary}, + {"column_index_and_offset_index", ReadStage::ColumnIndexAndOffsetIndex}, + {"offset_index", ReadStage::OffsetIndex}, + {"column_data_prefetch", ReadStage::ColumnDataPrefetch}, + {"column_data", ReadStage::ColumnData}, + }; + for (const auto & [key, weight] : ext.read_stage_weights) + { + if (weight < 0) + throw Exception(ErrorCodes::BAD_ARGUMENTS, + "Negative weight {} for '{}' in input_format_parquet_read_stage_weights", weight, key); + size_t dot = key.rfind('.'); + if (dot == String::npos) + throw Exception(ErrorCodes::BAD_ARGUMENTS, + "Key '{}' in input_format_parquet_read_stage_weights must have the form '.'", key); + std::string_view stage_name(key.data(), dot); + std::string_view resource(key.data() + dot + 1, key.size() - dot - 1); + auto it = stage_by_name.find(stage_name); + if (it == stage_by_name.end()) + throw Exception(ErrorCodes::BAD_ARGUMENTS, + "Unknown stage '{}' in input_format_parquet_read_stage_weights key '{}'", stage_name, key); + Stage & stage = stages[size_t(it->second)]; + if (resource == "memory") + stage.memory_target_fraction = weight; + else if (resource == "threads") + stage.thread_target_fraction = weight; + else + throw Exception(ErrorCodes::BAD_ARGUMENTS, + "Unknown resource '{}' in input_format_parquet_read_stage_weights key '{}' (expected 'memory' or 'threads')", + resource, key); + } + } double memory_sum = 0; double thread_sum = 0; @@ -108,6 +169,10 @@ void ReadManager::init(FormatParserSharedResourcesPtr parser_shared_resources_, memory_sum += stage.memory_target_fraction; thread_sum += stage.thread_target_fraction; } + if (memory_sum <= 0 || thread_sum <= 0) + throw Exception(ErrorCodes::BAD_ARGUMENTS, + "input_format_parquet_read_stage_weights leaves the total {} weight at zero", + memory_sum <= 0 ? "memory" : "thread"); for (Stage & stage : stages) { stage.memory_target_fraction /= memory_sum; diff --git a/src/Processors/Formats/Impl/ParquetV3BlockInputFormat.cpp b/src/Processors/Formats/Impl/ParquetV3BlockInputFormat.cpp index edbf421ccbeb..b4a64e089ee0 100644 --- a/src/Processors/Formats/Impl/ParquetV3BlockInputFormat.cpp +++ b/src/Processors/Formats/Impl/ParquetV3BlockInputFormat.cpp @@ -90,6 +90,7 @@ void ParquetV3BlockInputFormat::initializeIfNeeded() ext->total_memory_low_watermark = format_settings.parquet.memory_low_watermark; ext->total_memory_high_watermark = format_settings.parquet.memory_high_watermark; + ext->read_stage_weights = format_settings.parquet.read_stage_weights; parser_shared_resources->opaque = ext; }); From df7105e1d658c2c58ce06c9411c7b2f8c1fbc016 Mon Sep 17 00:00:00 2001 From: UnamedRus Date: Mon, 24 Aug 2026 12:15:38 +0300 Subject: [PATCH 3/4] Parquet v3: don't fragment no-op ColumnDataPrefetch tasks into many batches `ColumnDataPrefetch` tasks do all their work (`determinePagesToPrefetch` + `startPrefetch`) synchronously in `scheduleTask`; their `runTask` body is empty and the reads proceed asynchronously in the Prefetcher's io pool. Their run time is therefore ~0, but `cost_estimate_bytes` was set to the charged compressed bytes, which are large. `scheduleTasksIfNeeded` uses `cost_estimate_bytes` as a proxy for run time to group tiny tasks into batches. Reporting the large compressed-byte charge made the batcher split these zero-work tasks across up to `parsing_threads` batches - one thread-pool dispatch each - for tasks that do nothing on the thread. Report a cost of 0 for `ColumnDataPrefetch` tasks so they collapse into a single batch, removing the spurious per-subgroup dispatch overhead introduced by the prefetch/decode stage split. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/Processors/Formats/Impl/Parquet/ReadManager.cpp | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/src/Processors/Formats/Impl/Parquet/ReadManager.cpp b/src/Processors/Formats/Impl/Parquet/ReadManager.cpp index 92d6888695f3..3c77364202e4 100644 --- a/src/Processors/Formats/Impl/Parquet/ReadManager.cpp +++ b/src/Processors/Formats/Impl/Parquet/ReadManager.cpp @@ -882,8 +882,16 @@ void ReadManager::scheduleTask(Task task, bool is_first_in_group, MemoryUsageDif /// E.g. main data read task's memory estimate consists of the input page sizes and the output /// column size; the run time is also roughly proportional to these sizes. /// Hope it's a good enough proxy in all cases. + /// + /// Exception: ColumnDataPrefetch tasks do all their work (startPrefetch) synchronously here and + /// have an empty runTask, so their run time is ~0 regardless of how many compressed bytes they + /// charge. Reporting the (large) charged bytes as the cost would make the batching split them + /// across many batches - i.e. many thread-pool dispatches for tasks that do nothing on the + /// thread. Report cost 0 so they collapse into a single batch. ssize_t memory_after = diff.by_stage[size_t(diff.cur_stage)]; - task.cost_estimate_bytes = size_t(std::max(0l, memory_after - memory_before)); + task.cost_estimate_bytes = task.stage == ReadStage::ColumnDataPrefetch + ? 0 + : size_t(std::max(0l, memory_after - memory_before)); out_tasks.push_back(task); } From 89def513b24e23196a63681e72988b5e65d82c4b Mon Sep 17 00:00:00 2001 From: UnamedRus Date: Mon, 24 Aug 2026 12:43:45 +0300 Subject: [PATCH 4/4] Parquet v3: trim verbose scheduler comments Condense the oversized comment blocks added by the read-concurrency work (the stage-weight rationale in `ReadManager::init`, the ColumnDataPrefetch/ColumnData scheduleTask and runTask notes, and the cost-estimate exception) to a few lines each. No code changes. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../Formats/Impl/Parquet/ReadManager.cpp | 73 ++++++------------- 1 file changed, 21 insertions(+), 52 deletions(-) diff --git a/src/Processors/Formats/Impl/Parquet/ReadManager.cpp b/src/Processors/Formats/Impl/Parquet/ReadManager.cpp index 3c77364202e4..354fa0e700e3 100644 --- a/src/Processors/Formats/Impl/Parquet/ReadManager.cpp +++ b/src/Processors/Formats/Impl/Parquet/ReadManager.cpp @@ -76,37 +76,17 @@ void ReadManager::init(FormatParserSharedResourcesPtr parser_shared_resources_, stages[i].row_group_tasks_to_schedule.resize(num_row_groups); } - /// Distribute the memory and thread budgets among stages. - /// The distribution is static to make sure no stage gets starved if others eat all the resources. - /// E.g. if the budget was shared among all stages, maybe ColumnData could run far ahead and eat - /// all the memory, starving the small index reads that other row groups need to make progress. + /// Static per-stage memory and thread budgets so no stage starves the others (e.g. ColumnData + /// eating all memory and blocking the small index reads other row groups need). /// - /// The values below are relative weights, not fractions: they need not sum to 1 (the memory weights - /// sum to 20, the thread weights to 8). The normalization loop right after divides each by the - /// per-resource sum to turn them into the actual budget fractions stored in the Stage. A stage then - /// gets `weight/sum` of the query-global memory watermark and of the parsing thread pool (further - /// divided across files read in parallel, see `getLimitsPerReader`). - /// - /// Memory and threads are budgeted separately (an earlier single 0.2 fraction capped ColumnData at - /// 0.2 of *both*, so only ~2 row groups were read/decoded ahead) because they answer two different - /// questions, and the weights follow from those: - /// - /// Threads = "does a task in this stage occupy a CPU?" Index/bloom/prefetch tasks only *issue* an - /// async read - they return immediately and the actual IO runs in the Prefetcher's own io pool - so - /// 1 slot each is enough to keep issuing and extra slots buy nothing. Decode (ColumnData) is real - /// CPU work (decompress + decode), so it gets the majority (3 of 8) to overlap several row groups - /// without monopolizing the pool and starving the cheap read-issuing stages. - /// - /// Memory = "how much outstanding work does a byte here buy?" ColumnDataPrefetch holds *compressed* - /// pages in flight, i.e. the read-ahead depth (~bandwidth-delay product): high-RTT S3 needs many - /// GETs outstanding to hide latency, and compressed pages are small, so memory here buys the most - /// concurrency per byte - hence the largest share (9). ColumnData holds *decoded* columns, which are - /// far larger per row group, so it is bounded (6) to keep resident decoded groups from blowing the - /// budget. Index/bloom footprints are tiny (a few KB per row group), so they get just a floor to - /// keep a few row groups' indexes resident for pipelining; BloomFilterBlocksOrDictionary gets 2 vs - /// 1 because dictionary/bloom blocks are larger than headers and offset indexes. - /// - /// The exact integers are relative priorities tuned empirically on high-RTT S3, not derived constants. + /// Values are relative weights, not fractions; the normalization loop below divides each by the + /// per-resource sum to get the actual fraction of the memory watermark / parsing pool per stage. + /// Memory and threads are weighted separately (a single shared fraction coupled them, capping + /// read-ahead and decode together): ColumnDataPrefetch holds small compressed pages, so it gets + /// the most memory to keep many GETs outstanding and hide S3 latency; ColumnData holds large + /// decoded columns, so its memory is bounded but it gets most threads (decode is the real CPU + /// work). Index/bloom stages only issue async reads, so 1 thread and a small memory floor each. + /// Integers are empirical priorities, not derived constants. auto set_weights = [&](ReadStage s, double memory_weight, double thread_weight) { stages[size_t(s)].memory_target_fraction = memory_weight; @@ -822,11 +802,9 @@ void ReadManager::scheduleTask(Task task, bool is_first_in_group, MemoryUsageDif RowSubgroup & row_subgroup = row_group.subgroups.at(task.row_subgroup_idx); if (row_subgroup.filter.rows_pass == 0) break; - /// Determine which data pages this subgroup needs and queue their reads. The - /// startPrefetch at the end of this function issues them against the Prefetcher's io - /// pool and charges the compressed bytes to the ColumnDataPrefetch stage budget - - /// separate from the decoded-output budget (ColumnData) - so many row groups can have - /// their reads in flight (deep prefetch) while only a few are decoded at once. + /// Queue this subgroup's data-page reads; startPrefetch (below) issues them and charges + /// compressed bytes to the ColumnDataPrefetch budget, separate from the decode budget, + /// so many row groups prefetch ahead while only a few decode at once. reader.determinePagesToPrefetch(column, row_subgroup, row_group, prefetches); /// Side note: would be nice to avoid reading the dictionary if all dictionary-encoded @@ -851,9 +829,8 @@ void ReadManager::scheduleTask(Task task, bool is_first_in_group, MemoryUsageDif ColumnSubchunk & subchunk = row_subgroup.columns.at(task.column_idx); if (row_subgroup.filter.rows_pass == 0) break; - /// The data-page reads were already issued in ColumnDataPrefetch (and are in flight or - /// done in the Prefetcher). Here we only reserve the estimated decoded-output memory - /// against the ColumnData budget; runTask then decodes from those buffers. + /// Reads already issued in ColumnDataPrefetch; here just reserve estimated decoded-output + /// memory against the ColumnData budget (runTask decodes from those buffers). double bytes_per_row = reader.estimateColumnMemoryBytesPerRow(column, row_group, reader.primitive_columns.at(task.column_idx)); size_t column_memory = static_cast(bytes_per_row * static_cast(row_subgroup.filter.rows_pass)); subchunk.column_and_offsets_memory = MemoryUsageToken(column_memory, &diff); @@ -877,17 +854,10 @@ void ReadManager::scheduleTask(Task task, bool is_first_in_group, MemoryUsageDif reader.prefetcher.startPrefetch(prefetches, &diff); - /// We want to detect tiny tasks to group them together to reduce scheduling overhead. - /// Use the predicted memory usage as a rough estimate of how long a task will take. - /// E.g. main data read task's memory estimate consists of the input page sizes and the output - /// column size; the run time is also roughly proportional to these sizes. - /// Hope it's a good enough proxy in all cases. - /// - /// Exception: ColumnDataPrefetch tasks do all their work (startPrefetch) synchronously here and - /// have an empty runTask, so their run time is ~0 regardless of how many compressed bytes they - /// charge. Reporting the (large) charged bytes as the cost would make the batching split them - /// across many batches - i.e. many thread-pool dispatches for tasks that do nothing on the - /// thread. Report cost 0 so they collapse into a single batch. + /// Group tiny tasks to reduce scheduling overhead, using predicted memory as a proxy for run time. + /// Exception: ColumnDataPrefetch does its work (startPrefetch) here and has an empty runTask, so + /// its run time is ~0 no matter how many compressed bytes it charges; report cost 0 so these tasks + /// collapse into one batch instead of being split across many no-op thread-pool dispatches. ssize_t memory_after = diff.by_stage[size_t(diff.cur_stage)]; task.cost_estimate_bytes = task.stage == ReadStage::ColumnDataPrefetch ? 0 @@ -967,9 +937,8 @@ void ReadManager::runTask(Task task, bool last_in_batch, MemoryUsageDiff & diff) column.offset_index_prefetch.reset(&diff); break; case ReadStage::ColumnDataPrefetch: - /// The compressed data-page reads were already issued in scheduleTask (startPrefetch) - /// and proceed asynchronously in the Prefetcher's io pool. Nothing to do here; the - /// subgroup advances to ColumnData, which decodes from those buffers. + /// Reads were issued in scheduleTask (startPrefetch) and run async in the Prefetcher; + /// nothing to do here. The subgroup advances to ColumnData, which decodes them. break; case ReadStage::ColumnData: {