diff --git a/src/Common/ProfileEvents.cpp b/src/Common/ProfileEvents.cpp index 33875c64a7b7..fc8c5635f6e1 100644 --- a/src/Common/ProfileEvents.cpp +++ b/src/Common/ProfileEvents.cpp @@ -1476,8 +1476,14 @@ The server successfully detected this situation and will download merged part fr \ M(ParquetReadRowGroups, "The total number of row groups read from parquet data", ValueType::Number) \ M(ParquetPrunedRowGroups, "The total number of row groups pruned from parquet data", ValueType::Number) \ + M(ParquetConstantColumnChunks, "The total number of parquet column chunks materialized from a single value in their min/max statistics, without reading their data pages", ValueType::Number) \ + M(ParquetConstantColumnChunksWithNulls, "The total number of parquet column chunks holding a single value plus nulls (per their statistics), for which only the definition levels were decoded and the value was taken from the statistics", ValueType::Number) \ M(ParquetDecodingTasks, "Tasks issued by parquet reader", ValueType::Number) \ M(ParquetDecodingTaskBatches, "Task groups sent to a thread pool by parquet reader", ValueType::Number) \ + M(ParquetReadTasks, "Coalesced read tasks created by the Parquet reader. Divided by `ParquetReadRowGroups`, values below 1 mean one read spans several row groups, which serializes their delivery", ValueType::Number) \ + M(ParquetReadTaskBytes, "Bytes covered by `ParquetReadTasks`, including bytes read to close short gaps between requested ranges", ValueType::Bytes) \ + M(ParquetReadAheadSubgroups, "Row subgroups whose data-page reads the Parquet reader issued ahead, while the previous subgroup of the row group was still decoding (see `input_format_parquet_read_ahead_subgroups`)", ValueType::Number) \ + M(ParquetPrefetchStarvation, "Times a decoding thread asked for a range whose read had not finished. High relative to `ParquetReadTasks` means read-ahead is too shallow", ValueType::Number) \ M(ParquetPrefetcherReadRandomRead, "The total number of reads with ReadMode::RandomRead by DB::Parquet::Prefetcher", ValueType::Number) \ M(ParquetPrefetcherReadSeekAndRead, "The total number of reads with ReadMode::SeekAndRead by DB::Parquet::Prefetcher", ValueType::Number) \ M(ParquetPrefetcherReadEntireFile, "The total number of read with ReadMode::EntireFileIsInMemory by DB::Parquet::Prefetcher", ValueType::Number) \ diff --git a/src/Core/FormatFactorySettings.h b/src/Core/FormatFactorySettings.h index 70efdcb163ae..3cbd383c8d46 100644 --- a/src/Core/FormatFactorySettings.h +++ b/src/Core/FormatFactorySettings.h @@ -212,6 +212,12 @@ Skip pages using min/max values from column index. )", 0) \ DECLARE(Bool, input_format_parquet_use_offset_index, true, R"( Minor tweak to how pages are read from parquet file when no page filtering is used. +)", 0) \ + DECLARE(Bool, input_format_parquet_use_constant_column_optimization, true, R"( +When a Parquet column chunk provably holds a single value in every row (according to its min/max statistics), materialize that value directly instead of reading and decoding the column's data pages. Also covers chunks that are all null (materialized as a sparse column) and chunks holding a single value plus nulls (only the definition levels are decoded). +)", 0) \ + DECLARE(Float, input_format_parquet_constant_column_sparse_ratio, 0.9375, R"( +For `input_format_parquet_use_constant_column_optimization`: a Parquet column chunk holding a single value plus nulls is materialized as a sparse column (memory proportional to the non-null rows) when the fraction of nulls is at least this ratio, and as a dense `Nullable` column otherwise. `1` disables sparse materialization for such chunks. )", 0) \ DECLARE(Bool, input_format_parquet_verify_checksums, true, R"( Verify page checksums when reading parquet files. @@ -250,6 +256,42 @@ Min bytes required for local read (file) to do seek, instead of read with ignore )", 0) \ DECLARE(Bool, input_format_parquet_enable_row_group_prefetch, true, R"( Enable row group prefetching during parquet parsing. Currently, only single-threaded parsing can prefetch. +)", 0) \ + DECLARE(UInt64, input_format_parquet_max_io_threads, 0, R"( +Size of the thread pool that issues reads for the Parquet reader, shared by all files read by the +query. `0` derives it from `max_download_threads` and `max_parsing_threads`. + +With too few reads in flight to cover the storage's response time, decoding threads end up running +the reads themselves or waiting for them. +)", 0) \ + DECLARE(UInt64, input_format_parquet_max_active_files, 0, R"( +How many Parquet files may read ahead at the same time when a query reads many files. `0` means no +limit, which is the previous behaviour. + +All files share one IO pool, so with many files each gets too few reads in flight and none finish +early. Files that do not hold a slot still read the row group they must deliver next. +)", 0) \ + DECLARE(UInt64, input_format_parquet_read_ahead_subgroups, 0, R"( +How many row subgroups ahead the Parquet reader may issue data-page reads for within a row group. `0` +keeps the previous behaviour, where a subgroup's reads are issued only after its predecessor has been +decoded, so the reader waits out the storage's response time on every subgroup. `1` issues the next +subgroup's reads while the current one decodes. Subgroups are still decoded in order. + +Read-ahead is opportunistic: it stops when its memory budget +(`input_format_parquet_read_ahead_memory_fraction`) is used up. Only helps for files whose row groups +are split into several subgroups (see `input_format_parquet_max_block_size`) and that have an offset +index; otherwise a row group's data is already read as one range. +)", 0) \ + DECLARE(Double, input_format_parquet_read_ahead_memory_fraction, 0.25, R"( +Share of the Parquet reader's prefetch memory budget (`input_format_parquet_prefetch_memory_fraction`) +reserved for data pages read ahead within a row group (`input_format_parquet_read_ahead_subgroups`). +Range `[0, 1]`. The rest of the prefetch budget keeps other row groups reading ahead. +)", 0) \ + DECLARE(UInt64, input_format_parquet_bytes_per_read_task, 0, R"( +Target size of a single read issued by the Parquet reader; nearby column chunks are coalesced up to +this size. `0` derives it from the min-bytes-for-seek of the underlying storage. + +A read never spans two row groups regardless of this setting. )", 0) \ DECLARE(Bool, input_format_arrow_allow_missing_columns, true, R"( Allow missing columns while reading Arrow input formats diff --git a/src/Core/SettingsChangesHistory.cpp b/src/Core/SettingsChangesHistory.cpp index f1b2bab3ac3a..6f8efb878299 100644 --- a/src/Core/SettingsChangesHistory.cpp +++ b/src/Core/SettingsChangesHistory.cpp @@ -42,6 +42,13 @@ const VersionToSettingsChangesMap & getSettingsChangesHistory() addSettingsChanges(settings_changes_history, "26.6.2.20001.altinityantalya", { {"use_puffin_files_cache", false, true, "Enables cache of parsed Puffin file content such as deletion vectors."}, + {"input_format_parquet_use_constant_column_optimization", false, true, "New setting: when a Parquet column chunk provably holds a single value in every row (per its min/max statistics), materialize that value directly instead of reading and decoding the column's data pages (reader v3)."}, + {"input_format_parquet_constant_column_sparse_ratio", 1.0, 0.9375, "New setting: a Parquet column chunk holding a single value plus nulls is materialized as a sparse column when the fraction of nulls is at least this ratio (reader v3)."}, + {"input_format_parquet_max_io_threads", 0, 0, "New setting: size of the thread pool that issues reads for the Parquet reader. 0 derives it from `max_download_threads` and `max_parsing_threads`; the derived value is larger than the previous hard-coded `max_download_threads`, which defaults to 4 and was chosen for the URL engine."}, + {"input_format_parquet_bytes_per_read_task", 0, 0, "New setting: target size of a single read issued by the Parquet reader. 0 derives it from the min-bytes-for-seek of the underlying storage."}, + {"input_format_parquet_read_ahead_subgroups", 0, 0, "New setting: how many row subgroups ahead the Parquet reader may issue data-page reads for within a row group. 0 keeps the previous behavior of issuing a subgroup's reads only after its predecessor was decoded."}, + {"input_format_parquet_read_ahead_memory_fraction", 0.25, 0.25, "New setting: share of the Parquet prefetch memory budget reserved for data pages read ahead within a row group."}, + {"input_format_parquet_max_active_files", 0, 0, "New setting: how many Parquet files may read ahead at the same time when a query reads many of them. 0 means no limit, which is the previous behavior."}, }); addSettingsChanges(settings_changes_history, "26.6", diff --git a/src/Formats/FormatFactory.cpp b/src/Formats/FormatFactory.cpp index d6c30363c0d2..3c6f2854372b 100644 --- a/src/Formats/FormatFactory.cpp +++ b/src/Formats/FormatFactory.cpp @@ -219,6 +219,8 @@ FormatSettings getFormatSettings(const ContextPtr & context, const Settings & se format_settings.parquet.bloom_filter_push_down = settings[Setting::input_format_parquet_bloom_filter_push_down]; format_settings.parquet.page_filter_push_down = settings[Setting::input_format_parquet_page_filter_push_down]; format_settings.parquet.use_offset_index = settings[Setting::input_format_parquet_use_offset_index]; + format_settings.parquet.use_constant_column_optimization = settings[Setting::input_format_parquet_use_constant_column_optimization]; + format_settings.parquet.constant_column_sparse_ratio = settings[Setting::input_format_parquet_constant_column_sparse_ratio]; 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]; @@ -248,6 +250,11 @@ FormatSettings getFormatSettings(const ContextPtr & context, const Settings & se format_settings.parquet.bloom_filter_bits_per_value = settings[Setting::output_format_parquet_bloom_filter_bits_per_value]; format_settings.parquet.bloom_filter_flush_threshold_bytes = settings[Setting::output_format_parquet_bloom_filter_flush_threshold_bytes]; format_settings.parquet.local_read_min_bytes_for_seek = settings[Setting::input_format_parquet_local_file_min_bytes_for_seek]; + format_settings.parquet.max_io_threads = settings[Setting::input_format_parquet_max_io_threads]; + format_settings.parquet.bytes_per_read_task = settings[Setting::input_format_parquet_bytes_per_read_task]; + format_settings.parquet.max_active_files = settings[Setting::input_format_parquet_max_active_files]; + format_settings.parquet.read_ahead_subgroups = settings[Setting::input_format_parquet_read_ahead_subgroups]; + format_settings.parquet.read_ahead_memory_fraction = settings[Setting::input_format_parquet_read_ahead_memory_fraction]; format_settings.parquet.enable_row_group_prefetch = settings[Setting::input_format_parquet_enable_row_group_prefetch]; format_settings.parquet.verify_checksums = settings[Setting::input_format_parquet_verify_checksums]; format_settings.parquet.local_time_as_utc = settings[Setting::input_format_parquet_local_time_as_utc]; diff --git a/src/Formats/FormatParserSharedResources.cpp b/src/Formats/FormatParserSharedResources.cpp index bdf4d6ffca43..47bb70911db7 100644 --- a/src/Formats/FormatParserSharedResources.cpp +++ b/src/Formats/FormatParserSharedResources.cpp @@ -27,6 +27,24 @@ FormatParserSharedResourcesPtr FormatParserSharedResources::singleThreaded(const } +bool FormatParserSharedResources::tryAcquirePrefetchSlot(size_t max_active) +{ + if (max_active == 0) + return true; // unlimited + size_t cur = active_prefetch_readers.load(std::memory_order_relaxed); + while (cur < max_active) + { + if (active_prefetch_readers.compare_exchange_weak(cur, cur + 1, std::memory_order_acq_rel, std::memory_order_relaxed)) + return true; + } + return false; +} + +void FormatParserSharedResources::releasePrefetchSlot() +{ + active_prefetch_readers.fetch_sub(1, std::memory_order_release); +} + void FormatParserSharedResources::finishStream() { num_streams.fetch_sub(1, std::memory_order_relaxed); diff --git a/src/Formats/FormatParserSharedResources.h b/src/Formats/FormatParserSharedResources.h index 8cfadffa026a..e75996780221 100644 --- a/src/Formats/FormatParserSharedResources.h +++ b/src/Formats/FormatParserSharedResources.h @@ -23,6 +23,7 @@ struct FormatParserSharedResources const size_t max_io_threads = 0; std::atomic num_streams{0}; + std::atomic active_prefetch_readers{0}; ThreadPoolCallbackRunnerFast parsing_runner; ThreadPoolCallbackRunnerFast io_runner; @@ -35,6 +36,12 @@ struct FormatParserSharedResources void finishStream(); + /// See input_format_parquet_max_active_files. Spreading one IO pool across many files leaves + /// each with too few reads in flight to cover the storage's response time and none finishing + /// early. A reader without a slot still reads what it must deliver next, so this cannot stall. + bool tryAcquirePrefetchSlot(size_t max_active); + void releasePrefetchSlot(); + size_t getParsingThreadsPerReader() const; size_t getIOThreadsPerReader() const; diff --git a/src/Formats/FormatSettings.h b/src/Formats/FormatSettings.h index 745898c0c751..fa25006eb167 100644 --- a/src/Formats/FormatSettings.h +++ b/src/Formats/FormatSettings.h @@ -349,6 +349,8 @@ struct FormatSettings bool bloom_filter_push_down = true; bool page_filter_push_down = true; bool use_offset_index = true; + bool use_constant_column_optimization = true; + float constant_column_sparse_ratio = 0.9375f; bool enable_json_parsing = true; bool preserve_order = false; @@ -359,6 +361,16 @@ struct FormatSettings UInt64 max_block_size = DEFAULT_BLOCK_SIZE; size_t prefer_block_bytes = DEFAULT_BLOCK_SIZE * 256; size_t local_read_min_bytes_for_seek = 8192; + /// 0 = derive from max_download_threads / max_parsing_threads. + size_t max_io_threads = 0; + /// 0 = derive from the storage's min-bytes-for-seek. + size_t bytes_per_read_task = 0; + /// 0 = no limit on how many files prefetch ahead concurrently. + size_t max_active_files = 0; + /// 0 = issue a subgroup's reads only after its predecessor finished. + size_t read_ahead_subgroups = 0; + /// Share of the prefetch memory budget reserved for read-ahead within a row group. + double read_ahead_memory_fraction = 0.25; 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 diff --git a/src/Formats/insertNullAsDefaultIfNeeded.cpp b/src/Formats/insertNullAsDefaultIfNeeded.cpp index d62719375d61..887ee81d92b5 100644 --- a/src/Formats/insertNullAsDefaultIfNeeded.cpp +++ b/src/Formats/insertNullAsDefaultIfNeeded.cpp @@ -16,6 +16,12 @@ namespace DB bool insertNullAsDefaultIfNeeded(ColumnWithTypeAndName & input_column, const ColumnWithTypeAndName & header_column, size_t column_i, BlockMissingValues * block_missing_values) { + /// Input formats may produce sparse columns (e.g. the Parquet reader for chunks that are all + /// null or mostly null), and they survive the Native protocol from the client. The casts below + /// expect the concrete Nullable / Array / Tuple / Map columns, so expand first. + if (input_column.column->isSparse()) + input_column.column = input_column.column->convertToFullColumnIfSparse(); + if (isArray(input_column.type) && isArray(header_column.type)) { ColumnWithTypeAndName nested_input_column; diff --git a/src/Interpreters/AsynchronousInsertQueue.cpp b/src/Interpreters/AsynchronousInsertQueue.cpp index 88207be6ba60..2ff4d0cd69dd 100644 --- a/src/Interpreters/AsynchronousInsertQueue.cpp +++ b/src/Interpreters/AsynchronousInsertQueue.cpp @@ -1414,7 +1414,12 @@ Chunk AsynchronousInsertQueue::processPreprocessedEntries( auto columns = block_to_insert.getColumns(); for (size_t i = 0, s = columns.size(); i < s; ++i) - result_columns[i]->insertRangeFrom(*columns[i], 0, columns[i]->size()); + { + /// Blocks may carry ColumnConst / ColumnSparse (e.g. from an input format that materializes + /// provably-constant column chunks); insertRangeFrom needs the concrete column type. + auto full_column = columns[i]->convertToFullColumnIfConst()->convertToFullColumnIfSparse(); + result_columns[i]->insertRangeFrom(*full_column, 0, full_column->size()); + } total_rows += block_to_insert.rows(); diff --git a/src/Processors/Executors/StreamingFormatExecutor.cpp b/src/Processors/Executors/StreamingFormatExecutor.cpp index 67f93053e158..15bfcb86d9d0 100644 --- a/src/Processors/Executors/StreamingFormatExecutor.cpp +++ b/src/Processors/Executors/StreamingFormatExecutor.cpp @@ -149,7 +149,13 @@ size_t StreamingFormatExecutor::insertChunk(Chunk chunk, size_t num_bytes) auto columns = chunk.detachColumns(); for (size_t i = 0, s = columns.size(); i < s; ++i) - result_columns[i]->insertRangeFrom(*columns[i], 0, columns[i]->size()); + { + /// Input formats may produce ColumnConst / ColumnSparse (e.g. the Parquet reader for column + /// chunks whose statistics prove a single value or all nulls); insertRangeFrom into the full + /// result column requires the concrete column type. + auto full_column = columns[i]->convertToFullColumnIfConst()->convertToFullColumnIfSparse(); + result_columns[i]->insertRangeFrom(*full_column, 0, full_column->size()); + } return chunk_rows; } diff --git a/src/Processors/Formats/Impl/Parquet/Prefetcher.cpp b/src/Processors/Formats/Impl/Parquet/Prefetcher.cpp index 1141cfe870a2..ac693ebb4fbe 100644 --- a/src/Processors/Formats/Impl/Parquet/Prefetcher.cpp +++ b/src/Processors/Formats/Impl/Parquet/Prefetcher.cpp @@ -19,6 +19,9 @@ namespace DB::ErrorCodes namespace ProfileEvents { extern const Event ParquetFetchWaitTimeMicroseconds; + extern const Event ParquetReadTasks; + extern const Event ParquetReadTaskBytes; + extern const Event ParquetPrefetchStarvation; extern const Event ParquetPrefetcherReadRandomRead; extern const Event ParquetPrefetcherReadSeekAndRead; extern const Event ParquetPrefetcherReadEntireFile; @@ -31,11 +34,50 @@ void Prefetcher::init(ReadBuffer * reader_, const ReadOptions & options, FormatP { min_bytes_for_seek = options.min_bytes_for_seek; bytes_per_read_task = options.bytes_per_read_task; + /// Below `min_bytes_for_seek` a read stops amortizing its round trip. + min_bytes_per_read_task = std::max(min_bytes_for_seek, bytes_per_read_task / 4); parser_shared_resources = parser_shared_resources_; + if (parser_shared_resources) + io_concurrency_target = std::max(size_t(1), parser_shared_resources->max_io_threads); determineReadModeAndFileSize(reader_, options); range_sets.resize(1); } +void Prefetcher::setRowGroupRanges(std::vector> ranges) +{ + chassert(std::is_sorted(ranges.begin(), ranges.end())); + std::lock_guard lock(mutex); + row_group_ranges = std::move(ranges); +} + +std::optional Prefetcher::rowGroupIndexFor(size_t offset) const +{ + if (row_group_ranges.empty()) + return std::nullopt; + /// First range starting after `offset`; the candidate is the one before it. + auto hi = std::upper_bound(row_group_ranges.begin(), row_group_ranges.end(), offset, + [](size_t off, const std::pair & range) { return off < range.first; }); + if (hi == row_group_ranges.begin()) + return std::nullopt; + const auto & range = *(hi - 1); + if (offset >= range.second) + return std::nullopt; // gap between row groups, or after the last one + return size_t(hi - 1 - row_group_ranges.begin()); +} + +size_t Prefetcher::currentReadTaskBudget() const +{ + if (min_bytes_per_read_task >= bytes_per_read_task) + return bytes_per_read_task; + /// Hysteresis, so the size doesn't flap around the threshold. + size_t in_flight = tasks_in_flight.load(std::memory_order_relaxed); + if (in_flight >= io_concurrency_target * 2) + return bytes_per_read_task; + if (in_flight < io_concurrency_target) + return min_bytes_per_read_task; + return (min_bytes_per_read_task + bytes_per_read_task) / 2; +} + Prefetcher::~Prefetcher() { shutdown->shutdown(); @@ -307,13 +349,28 @@ void Prefetcher::pickRangesAndCreateTaskIfNotExists(RequestState * initial_req, size_t end_idx = range_idx + 1; size_t total_length_of_covered_ranges = end_offset - start_offset; + /// Compared against the resulting span, not the distance from the seed range in each direction: + /// the latter let a task reach `seed length + 2 * bytes_per_read_task`. + const size_t task_budget = currentReadTaskBudget(); + + /// One read must not span two row groups: `getRangeData` waits for the whole task, so the + /// earlier row group would wait for the later one's bytes and delivery would serialize. + /// Bytes outside every row group (page indexes, bloom filters, footer) don't count: they may join + /// whichever row group the task already covers. `claimed_row_group` is the one it covers so far. + std::optional claimed_row_group = rowGroupIndexFor(start_offset); + auto other_row_group = [&](size_t offset) + { + auto rg = rowGroupIndexFor(offset); + return rg.has_value() && claimed_row_group.has_value() && *rg != *claimed_row_group; + }; + /// Go left. - size_t initial_offset = start_offset; for (size_t idx = range_idx; idx > 0; --idx) { const RangeState & r = ranges[idx - 1]; if (r.end + min_bytes_for_seek <= start_offset || // short gap - r.start + bytes_per_read_task <= initial_offset || // task not too big + other_row_group(r.start) || // would reach into another row group + end_offset - std::min(r.start, start_offset) > task_budget || // task not too big !r.request->allow_incidental_read.load(std::memory_order_relaxed)) // range wants to be coalesced break; @@ -321,6 +378,8 @@ void Prefetcher::pickRangesAndCreateTaskIfNotExists(RequestState * initial_req, if (s == RequestState::State::HasRange) { /// Include this range in the task. + if (!claimed_row_group.has_value()) + claimed_row_group = rowGroupIndexFor(r.start); start_idx = idx - 1; total_length_of_covered_ranges += r.length(); start_offset = std::min(start_offset, r.start); @@ -343,18 +402,20 @@ void Prefetcher::pickRangesAndCreateTaskIfNotExists(RequestState * initial_req, } /// Go right. - initial_offset = end_offset; for (size_t idx = range_idx + 1; idx < ranges.size(); ++idx) { const RangeState & r = ranges[end_idx]; if (end_offset + min_bytes_for_seek <= r.start || - initial_offset + bytes_per_read_task <= r.end || + (r.end > r.start && other_row_group(r.end - 1)) || // would reach into another row group + std::max(r.end, end_offset) - start_offset > task_budget || !r.request->allow_incidental_read.load(std::memory_order_relaxed)) break; const auto s = r.request->state.load(std::memory_order_relaxed); if (s == RequestState::State::HasRange) { + if (!claimed_row_group.has_value() && r.end > r.start) + claimed_row_group = rowGroupIndexFor(r.end - 1); end_idx = idx + 1; total_length_of_covered_ranges += r.length(); end_offset = std::max(end_offset, r.end); @@ -370,8 +431,11 @@ void Prefetcher::pickRangesAndCreateTaskIfNotExists(RequestState * initial_req, /// Create task. Task & task = tasks.emplace_back(); + task.owner = this; task.offset = start_offset; task.length = end_offset - task.offset; + ProfileEvents::increment(ProfileEvents::ParquetReadTasks); + ProfileEvents::increment(ProfileEvents::ParquetReadTaskBytes, task.length); task.memory_amplification = 1. * static_cast(task.length) / static_cast(total_length_of_covered_ranges); size_t initial_refcount = end_idx - start_idx + 1; task.refcount.store(initial_refcount); @@ -406,15 +470,22 @@ void Prefetcher::decreaseTaskRefcount(Task * task, size_t amount) if (c != amount) return; - if (task->state.exchange(Task::State::Deallocated) != Task::State::Running) + const auto prev = task->state.exchange(Task::State::Deallocated); + if (prev != Task::State::Running) { task->buf = {}; task->cached_region.reset(); } + /// Cancelled before any thread picked it up, so nothing else will account for it. Running is + /// accounted by whoever runs it; Done was accounted when it finished. + if (prev == Task::State::Scheduled && task->owner) + task->owner->tasks_in_flight.fetch_sub(1, std::memory_order_relaxed); } void Prefetcher::scheduleTask(Task * task) { + /// Counted from queueing, not from when a thread picks it up: it is already committed work. + tasks_in_flight.fetch_add(1, std::memory_order_relaxed); if (parser_shared_resources && !parser_shared_resources->io_runner.isDisabled()) parser_shared_resources->io_runner([this, task, _shutdown = shutdown] { @@ -435,6 +506,10 @@ std::span Prefetcher::getRangeData(const PrefetchHandle & request) { Stopwatch wait_time; + /// Read-ahead didn't stay in front of decoding: this thread either runs the read inline or + /// parks until it lands. + ProfileEvents::increment(ProfileEvents::ParquetPrefetchStarvation); + if (s == Task::State::Scheduled) { s = runTask(task); @@ -542,6 +617,8 @@ Prefetcher::Task::State Prefetcher::runTask(Task * task) task->cached_region.reset(); } + tasks_in_flight.fetch_sub(1, std::memory_order_relaxed); + task->completion.notify(); return s; diff --git a/src/Processors/Formats/Impl/Parquet/Prefetcher.h b/src/Processors/Formats/Impl/Parquet/Prefetcher.h index 40796dd10342..f82e3e127a02 100644 --- a/src/Processors/Formats/Impl/Parquet/Prefetcher.h +++ b/src/Processors/Formats/Impl/Parquet/Prefetcher.h @@ -41,6 +41,11 @@ class Prefetcher /// Called at most once, after all registerRange calls and before all enqueue/getRangeData calls. void finalizeRanges(); + /// Keeps one read from covering parts of two row groups. `ranges` are the [start, end) byte + /// ranges of the row groups, sorted by start. Bytes outside every range (footer, page indexes, + /// bloom filters) belong to no row group and may be coalesced with either neighbour. + void setRowGroupRanges(std::vector> ranges); + /// Replace a requested range with a set of disjoint smaller ranges contained within it. /// `subranges` must be sorted. std::vector splitRange( @@ -144,6 +149,9 @@ class Prefetcher }; std::optional cached_region; + /// `decreaseTaskRefcount` is static but has to account for a task cancelled before it ran. + Prefetcher * owner = nullptr; + std::atomic state {State::Scheduled}; /// How many RequestState-s in HasTask state point to this Task. std::atomic refcount {}; @@ -179,6 +187,15 @@ class Prefetcher size_t min_bytes_for_seek{}; size_t bytes_per_read_task{}; + /// See setRowGroupRanges. Empty until it is called. + std::vector> row_group_ranges; + + /// Reads running or queued. Drives read-task size: smaller reads fill an idle pool faster, larger + /// ones amortize the round trip once it is busy. + std::atomic tasks_in_flight {0}; + size_t io_concurrency_target = 1; + size_t min_bytes_per_read_task{}; + std::shared_ptr shutdown = std::make_shared(); /// Locked when creating a Task. @@ -194,6 +211,11 @@ class Prefetcher /// (One mutex for all tasks because it's not used frequently.) std::mutex exception_mutex; + /// Index of the row group whose byte range contains `offset`, or nullopt if `offset` is outside + /// every row group (metadata) or the layout isn't known yet. Called with `mutex` held. + std::optional rowGroupIndexFor(size_t offset) const; + size_t currentReadTaskBudget() const; + void determineReadModeAndFileSize(ReadBuffer * reader_, const ReadOptions & options); /// Creates and starts a Task covering this request and possibly other nearby ranges. /// diff --git a/src/Processors/Formats/Impl/Parquet/ReadCommon.h b/src/Processors/Formats/Impl/Parquet/ReadCommon.h index cbaf7f095ec6..4701af2432b7 100644 --- a/src/Processors/Formats/Impl/Parquet/ReadCommon.h +++ b/src/Processors/Formats/Impl/Parquet/ReadCommon.h @@ -92,6 +92,10 @@ enum class ReadStage /// row groups prefetch ahead while only a few decode at once. Decouples fetch from decode depth. ColumnDataPrefetch, ColumnData, + /// Accounting only, never has tasks: compressed data pages issued ahead for later row subgroups + /// of a row group (see input_format_parquet_read_ahead_subgroups). Gets its own memory budget so + /// read-ahead cannot eat the ColumnDataPrefetch share that keeps other row groups moving. + ColumnDataReadAhead, Deliver, diff --git a/src/Processors/Formats/Impl/Parquet/ReadManager.cpp b/src/Processors/Formats/Impl/Parquet/ReadManager.cpp index b6375ccb96c2..84e20822f2ca 100644 --- a/src/Processors/Formats/Impl/Parquet/ReadManager.cpp +++ b/src/Processors/Formats/Impl/Parquet/ReadManager.cpp @@ -26,6 +26,7 @@ namespace ProfileEvents { extern const Event ParquetDecodingTasks; extern const Event ParquetDecodingTaskBatches; + extern const Event ParquetReadAheadSubgroups; extern const Event ParquetReadRowGroups; extern const Event ParquetPrunedRowGroups; } @@ -82,10 +83,14 @@ void ReadManager::init(FormatParserSharedResourcesPtr parser_shared_resources_, /// 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 read_ahead_memory_fraction = reader.options.format.parquet.read_ahead_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 (!(read_ahead_memory_fraction >= 0 && read_ahead_memory_fraction <= 1)) + throw Exception(ErrorCodes::BAD_ARGUMENTS, + "input_format_parquet_read_ahead_memory_fraction must be in [0, 1], got {}", read_ahead_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); @@ -102,7 +107,10 @@ void ReadManager::init(FormatParserSharedResourcesPtr parser_shared_resources_, 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); + /// Read-ahead is carved out of the prefetch share, so enabling it doesn't change the decode budget. + const double prefetch_memory = data_memory_fraction * prefetch_memory_fraction; + set_fractions(ReadStage::ColumnDataPrefetch, prefetch_memory * (1.0 - read_ahead_memory_fraction), issuer_thread_fraction); + set_fractions(ReadStage::ColumnDataReadAhead, prefetch_memory * read_ahead_memory_fraction, 0); set_fractions(ReadStage::ColumnData, data_memory_fraction * (1.0 - prefetch_memory_fraction), decode_thread_fraction); set_fractions(ReadStage::Deliver, 0, 0); @@ -130,6 +138,8 @@ void ReadManager::init(FormatParserSharedResourcesPtr parser_shared_resources_, ReadManager::~ReadManager() { shutdown->shutdown(); + if (holds_prefetch_slot.exchange(false) && parser_shared_resources) + parser_shared_resources->releasePrefetchSlot(); } void ReadManager::cancel() noexcept @@ -175,6 +185,7 @@ void ReadManager::finishRowGroupStage(size_t row_group_idx, ReadStage stage, Mem case ReadStage::NotStarted: case ReadStage::ColumnDataPrefetch: case ReadStage::ColumnData: + case ReadStage::ColumnDataReadAhead: case ReadStage::Deliver: chassert(false); break; @@ -328,9 +339,8 @@ void ReadManager::addTasksToReadColumns(size_t row_group_idx, size_t row_subgrou LOG_TEST(getLogger("ParquetReadManager"), "addTasksToReadColumns: not added due locations empty i={} step_idx={} row_group_idx={} row_subgroup_idx={}", i, step_idx, row_group_idx, row_subgroup_idx); } } - else + else if (stage == ReadStage::ColumnData) { - /// `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 = stage, @@ -341,6 +351,29 @@ void ReadManager::addTasksToReadColumns(size_t row_group_idx, size_t row_subgrou } } + if (stage == ReadStage::ColumnDataPrefetch && row_subgroup.reads_issued_ahead && step_idx == firstStep()) + { + /// The previous subgroup's ColumnDataPrefetch already issued this subgroup's first-step + /// reads (read-ahead); nothing left to issue, go straight to decoding. + chassert(add_tasks.empty()); + stage = ReadStage::ColumnData; + continue; + } + + if (stage == ReadStage::ColumnDataPrefetch) + { + /// Issuing this subgroup's reads is one unit of work for the whole subgroup, not one per + /// column: it runs on the scheduling thread (see scheduleTask) and its runTask is empty, + /// so a task per column only bought a thread-pool round trip each and a wider barrier. + chassert(add_tasks.empty()); + add_tasks.push_back(Task { + .stage = stage, + .step_idx = step_idx, + .row_group_idx = row_group_idx, + .row_subgroup_idx = row_subgroup_idx, + .column_idx = UINT64_MAX}); + } + if (add_tasks.empty() && is_offset_index) { /// Don't need to read offset index, move on to the next stage (ColumnDataPrefetch). @@ -369,6 +402,12 @@ void ReadManager::addTasksToReadColumns(size_t row_group_idx, size_t row_subgrou } } +size_t ReadManager::firstStep() const +{ + /// 1 if there are prewhere steps, 0 otherwise. + return reader.steps.empty() ? 0 : 1; +} + void ReadManager::finishRowSubgroupStage(size_t row_group_idx, size_t row_subgroup_idx, ReadStage stage, size_t step_idx, MemoryUsageDiff & diff) { RowGroup & row_group = reader.row_groups[row_group_idx]; @@ -383,8 +422,7 @@ void ReadManager::finishRowSubgroupStage(size_t row_group_idx, size_t row_subgro { case ReadStage::NotStarted: { - /// 1 if there are prewhere steps, 0 otherwise - size_t first_step = reader.steps.empty() ? 0 : 1; + size_t first_step = firstStep(); if (first_step < reader.steps.size() + 1) { addTasksToReadColumns(row_group_idx, row_subgroup_idx, ReadStage::OffsetIndex, first_step, diff); @@ -466,6 +504,7 @@ void ReadManager::finishRowSubgroupStage(size_t row_group_idx, size_t row_subgro addTasksToReadColumns(row_group_idx, row_subgroup_idx, ReadStage::ColumnData, step_idx, diff); return; } + case ReadStage::ColumnDataReadAhead: case ReadStage::Deallocated: chassert(false); break; @@ -481,6 +520,11 @@ void ReadManager::finishRowSubgroupStage(size_t row_group_idx, size_t row_subgro while (main_ptr < row_group.subgroups.size()) { RowSubgroup & next_subgroup = row_group.subgroups[main_ptr]; + /// `main_ptr` is either a subgroup nobody has started, or the current subgroup itself when + /// PREWHERE just dropped all of its rows (the ColumnData case above breaks out with + /// rows_pass == 0 without advancing read_ptr); the branch below then deallocates it and + /// moves on. Only one subgroup of a row group is in progress at a time, so the exchange + /// cannot hit anything else. ReadStage next_subgroup_stage = next_subgroup.stage.load(); if (!next_subgroup.stage.compare_exchange_strong( next_subgroup_stage, ReadStage::OffsetIndex)) @@ -488,8 +532,7 @@ void ReadManager::finishRowSubgroupStage(size_t row_group_idx, size_t row_subgro if (next_subgroup.filter.rows_pass > 0) { - size_t first_step = reader.steps.empty() ? 0 : 1; - addTasksToReadColumns(row_group_idx, main_ptr, ReadStage::OffsetIndex, first_step, diff); + addTasksToReadColumns(row_group_idx, main_ptr, ReadStage::OffsetIndex, firstStep(), diff); break; } else @@ -583,6 +626,10 @@ void ReadManager::flushMemoryUsageDiff(MemoryUsageDiff && diff) stages[i].memory_usage.fetch_add(d, std::memory_order_relaxed); } + /// Accounting only; nothing is ever scheduled on it. + if (i == size_t(ReadStage::ColumnDataReadAhead)) + continue; + bool should_schedule = (diff.stages_to_schedule & (1ul << i)) != 0; if (!should_schedule && d < 0) { @@ -649,6 +696,19 @@ void ReadManager::scheduleTasksIfNeeded(ReadStage stage_idx) if (!can_schedule && !is_privileged) break; + /// Read ahead only while this file holds an active-file slot. Files without one still read + /// the row group they must deliver next, so this cannot stall the query. + if (stage_idx == ReadStage::ColumnDataPrefetch && !is_privileged + && reader.options.format.parquet.max_active_files != 0 + && !holds_prefetch_slot.load(std::memory_order_relaxed)) + { + bool expected = false; + if (!parser_shared_resources->tryAcquirePrefetchSlot(reader.options.format.parquet.max_active_files)) + break; + if (!holds_prefetch_slot.compare_exchange_strong(expected, true)) + parser_shared_resources->releasePrefetchSlot(); // another thread got one first + } + if (!stage.schedulable_row_groups.unset(row_group_idx, std::memory_order_acquire)) { LOG_TEST(getLogger("ParquetReadManager"), "scheduleTasksIfNeeded: another thread got row group {}", row_group_idx); @@ -735,7 +795,35 @@ void ReadManager::scheduleTask(Task task, bool is_first_in_group, MemoryUsageDif std::vector prefetches; RowGroup & row_group = reader.row_groups[task.row_group_idx]; ssize_t memory_before = diff.by_stage[size_t(diff.cur_stage)]; - if (task.column_idx != UINT64_MAX) + + if (task.stage == ReadStage::ColumnDataPrefetch) + { + /// Queue every column's data-page reads for this subgroup. Charged to the ColumnDataPrefetch + /// budget, which is separate from the decode budget and bounds how many row groups prefetch. + RowSubgroup & row_subgroup = row_group.subgroups.at(task.row_subgroup_idx); + if (row_subgroup.filter.rows_pass > 0) + { + for (size_t i = 0; i < reader.primitive_columns.size(); ++i) + { + if (reader.primitive_columns[i].first_step_to_calculate != task.step_idx) + continue; + ColumnChunk & column = row_group.columns.at(i); + reader.determinePagesToPrefetch(column, row_subgroup, row_group, prefetches); + + /// Side note: would be nice to avoid reading the dictionary if all dictionary-encoded + /// pages were filtered out (e.g. if it's a 100 MB column chunk with unique long strings, + /// typically only the first ~1 MB would be dictionary-encoded; if we only need a few + /// rows, we likely won't hit that 1 MB). But AFAICT parquet metadata doesn't have + /// enough information for that (there's no page encoding in offset/column indexes). + if (!column.dictionary.isInitialized() && column.dictionary_page_prefetch) + prefetches.push_back(&column.dictionary_page_prefetch); + + if (column.data_pages.empty()) + prefetches.push_back(&column.data_pages_prefetch); + } + } + } + else if (task.column_idx != UINT64_MAX) { ColumnChunk & column = row_group.columns.at(task.column_idx); switch (task.stage) @@ -758,32 +846,6 @@ void ReadManager::scheduleTask(Task task, bool is_first_in_group, MemoryUsageDif case ReadStage::OffsetIndex: prefetches.push_back(&column.offset_index_prefetch); break; - case ReadStage::ColumnDataPrefetch: - { - RowSubgroup & row_subgroup = row_group.subgroups.at(task.row_subgroup_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 - /// pages were filtered out (e.g. if it's a 100 MB column chunk with unique long strings, - /// typically only the first ~1 MB would be dictionary-encoded; if we only need a few - /// rows, we likely won't hit that 1 MB). But AFAICT parquet metadata doesn't have - /// enough information for that (there's no page encoding in offset/column indexes). - if (!column.dictionary.isInitialized() && column.dictionary_page_prefetch) - { - prefetches.push_back(&column.dictionary_page_prefetch); - } - - if (column.data_pages.empty()) - { - prefetches.push_back(&column.data_pages_prefetch); - } - break; - } case ReadStage::ColumnData: { RowSubgroup & row_subgroup = row_group.subgroups.at(task.row_subgroup_idx); @@ -797,6 +859,8 @@ void ReadManager::scheduleTask(Task task, bool is_first_in_group, MemoryUsageDif subchunk.column_and_offsets_memory = MemoryUsageToken(column_memory, &diff); break; } + case ReadStage::ColumnDataPrefetch: // handled above, for the whole subgroup at once + case ReadStage::ColumnDataReadAhead: case ReadStage::NotStarted: case ReadStage::Deliver: case ReadStage::Deallocated: @@ -815,10 +879,53 @@ void ReadManager::scheduleTask(Task task, bool is_first_in_group, MemoryUsageDif reader.prefetcher.startPrefetch(prefetches, &diff); + /// Read-ahead: also issue the next subgroups' first-step reads now, so the storage's + /// response time overlaps this subgroup's decoding instead of being paid again after it. + /// Subgroups of a row group are decoded strictly in order and share one page cursor per + /// column chunk, so we only issue reads here; the next subgroup is admitted and decoded by + /// the normal path (finishRowSubgroupStage), which then skips its own ColumnDataPrefetch + /// stage (`reads_issued_ahead`). Idempotent by construction: determinePagesToPrefetch + /// advances `data_pages_prefetch_idx` past the pages it handed out, and startPrefetch + /// skips handles that already have a task. Charged to the ColumnDataReadAhead budget. + /// Without an offset index there is one range per column chunk and this is a no-op. + if (task.stage == ReadStage::ColumnDataPrefetch && task.step_idx == firstStep() + && row_group.subgroups.at(task.row_subgroup_idx).filter.rows_pass > 0) + { + std::vector read_ahead_prefetches; + const Stage & read_ahead_stage = stages[size_t(ReadStage::ColumnDataReadAhead)]; + const auto read_ahead_limits = SharedResourcesExt::getLimitsPerReader( + *parser_shared_resources, read_ahead_stage.memory_target_fraction, /*thread_fraction=*/ 0); + const size_t max_ahead = reader.options.format.parquet.read_ahead_subgroups; + for (size_t k = 1; k <= max_ahead && task.row_subgroup_idx + k < row_group.subgroups.size(); ++k) + { + size_t read_ahead_memory = read_ahead_stage.memory_usage.load(std::memory_order_relaxed) + + size_t(std::max(0, diff.by_stage[size_t(ReadStage::ColumnDataReadAhead)])); + if (read_ahead_memory >= read_ahead_limits.memory_high_watermark) + break; // read-ahead budget used up + + RowSubgroup & next_subgroup = row_group.subgroups[task.row_subgroup_idx + k]; + if (next_subgroup.filter.rows_pass == 0) + continue; // skipped by the normal path, nothing to read + for (size_t i = 0; i < reader.primitive_columns.size(); ++i) + { + if (reader.primitive_columns[i].first_step_to_calculate != task.step_idx) + continue; + reader.determinePagesToPrefetch(row_group.columns.at(i), next_subgroup, row_group, read_ahead_prefetches); + } + next_subgroup.reads_issued_ahead = true; + + /// Charge as we go so the budget check above sees this subgroup's bytes. + const ReadStage saved_stage = std::exchange(diff.cur_stage, ReadStage::ColumnDataReadAhead); + reader.prefetcher.startPrefetch(read_ahead_prefetches, &diff); + diff.cur_stage = saved_stage; + ProfileEvents::increment(ProfileEvents::ParquetReadAheadSubgroups); + read_ahead_prefetches.clear(); + } + } + /// 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. + /// ColumnDataPrefetch is the exception: its work happened above and its runTask is empty, so the + /// compressed bytes it charges say nothing about how long it takes to run. ssize_t memory_after = diff.by_stage[size_t(diff.cur_stage)]; task.cost_estimate_bytes = task.stage == ReadStage::ColumnDataPrefetch ? 0 @@ -927,6 +1034,7 @@ void ReadManager::runTask(Task task, bool last_in_batch, MemoryUsageDiff & diff) break; } case ReadStage::NotStarted: + case ReadStage::ColumnDataReadAhead: case ReadStage::Deliver: case ReadStage::Deallocated: chassert(false); diff --git a/src/Processors/Formats/Impl/Parquet/ReadManager.h b/src/Processors/Formats/Impl/Parquet/ReadManager.h index 7073492d2174..f31c7dc623d4 100644 --- a/src/Processors/Formats/Impl/Parquet/ReadManager.h +++ b/src/Processors/Formats/Impl/Parquet/ReadManager.h @@ -33,6 +33,12 @@ class ReadManager public: Reader reader; + /// See input_format_parquet_max_active_files. Progress never depends on holding a slot. + std::atomic holds_prefetch_slot {false}; + + /// Index of the first PREWHERE step (1), or 0 when there are no steps. + size_t firstStep() const; + /// To initialize ReadManager: /// 1. call manager.reader.prefetcher.init /// 2. call manager.reader.init diff --git a/src/Processors/Formats/Impl/Parquet/Reader.cpp b/src/Processors/Formats/Impl/Parquet/Reader.cpp index a3c91158f695..e8e279da5482 100644 --- a/src/Processors/Formats/Impl/Parquet/Reader.cpp +++ b/src/Processors/Formats/Impl/Parquet/Reader.cpp @@ -1,4 +1,6 @@ #include +#include +#include #include #include #include @@ -42,6 +44,8 @@ namespace ProfileEvents { extern const Event ParquetRowsFilterExpression; extern const Event ParquetColumnsFilterExpression; + extern const Event ParquetConstantColumnChunks; + extern const Event ParquetConstantColumnChunksWithNulls; } namespace DB::Parquet @@ -456,12 +460,45 @@ void Reader::prefilterAndInitRowGroups(const std::optionalmeta_data.statistics.__isset.null_count && column.meta->meta_data.statistics.null_count == 0; column.need_null_map = is_nullable && !null_count_is_known_to_be_zero; + + detectConstantColumn(column, primitive_columns[column_idx]); } } if (row_groups.empty()) return; // all row groups were skipped + /// So a single read never covers parts of two row groups; see Prefetcher::setRowGroupRanges. + { + std::vector> bounds; + bounds.reserve(file_metadata.row_groups.size()); + for (const auto & rg : file_metadata.row_groups) + { + size_t start = std::numeric_limits::max(); + size_t end = 0; + for (const auto & col : rg.columns) + { + /// Some writers leave dictionary_page_offset unset even with a dictionary present, + /// in which case data_page_offset already points at it. + size_t col_start = size_t(col.meta_data.data_page_offset); + if (col.meta_data.__isset.dictionary_page_offset && col.meta_data.dictionary_page_offset > 0) + col_start = std::min(col_start, size_t(col.meta_data.dictionary_page_offset)); + start = std::min(start, col_start); + end = std::max(end, col_start + size_t(col.meta_data.total_compressed_size)); + } + if (start == std::numeric_limits::max() || end <= start) + continue; // unusable metadata + if (!bounds.empty() && start < bounds.back().second) + { + bounds.clear(); // not laid out in order, or overlapping; don't guess + break; + } + bounds.emplace_back(start, end); + } + if (!bounds.empty()) + prefetcher.setRowGroupRanges(std::move(bounds)); + } + if (options.format.parquet.bloom_filter_push_down && format_filter_info->key_condition) prepareBloomFilterCondition(); @@ -578,7 +615,9 @@ void Reader::initializePrefetches() /// Dictionary page. size_t dict_page_length = 0; - if (column.meta->meta_data.__isset.dictionary_page_offset) + /// A Const/AllDefault column chunk is materialized without reading any pages (data or + /// dictionary), so don't prefetch its dictionary page either. + if (column.meta->meta_data.__isset.dictionary_page_offset && !constantKindSkipsDataPages(column.constant_kind)) { /// We assume that the dictionary page is immediately followed by the first data page. size_t start = size_t(column.meta->meta_data.dictionary_page_offset); @@ -626,8 +665,14 @@ void Reader::initializePrefetches() max_header_length, /*likely_to_be_used=*/ true); } + /// A Const/AllDefault column chunk is materialized without reading any of its pages (see + /// detectConstantColumn and decodePrimitiveColumn), so it needs neither the offset index + /// nor the column index nor the data pages. Page-level pruning would be redundant: every + /// page of the chunk holds the same value (or only nulls), so the key condition has already + /// been decided at the row group level. + /// Offset index. - if (use_offset_index && + if (use_offset_index && !constantKindSkipsDataPages(column.constant_kind) && column.meta->__isset.offset_index_offset && column.meta->__isset.offset_index_length) { column.offset_index_prefetch = prefetcher.registerRange( @@ -636,7 +681,8 @@ void Reader::initializePrefetches() } /// Column index. - column.use_column_index = primitive_columns[column_idx].column_index_condition + column.use_column_index = !constantKindSkipsDataPages(column.constant_kind) + && primitive_columns[column_idx].column_index_condition && column.offset_index_prefetch && column.meta->__isset.column_index_offset && column.meta->__isset.column_index_length; if (column.use_column_index) @@ -656,10 +702,11 @@ void Reader::initializePrefetches() if (file_metadata.created_by == "parquet-mr" && !column.meta->meta_data.__isset.dictionary_page_offset && !column.meta->__isset.offset_index_offset) data_pages_extra_bytes = std::min(100ul, prefetcher.getFileSize() - size_t(column.meta->meta_data.data_page_offset) - column.data_pages_bytes); - column.data_pages_prefetch = prefetcher.registerRange( - size_t(column.meta->meta_data.data_page_offset), - column.data_pages_bytes + data_pages_extra_bytes, - /*likely_to_be_used=*/ true); + if (!constantKindSkipsDataPages(column.constant_kind)) + column.data_pages_prefetch = prefetcher.registerRange( + size_t(column.meta->meta_data.data_page_offset), + column.data_pages_bytes + data_pages_extra_bytes, + /*likely_to_be_used=*/ true); } } @@ -1224,6 +1271,8 @@ void Reader::decodeOffsetIndex(ColumnChunk & column, const RowGroup & row_group) void Reader::determinePagesToPrefetch(ColumnChunk & column, const RowSubgroup & row_subgroup, const RowGroup & row_group, std::vector & out) { chassert(row_subgroup.filter.rows_pass > 0); + if (constantKindSkipsDataPages(column.constant_kind)) + return; // Const/AllDefault column: data pages are never read if (column.offset_index.page_locations.empty()) return; // no offset index, can't prefetch individual pages @@ -1341,6 +1390,19 @@ double Reader::estimateAverageStringLengthPerRow(const ColumnChunk & column, con double Reader::estimateColumnMemoryBytesPerRow(const ColumnChunk & column, const RowGroup & row_group, const PrimitiveColumnInfo & column_info) const { + /// Const/AllDefault chunks are materialized in O(1) memory; SparseNulls in O(non-null rows) + /// (8-byte offset + value each). Estimating them at full size would make the memory scheduler + /// under-parallelize exactly the files this optimization targets. + if (constantKindSkipsDataPages(column.constant_kind)) + return 0; + if (column.constant_kind == ConstantKind::SparseNulls) + { + const auto & stats = column.meta->meta_data.statistics; + double non_null_ratio = 1. - static_cast(stats.null_count) / static_cast(std::max(1, column.meta->meta_data.num_values)); + double value_size = column_info.output_type->haveMaximumSizeOfValue() ? static_cast(column_info.output_type->getMaximumSizeOfValueInMemory()) : 32.; + return non_null_ratio * (8. + value_size); + } + double res = 0; if (column_info.output_type->haveMaximumSizeOfValue()) /// Fixed-size values, e.g. numbers or FixedString. @@ -1361,8 +1423,172 @@ double Reader::estimateColumnMemoryBytesPerRow(const ColumnChunk & column, const return res; } +bool Reader::isConstantColumnCandidate(const PrimitiveColumnInfo & column_info) const +{ + if (!options.format.parquet.use_constant_column_optimization) + return false; + /// We rely on column chunk min/max statistics being both present and decodable. + if (!column_info.decoder.allow_stats) + return false; + + /// Only flat, top-level primitive columns, so that one parquet value maps 1:1 to one output row + /// and formOutputColumn can materialize the value directly. Exclude: + /// - arrays (leaf repetition level > 0, or any array level: max_array_def > 0), + /// - physically-nullable structs read as Nullable(Tuple(...)) (group_nullable), + /// - leaves nested inside a Tuple/Map/Array output column (the output column is not primitive). + /// A plain Nullable(T) is fine: it adds a definition level but no repetition, and its output + /// column is still primitive; the no-nulls check below and the output_nullable wrap handle it. + if (column_info.levels.back().rep != 0 || column_info.max_array_def != 0 || column_info.group_nullable) + return false; + if (column_info.idx_in_output_block >= sample_block_to_output_columns_idx.size()) + return false; + const auto & output_idx = sample_block_to_output_columns_idx.at(column_info.idx_in_output_block); + if (!output_idx.has_value() || !output_columns[output_idx.value()].is_primitive) + return false; + /// The value comes from decodeField, whose conversions are chosen from the requested type for + /// the purpose of comparing against statistics (allow_stats) and are not guaranteed to match + /// what decoding a page and then castColumn-ing it to the requested type would produce: e.g. + /// FIXED_LEN_BYTE_ARRAY read as String keeps its zero padding in the statistics but not in the + /// decoded column, and Date32 read as Enum8 is accepted for statistics but rejected by the cast. + /// So only take the shortcut when no cast is involved, i.e. the decoded type is the output type. + return !output_columns[output_idx.value()].needs_cast; +} + +Reader::ConstantKind Reader::chooseConstantKind(const PrimitiveColumnInfo & column_info, const DataTypePtr & final_output_type, Int64 num_values, std::optional null_count, bool single_value) const +{ + if (num_values <= 0) + return ConstantKind::None; + const bool physically_nullable = column_info.levels.back().def > 0; + /// A REQUIRED leaf (definition level 0) cannot hold nulls, and writers commonly omit null_count + /// for it. A physically nullable leaf must prove its null count. + const Int64 nulls = physically_nullable ? null_count.value_or(-1) : 0; + if (nulls < 0 || nulls > num_values) + return ConstantKind::None; + + if (nulls == 0) + return single_value ? ConstantKind::Const : ConstantKind::None; + + /// Nulls are present. A non-nullable output can only take them under null_as_default, where + /// they become the type default; otherwise the normal decode path reports the error. + const bool null_as_default = options.format.null_as_default && !column_info.output_nullable; + if (!column_info.output_nullable && !null_as_default) + return ConstantKind::None; + /// Sparse kinds are materialized directly in the final output type (see decodePrimitiveColumn + /// and formOutputColumn), so that type - not the decoder's - must support being sparse; e.g. + /// LowCardinality does not. + const bool can_be_sparse = final_output_type->canBeInsideSparseColumns(); + + if (nulls == num_values) + return can_be_sparse ? ConstantKind::AllDefault : ConstantKind::None; + + if (!single_value) + return ConstantKind::None; + const double null_ratio = static_cast(nulls) / static_cast(num_values); + if (can_be_sparse && null_ratio >= static_cast(options.format.parquet.constant_column_sparse_ratio)) + return ConstantKind::SparseNulls; + return ConstantKind::DenseNulls; +} + +void Reader::detectConstantColumn(ColumnChunk & column, const PrimitiveColumnInfo & column_info) const +{ + if (!isConstantColumnCandidate(column_info)) + return; + + const auto & meta_data = column.meta->meta_data; + if (!meta_data.__isset.statistics) + return; + const auto & stats = meta_data.statistics; + std::optional null_count; + if (stats.__isset.null_count) + null_count = stats.null_count; + + /// Is min == max, and can we trust that to mean "one value"? + /// + /// A writer may store truncated min/max for variable- or opaque-length physical types (BYTE_ARRAY, + /// FIXED_LEN_BYTE_ARRAY), which could make two different values compare equal. Allowlist only the + /// fixed-width integer physical types, whose min/max are never truncated, as unconditionally + /// trustworthy; anything else (BYTE_ARRAY, FIXED_LEN_BYTE_ARRAY, and any physical type added in the + /// future) must present the writer's is_*_value_exact flags before min == max is trusted. Fails + /// closed: an unrecognized type is treated as possibly-truncated rather than blindly trusted. + /// + /// FLOAT and DOUBLE are deliberately excluded (below) even though they are fixed-width. parquet.thrift + /// says NaN values are not written to min/max ("When looking for NaN values, min and max should + /// be ignored"), and both arrow and our own writer drop NaN when computing them, so a chunk like + /// [1.0, NaN, 1.0] has min == max == 1.0 with null_count == 0 and is not constant. Also + /// "if the min is +0, the row group may contain -0 values as well", and -0.0 is a distinct + /// GROUP BY key in ClickHouse. No statistic in the thrift version we ship proves the absence of + /// NaN; revisit once `Statistics::nan_count` (parquet-format 2.11) is available: require + /// nan_count == 0 and a nonzero decoded value. + const bool never_truncated = + meta_data.type == parq::Type::BOOLEAN + || meta_data.type == parq::Type::INT32 + || meta_data.type == parq::Type::INT64 + || meta_data.type == parq::Type::INT96; + /// is_*_value_exact is an optional thrift bool; guard on __isset so an absent flag fails closed + /// (treated as not-exact) rather than reading a possibly-uninitialized value and trusting a + /// truncated min/max. + const bool min_max_marked_exact = + stats.__isset.is_min_value_exact && stats.is_min_value_exact + && stats.__isset.is_max_value_exact && stats.is_max_value_exact; + /// Floats are rejected outright: the exactness flags say nothing about NaN, so they must not + /// reopen the door that the allowlist closes. + const bool is_float = meta_data.type == parq::Type::FLOAT || meta_data.type == parq::Type::DOUBLE; + const bool single_value = !is_float + && stats.__isset.min_value && stats.__isset.max_value + && stats.min_value == stats.max_value + && (never_truncated || min_max_marked_exact); + + const auto & output_idx = sample_block_to_output_columns_idx.at(column_info.idx_in_output_block); + const OutputColumnInfo & output_info = output_columns.at(output_idx.value()); + ConstantKind kind = chooseConstantKind(column_info, output_info.output_type, meta_data.num_values, null_count, single_value); + if (kind == ConstantKind::None) + return; + + if (kind != ConstantKind::AllDefault) + { + /// decodeField yields the value in the requested (output) type's domain, which + /// isConstantColumnCandidate guarantees to be the decoded type as well (no cast involved), so + /// the value is inserted straight into a column of output_info.output_type. decodeField leaves + /// `value` Null when the physical type is unsupported for stats, in which case the optimization + /// does not fire. + Field value; + column_info.decoder.decodeField(stats.min_value, /*is_max=*/ false, value); + if (value.isNull()) + return; + column.constant_value = std::move(value); + } + + column.constant_kind = kind; + if (constantKindSkipsDataPages(kind)) + ProfileEvents::increment(ProfileEvents::ParquetConstantColumnChunks); + else + ProfileEvents::increment(ProfileEvents::ParquetConstantColumnChunksWithNulls); +} + void Reader::decodePrimitiveColumn(ColumnChunk & column, const PrimitiveColumnInfo & column_info, ColumnSubchunk & subchunk, const RowGroup & row_group, RowSubgroup & row_subgroup, MemoryUsageDiff & diff) { + if (constantKindSkipsDataPages(column.constant_kind)) + { + /// This chunk is provably one value in every row, or all null (see detectConstantColumn), + /// and its data pages were never fetched. Skip all decoding and hand the already-decoded + /// value to formOutputColumn, which materializes it directly as ColumnConst / ColumnSparse. + /// We still run the per-output-column bookkeeping below so the output column is formed once + /// the last of its primitive columns is done. + subchunk.constant_kind = column.constant_kind; + subchunk.constant_value = column.constant_value; + + 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); + chassert(prev_count > 0); + if (prev_count == 1) + { + const auto & output_idx = sample_block_to_output_columns_idx.at(column_info.idx_in_output_block); + state.column = formOutputColumn(row_subgroup, output_idx.value(), row_subgroup.filter.rows_pass); + } + return; + } + /// Allocate columns for values, null map, and array offsets. size_t output_num_values_estimate = 0; @@ -1388,9 +1614,11 @@ void Reader::decodePrimitiveColumn(ColumnChunk & column, const PrimitiveColumnIn subchunk.null_map->reserve(output_num_values_estimate); } + const bool skip_values = constantKindSkipsValues(column.constant_kind); subchunk.column = column_info.decoded_type->createColumn(); - subchunk.column->reserve(output_num_values_estimate); - if (auto * string_column = typeid_cast(subchunk.column.get())) + if (!skip_values) + subchunk.column->reserve(output_num_values_estimate); + if (auto * string_column = typeid_cast(subchunk.column.get()); string_column && !skip_values) { double avg_len = estimateAverageStringLengthPerRow(column, row_group); size_t bytes_to_reserve = size_t(1.2 * avg_len * static_cast(row_subgroup.filter.rows_pass)); @@ -1493,6 +1721,53 @@ void Reader::decodePrimitiveColumn(ColumnChunk & column, const PrimitiveColumnIn throw Exception(ErrorCodes::INCORRECT_DATA, "Invalid repetition/definition levels for arrays in column {}", column_info.name); } + if (skip_values) + { + /// Value decoding was skipped for this chunk (see readRowsInPage): the single non-null value + /// is known from the statistics and only the definition levels were read, into `null_map`. + /// `constant_value` is in the FINAL output type's domain (see detectConstantColumn), so the + /// column is built directly as output_info.output_type - Nullable wrapper, LowCardinality and + /// all - and formOutputColumn skips the decoded_type -> output_type cast for it. The null map + /// is kept only under null_as_default, where formOutputColumn feeds it to block_missing_values + /// (a Nullable output holds real NULLs, which are not "missing"). + chassert(subchunk.null_map); + chassert(subchunk.column->empty()); + const auto & output_idx = sample_block_to_output_columns_idx.at(column_info.idx_in_output_block); + const OutputColumnInfo & output_info = output_columns.at(output_idx.value()); + const auto & null_map = assert_cast(*subchunk.null_map).getData(); + const size_t num_rows = null_map.size(); + const size_t non_null_count = num_rows - countBytesInFilter(null_map.data(), 0, num_rows); + + if (column.constant_kind == ConstantKind::SparseNulls) + { + /// ColumnSparse: `values` holds the type default (NULL for a Nullable output; the default + /// under null_as_default) at index 0 followed by one copy of the constant per non-null row, + /// `offsets` lists the non-null rows. O(non-null rows) memory. + MutableColumnPtr values = output_info.output_type->createColumn(); + values->insertDefault(); + values->insertMany(column.constant_value, non_null_count); + auto offsets = ColumnUInt64::create(); + auto & offsets_data = offsets->getData(); + offsets_data.reserve(non_null_count); + for (size_t i = 0; i < num_rows; ++i) + if (!null_map[i]) + offsets_data.push_back(i); + MutableColumnPtr offsets_ptr = std::move(offsets); + subchunk.column = ColumnSparse::create(std::move(values), std::move(offsets_ptr), num_rows); + } + else + { + /// Dense: one copy of the constant per non-null row, then expand() inserts the type default + /// (NULL for a Nullable output) at the null positions. + subchunk.column = output_info.output_type->createColumn(); + subchunk.column->insertMany(column.constant_value, non_null_count); + subchunk.column->expand(null_map, /*inverted*/ true); + } + if (column_info.output_nullable) + subchunk.null_map.reset(); + subchunk.constant_kind = column.constant_kind; + } + if (subchunk.null_map && !column_info.output_nullable && !column_info.group_nullable && !options.format.null_as_default) { const auto & null_map = assert_cast(*subchunk.null_map).getData(); @@ -1503,7 +1778,7 @@ void Reader::decodePrimitiveColumn(ColumnChunk & column, const PrimitiveColumnIn subchunk.null_map = nullptr; } - if (subchunk.null_map) + if (subchunk.null_map && !skip_values) { const auto & null_map = assert_cast(*subchunk.null_map).getData(); /// Fill defaults at null rows so the column reaches full size. For a group_nullable leaf, @@ -1526,7 +1801,7 @@ void Reader::decodePrimitiveColumn(ColumnChunk & column, const PrimitiveColumnIn if (subchunk.arrays_offsets.empty() && subchunk.column->size() != row_subgroup.filter.rows_pass) throw Exception(ErrorCodes::LOGICAL_ERROR, "Unexpected number of rows in column subchunk {} {}", subchunk.column->size(), row_subgroup.filter.rows_pass); - if (column_info.output_nullable) + if (column_info.output_nullable && !skip_values) { if (!subchunk.null_map) subchunk.null_map = ColumnUInt8::create(subchunk.column->size(), false); @@ -1534,7 +1809,7 @@ void Reader::decodePrimitiveColumn(ColumnChunk & column, const PrimitiveColumnIn subchunk.null_map.reset(); } - chassert(subchunk.column->getDataType() == column_info.output_type->getColumnType()); + chassert(skip_values || 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 @@ -2122,7 +2397,8 @@ void Reader::readRowsInPage(size_t end_row_idx, ColumnSubchunk & subchunk, Colum /// See if we can decompress the whole page directly into IColumn's memory. /// Skip when filter is set: direct read bypasses decode and would write all values without applying the filter. const bool has_filter = row_subgroup && !row_subgroup->filter.filter.empty(); - if (!has_filter && !page.is_dictionary_encoded && prev_value_idx == 0 && page.value_idx == page.num_values && + const bool skip_values = constantKindSkipsValues(column.constant_kind); + if (!has_filter && !skip_values && !page.is_dictionary_encoded && prev_value_idx == 0 && page.value_idx == page.num_values && page.codec != parq::CompressionCodec::UNCOMPRESSED) { std::span span; @@ -2135,7 +2411,10 @@ void Reader::readRowsInPage(size_t end_row_idx, ColumnSubchunk & subchunk, Colum } } - if (encoded_values_to_read > 0) + /// For SparseNulls/DenseNulls the single non-null value is known from the statistics, so only the + /// definition levels (processed above) are needed; skip decompressing (DATA_PAGE_V2) and decoding + /// the values. decodePrimitiveColumn fills the column from `constant_value` and the null map. + if (encoded_values_to_read > 0 && !skip_values) { decompressPageIfCompressed(page); if (!page.decoder) @@ -2190,6 +2469,7 @@ MutableColumnPtr Reader::formOutputColumn(RowSubgroup & row_subgroup, size_t out const OutputColumnInfo & output_info = output_columns.at(output_column_idx); MutableColumnPtr res; + bool already_output_type = false; if (output_info.is_missing_column) { @@ -2234,7 +2514,46 @@ MutableColumnPtr Reader::formOutputColumn(RowSubgroup & row_subgroup, size_t out chassert(output_info.primitive_start + 1 == output_info.primitive_end); size_t primitive_idx = output_info.primitive_start; ColumnSubchunk & subchunk = row_subgroup.columns.at(primitive_idx); + + if (constantKindSkipsDataPages(subchunk.constant_kind)) + { + ColumnPtr result; + if (subchunk.constant_kind == ConstantKind::AllDefault) + { + /// All-null chunk: every row is the output type's default (Null for a Nullable output, + /// the type default under null_as_default). ColumnSparse with no non-default rows is + /// O(1) and, unlike ColumnConst, stays writable for AddingDefaultsTransform (see + /// ConstantKind). chooseConstantKind guarantees the type can be inside a sparse column. + auto sparse = ColumnSparse::create(output_info.output_type->createColumn()); + sparse->insertManyDefaults(num_rows); + result = std::move(sparse); + + /// Under null_as_default the rows are "missing" for AddingDefaultsTransform, as the + /// normal decode path records them from the null map. A Nullable output holds real + /// NULLs, which must not be replaced by column defaults. + const bool null_as_default = options.format.null_as_default && !output_info.output_type->isNullable(); + if (null_as_default && output_info.idx_in_output_block.has_value() + && *output_info.idx_in_output_block < row_subgroup.block_missing_values.getNumColumns()) + row_subgroup.block_missing_values.setBits(*output_info.idx_in_output_block, num_rows); + } + else + { + /// Single non-null value in every row: ColumnConst, so the const-ness propagates + /// downstream (a PREWHERE/WHERE predicate is computed from the one value, GROUP BY gets + /// a const key). constant_value is already in the final output type's domain (see + /// detectConstantColumn), so no castColumn is applied. + MutableColumnPtr single_value = output_info.output_type->createColumn(); + single_value->insert(subchunk.constant_value); + result = ColumnConst::create(std::move(single_value), num_rows); + } + + return IColumn::mutate(std::move(result)); + } + res = std::move(subchunk.column); + /// SparseNulls/DenseNulls columns were built directly in output_type (see decodePrimitiveColumn). + if (constantKindSkipsValues(subchunk.constant_kind)) + already_output_type = true; if (output_info.idx_in_output_block.has_value() && *output_info.idx_in_output_block < row_subgroup.block_missing_values.getNumColumns() && @@ -2293,6 +2612,9 @@ MutableColumnPtr Reader::formOutputColumn(RowSubgroup & row_subgroup, size_t out res = ColumnNullable::create(std::move(res), std::move(nullable_group_null_map)); } + if (already_output_type) + return res; + chassert(res->getDataType() == output_info.input_type->getColumnType()); if (output_info.needs_cast) diff --git a/src/Processors/Formats/Impl/Parquet/Reader.h b/src/Processors/Formats/Impl/Parquet/Reader.h index 105cb07a3061..5a841f6a9a25 100644 --- a/src/Processors/Formats/Impl/Parquet/Reader.h +++ b/src/Processors/Formats/Impl/Parquet/Reader.h @@ -302,6 +302,39 @@ struct Reader MutableColumnPtr indices_column; // if is_dictionary_encoded; ColumnUInt32 }; + /// How a column chunk whose statistics prove (almost) all of its content is materialized, chosen + /// once per chunk by detectConstantColumn from the chunk statistics alone. Each kind maps to an + /// existing column representation that every downstream consumer already handles: + /// Const - ColumnConst: one non-null value in every row (null_count == 0, min == max). + /// Const-ness propagates to WHERE/GROUP BY. Data pages are not read. + /// AllDefault - ColumnSparse with no non-default rows: every row is null (null_count == + /// num_values), which is the type default (Null for Nullable, or the output type's + /// default under null_as_default). Data pages are not read. Sparse rather than Const + /// because a sparse column stays writable (ColumnConst silently drops inserts), which + /// matters when block_missing_values makes AddingDefaultsTransform mix defaults in. + /// SparseNulls - ColumnSparse: nulls dominate (null_count / num_values >= constant_column_sparse_ratio) + /// and the non-null rows all hold one value. Data pages are read only for their + /// definition levels (the null map); values are neither decompressed (DATA_PAGE_V2) + /// nor decoded. Memory is O(non-null rows). + /// DenseNulls - ColumnNullable: nulls are present but few; same page handling as SparseNulls, + /// materialized as a dense column filled with the value plus the decoded null map. + /// Invariant: Const <=> a single non-default or default value with no nulls; sparse kinds <=> the + /// default (null) dominates; a non-null constant equal to the type default is still Const, since + /// const propagation beats sparse there and there is no null map to mix. + enum class ConstantKind : UInt8 + { + None, + Const, + AllDefault, + SparseNulls, + DenseNulls, + }; + + /// Kinds for which the chunk's data pages are neither prefetched nor decoded. + static bool constantKindSkipsDataPages(ConstantKind kind) { return kind == ConstantKind::Const || kind == ConstantKind::AllDefault; } + /// Kinds for which the data pages are read for definition levels only (value decoding skipped). + static bool constantKindSkipsValues(ConstantKind kind) { return kind == ConstantKind::SparseNulls || kind == ConstantKind::DenseNulls; } + struct ColumnChunk { const parq::ColumnChunk * meta{}; @@ -311,6 +344,13 @@ struct Reader bool use_column_index = false; bool need_null_map = false; + /// See ConstantKind and detectConstantColumn. For Const/SparseNulls/DenseNulls, + /// `constant_value` is the single non-null value, already decoded (not the raw parquet-encoded + /// bytes) and in the FINAL output type's domain (OutputColumnInfo::output_type, not + /// decoded_type - see detectConstantColumn). Unused for None/AllDefault. + ConstantKind constant_kind = ConstantKind::None; + Field constant_value; + /// Prefetches. /// TODO [parquet]: Check that all handles and tokens are reset after correct stages. PrefetchHandle bloom_filter_header_prefetch; @@ -361,6 +401,14 @@ struct Reader /// Primitive column. MutableColumnPtr column; + /// Mirror of ColumnChunk::constant_kind, set by decodePrimitiveColumn. For Const/AllDefault + /// `column` is left empty and formOutputColumn materializes the chunk directly from + /// `constant_value`. For SparseNulls/DenseNulls `column` is materialized by + /// decodePrimitiveColumn directly in the final output type (no cast needed). `constant_value` + /// is in the final output type's domain, as in ColumnChunk. + ConstantKind constant_kind = ConstantKind::None; + Field constant_value; + MutableColumnPtr null_map; /// For a leaf of a physically-nullable struct read as Nullable(Tuple(...)) (see @@ -416,6 +464,11 @@ struct Reader std::atomic stage {ReadStage::NotStarted}; std::atomic stage_tasks_remaining {0}; + + /// Set when the previous subgroup's ColumnDataPrefetch task also issued this subgroup's + /// first-step data-page reads (read-ahead), so this subgroup skips its own ColumnDataPrefetch + /// stage. Written before this subgroup is admitted (its `stage` CAS orders the read). + bool reads_issued_ahead = false; }; struct RowGroup @@ -534,6 +587,22 @@ struct Reader void decodePrimitiveColumn(ColumnChunk & column, const PrimitiveColumnInfo & column_info, ColumnSubchunk & subchunk, const RowGroup & row_group, RowSubgroup & row_subgroup, MemoryUsageDiff & diff); + /// Shape/eligibility gate for detectConstantColumn: a flat, top-level primitive column whose + /// chunk statistics we can read (excludes arrays, physically-nullable structs, and leaves nested + /// in a Tuple/Map/Array output). Only such a column maps one parquet value 1:1 to one output row. + bool isConstantColumnCandidate(const PrimitiveColumnInfo & column_info) const; + + /// If the column chunk statistics prove its content (one value, all null, or one value plus + /// nulls), sets column.constant_kind and column.constant_value. Uses column chunk min/max and + /// null_count statistics (Tier 1). Only applies to flat, top-level primitive columns; see the + /// implementation and ConstantKind for the exact conditions and representations. + void detectConstantColumn(ColumnChunk & column, const PrimitiveColumnInfo & column_info) const; + + /// Pure decision from statistics + output type: which ConstantKind to use for a chunk with + /// `num_values` values of which `null_count` are null and whose non-null values are all equal + /// (`single_value`). Returns None when nothing applies. + ConstantKind chooseConstantKind(const PrimitiveColumnInfo & column_info, const DataTypePtr & final_output_type, Int64 num_values, std::optional null_count, bool single_value) const; + /// Returns mutable column because some of the recursive calls require it, /// e.g. ColumnArray::create does assumeMutable() on the nested columns. /// Moves the column out of ColumnSubchunk-s, leaving nullptrs in ColumnSubchunk::column. diff --git a/src/Processors/Formats/Impl/ParquetV3BlockInputFormat.cpp b/src/Processors/Formats/Impl/ParquetV3BlockInputFormat.cpp index 6a238834caec..2ecac08466ec 100644 --- a/src/Processors/Formats/Impl/ParquetV3BlockInputFormat.cpp +++ b/src/Processors/Formats/Impl/ParquetV3BlockInputFormat.cpp @@ -57,7 +57,11 @@ ParquetV3BlockInputFormat::ParquetV3BlockInputFormat( , object_with_metadata(object_with_metadata_) { read_options.min_bytes_for_seek = min_bytes_for_seek; - read_options.bytes_per_read_task = min_bytes_for_seek * 4; + /// min_bytes_for_seek says when reading across a gap beats another request; this says how big one + /// request should get, which is bounded by how many we want in flight. + read_options.bytes_per_read_task = format_settings.parquet.bytes_per_read_task != 0 + ? format_settings.parquet.bytes_per_read_task + : min_bytes_for_seek * 4; if (!format_filter_info) format_filter_info = std::make_shared(); @@ -70,9 +74,17 @@ void ParquetV3BlockInputFormat::initializeIfNeeded() format_filter_info->initKeyConditionOnce(getPort().getHeader()); parser_shared_resources->initOnce([&] { - if (format_settings.parquet.enable_row_group_prefetch && parser_shared_resources->max_io_threads > 0) + /// `max_download_threads` defaults to 4, picked for the URL engine; on object storage + /// that rarely keeps the decoding threads fed. + size_t io_threads = format_settings.parquet.max_io_threads; + if (io_threads == 0) + io_threads = std::max( + parser_shared_resources->max_io_threads, + std::min(parser_shared_resources->max_parsing_threads, 16)); + if (format_settings.parquet.enable_row_group_prefetch && io_threads > 0 + && parser_shared_resources->max_io_threads > 0) parser_shared_resources->io_runner.initThreadPool( - getFormatParsingThreadPool().get(), parser_shared_resources->max_io_threads, ThreadName::PARQUET_PREFETCH, CurrentThread::getGroup()); + getFormatParsingThreadPool().get(), io_threads, ThreadName::PARQUET_PREFETCH, CurrentThread::getGroup()); /// Unfortunately max_parsing_threads setting doesn't have a value for /// "do parsing in the same thread as the rest of query processing diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp index 24930b88462a..3f6ffff65b9d 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp @@ -1471,7 +1471,11 @@ void IcebergMetadata::addDeleteTransformers( Columns delete_columns = delete_chunk.detachColumns(); for (size_t i = 0; i < equality_indexes_delete_file.size(); i++) { - mutable_columns_for_set[i]->insertRangeFrom(*delete_columns[equality_indexes_delete_file[i]], 0, rows); + /// The format may return ColumnConst / ColumnSparse for column chunks whose + /// statistics prove a single value or all nulls; insertRangeFrom needs the + /// concrete column type. + auto full_column = delete_columns[equality_indexes_delete_file[i]]->convertToFullColumnIfConst()->convertToFullColumnIfSparse(); + mutable_columns_for_set[i]->insertRangeFrom(*full_column, 0, rows); } } block_for_set.setColumns(std::move(mutable_columns_for_set)); diff --git a/tests/queries/0_stateless/03408_parquet_checksums.sh b/tests/queries/0_stateless/03408_parquet_checksums.sh index d527e606252f..bcddc4ae49c4 100755 --- a/tests/queries/0_stateless/03408_parquet_checksums.sh +++ b/tests/queries/0_stateless/03408_parquet_checksums.sh @@ -24,9 +24,11 @@ ${CLICKHOUSE_LOCAL} -q " select * from file('$F');" corrupt_file +# The single-row column chunk is provably constant from its statistics, so the reader would not read +# (and therefore could not verify) its data page at all; disable that shortcut to exercise the checksum. ${CLICKHOUSE_LOCAL} -q " - select * from file('$F') settings input_format_parquet_verify_checksums=1 + select * from file('$F') settings input_format_parquet_verify_checksums=1, input_format_parquet_use_constant_column_optimization=0 " 2>&1 | grep -o 'CRC checksum verification failed' || echo 'got no checksum error, unexpected' ${CLICKHOUSE_LOCAL} -q " @@ -42,5 +44,5 @@ ${CLICKHOUSE_LOCAL} -q " corrupt_file ${CLICKHOUSE_LOCAL} -q " - select * from file('$F') settings input_format_parquet_verify_checksums=1 + select * from file('$F') settings input_format_parquet_verify_checksums=1, input_format_parquet_use_constant_column_optimization=0 " 2>&1 | grep -o 'CRC checksum verification failed' || echo 'no checksum error, as expected' diff --git a/tests/queries/0_stateless/04811_parquet_constant_column_optimization.reference b/tests/queries/0_stateless/04811_parquet_constant_column_optimization.reference new file mode 100644 index 000000000000..1019a28c9acb --- /dev/null +++ b/tests/queries/0_stateless/04811_parquet_constant_column_optimization.reference @@ -0,0 +1,14 @@ +-- values, optimization on +42 hello 2020-01-02 03:04:05 7 1000 +-- values, optimization off (must be identical) +42 hello 2020-01-02 03:04:05 7 1000 +-- the varying column is read correctly (not treated as constant) +499500 0 999 1000 +-- filters on a constant column still work +1000 +0 +-- optimization fired only when enabled +1 +1 +-- float chunks are never treated as constant: NaN and -0.0 are invisible to min/max statistics +1 1 1 100 diff --git a/tests/queries/0_stateless/04811_parquet_constant_column_optimization.sh b/tests/queries/0_stateless/04811_parquet_constant_column_optimization.sh new file mode 100755 index 000000000000..87c53ee3a3cb --- /dev/null +++ b/tests/queries/0_stateless/04811_parquet_constant_column_optimization.sh @@ -0,0 +1,86 @@ +#!/usr/bin/env bash +# Tags: no-fasttest + +CUR_DIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd) +# shellcheck source=../shell_config.sh +. "$CUR_DIR"/../shell_config.sh + +USER_FILES_PATH=$(${CLICKHOUSE_CLIENT} -q "SELECT value FROM system.server_settings WHERE name = 'user_files_path'" | sed 's|/$||') +WORKING_DIR="${USER_FILES_PATH}/${CLICKHOUSE_TEST_UNIQUE_NAME}" +mkdir -p "${WORKING_DIR}" +DATA_FILE="${WORKING_DIR}/const.parquet" + +# 1000 rows, 100 rows per row group => 10 row groups. `k` varies; the other four columns each hold a +# single value in every row, so their per-chunk min/max statistics have min == max and no nulls. +# `c_dt` is written as TIMESTAMP_MILLIS and read back with a DateTime hint; that needs a cast from the +# decoded DateTime64(3), so the optimization deliberately does not apply to it and it goes through the +# normal decode path (the result must still be identical). +${CLICKHOUSE_CLIENT} -q " + INSERT INTO FUNCTION file('${DATA_FILE}', Parquet) + SELECT + number AS k, + 42::Int64 AS c_int, + 'hello' AS c_str, + toDateTime('2020-01-02 03:04:05') AS c_dt, + 7::Nullable(Int64) AS c_nullable + FROM numbers(1000) + SETTINGS engine_file_truncate_on_insert = 1, output_format_parquet_row_group_size = 100 +" + +STRUCTURE="k UInt64, c_int Int64, c_str String, c_dt DateTime, c_nullable Nullable(Int64)" + +qid_on="${CLICKHOUSE_TEST_UNIQUE_NAME}_on" +qid_off="${CLICKHOUSE_TEST_UNIQUE_NAME}_off" + +echo "-- values, optimization on" +${CLICKHOUSE_CLIENT} --query_id="${qid_on}" -q " + SELECT c_int, c_str, c_dt, c_nullable, count() + FROM file('${DATA_FILE}', Parquet, '${STRUCTURE}') + GROUP BY 1, 2, 3, 4 +" + +echo "-- values, optimization off (must be identical)" +${CLICKHOUSE_CLIENT} --query_id="${qid_off}" -q " + SELECT c_int, c_str, c_dt, c_nullable, count() + FROM file('${DATA_FILE}', Parquet, '${STRUCTURE}') + GROUP BY 1, 2, 3, 4 + SETTINGS input_format_parquet_use_constant_column_optimization = 0 +" + +echo "-- the varying column is read correctly (not treated as constant)" +${CLICKHOUSE_CLIENT} -q "SELECT sum(k), min(k), max(k), count() FROM file('${DATA_FILE}', Parquet, '${STRUCTURE}')" + +echo "-- filters on a constant column still work" +${CLICKHOUSE_CLIENT} -q "SELECT count() FROM file('${DATA_FILE}', Parquet, '${STRUCTURE}') WHERE c_int = 42" +${CLICKHOUSE_CLIENT} -q "SELECT count() FROM file('${DATA_FILE}', Parquet, '${STRUCTURE}') WHERE c_int = 43" + +echo "-- optimization fired only when enabled" +${CLICKHOUSE_CLIENT} -q " + SYSTEM FLUSH LOGS query_log; + SELECT ProfileEvents['ParquetConstantColumnChunks'] > 0 + FROM system.query_log + WHERE event_date >= yesterday() AND event_time >= now() - 600 + AND query_id = '${qid_on}' AND type = 'QueryFinish' AND current_database = currentDatabase(); + SELECT ProfileEvents['ParquetConstantColumnChunks'] = 0 + FROM system.query_log + WHERE event_date >= yesterday() AND event_time >= now() - 600 + AND query_id = '${qid_off}' AND type = 'QueryFinish' AND current_database = currentDatabase(); +" + +echo "-- float chunks are never treated as constant: NaN and -0.0 are invisible to min/max statistics" +NAN_FILE="${WORKING_DIR}/nan.parquet" +${CLICKHOUSE_CLIENT} -q " + INSERT INTO FUNCTION file('${NAN_FILE}', Parquet) + SELECT + if(number = 1, nan, 1.0)::Float64 AS f, + if(number = 1, -0.0, 0.0)::Float64 AS z, + if(number = 1, nan, 1.0)::Float32 AS f32 + FROM numbers(100) + SETTINGS engine_file_truncate_on_insert = 1, output_format_parquet_row_group_size = 100 +" +${CLICKHOUSE_CLIENT} -q " + SELECT countIf(isNaN(f)), countIf(toString(z) = '-0'), countIf(isNaN(f32)), count() + FROM file('${NAN_FILE}', Parquet, 'f Float64, z Float64, f32 Float32') +" + +rm -rf "${WORKING_DIR}" diff --git a/tests/queries/0_stateless/04812_parquet_constant_column_kinds.reference b/tests/queries/0_stateless/04812_parquet_constant_column_kinds.reference new file mode 100644 index 000000000000..ae73a5468a3c --- /dev/null +++ b/tests/queries/0_stateless/04812_parquet_constant_column_kinds.reference @@ -0,0 +1,27 @@ +-- optimization on (sparse + dense kinds) +13606684125528126333 0 10 10 10 990 10 499500 +-- optimization on, sparse disabled via ratio = 1 (dense kind only) +13606684125528126333 0 10 10 10 990 10 499500 +-- optimization off (must be identical) +13606684125528126333 0 10 10 10 990 10 499500 +-- filters and aggregation over sparse / all-null columns +10 +990 +7 x 10 +\N \N 990 +0 \N 7 \N x +1 \N \N 7 \N +100 \N 7 \N x +999 \N \N 7 \N +-- null_as_default with non-nullable output types +0 70 990 6930 990 10 +-- DEFAULT expressions fill the null rows (AddingDefaultsTransform over sparse columns) +5000 8980 7040 990 10 +-- INSERT ... FORMAT Parquet through the client: sparse columns travel over the Native protocol +0 70 6930 990 10 1000 +-- LowCardinality output cannot be sparse: falls back to dense / normal decode with the same result +0 10 10 10 10 +-- profile events: all-null chunks skip pages; single-value-plus-nulls chunks skip values; nothing when disabled +10 30 +10 30 +0 0 diff --git a/tests/queries/0_stateless/04812_parquet_constant_column_kinds.sh b/tests/queries/0_stateless/04812_parquet_constant_column_kinds.sh new file mode 100755 index 000000000000..98038c3e74a4 --- /dev/null +++ b/tests/queries/0_stateless/04812_parquet_constant_column_kinds.sh @@ -0,0 +1,99 @@ +#!/usr/bin/env bash +# Tags: no-fasttest + +CUR_DIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd) +# shellcheck source=../shell_config.sh +. "$CUR_DIR"/../shell_config.sh + +USER_FILES_PATH=$(${CLICKHOUSE_CLIENT} -q "SELECT value FROM system.server_settings WHERE name = 'user_files_path'" | sed 's|/$||') +REL_DIR="${CLICKHOUSE_TEST_UNIQUE_NAME}" +WORKING_DIR="${USER_FILES_PATH}/${REL_DIR}" +mkdir -p "${WORKING_DIR}" +DATA_FILE="${WORKING_DIR}/kinds.parquet" + +# 1000 rows, 100 rows per row group => 10 row groups. Per chunk: +# n_all - every row NULL -> all-default (sparse), pages not read +# n_sparse - 7 in 1 row of 100, NULL otherwise -> single value + 99% nulls -> sparse, values not decoded +# n_dense - NULL in 1 row of 100, 7 otherwise -> single value + 1% nulls -> dense Nullable, values not decoded +# s_sparse - 'x' in 1 row of 100, NULL otherwise -> BYTE_ARRAY with exact min/max flags -> sparse +# k - varies -> normal decode +${CLICKHOUSE_CLIENT} -q " + INSERT INTO FUNCTION file('${DATA_FILE}', Parquet) + SELECT + number AS k, + NULL::Nullable(Int64) AS n_all, + if(number % 100 = 0, 7, NULL)::Nullable(Int64) AS n_sparse, + if(number % 100 = 0, NULL, 7)::Nullable(Int64) AS n_dense, + if(number % 100 = 0, 'x', NULL)::Nullable(String) AS s_sparse + FROM numbers(1000) + SETTINGS engine_file_truncate_on_insert = 1, output_format_parquet_row_group_size = 100 +" + +STRUCTURE="k UInt64, n_all Nullable(Int64), n_sparse Nullable(Int64), n_dense Nullable(Int64), s_sparse Nullable(String)" +FULL_HASH="SELECT sum(cityHash64(k, ifNull(n_all, -1), ifNull(n_sparse, -1), ifNull(n_dense, -1), ifNull(s_sparse, ''))), count(n_all), countIf(n_sparse = 7), count(n_sparse), countIf(n_dense IS NULL), count(n_dense), countIf(s_sparse = 'x'), sum(k) FROM file('${DATA_FILE}', Parquet, '${STRUCTURE}')" + +qid_on="${CLICKHOUSE_TEST_UNIQUE_NAME}_on" +qid_dense="${CLICKHOUSE_TEST_UNIQUE_NAME}_dense" +qid_off="${CLICKHOUSE_TEST_UNIQUE_NAME}_off" + +echo "-- optimization on (sparse + dense kinds)" +${CLICKHOUSE_CLIENT} --query_id="${qid_on}" -q "${FULL_HASH}" +echo "-- optimization on, sparse disabled via ratio = 1 (dense kind only)" +${CLICKHOUSE_CLIENT} --query_id="${qid_dense}" -q "${FULL_HASH} SETTINGS input_format_parquet_constant_column_sparse_ratio = 1" +echo "-- optimization off (must be identical)" +${CLICKHOUSE_CLIENT} --query_id="${qid_off}" -q "${FULL_HASH} SETTINGS input_format_parquet_use_constant_column_optimization = 0" + +echo "-- filters and aggregation over sparse / all-null columns" +${CLICKHOUSE_CLIENT} -q "SELECT count() FROM file('${DATA_FILE}', Parquet, '${STRUCTURE}') WHERE n_sparse = 7" +${CLICKHOUSE_CLIENT} -q "SELECT count() FROM file('${DATA_FILE}', Parquet, '${STRUCTURE}') WHERE n_sparse IS NULL AND n_all IS NULL" +${CLICKHOUSE_CLIENT} -q "SELECT n_sparse, s_sparse, count() FROM file('${DATA_FILE}', Parquet, '${STRUCTURE}') GROUP BY 1, 2 ORDER BY 1, 2" +${CLICKHOUSE_CLIENT} -q "SELECT k, n_all, n_sparse, n_dense, s_sparse FROM file('${DATA_FILE}', Parquet, '${STRUCTURE}') WHERE k IN (0, 1, 100, 999) ORDER BY k" + +echo "-- null_as_default with non-nullable output types" +${CLICKHOUSE_CLIENT} -q " + SELECT sum(n_all), sum(n_sparse), countIf(n_sparse = 0), sum(n_dense), countIf(s_sparse = ''), countIf(s_sparse = 'x') + FROM file('${DATA_FILE}', Parquet, 'k UInt64, n_all Int64, n_sparse Int64, n_dense Int64, s_sparse String') + SETTINGS input_format_null_as_default = 1 +" + +echo "-- DEFAULT expressions fill the null rows (AddingDefaultsTransform over sparse columns)" +${CLICKHOUSE_CLIENT} -q "DROP TABLE IF EXISTS t_kinds_file" +${CLICKHOUSE_CLIENT} -q " + CREATE TABLE t_kinds_file (k UInt64, n_all Int64 DEFAULT 5, n_sparse Int64 DEFAULT 9, n_dense Int64 DEFAULT 11, s_sparse String DEFAULT 'd') + ENGINE = File(Parquet, '${REL_DIR}/kinds.parquet') +" +${CLICKHOUSE_CLIENT} -q "SELECT sum(n_all), sum(n_sparse), sum(n_dense), countIf(s_sparse = 'd'), countIf(s_sparse = 'x') FROM t_kinds_file SETTINGS input_format_null_as_default = 1" +${CLICKHOUSE_CLIENT} -q "DROP TABLE t_kinds_file" + +echo "-- INSERT ... FORMAT Parquet through the client: sparse columns travel over the Native protocol" +${CLICKHOUSE_CLIENT} -q "DROP TABLE IF EXISTS t_kinds_ins" +${CLICKHOUSE_CLIENT} -q "CREATE TABLE t_kinds_ins (k UInt64, n_all Int64, n_sparse Int64, n_dense Int64, s_sparse String) ENGINE = Memory" +${CLICKHOUSE_CLIENT} --input_format_null_as_default=1 -q "INSERT INTO t_kinds_ins FORMAT Parquet" < "${DATA_FILE}" +${CLICKHOUSE_CLIENT} -q "SELECT sum(n_all), sum(n_sparse), sum(n_dense), countIf(s_sparse = ''), countIf(s_sparse = 'x'), count() FROM t_kinds_ins" +${CLICKHOUSE_CLIENT} -q "DROP TABLE t_kinds_ins" + +echo "-- LowCardinality output cannot be sparse: falls back to dense / normal decode with the same result" +${CLICKHOUSE_CLIENT} -q " + SELECT count(n_all), countIf(n_sparse = 7), count(n_sparse), countIf(s_sparse = 'x'), count(s_sparse) + FROM file('${DATA_FILE}', Parquet, 'k UInt64, n_all LowCardinality(Nullable(Int64)), n_sparse LowCardinality(Nullable(Int64)), n_dense LowCardinality(Nullable(Int64)), s_sparse LowCardinality(Nullable(String))') + SETTINGS allow_suspicious_low_cardinality_types = 1 +" + +echo "-- profile events: all-null chunks skip pages; single-value-plus-nulls chunks skip values; nothing when disabled" +${CLICKHOUSE_CLIENT} -q " + SYSTEM FLUSH LOGS query_log; + SELECT ProfileEvents['ParquetConstantColumnChunks'], ProfileEvents['ParquetConstantColumnChunksWithNulls'] + FROM system.query_log + WHERE event_date >= yesterday() AND event_time >= now() - 600 + AND query_id = '${qid_on}' AND type = 'QueryFinish' AND current_database = currentDatabase(); + SELECT ProfileEvents['ParquetConstantColumnChunks'], ProfileEvents['ParquetConstantColumnChunksWithNulls'] + FROM system.query_log + WHERE event_date >= yesterday() AND event_time >= now() - 600 + AND query_id = '${qid_dense}' AND type = 'QueryFinish' AND current_database = currentDatabase(); + SELECT ProfileEvents['ParquetConstantColumnChunks'], ProfileEvents['ParquetConstantColumnChunksWithNulls'] + FROM system.query_log + WHERE event_date >= yesterday() AND event_time >= now() - 600 + AND query_id = '${qid_off}' AND type = 'QueryFinish' AND current_database = currentDatabase(); +" + +rm -rf "${WORKING_DIR}" diff --git a/tests/queries/0_stateless/04813_parquet_read_ahead_subgroups.reference b/tests/queries/0_stateless/04813_parquet_read_ahead_subgroups.reference new file mode 100644 index 000000000000..efc23e6ff9f0 --- /dev/null +++ b/tests/queries/0_stateless/04813_parquet_read_ahead_subgroups.reference @@ -0,0 +1,20 @@ +-- read_ahead_subgroups = 0 +300000 44999850000 149850000 13518539653315467104 9900000 100000 +20000 2998470000 268166841057690648 990000 +20000 2998470000 268166841057690648 990000 +-- read_ahead_subgroups = 1 +300000 44999850000 149850000 13518539653315467104 9900000 100000 +20000 2998470000 268166841057690648 990000 +20000 2998470000 268166841057690648 990000 +-- read_ahead_subgroups = 3 +300000 44999850000 149850000 13518539653315467104 9900000 100000 +20000 2998470000 268166841057690648 990000 +20000 2998470000 268166841057690648 990000 +-- read_ahead_memory_fraction = 0 disables read-ahead even when subgroups > 0 +300000 44999850000 149850000 13518539653315467104 9900000 100000 +-- read-ahead fired only when enabled and budgeted (ParquetReadAheadSubgroups > 0) +filtered_1 1 +full_0 0 +full_1 1 +full_3 1 +nobudget 0 diff --git a/tests/queries/0_stateless/04813_parquet_read_ahead_subgroups.sh b/tests/queries/0_stateless/04813_parquet_read_ahead_subgroups.sh new file mode 100755 index 000000000000..2eb119eeac9d --- /dev/null +++ b/tests/queries/0_stateless/04813_parquet_read_ahead_subgroups.sh @@ -0,0 +1,57 @@ +#!/usr/bin/env bash +# Tags: no-fasttest + +CUR_DIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd) +# shellcheck source=../shell_config.sh +. "$CUR_DIR"/../shell_config.sh + +USER_FILES_PATH=$(${CLICKHOUSE_CLIENT} -q "SELECT value FROM system.server_settings WHERE name = 'user_files_path'" | sed 's|/$||') +WORKING_DIR="${USER_FILES_PATH}/${CLICKHOUSE_TEST_UNIQUE_NAME}" +mkdir -p "${WORKING_DIR}" +DATA_FILE="${WORKING_DIR}/ra.parquet" + +# 3 row groups of 100k rows, small data pages and a page index, so that each row group is read as many +# subgroups (input_format_parquet_max_block_size below) with several pages per subgroup and column. +${CLICKHOUSE_CLIENT} -q " + INSERT INTO FUNCTION file('${DATA_FILE}', Parquet) + SELECT + number AS k, + number * 7 % 1000 AS v, + toString(number % 5000) AS s, + if(number % 3 = 0, NULL, number % 100)::Nullable(UInt8) AS n + FROM numbers(300000) + SETTINGS engine_file_truncate_on_insert = 1, output_format_parquet_row_group_size = 100000, + output_format_parquet_data_page_size = 8192, output_format_parquet_write_page_index = 1 +" + +STRUCTURE="k UInt64, v UInt64, s String, n Nullable(UInt8)" +# Small subgroups and a low decode watermark, so the scheduler throttles non-first row groups and +# subgroups queue up behind each other (the situation read-ahead is for). +COMMON="input_format_parquet_max_block_size = 4096, input_format_parquet_prefer_block_bytes = 0, input_format_parquet_memory_high_watermark = 4194304, input_format_parquet_memory_low_watermark = 1048576" + +FULL="SELECT count(), sum(k), sum(v), sum(cityHash64(s)), sum(n), countIf(n IS NULL) FROM file('${DATA_FILE}', Parquet, '${STRUCTURE}')" +FILTERED="SELECT count(), sum(k), sum(cityHash64(s)), sum(n) FROM file('${DATA_FILE}', Parquet, '${STRUCTURE}') WHERE v < 100 AND n IS NOT NULL" + +for ahead in 0 1 3; do + echo "-- read_ahead_subgroups = ${ahead}" + ${CLICKHOUSE_CLIENT} --query_id="${CLICKHOUSE_TEST_UNIQUE_NAME}_full_${ahead}" -q "${FULL} SETTINGS ${COMMON}, input_format_parquet_read_ahead_subgroups = ${ahead}" + ${CLICKHOUSE_CLIENT} --query_id="${CLICKHOUSE_TEST_UNIQUE_NAME}_filtered_${ahead}" -q "${FILTERED} SETTINGS ${COMMON}, input_format_parquet_read_ahead_subgroups = ${ahead}" + # Single-threaded parsing exercises the same-thread scheduling path. + ${CLICKHOUSE_CLIENT} -q "${FILTERED} SETTINGS ${COMMON}, input_format_parquet_read_ahead_subgroups = ${ahead}, max_parsing_threads = 1, max_threads = 1" +done + +echo "-- read_ahead_memory_fraction = 0 disables read-ahead even when subgroups > 0" +${CLICKHOUSE_CLIENT} --query_id="${CLICKHOUSE_TEST_UNIQUE_NAME}_nobudget" -q "${FULL} SETTINGS ${COMMON}, input_format_parquet_read_ahead_subgroups = 1, input_format_parquet_read_ahead_memory_fraction = 0" + +echo "-- read-ahead fired only when enabled and budgeted (ParquetReadAheadSubgroups > 0)" +${CLICKHOUSE_CLIENT} -q " + SYSTEM FLUSH LOGS query_log; + SELECT replaceOne(query_id, '${CLICKHOUSE_TEST_UNIQUE_NAME}_', ''), ProfileEvents['ParquetReadAheadSubgroups'] > 0 + FROM system.query_log + WHERE event_date >= yesterday() AND event_time >= now() - 600 AND type = 'QueryFinish' + AND current_database = currentDatabase() + AND query_id IN ('${CLICKHOUSE_TEST_UNIQUE_NAME}_full_0', '${CLICKHOUSE_TEST_UNIQUE_NAME}_full_1', '${CLICKHOUSE_TEST_UNIQUE_NAME}_full_3', '${CLICKHOUSE_TEST_UNIQUE_NAME}_filtered_1', '${CLICKHOUSE_TEST_UNIQUE_NAME}_nobudget') + ORDER BY 1 +" + +rm -rf "${WORKING_DIR}"