From 79d3658690cbd2f798a02da5cbf87cbe28c8cf60 Mon Sep 17 00:00:00 2001 From: UnamedRus Date: Tue, 18 Aug 2026 16:45:42 +0300 Subject: [PATCH 01/10] 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 02/10] 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 03/10] 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 04/10] 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: { From 949ef5cde65227942f8c604ee1397aa4cad1adc4 Mon Sep 17 00:00:00 2001 From: UnamedRus Date: Tue, 25 Aug 2026 11:32:05 +0300 Subject: [PATCH 05/10] Parquet v3: replace read-stage-weights Map with two Float scheduler knobs The `input_format_parquet_read_stage_weights` `Map` setting failed to build: `Map` is not in `FORMAT_FACTORY_SETTINGS_SUPPORTED_TYPES` (nor in the supported types of the collections that embed the format settings, e.g. `DataLakeStorageSettings`), so the settings-layout machinery has no `SettingTypeTag::Map` for those collections: DataLakeStorageSettings.cpp: error: no member named 'Map' in 'DataLakeStorageSettingsTraitsDetail::SettingTypeTag' ...expanded from DECLARE(Map, input_format_parquet_read_stage_weights, ...) Adding `Map` to every such collection's supported-types list is invasive and is why no format setting is a `Map` today. Replace the single `Map` with two `Float` settings (`Float` is already a supported format-settings type): - `input_format_parquet_prefetch_memory_fraction` (default 0.6): of the memory budget reserved for column data, the share given to compressed read-ahead (`ColumnDataPrefetch`) vs decoded output (`ColumnData`). - `input_format_parquet_decode_thread_fraction` (default 0.375): `ColumnData`'s share of the parsing thread pool; the read-issuing stages split the rest. These cover the two binding constraints on decode-ahead depth (memory on latency-bound remote reads, threads on CPU-bound local decode) without exposing the dead per-stage knobs for the read-issuing stages. `ReadManager::init` now computes per-stage fractions directly from the two knobs; the defaults reproduce the previously hard-coded fractions exactly, so behavior is unchanged by default. The index/bloom stages keep fixed shares. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/Core/FormatFactorySettings.h | 9 +- src/Core/SettingsChangesHistory.cpp | 3 +- src/Formats/FormatFactory.cpp | 27 +---- src/Formats/FormatSettings.h | 10 +- .../Formats/Impl/Parquet/ReadCommon.h | 8 +- .../Formats/Impl/Parquet/ReadManager.cpp | 107 +++++++----------- .../Impl/ParquetV3BlockInputFormat.cpp | 3 +- 7 files changed, 59 insertions(+), 108 deletions(-) diff --git a/src/Core/FormatFactorySettings.h b/src/Core/FormatFactorySettings.h index 38cb6d71cf50..0b06c32a8867 100644 --- a/src/Core/FormatFactorySettings.h +++ b/src/Core/FormatFactorySettings.h @@ -201,9 +201,12 @@ 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(Float, input_format_parquet_prefetch_memory_fraction, 0.6, R"( +Advanced tuning knob for the Parquet reader v3 scheduler. Of the memory budget reserved for column data, the fraction given to compressed read-ahead (the `ColumnDataPrefetch` stage) versus decoded output (the `ColumnData` stage); the rest goes to decode. A higher value keeps more compressed pages in flight to hide read latency (useful on high-latency storage such as S3); a lower value caps read-ahead and leaves more budget for decoded columns. Must be in [0, 1]. The index and bloom-filter stages keep a fixed share of the memory budget regardless of this setting. +)", 0) \ + DECLARE(Float, input_format_parquet_decode_thread_fraction, 0.375, R"( +Advanced tuning knob for the Parquet reader v3 scheduler. The fraction of the Parquet parsing thread pool dedicated to column decoding (the `ColumnData` stage); the remaining stages, which only issue asynchronous reads, share the rest. Raise it to give decoding (the only CPU-bound stage) more parallelism on fast/local storage; the default suits latency-bound remote reads where memory, not threads, limits concurrency. Must be in [0, 1]. +)", 0) \ 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 3f656d4ebfd7..8ed2c0b7daa7 100644 --- a/src/Core/SettingsChangesHistory.cpp +++ b/src/Core/SettingsChangesHistory.cpp @@ -49,7 +49,8 @@ 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."}, + {"input_format_parquet_prefetch_memory_fraction", 0.6, 0.6, "New setting to tune the Parquet reader v3 split of the column-data memory budget between compressed read-ahead and decode."}, + {"input_format_parquet_decode_thread_fraction", 0.375, 0.375, "New setting to tune the Parquet reader v3 share of the parsing thread pool given to column decoding."}, {"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 02c67cd6aa4a..d6c30363c0d2 100644 --- a/src/Formats/FormatFactory.cpp +++ b/src/Formats/FormatFactory.cpp @@ -223,31 +223,8 @@ 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.prefetch_memory_fraction = settings[Setting::input_format_parquet_prefetch_memory_fraction]; + format_settings.parquet.decode_thread_fraction = settings[Setting::input_format_parquet_decode_thread_fraction]; 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 1620e4603451..d031e41048c0 100644 --- a/src/Formats/FormatSettings.h +++ b/src/Formats/FormatSettings.h @@ -5,8 +5,6 @@ #include #include -#include - namespace DB { @@ -363,10 +361,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; + /// Reader v3 scheduler knobs: share of the column-data memory budget given to compressed + /// read-ahead vs decode, and ColumnData's share of the parsing thread pool. + double prefetch_memory_fraction = 0.6; + double decode_thread_fraction = 0.375; /// 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 b44723e5bf00..578987562eb6 100644 --- a/src/Processors/Formats/Impl/Parquet/ReadCommon.h +++ b/src/Processors/Formats/Impl/Parquet/ReadCommon.h @@ -42,10 +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; + /// Reader scheduler knobs applied in ReadManager::init. See + /// input_format_parquet_prefetch_memory_fraction / input_format_parquet_decode_thread_fraction. + double prefetch_memory_fraction = 0.6; + double decode_thread_fraction = 0.375; struct Limits { diff --git a/src/Processors/Formats/Impl/Parquet/ReadManager.cpp b/src/Processors/Formats/Impl/Parquet/ReadManager.cpp index 354fa0e700e3..e8a5b2e6ae40 100644 --- a/src/Processors/Formats/Impl/Parquet/ReadManager.cpp +++ b/src/Processors/Formats/Impl/Parquet/ReadManager.cpp @@ -13,7 +13,6 @@ #include #include #include -#include #include namespace DB::ErrorCodes @@ -76,72 +75,48 @@ void ReadManager::init(FormatParserSharedResourcesPtr parser_shared_resources_, stages[i].row_group_tasks_to_schedule.resize(num_row_groups); } - /// 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). + /// Static per-stage memory and thread budgets (fractions summing to 1 within each resource) so no + /// stage starves the others (e.g. ColumnData eating all memory and blocking the small index reads + /// other row groups need). Memory and threads are budgeted separately because a single shared + /// fraction coupled read-ahead and decode: ColumnDataPrefetch holds small compressed pages, so it + /// gets most of the memory to keep many reads outstanding and hide 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 they get one thread-share each and a small memory + /// floor. /// - /// 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; - stages[size_t(s)].thread_target_fraction = thread_weight; - }; - 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. + /// Two knobs re-balance the data stages (index/bloom shares are fixed): + /// - prefetch_memory_fraction: of the memory budget reserved for column data (0.75 of the total), + /// the share given to ColumnDataPrefetch (compressed read-ahead) vs ColumnData (decoded output). + /// - decode_thread_fraction: ColumnData's share of the thread pool; the five read-issuing stages + /// split the remainder equally. + /// The defaults (0.6, 0.375) reproduce the previously hard-coded fractions. 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); - } - } + const double mem_frac = ext.prefetch_memory_fraction; + const double dec_thr = ext.decode_thread_fraction; + if (!(mem_frac >= 0 && mem_frac <= 1)) + throw Exception(ErrorCodes::BAD_ARGUMENTS, + "input_format_parquet_prefetch_memory_fraction must be in [0, 1], got {}", mem_frac); + if (!(dec_thr >= 0 && dec_thr <= 1)) + throw Exception(ErrorCodes::BAD_ARGUMENTS, + "input_format_parquet_decode_thread_fraction must be in [0, 1], got {}", dec_thr); + auto set_fractions = [&](ReadStage 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; + }; + const double data_mem = 0.75; // index/bloom take the remaining 0.25 + const double issuer_thr = (1.0 - dec_thr) / 5.0; // five read-issuing stages split the rest + set_fractions(ReadStage::NotStarted, 0, 0); + set_fractions(ReadStage::BloomFilterHeader, 0.05, issuer_thr); + set_fractions(ReadStage::BloomFilterBlocksOrDictionary, 0.10, issuer_thr); + set_fractions(ReadStage::ColumnIndexAndOffsetIndex, 0.05, issuer_thr); + set_fractions(ReadStage::OffsetIndex, 0.05, issuer_thr); + set_fractions(ReadStage::ColumnDataPrefetch, data_mem * mem_frac, issuer_thr); + set_fractions(ReadStage::ColumnData, data_mem * (1.0 - mem_frac), dec_thr); + set_fractions(ReadStage::Deliver, 0, 0); + + /// Normalize (defensive: the fractions already sum to 1 within each resource). double memory_sum = 0; double thread_sum = 0; for (const Stage & stage : stages) @@ -149,10 +124,6 @@ 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 aa42fa65d47a..6076c54089cc 100644 --- a/src/Processors/Formats/Impl/ParquetV3BlockInputFormat.cpp +++ b/src/Processors/Formats/Impl/ParquetV3BlockInputFormat.cpp @@ -91,7 +91,8 @@ 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; + ext->prefetch_memory_fraction = format_settings.parquet.prefetch_memory_fraction; + ext->decode_thread_fraction = format_settings.parquet.decode_thread_fraction; parser_shared_resources->opaque = ext; }); From 6f1751b123087906903dfb969ea253e81d331ffb Mon Sep 17 00:00:00 2001 From: UnamedRus Date: Tue, 25 Aug 2026 11:36:22 +0300 Subject: [PATCH 06/10] Parquet v3: shorten the scheduler-budget comment Co-Authored-By: Claude Opus 4.8 (1M context) --- .../Formats/Impl/Parquet/ReadManager.cpp | 21 ++++++------------- 1 file changed, 6 insertions(+), 15 deletions(-) diff --git a/src/Processors/Formats/Impl/Parquet/ReadManager.cpp b/src/Processors/Formats/Impl/Parquet/ReadManager.cpp index e8a5b2e6ae40..7b6b3295c961 100644 --- a/src/Processors/Formats/Impl/Parquet/ReadManager.cpp +++ b/src/Processors/Formats/Impl/Parquet/ReadManager.cpp @@ -75,21 +75,12 @@ void ReadManager::init(FormatParserSharedResourcesPtr parser_shared_resources_, stages[i].row_group_tasks_to_schedule.resize(num_row_groups); } - /// Static per-stage memory and thread budgets (fractions summing to 1 within each resource) so no - /// stage starves the others (e.g. ColumnData eating all memory and blocking the small index reads - /// other row groups need). Memory and threads are budgeted separately because a single shared - /// fraction coupled read-ahead and decode: ColumnDataPrefetch holds small compressed pages, so it - /// gets most of the memory to keep many reads outstanding and hide 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 they get one thread-share each and a small memory - /// floor. - /// - /// Two knobs re-balance the data stages (index/bloom shares are fixed): - /// - prefetch_memory_fraction: of the memory budget reserved for column data (0.75 of the total), - /// the share given to ColumnDataPrefetch (compressed read-ahead) vs ColumnData (decoded output). - /// - decode_thread_fraction: ColumnData's share of the thread pool; the five read-issuing stages - /// split the remainder equally. - /// The defaults (0.6, 0.375) reproduce the previously hard-coded fractions. + /// Per-stage memory/thread budgets (each resource sums to 1) so no stage starves the others. + /// Prefetch holds small compressed pages -> most memory (keep reads outstanding, hide latency); + /// decode holds large columns -> bounded memory but most threads (only CPU-bound stage); + /// index/bloom only issue async reads -> fixed small shares. Two knobs re-balance the data stages: + /// prefetch_memory_fraction splits the 0.75 data-memory budget prefetch/decode; decode_thread_fraction + /// is decode's thread share (issuers split the rest). Defaults preserve the old hard-coded fractions. const auto & ext = *static_cast(parser_shared_resources->opaque.get()); const double mem_frac = ext.prefetch_memory_fraction; const double dec_thr = ext.decode_thread_fraction; From 6ba63a166fe88f7147561beab116957889729c2d Mon Sep 17 00:00:00 2001 From: UnamedRus Date: Tue, 25 Aug 2026 11:40:26 +0300 Subject: [PATCH 07/10] Parquet: drop "v3" from reader setting descriptions The native reader is now the only one (input_format_parquet_use_native_reader_v3 is obsolete/always-on), so the "v3" qualifier in the user-facing setting descriptions and code comments is stale. Identifiers and the obsolete-setting history entries keep their names. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/Core/FormatFactorySettings.h | 6 +++--- src/Core/SettingsChangesHistory.cpp | 4 ++-- src/Formats/FormatSettings.h | 2 +- 3 files changed, 6 insertions(+), 6 deletions(-) diff --git a/src/Core/FormatFactorySettings.h b/src/Core/FormatFactorySettings.h index 0b06c32a8867..1cde10c661cc 100644 --- a/src/Core/FormatFactorySettings.h +++ b/src/Core/FormatFactorySettings.h @@ -199,13 +199,13 @@ When reading Parquet files, parse JSON columns as ClickHouse JSON Column. Schedule prefetches more aggressively if memory usage is below than threshold. Potentially useful e.g. if there are many small bloom filters to read over network. )", 0) \ 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. +Approximate memory limit for the Parquet reader. 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(Float, input_format_parquet_prefetch_memory_fraction, 0.6, R"( -Advanced tuning knob for the Parquet reader v3 scheduler. Of the memory budget reserved for column data, the fraction given to compressed read-ahead (the `ColumnDataPrefetch` stage) versus decoded output (the `ColumnData` stage); the rest goes to decode. A higher value keeps more compressed pages in flight to hide read latency (useful on high-latency storage such as S3); a lower value caps read-ahead and leaves more budget for decoded columns. Must be in [0, 1]. The index and bloom-filter stages keep a fixed share of the memory budget regardless of this setting. +Advanced tuning knob for the Parquet reader scheduler. Of the memory budget reserved for column data, the fraction given to compressed read-ahead (the `ColumnDataPrefetch` stage) versus decoded output (the `ColumnData` stage); the rest goes to decode. A higher value keeps more compressed pages in flight to hide read latency (useful on high-latency storage such as S3); a lower value caps read-ahead and leaves more budget for decoded columns. Must be in [0, 1]. The index and bloom-filter stages keep a fixed share of the memory budget regardless of this setting. )", 0) \ DECLARE(Float, input_format_parquet_decode_thread_fraction, 0.375, R"( -Advanced tuning knob for the Parquet reader v3 scheduler. The fraction of the Parquet parsing thread pool dedicated to column decoding (the `ColumnData` stage); the remaining stages, which only issue asynchronous reads, share the rest. Raise it to give decoding (the only CPU-bound stage) more parallelism on fast/local storage; the default suits latency-bound remote reads where memory, not threads, limits concurrency. Must be in [0, 1]. +Advanced tuning knob for the Parquet reader scheduler. The fraction of the Parquet parsing thread pool dedicated to column decoding (the `ColumnData` stage); the remaining stages, which only issue asynchronous reads, share the rest. Raise it to give decoding (the only CPU-bound stage) more parallelism on fast/local storage; the default suits latency-bound remote reads where memory, not threads, limits concurrency. Must be in [0, 1]. )", 0) \ DECLARE(Bool, input_format_parquet_page_filter_push_down, true, R"( Skip pages using min/max values from column index. diff --git a/src/Core/SettingsChangesHistory.cpp b/src/Core/SettingsChangesHistory.cpp index 8ed2c0b7daa7..f1b2bab3ac3a 100644 --- a/src/Core/SettingsChangesHistory.cpp +++ b/src/Core/SettingsChangesHistory.cpp @@ -49,8 +49,8 @@ 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_prefetch_memory_fraction", 0.6, 0.6, "New setting to tune the Parquet reader v3 split of the column-data memory budget between compressed read-ahead and decode."}, - {"input_format_parquet_decode_thread_fraction", 0.375, 0.375, "New setting to tune the Parquet reader v3 share of the parsing thread pool given to column decoding."}, + {"input_format_parquet_prefetch_memory_fraction", 0.6, 0.6, "New setting to tune the Parquet reader split of the column-data memory budget between compressed read-ahead and decode."}, + {"input_format_parquet_decode_thread_fraction", 0.375, 0.375, "New setting to tune the Parquet reader share of the parsing thread pool given to column decoding."}, {"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/FormatSettings.h b/src/Formats/FormatSettings.h index d031e41048c0..745898c0c751 100644 --- a/src/Formats/FormatSettings.h +++ b/src/Formats/FormatSettings.h @@ -361,7 +361,7 @@ 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; - /// Reader v3 scheduler knobs: share of the column-data memory budget given to compressed + /// Reader scheduler knobs: share of the column-data memory budget given to compressed /// read-ahead vs decode, and ColumnData's share of the parsing thread pool. double prefetch_memory_fraction = 0.6; double decode_thread_fraction = 0.375; From 18afd479377d39909fbccaa72c73ed86c407df0f Mon Sep 17 00:00:00 2001 From: UnamedRus Date: Tue, 25 Aug 2026 12:02:26 +0300 Subject: [PATCH 08/10] fix double to float implicit conversion --- src/Core/FormatFactorySettings.h | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/Core/FormatFactorySettings.h b/src/Core/FormatFactorySettings.h index 1cde10c661cc..ad6c13fa5998 100644 --- a/src/Core/FormatFactorySettings.h +++ b/src/Core/FormatFactorySettings.h @@ -201,10 +201,10 @@ 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 the Parquet reader. 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(Float, input_format_parquet_prefetch_memory_fraction, 0.6, R"( + DECLARE(Float, input_format_parquet_prefetch_memory_fraction, 0.6f, R"( Advanced tuning knob for the Parquet reader scheduler. Of the memory budget reserved for column data, the fraction given to compressed read-ahead (the `ColumnDataPrefetch` stage) versus decoded output (the `ColumnData` stage); the rest goes to decode. A higher value keeps more compressed pages in flight to hide read latency (useful on high-latency storage such as S3); a lower value caps read-ahead and leaves more budget for decoded columns. Must be in [0, 1]. The index and bloom-filter stages keep a fixed share of the memory budget regardless of this setting. )", 0) \ - DECLARE(Float, input_format_parquet_decode_thread_fraction, 0.375, R"( + DECLARE(Float, input_format_parquet_decode_thread_fraction, 0.375f, R"( Advanced tuning knob for the Parquet reader scheduler. The fraction of the Parquet parsing thread pool dedicated to column decoding (the `ColumnData` stage); the remaining stages, which only issue asynchronous reads, share the rest. Raise it to give decoding (the only CPU-bound stage) more parallelism on fast/local storage; the default suits latency-bound remote reads where memory, not threads, limits concurrency. Must be in [0, 1]. )", 0) \ DECLARE(Bool, input_format_parquet_page_filter_push_down, true, R"( From 62b46d40834b4c81ad441c8996deb25c427ffdbd Mon Sep 17 00:00:00 2001 From: UnamedRus Date: Tue, 25 Aug 2026 12:08:02 +0300 Subject: [PATCH 09/10] Parquet: read scheduler fractions from options, not a duplicate ext copy The two knobs were stored both in FormatSettings.parquet and copied into SharedResourcesExt (mirroring the memory watermarks). The watermarks live in the ext because getLimitsPerReader (a static method without options access) reads them; the fractions are only consumed in ReadManager::init, which already has reader.options.format.parquet. Drop the duplicate ext fields and the copy, read the fractions directly, and spell the locals out in full. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../Formats/Impl/Parquet/ReadCommon.h | 4 --- .../Formats/Impl/Parquet/ReadManager.cpp | 29 +++++++++---------- .../Impl/ParquetV3BlockInputFormat.cpp | 2 -- 3 files changed, 14 insertions(+), 21 deletions(-) diff --git a/src/Processors/Formats/Impl/Parquet/ReadCommon.h b/src/Processors/Formats/Impl/Parquet/ReadCommon.h index 578987562eb6..cbaf7f095ec6 100644 --- a/src/Processors/Formats/Impl/Parquet/ReadCommon.h +++ b/src/Processors/Formats/Impl/Parquet/ReadCommon.h @@ -42,10 +42,6 @@ struct SharedResourcesExt { size_t total_memory_low_watermark = 0; size_t total_memory_high_watermark = 0; - /// Reader scheduler knobs applied in ReadManager::init. See - /// input_format_parquet_prefetch_memory_fraction / input_format_parquet_decode_thread_fraction. - double prefetch_memory_fraction = 0.6; - double decode_thread_fraction = 0.375; struct Limits { diff --git a/src/Processors/Formats/Impl/Parquet/ReadManager.cpp b/src/Processors/Formats/Impl/Parquet/ReadManager.cpp index 7b6b3295c961..b6375ccb96c2 100644 --- a/src/Processors/Formats/Impl/Parquet/ReadManager.cpp +++ b/src/Processors/Formats/Impl/Parquet/ReadManager.cpp @@ -81,30 +81,29 @@ void ReadManager::init(FormatParserSharedResourcesPtr parser_shared_resources_, /// index/bloom only issue async reads -> fixed small shares. Two knobs re-balance the data stages: /// prefetch_memory_fraction splits the 0.75 data-memory budget prefetch/decode; decode_thread_fraction /// is decode's thread share (issuers split the rest). Defaults preserve the old hard-coded fractions. - const auto & ext = *static_cast(parser_shared_resources->opaque.get()); - const double mem_frac = ext.prefetch_memory_fraction; - const double dec_thr = ext.decode_thread_fraction; - if (!(mem_frac >= 0 && mem_frac <= 1)) + const double prefetch_memory_fraction = reader.options.format.parquet.prefetch_memory_fraction; + const double decode_thread_fraction = reader.options.format.parquet.decode_thread_fraction; + if (!(prefetch_memory_fraction >= 0 && prefetch_memory_fraction <= 1)) throw Exception(ErrorCodes::BAD_ARGUMENTS, - "input_format_parquet_prefetch_memory_fraction must be in [0, 1], got {}", mem_frac); - if (!(dec_thr >= 0 && dec_thr <= 1)) + "input_format_parquet_prefetch_memory_fraction must be in [0, 1], got {}", prefetch_memory_fraction); + if (!(decode_thread_fraction >= 0 && decode_thread_fraction <= 1)) throw Exception(ErrorCodes::BAD_ARGUMENTS, - "input_format_parquet_decode_thread_fraction must be in [0, 1], got {}", dec_thr); + "input_format_parquet_decode_thread_fraction must be in [0, 1], got {}", decode_thread_fraction); auto set_fractions = [&](ReadStage 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; }; - const double data_mem = 0.75; // index/bloom take the remaining 0.25 - const double issuer_thr = (1.0 - dec_thr) / 5.0; // five read-issuing stages split the rest + const double data_memory_fraction = 0.75; // index/bloom take the remaining 0.25 + const double issuer_thread_fraction = (1.0 - decode_thread_fraction) / 5.0; // five read-issuing stages split the rest set_fractions(ReadStage::NotStarted, 0, 0); - set_fractions(ReadStage::BloomFilterHeader, 0.05, issuer_thr); - set_fractions(ReadStage::BloomFilterBlocksOrDictionary, 0.10, issuer_thr); - set_fractions(ReadStage::ColumnIndexAndOffsetIndex, 0.05, issuer_thr); - set_fractions(ReadStage::OffsetIndex, 0.05, issuer_thr); - set_fractions(ReadStage::ColumnDataPrefetch, data_mem * mem_frac, issuer_thr); - set_fractions(ReadStage::ColumnData, data_mem * (1.0 - mem_frac), dec_thr); + set_fractions(ReadStage::BloomFilterHeader, 0.05, issuer_thread_fraction); + set_fractions(ReadStage::BloomFilterBlocksOrDictionary, 0.10, issuer_thread_fraction); + set_fractions(ReadStage::ColumnIndexAndOffsetIndex, 0.05, issuer_thread_fraction); + set_fractions(ReadStage::OffsetIndex, 0.05, issuer_thread_fraction); + set_fractions(ReadStage::ColumnDataPrefetch, data_memory_fraction * prefetch_memory_fraction, issuer_thread_fraction); + set_fractions(ReadStage::ColumnData, data_memory_fraction * (1.0 - prefetch_memory_fraction), decode_thread_fraction); set_fractions(ReadStage::Deliver, 0, 0); /// Normalize (defensive: the fractions already sum to 1 within each resource). diff --git a/src/Processors/Formats/Impl/ParquetV3BlockInputFormat.cpp b/src/Processors/Formats/Impl/ParquetV3BlockInputFormat.cpp index 6076c54089cc..6a238834caec 100644 --- a/src/Processors/Formats/Impl/ParquetV3BlockInputFormat.cpp +++ b/src/Processors/Formats/Impl/ParquetV3BlockInputFormat.cpp @@ -91,8 +91,6 @@ 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->prefetch_memory_fraction = format_settings.parquet.prefetch_memory_fraction; - ext->decode_thread_fraction = format_settings.parquet.decode_thread_fraction; parser_shared_resources->opaque = ext; }); From 8d9e0517a31fdb73674223822cba661b1d7242f5 Mon Sep 17 00:00:00 2001 From: UnamedRus Date: Tue, 25 Aug 2026 12:28:24 +0300 Subject: [PATCH 10/10] Change Float to Double for Parquet settings --- src/Core/FormatFactorySettings.h | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/Core/FormatFactorySettings.h b/src/Core/FormatFactorySettings.h index ad6c13fa5998..70efdcb163ae 100644 --- a/src/Core/FormatFactorySettings.h +++ b/src/Core/FormatFactorySettings.h @@ -201,10 +201,10 @@ 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 the Parquet reader. 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(Float, input_format_parquet_prefetch_memory_fraction, 0.6f, R"( + DECLARE(Double, input_format_parquet_prefetch_memory_fraction, 0.6, R"( Advanced tuning knob for the Parquet reader scheduler. Of the memory budget reserved for column data, the fraction given to compressed read-ahead (the `ColumnDataPrefetch` stage) versus decoded output (the `ColumnData` stage); the rest goes to decode. A higher value keeps more compressed pages in flight to hide read latency (useful on high-latency storage such as S3); a lower value caps read-ahead and leaves more budget for decoded columns. Must be in [0, 1]. The index and bloom-filter stages keep a fixed share of the memory budget regardless of this setting. )", 0) \ - DECLARE(Float, input_format_parquet_decode_thread_fraction, 0.375f, R"( + DECLARE(Double, input_format_parquet_decode_thread_fraction, 0.375, R"( Advanced tuning knob for the Parquet reader scheduler. The fraction of the Parquet parsing thread pool dedicated to column decoding (the `ColumnData` stage); the remaining stages, which only issue asynchronous reads, share the rest. Raise it to give decoding (the only CPU-bound stage) more parallelism on fast/local storage; the default suits latency-bound remote reads where memory, not threads, limits concurrency. Must be in [0, 1]. )", 0) \ DECLARE(Bool, input_format_parquet_page_filter_push_down, true, R"(