diff --git a/src/Core/FormatFactorySettings.h b/src/Core/FormatFactorySettings.h index c37da84008e3..12fad80b961b 100644 --- a/src/Core/FormatFactorySettings.h +++ b/src/Core/FormatFactorySettings.h @@ -216,7 +216,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(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(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"( Skip pages using min/max values from column index. diff --git a/src/Core/SettingsChangesHistory.cpp b/src/Core/SettingsChangesHistory.cpp index 0a3ddb973f7b..e157165f56a7 100644 --- a/src/Core/SettingsChangesHistory.cpp +++ b/src/Core/SettingsChangesHistory.cpp @@ -224,6 +224,10 @@ 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."}, {"cloud_mode_database_engine", 1, 1, "Obsolete setting, the database engine in Cloud no longer depends on it."}, + {"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 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/FormatFactory.cpp b/src/Formats/FormatFactory.cpp index 17befc8e3313..34c81d515589 100644 --- a/src/Formats/FormatFactory.cpp +++ b/src/Formats/FormatFactory.cpp @@ -234,6 +234,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]; + 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 52ac7ff55317..dfd95af1fd98 100644 --- a/src/Formats/FormatSettings.h +++ b/src/Formats/FormatSettings.h @@ -384,6 +384,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; + /// 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; /// Write. UInt64 row_group_rows = 1000000; 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 868edfab67f2..60c7418672c4 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 be0e21c4529c..20e1c113ae7b 100644 --- a/src/Processors/Formats/Impl/Parquet/ReadManager.cpp +++ b/src/Processors/Formats/Impl/Parquet/ReadManager.cpp @@ -17,6 +17,7 @@ namespace DB::ErrorCodes { + extern const int BAD_ARGUMENTS; extern const int LOGICAL_ERROR; extern const int QUERY_WAS_CANCELLED; } @@ -74,17 +75,50 @@ 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; + /// 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 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 {}", 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 {}", 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_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_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). + 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); @@ -145,6 +179,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); @@ -301,9 +336,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, @@ -313,8 +349,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; } @@ -325,7 +361,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, @@ -426,6 +462,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; } @@ -550,7 +593,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); @@ -568,7 +611,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); @@ -721,12 +764,14 @@ 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; + /// 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 @@ -743,7 +788,16 @@ 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; + /// 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); @@ -767,13 +821,14 @@ 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. + /// 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 = 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); } @@ -831,7 +886,7 @@ PruningMemoryReservation ReadManager::pruningMemoryReservation(const MemoryUsage /// per-reader - each reserve the entire watermark, breaking its documented "total across those files" /// contract and overshooting the cap before scheduler throttling has a chance to help. size_t watermark = SharedResourcesExt::getLimitsPerReader( - *parser_shared_resources, stages[idx].memory_target_fraction).memory_high_watermark; + *parser_shared_resources, stages[idx].memory_target_fraction, stages[idx].thread_target_fraction).memory_high_watermark; /// Never let a tiny per-reader budget round down to 0, which `PruningMemoryReservation` reads as /// "unbounded"; a near-zero budget must instead mean "reserve nothing", i.e. skip pruning (full scan). watermark = std::max(watermark, size_t(1)); @@ -893,6 +948,10 @@ 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: + /// 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: { RowSubgroup & row_subgroup = row_group.subgroups.at(task.row_subgroup_idx); @@ -910,7 +969,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 986dee8ce728..09f4ad3a69b8 100644 --- a/src/Processors/Formats/Impl/Parquet/ReadManager.h +++ b/src/Processors/Formats/Impl/Parquet/ReadManager.h @@ -93,7 +93,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 1ee01ab2bbd5..0a9d40ec7034 100644 --- a/src/Processors/Formats/Impl/Parquet/Reader.cpp +++ b/src/Processors/Formats/Impl/Parquet/Reader.cpp @@ -2383,7 +2383,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. @@ -2558,6 +2558,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 69a4b98c9c15..cd3debc58656 100644 --- a/src/Processors/Formats/Impl/Parquet/Reader.h +++ b/src/Processors/Formats/Impl/Parquet/Reader.h @@ -614,7 +614,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.