diff --git a/src/Core/FormatFactorySettings.h b/src/Core/FormatFactorySettings.h index 869b03a88246e..c9ee5a524f977 100644 --- a/src/Core/FormatFactorySettings.h +++ b/src/Core/FormatFactorySettings.h @@ -208,6 +208,20 @@ When reading Parquet files, skip whole row groups based on the WHERE expressions )", 0) \ DECLARE(UInt64, input_format_parquet_dictionary_filter_push_down, 1024 * 1024, R"( When reading Parquet files (with reader v3), skip whole row groups based on the WHERE/PREWHERE expressions and the dictionary page contents, when all data pages of a column chunk are dictionary-encoded. The value is the maximum dictionary page size (in bytes) for which this optimization is applied; set to 0 to disable. This takes precedence over the bloom filter when both are available. +)", 0) \ + DECLARE(UInt64, input_format_parquet_bloom_filter_coalesce_max_bytes, 64 * 1024 * 1024, R"( +Max bytes of one read when loading Parquet bloom filters. + +ClickHouse writes those filters back-to-back, after the row-group data, so nothing sits between neighboring filters. Filters in that block are read together, up to this many bytes. If the file itself is not larger than this value and it has enough filters that separate reads would cost more than one read of the file, the whole file is read once and later column reads reuse that buffer. `0` keeps one read per bloom filter. + +This is separate from `remote_read_min_bytes_for_seek` (and from `input_format_parquet_local_file_min_bytes_for_seek` for a local file): + +- `remote_read_min_bytes_for_seek` merges a neighboring range only when that range is shorter than the seek threshold. A bloom filter longer than the threshold stays its own read. +- Raising that threshold far enough to cover a large bloom filter also applies to dictionary pages, indexes, and column data, so unrelated bytes are read with the filters. +- An ordinary coalesced read may grow to four times the seek threshold. This setting is its own cap, and it applies only to bloom-filter ranges. +- Enough filters in a file no larger than this setting select one read of the whole file. The seek threshold has no such rule. + +A bloom body kept only as a fallback for the dictionary filter is not part of this read. Its header is still read. )", 0) \ DECLARE(Bool, input_format_parquet_enable_json_parsing, true, R"( When reading Parquet files, parse JSON columns as ClickHouse JSON Column. diff --git a/src/Core/SettingsChangesHistory.cpp b/src/Core/SettingsChangesHistory.cpp index 467099d10b865..c07524d09ee57 100644 --- a/src/Core/SettingsChangesHistory.cpp +++ b/src/Core/SettingsChangesHistory.cpp @@ -41,6 +41,10 @@ const VersionToSettingsChangesMap & getSettingsChangesHistory() /// controls new feature and it's 'true' by default, use 'false' as previous_value). /// It's used to implement `compatibility` setting (see https://github.com/ClickHouse/ClickHouse/issues/35972) /// Note: please check if the key already exists to prevent duplicate entries. + addSettingsChanges(settings_changes_history, "26.8.6.20001.altinityantalya", + { + {"input_format_parquet_bloom_filter_coalesce_max_bytes", 0, 64 * 1024 * 1024, "Read contiguous Parquet bloom filters, or a small file that has many of them, in one request."}, + }); addSettingsChanges(settings_changes_history, "26.10", { {"azure_validate_etag_on_read", false, true, "New setting to detect concurrent in-place overwrites of Azure blobs during a read by pinning every `GET` to the listed `ETag` with `If-Match` and validating the `ETag` of the response, like `s3_validate_etag_on_read` does for S3. `compatibility` with versions before 26.10 restores the previous behavior (no validation)."}, diff --git a/src/Formats/FormatFactory.cpp b/src/Formats/FormatFactory.cpp index 8906a9924a13a..bdb135e1057e3 100644 --- a/src/Formats/FormatFactory.cpp +++ b/src/Formats/FormatFactory.cpp @@ -227,6 +227,7 @@ FormatSettings getFormatSettings(const ContextPtr & context, const Settings & se format_settings.parquet.filter_push_down = settings[Setting::input_format_parquet_filter_push_down]; format_settings.parquet.bloom_filter_push_down = settings[Setting::input_format_parquet_bloom_filter_push_down]; format_settings.parquet.dictionary_filter_push_down = settings[Setting::input_format_parquet_dictionary_filter_push_down]; + format_settings.parquet.bloom_filter_coalesce_max_bytes = settings[Setting::input_format_parquet_bloom_filter_coalesce_max_bytes]; format_settings.parquet.page_filter_push_down = settings[Setting::input_format_parquet_page_filter_push_down]; format_settings.parquet.spatial_filter_push_down = settings[Setting::input_format_parquet_spatial_filter_push_down]; format_settings.parquet.use_offset_index = settings[Setting::input_format_parquet_use_offset_index]; diff --git a/src/Formats/FormatSettings.h b/src/Formats/FormatSettings.h index 529d763929b7b..eb769b8fe5d1a 100644 --- a/src/Formats/FormatSettings.h +++ b/src/Formats/FormatSettings.h @@ -374,6 +374,8 @@ struct FormatSettings bool filter_push_down = true; bool bloom_filter_push_down = true; size_t dictionary_filter_push_down = 1024 * 1024; + /// 0 disables coalescing bloom-filter reads. See `input_format_parquet_bloom_filter_coalesce_max_bytes`. + size_t bloom_filter_coalesce_max_bytes = 64 * 1024 * 1024; bool page_filter_push_down = true; bool use_offset_index = true; diff --git a/src/Processors/Formats/Impl/Parquet/Prefetcher.cpp b/src/Processors/Formats/Impl/Parquet/Prefetcher.cpp index c673a467b7279..32b2f978b37e8 100644 --- a/src/Processors/Formats/Impl/Parquet/Prefetcher.cpp +++ b/src/Processors/Formats/Impl/Parquet/Prefetcher.cpp @@ -129,13 +129,14 @@ void Prefetcher::readSync(char * to, size_t n, size_t offset) throw Exception(ErrorCodes::INCORRECT_DATA, "Unexpected eof: offset {}, length {}, bytes read {}, expected file size {}", offset, n, nread, file_size); } -PrefetchHandle Prefetcher::registerRange(size_t offset, size_t length, bool likely_to_be_used) +PrefetchHandle Prefetcher::registerRange(size_t offset, size_t length, bool likely_to_be_used, uint8_t coalesce_group) { chassert(!ranges_finalized.load(std::memory_order_relaxed)); if (offset > file_size || length > file_size - offset) throw Exception(ErrorCodes::INCORRECT_DATA, "Range out of bounds: offset {}, length {}, file size {}", offset, length, file_size); RequestState & req = requests.emplace_back(); req.length = length; + req.coalesce_group = coalesce_group_for_all_ranges ? coalesce_group_for_all_ranges : coalesce_group; req.allow_incidental_read.store(likely_to_be_used || length < min_bytes_for_seek, std::memory_order_relaxed); range_sets[0].ranges.push_back(RangeState {.request = &req, .start = offset, .end = offset + length}); return PrefetchHandle(&req); @@ -148,7 +149,11 @@ void Prefetcher::finalizeRanges() auto & ranges = range_sets[0].ranges; std::sort(ranges.begin(), ranges.end(), [](const RangeState & a, const RangeState & b) { - return a.start < b.start; + if (a.start != b.start) + return a.start < b.start; + /// Longer range first. A bloom-filter data range and its header share `start`, and + /// `std::sort` would otherwise leave that order unspecified. + return a.end > b.end; }); for (size_t i = 0; i < ranges.size(); ++i) { @@ -179,12 +184,29 @@ void Prefetcher::startPrefetch(const std::vector & requests_to chassert(req); pickRangesAndCreateTaskIfNotExists(req, *handle, /*splitting=*/ false, 0, 0, std::unique_lock(mutex)); chassert(req->state.load(std::memory_order_relaxed) == RequestState::State::HasTask); - const Task * task = req->task; + Task * task = req->task; if (!handle->memory) { size_t memory_usage = static_cast(static_cast(req->length) * task->memory_amplification); - handle->memory = MemoryUsageToken(memory_usage, diff); + /// A coalesced buffer outlives the handle that started it: the bloom-filter header is reset + /// as soon as it is parsed, while the other headers and block handles still pin `buf`. + /// Charge the buffer on the task until the last of those handles drops it. + if (req->coalesce_group != 0 || task->coalesce_memory_state.load(std::memory_order_acquire) == 1) + { + if (req->coalesce_group != 0 && task->length > 0) + { + uint8_t expected = 0; + if (task->coalesce_memory_state.compare_exchange_strong(expected, 2, std::memory_order_acq_rel)) + { + task->coalesce_memory = MemoryUsageToken(task->length, diff); + task->coalesce_memory_state.store(1, std::memory_order_release); + } + } + memory_usage = 0; + } + if (memory_usage != 0) + handle->memory = MemoryUsageToken(memory_usage, diff); } } } @@ -328,9 +350,26 @@ void Prefetcher::pickRangesAndCreateTaskIfNotExists(RequestState * initial_req, 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 - !r.request->allow_incidental_read.load(std::memory_order_relaxed)) // range wants to be coalesced + const bool same_group = initial_req->coalesce_group != 0 && r.request->coalesce_group == initial_req->coalesce_group; + /// A bloom-filter group stays inside itself. Column data must not merge with a filter, whichever range starts the read. + /// `across_gaps` is the small-file path: one read covers the whole object. + /// The same-group span is the configured cap, even when that cap is below `bytes_per_read_task`. + /// The cap is the whole task. A bloom body included on the left must share it with later filters. + const size_t gap_limit = (same_group && coalesce_group_across_gaps) ? coalesce_group_span_limit : min_bytes_for_seek; + const size_t span_limit = same_group ? coalesce_group_span_limit : bytes_per_read_task; + const size_t prospective_start = std::min(start_offset, r.start); + const size_t prospective_end = std::max(end_offset, r.end); + const bool end_past_cap = same_group && prospective_end - prospective_start > span_limit; + const bool group_boundary = initial_req->coalesce_group != r.request->coalesce_group + && (initial_req->coalesce_group != 0 || r.request->coalesce_group != 0); + /// A zero-byte gap is contiguous. `min_bytes_for_seek` of 0 must not split filters written back-to-back. + const bool positive_gap = same_group ? r.end < start_offset : true; + if (group_boundary || + (positive_gap && r.end + gap_limit <= start_offset) || // short gap + /// `end_past_cap` is the same-group cap. This addition wraps when the cap is `UInt64` max. + (!same_group && r.start + span_limit <= initial_offset) || // task not too big + end_past_cap || + (!same_group && !r.request->allow_incidental_read.load(std::memory_order_relaxed))) // range wants to be coalesced break; const auto s = r.request->state.load(std::memory_order_relaxed); @@ -363,9 +402,21 @@ void Prefetcher::pickRangesAndCreateTaskIfNotExists(RequestState * initial_req, 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.request->allow_incidental_read.load(std::memory_order_relaxed)) + const bool same_group = initial_req->coalesce_group != 0 && r.request->coalesce_group == initial_req->coalesce_group; + const size_t gap_limit = (same_group && coalesce_group_across_gaps) ? coalesce_group_span_limit : min_bytes_for_seek; + const size_t span_limit = same_group ? coalesce_group_span_limit : bytes_per_read_task; + const size_t prospective_start = std::min(start_offset, r.start); + const size_t prospective_end = std::max(end_offset, r.end); + const bool end_past_cap = same_group && prospective_end - prospective_start > span_limit; + const bool group_boundary = initial_req->coalesce_group != r.request->coalesce_group + && (initial_req->coalesce_group != 0 || r.request->coalesce_group != 0); + /// Same zero-gap rule as the left scan. + const bool positive_gap = same_group ? end_offset < r.start : true; + if (group_boundary || + (positive_gap && end_offset + gap_limit <= r.start) || + (!same_group && initial_offset + span_limit <= r.end) || + end_past_cap || + (!same_group && !r.request->allow_incidental_read.load(std::memory_order_relaxed))) break; const auto s = r.request->state.load(std::memory_order_relaxed); @@ -408,20 +459,33 @@ void Prefetcher::pickRangesAndCreateTaskIfNotExists(RequestState * initial_req, } chassert(actual_refcount > 0); - decreaseTaskRefcount(&task, initial_refcount - actual_refcount); + decreaseTaskRefcount(&task, initial_refcount - actual_refcount, nullptr); lock.unlock(); scheduleTask(&task); } -void Prefetcher::decreaseTaskRefcount(Task * task, size_t amount) +void Prefetcher::decreaseTaskRefcount(Task * task, size_t amount, MemoryUsageDiff * diff) { size_t c = task->refcount.fetch_sub(amount, std::memory_order_acq_rel); chassert(c >= amount); if (c != amount) return; + /// Last handle. Drop the coalesced-buffer charge together with the buffer itself. + if (task->coalesce_memory_state.load(std::memory_order_acquire) == 1) + { + if (diff) + task->coalesce_memory.reset(diff); + else + { + /// Shutdown abandons per-stage accounting; drop the token without recording it. + MemoryUsageToken dropped(std::move(task->coalesce_memory)); + (void)dropped; + } + } + if (task->state.exchange(Task::State::Deallocated) != Task::State::Running) { task->buf = {}; @@ -607,7 +671,7 @@ void PrefetchHandle::reset(MemoryUsageDiff * diff) memory.reset(diff); if (request->state.exchange(RequestState::State::Cancelled) == RequestState::State::HasTask) - Prefetcher::decreaseTaskRefcount(request->task, 1); + Prefetcher::decreaseTaskRefcount(request->task, 1, diff); request = nullptr; } diff --git a/src/Processors/Formats/Impl/Parquet/Prefetcher.h b/src/Processors/Formats/Impl/Parquet/Prefetcher.h index 135492aff741b..d0734ff23708c 100644 --- a/src/Processors/Formats/Impl/Parquet/Prefetcher.h +++ b/src/Processors/Formats/Impl/Parquet/Prefetcher.h @@ -40,7 +40,23 @@ class Prefetcher /// (We use overlap to simplify bloom filter header reading a little.) /// If likely_to_be_used is true, Prefetcher will be more eager to piggy-back this range when /// reading other ranges. - PrefetchHandle registerRange(size_t offset, size_t length, bool likely_to_be_used); + /// `coalesce_group` != 0: this range may be read together with other ranges of the same group. + /// See configureCoalescing. + PrefetchHandle registerRange(size_t offset, size_t length, bool likely_to_be_used, uint8_t coalesce_group = 0); + + /// Subsequent ranges in one group are read in one request, up to `span_limit` bytes. + /// `across_gaps` also merges ranges farther apart than min_bytes_for_seek (a small file, one read). + /// `for_all_ranges` puts every range into that group, so later column reads reuse the same buffer. + void configureCoalescing(size_t span_limit, bool across_gaps, bool for_all_ranges) + { + coalesce_group_span_limit = span_limit; + coalesce_group_across_gaps = across_gaps; + coalesce_group_for_all_ranges = for_all_ranges ? 1 : 0; + } + + /// Group id to pass to registerRange for bloom-filter ranges. 0 if coalescing is off. + /// When every range is already in the group, this is also non-zero so bloom ranges match. + uint8_t bloomFilterCoalesceGroup() const { return coalesce_group_span_limit > 0 ? 1 : 0; } /// Called at most once, after all registerRange calls and before all enqueue/getRangeData calls. void finalizeRanges(); @@ -98,6 +114,8 @@ class Prefetcher size_t range_idx = UINT64_MAX; size_t length = 0; size_t task_offset = 0; + /// Non-zero: merge with other ranges of this group. See configureCoalescing. + uint8_t coalesce_group = 0; }; /// Range that the user wants, before coalescing. Overlapping ranges are allowed, but are not @@ -151,6 +169,10 @@ class Prefetcher std::atomic state {State::Scheduled}; /// How many RequestState-s in HasTask state point to this Task. std::atomic refcount {}; + /// Coalesced reads charge `length` once, for as long as `buf` stays pinned by any handle. + /// 0 = not charged, 1 = `coalesce_memory` is published. + std::atomic coalesce_memory_state {0}; + MemoryUsageToken coalesce_memory; /// Notified when the state changes from Running to Done or Exception. CompletionNotification completion; std::exception_ptr exception; @@ -182,6 +204,9 @@ class Prefetcher size_t file_size{}; size_t min_bytes_for_seek{}; size_t bytes_per_read_task{}; + size_t coalesce_group_span_limit = 0; + bool coalesce_group_across_gaps = false; + uint8_t coalesce_group_for_all_ranges = 0; std::shared_ptr shutdown = std::make_shared(); @@ -204,7 +229,9 @@ class Prefetcher /// If splitting, the request is being cancelled and replaced by a smaller range /// (splitAndPrefetchRange), and only subrange [subrange_start, subrange_end) needs to be read. void pickRangesAndCreateTaskIfNotExists(RequestState *, const PrefetchHandle &, bool splitting, size_t start_offset, size_t end_offset, std::unique_lock lock); - static void decreaseTaskRefcount(Task * task, size_t amount); + /// `diff` records the release of a coalesced-buffer charge when the last reference drops. + /// Nullptr on shutdown, where that accounting is abandoned with the reader. + static void decreaseTaskRefcount(Task * task, size_t amount, MemoryUsageDiff * diff); void scheduleTask(Task * task); Task::State runTask(Task * task); [[noreturn]] void rethrowException(Task * task); diff --git a/src/Processors/Formats/Impl/Parquet/ReadManager.cpp b/src/Processors/Formats/Impl/Parquet/ReadManager.cpp index 20e1c113ae7be..8ada36fef5167 100644 --- a/src/Processors/Formats/Impl/Parquet/ReadManager.cpp +++ b/src/Processors/Formats/Impl/Parquet/ReadManager.cpp @@ -628,10 +628,12 @@ void ReadManager::scheduleTasksIfNeeded(ReadStage stage_idx) if (row_group_idx != i) return false; const RowGroup & row_group = reader.row_groups[row_group_idx]; - /// Must check stage first so that read_ptr is meaningful (we start advancing it in finishRowSubgroupStage). + /// Before column data, read_ptr is not advanced yet. This row group still has to run: a coalesced + /// prefetch is charged until every handle that pins it has finished, including handles of later + /// row groups. Blocking this row group would leave that charge stuck. /// Using acquire ordering to synchronize with the release (seq_cst) store in `finishRowGroupStage`. if (row_group.stage.load(std::memory_order_acquire) < ReadStage::ColumnData) - return false; + return true; return row_group.read_ptr.load() == row_group.delivery_ptr.load(); }; diff --git a/src/Processors/Formats/Impl/Parquet/Reader.cpp b/src/Processors/Formats/Impl/Parquet/Reader.cpp index cbc1bb460c468..c71cc729f5a5c 100644 --- a/src/Processors/Formats/Impl/Parquet/Reader.cpp +++ b/src/Processors/Formats/Impl/Parquet/Reader.cpp @@ -1032,6 +1032,46 @@ void Reader::prepareBloomFilterCondition() void Reader::initializePrefetches() { + /// One S3 GET is one contiguous range. Contiguous bloom filters (ClickHouse writer) are read together. + /// A small file with many filters is read once, and later column reads reuse that buffer. + const size_t coalesce_max = options.format.parquet.bloom_filter_coalesce_max_bytes; + if (coalesce_max > 0 && options.format.parquet.bloom_filter_push_down) + { + size_t bloom_count = 0; + for (const RowGroup & row_group : row_groups) + { + if (!row_group.need_to_process) + continue; + for (size_t column_idx = 0; column_idx < primitive_columns.size(); ++column_idx) + { + /// Same gates as the bloom-filter registration below. A dictionary-filtered chunk still + /// reads its bloom filter as a fallback, so it counts. A chunk that will not register + /// a range must not turn on a whole-file read. + const auto & column_info = primitive_columns[column_idx]; + if (!column_info.use_bloom_filter || column_info.bloom_filter_hashes.empty()) + continue; + const auto & meta = row_group.columns[column_idx].meta->meta_data; + const bool nullable = column_info.levels.back().def > 0; + const bool can_be_null = !meta.statistics.__isset.null_count || meta.statistics.null_count != 0; + if (nullable && can_be_null && !column_info.output_nullable) + continue; + /// `bloom_filter_length` is optional (Parquet 2.10). An offset without it is still a + /// filter; the hard-way path below derives the length and registers that range. + const bool length_missing_or_positive = !meta.__isset.bloom_filter_length || meta.bloom_filter_length > 0; + if (meta.__isset.bloom_filter_offset && length_missing_or_positive) + ++bloom_count; + } + } + if (bloom_count >= 2) + { + const size_t file_size = prefetcher.getFileSize(); + const size_t seek = options.min_bytes_for_seek; + const bool whole_file = file_size <= coalesce_max && seek > 0 && bloom_count >= (file_size + seek - 1) / seek; + prefetcher.configureCoalescing(whole_file ? file_size : coalesce_max, whole_file, whole_file); + } + } + const uint8_t bloom_group = prefetcher.bloomFilterCoalesceGroup(); + bool use_offset_index = options.format.parquet.use_offset_index || format_filter_info->prewhere_info || format_filter_info->row_level_filter || format_filter_info->rows_to_read || std::any_of(primitive_columns.begin(), primitive_columns.end(), [](const auto & c) { return !c.column_index_conditions.empty(); }); @@ -1098,7 +1138,10 @@ void Reader::initializePrefetches() || column.meta->meta_data.statistics.null_count != 0; bool bloom_filter_null_safe = !nullable || !can_be_null || primitive_columns[column_idx].output_nullable; - if (options.format.parquet.bloom_filter_push_down && + /// A row group outside the requested subset produces no rows. Do not register its filter: + /// the count above skips it, and a registered filter would still be read. + if (row_group.need_to_process && + options.format.parquet.bloom_filter_push_down && primitive_columns[column_idx].use_bloom_filter && !primitive_columns[column_idx].bloom_filter_hashes.empty() && bloom_filter_null_safe && @@ -1113,15 +1156,20 @@ void Reader::initializePrefetches() size_t len = size_t(column.meta->meta_data.bloom_filter_length); max_header_length = std::min(max_header_length, len); column.bloom_filter_data_bytes = len; + /// The body is only a fallback when the dictionary filter can still rule the chunk out. + /// Leave it out of the bloom coalesce group: the header is read first and would + /// otherwise download the body before that filter is tried. A whole-file read still + /// assigns every range to the group. + const uint8_t bloom_data_group = column.use_dictionary_filter ? uint8_t{0} : bloom_group; column.bloom_filter_data_prefetch = prefetcher.registerRange( size_t(column.meta->meta_data.bloom_filter_offset), - len, /*likely_to_be_used=*/ false); + len, /*likely_to_be_used=*/ false, bloom_data_group); } /// bloom_filter_header_prefetch and bloom_filter_data_prefetch overlap, that's ok. column.use_bloom_filter = true; column.bloom_filter_header_prefetch = prefetcher.registerRange( size_t(column.meta->meta_data.bloom_filter_offset), - max_header_length, /*likely_to_be_used=*/ true); + max_header_length, /*likely_to_be_used=*/ true, bloom_group); } /// Offset index. @@ -1199,8 +1247,10 @@ void Reader::initializePrefetches() size_t end = it == all_offsets.end() ? prefetcher.getFileSize() : *it; column.bloom_filter_data_bytes = end - offset; + /// Same group rule as the length-known path above. + const uint8_t bloom_data_group = column.use_dictionary_filter ? uint8_t{0} : bloom_group; column.bloom_filter_data_prefetch = prefetcher.registerRange( - offset, end - offset, /*likely_to_be_used=*/ false); + offset, end - offset, /*likely_to_be_used=*/ false, bloom_data_group); } } } diff --git a/src/Processors/Formats/Impl/ParquetV3BlockInputFormat.cpp b/src/Processors/Formats/Impl/ParquetV3BlockInputFormat.cpp index 452a0a102c85d..2f719f00dd70f 100644 --- a/src/Processors/Formats/Impl/ParquetV3BlockInputFormat.cpp +++ b/src/Processors/Formats/Impl/ParquetV3BlockInputFormat.cpp @@ -621,6 +621,7 @@ To exchange data with Hadoop, you can use the [`HDFS table engine`](/reference/e | `input_format_parquet_filter_push_down` | When reading Parquet files, skip whole row groups based on the WHERE/PREWHERE expressions and min/max statistics in the Parquet metadata. | `1` | | `input_format_parquet_bloom_filter_push_down` | When reading Parquet files, skip whole row groups based on the WHERE expressions and bloom filter in the Parquet metadata. | `0` | | `input_format_parquet_dictionary_filter_push_down` | When reading Parquet files (with reader v3), skip whole row groups based on the WHERE/PREWHERE expressions and the dictionary page contents, for equality and `IN` conditions, when all data pages of a column chunk are dictionary-encoded. The value is the maximum dictionary page size (in bytes) for which this optimization is applied; set to `0` to disable. Takes precedence over the bloom filter when both are available. | `1048576` | +| `input_format_parquet_bloom_filter_coalesce_max_bytes` | Max bytes of one read when loading Parquet bloom filters. Contiguous filters are read together. A file not larger than this, with many filters, is read once. `0` keeps one read per bloom filter. Separate from `remote_read_min_bytes_for_seek`; see [Bloom filter reads](#bloom-filter-reads). | `67108864` | | `input_format_parquet_allow_missing_columns` | Allow missing columns while reading Parquet input formats | `1` | | `input_format_parquet_local_file_min_bytes_for_seek` | Min bytes required for local read (file) to do seek, instead of read with ignore in Parquet input format | `8192` | | `input_format_parquet_enable_row_group_prefetch` | Enable row group prefetching during parquet parsing. Currently, only single-threaded parsing can prefetch. | `1` | @@ -642,6 +643,19 @@ To exchange data with Hadoop, you can use the [`HDFS table engine`](/reference/e | `output_format_parquet_geometadata` | Write GeoParquet `geo` metadata into the Parquet file footer and encode top-level ClickHouse geo columns ([`Point`](/reference/data-types/geo#point), [`MultiPoint`](/reference/data-types/geo#multipoint), [`LineString`](/reference/data-types/geo#linestring), [`Polygon`](/reference/data-types/geo#polygon), [`MultiLineString`](/reference/data-types/geo#multilinestring), [`MultiPolygon`](/reference/data-types/geo#multipolygon)) as WKB. If `0`, those columns are written using their native underlying representation (e.g. `Point` as `Tuple(Float64, Float64)`) and no GeoParquet metadata is emitted. | `1` | | `input_format_parquet_import_nested` | Obsolete setting, does nothing. | `0` | | `input_format_parquet_local_time_as_utc` | true | Determines the data type used by schema inference for Parquet timestamps with isAdjustedToUTC=false. If true: DateTime64(..., 'UTC'), if false: DateTime64(...). Neither behavior is fully correct as ClickHouse doesn't have a data type for local wall-clock time. Counterintuitively, 'true' is probably the less incorrect option, because formatting the 'UTC' timestamp as String will produce representation of the correct local time. | + +## Bloom filter reads {#bloom-filter-reads} + +`input_format_parquet_bloom_filter_coalesce_max_bytes` is the maximum size of one read that loads Parquet bloom filters. ClickHouse writes those filters back-to-back, after the row-group data, so nothing sits between neighboring filters. Filters in that block are read together, up to this many bytes. If the file itself is not larger than this value and it has enough filters that separate reads would cost more than one read of the file, the whole file is read once and later column reads reuse that buffer. `0` keeps one read per bloom filter. + +`remote_read_min_bytes_for_seek` can already merge the small filters, because they are adjacent and shorter than a large seek threshold. It is a different control, and so is `input_format_parquet_local_file_min_bytes_for_seek` for a local file: + +- `remote_read_min_bytes_for_seek` merges a neighboring range only when that range is shorter than the seek threshold. A bloom filter longer than the threshold stays its own read. +- Raising that threshold far enough to cover a large bloom filter also applies to dictionary pages, indexes, and column data, so unrelated bytes are read with the filters. +- An ordinary coalesced read may grow to four times the seek threshold. `input_format_parquet_bloom_filter_coalesce_max_bytes` is its own cap, and it applies only to bloom-filter ranges. +- Enough filters in a file no larger than this setting select one read of the whole file. The seek threshold has no such rule. + +A bloom body kept only as a fallback for the dictionary filter is not part of this read. Its header is still read. )DOCS_MD"}); } diff --git a/tests/queries/0_stateless/05297_parquet_bloom_filter_coalesce.reference b/tests/queries/0_stateless/05297_parquet_bloom_filter_coalesce.reference new file mode 100644 index 0000000000000..d86bac9de59ab --- /dev/null +++ b/tests/queries/0_stateless/05297_parquet_bloom_filter_coalesce.reference @@ -0,0 +1 @@ +OK diff --git a/tests/queries/0_stateless/05297_parquet_bloom_filter_coalesce.sh b/tests/queries/0_stateless/05297_parquet_bloom_filter_coalesce.sh new file mode 100755 index 0000000000000..2437c1a21368b --- /dev/null +++ b/tests/queries/0_stateless/05297_parquet_bloom_filter_coalesce.sh @@ -0,0 +1,109 @@ +#!/usr/bin/env bash +# Tags: no-fasttest + +# Many row groups, bloom filters flushed as one block. A selective WHERE must not issue one read per row group. +# Dictionary-filter pushdown is off: a dictionary-eligible bloom body is a fallback and stays out of this read. + +CUR_DIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd) +# shellcheck source=../shell_config.sh +. "$CUR_DIR"/../shell_config.sh + +WORKING_DIR="${CLICKHOUSE_TMP}/${CLICKHOUSE_TEST_UNIQUE_NAME}" +mkdir -p "${WORKING_DIR}" +FILE_PATH="${WORKING_DIR}/bloom.parquet" + +${CLICKHOUSE_LOCAL} -q " + INSERT INTO FUNCTION file('${FILE_PATH}', Parquet) + SELECT if(number % 1000 IN (0, 1), + if(number % 1000 = 0, toUInt32(0), toUInt32(1000000)), + toUInt32(intDiv(number, 1000) * 10000 + (number % 1000))) AS k + FROM numbers(50000) + SETTINGS output_format_parquet_row_group_size = 1000, output_format_parquet_write_bloom_filter = 1 +" + +# Every row group has min 0 and max 1000000, so min/max cannot skip. Value 42 exists in one row group. +read_stats() +{ + local coalesce_max=$1 + local min_seek=$2 + ${CLICKHOUSE_LOCAL} --print-profile-events --profile-events-delay-ms=-1 -q " + SELECT count() + FROM file('${FILE_PATH}', Parquet) + WHERE k = 42 + SETTINGS + max_threads = 1, + use_cache_for_count_from_files = 0, + input_format_parquet_filter_push_down = 1, + input_format_parquet_bloom_filter_push_down = 1, + input_format_parquet_page_filter_push_down = 0, + input_format_parquet_use_offset_index = 0, + input_format_parquet_dictionary_filter_push_down = 0, + input_format_parquet_local_file_min_bytes_for_seek = ${min_seek}, + input_format_parquet_bloom_filter_coalesce_max_bytes = ${coalesce_max} + " 2>&1 | awk ' + $0 ~ /ParquetPrefetcherReadRandomRead:/ { reads += $(NF-1) } + $0 ~ /ReadBufferFromFileDescriptorReadBytes:/ { bytes += $(NF-1) } + $1 == "1" { rows = 1 } + END { + if (rows != 1) exit 2 + print reads, bytes + 0 + } + ' +} + +# min_bytes_for_seek = 1 disables the old gap coalescing, so each bloom filter is its own read. +SEPARATE=$(read_stats 0 1) || { echo "separate read failed"; exit 1; } +CLUSTER=$(read_stats 67108864 1) || { echo "cluster read failed"; exit 1; } +# Seek of 0 used to treat a zero-byte gap as a split. ClickHouse writes the filters back-to-back. +TOUCH=$(read_stats 67108864 0) || { echo "zero-seek coalesce failed"; exit 1; } +# UInt64 max used to wrap the left-scan cap and split a filter header from its body. +UNLIMITED=$(read_stats 18446744073709551615 0) || { echo "unlimited coalesce failed"; exit 1; } +# Enough filters that the seek budget covers the file: one read of the whole object. +WHOLE=$(read_stats 67108864 1000000000) || { echo "whole-file read failed"; exit 1; } +# Cap of 1 byte, seek of 1 MiB: bytes_per_read_task is 4 MiB and would swallow this file if the cap were raised. +CAPPED=$(read_stats 1 1000000) || { echo "capped read failed"; exit 1; } +# Cap below one bloom filter. The header and the filter body share a start offset; the body must not be pulled in. +NARROW=$(read_stats 512 1) || { echo "narrow read failed"; exit 1; } + +SEPARATE_READS=${SEPARATE%% *} +SEPARATE_BYTES=${SEPARATE##* } +CLUSTER_READS=${CLUSTER%% *} +TOUCH_READS=${TOUCH%% *} +UNLIMITED_READS=${UNLIMITED%% *} +WHOLE_READS=${WHOLE%% *} +CAPPED_READS=${CAPPED%% *} +NARROW_READS=${NARROW%% *} +NARROW_BYTES=${NARROW##* } + +if [ "$CLUSTER_READS" -ge "$SEPARATE_READS" ]; then + echo "cluster coalesce did not reduce reads: cluster=$CLUSTER separate=$SEPARATE" + exit 1 +fi +if [ "$TOUCH_READS" -ge "$SEPARATE_READS" ]; then + echo "zero seek split back-to-back filters: touch=$TOUCH separate=$SEPARATE cluster=$CLUSTER" + exit 1 +fi +if [ "$UNLIMITED_READS" -ge "$SEPARATE_READS" ]; then + echo "max coalesce cap split the bloom header from its body: unlimited=$UNLIMITED separate=$SEPARATE touch=$TOUCH" + exit 1 +fi +if [ "$WHOLE_READS" -gt 4 ]; then + echo "whole-file read used too many requests: whole=$WHOLE cluster=$CLUSTER separate=$SEPARATE" + exit 1 +fi +if [ "$CAPPED_READS" -le 20 ]; then + echo "small coalesce cap was raised: capped=$CAPPED separate=$SEPARATE cluster=$CLUSTER whole=$WHOLE" + exit 1 +fi +# The header and the filter body share a start offset. A cap smaller than the body must not read the body. +if [ "$NARROW_BYTES" -gt $((SEPARATE_BYTES * 2)) ]; then + echo "bloom body was read with the header: narrow=$NARROW separate=$SEPARATE" + exit 1 +fi +if [ "$NARROW_READS" -le 20 ]; then + echo "narrow cap collapsed reads: narrow=$NARROW separate=$SEPARATE" + exit 1 +fi + +rm -rf "${WORKING_DIR}" +echo OK diff --git a/tests/queries/0_stateless/05298_parquet_bloom_filter_s3_gets.reference b/tests/queries/0_stateless/05298_parquet_bloom_filter_s3_gets.reference new file mode 100644 index 0000000000000..d86bac9de59ab --- /dev/null +++ b/tests/queries/0_stateless/05298_parquet_bloom_filter_s3_gets.reference @@ -0,0 +1 @@ +OK diff --git a/tests/queries/0_stateless/05298_parquet_bloom_filter_s3_gets.sh b/tests/queries/0_stateless/05298_parquet_bloom_filter_s3_gets.sh new file mode 100755 index 0000000000000..e6320a3a6860b --- /dev/null +++ b/tests/queries/0_stateless/05298_parquet_bloom_filter_s3_gets.sh @@ -0,0 +1,79 @@ +#!/usr/bin/env bash +# Tags: no-fasttest + +# Selective Parquet read from S3. Bloom filters for many row groups must not cost one GET each. +# A small object is otherwise prefetched in one request, which hides the per-range GETs. +# max_download_buffer_size keeps that prefetch off. remote_read_min_bytes_for_seek is what the +# Parquet reader uses as the gap it will still merge, so 1 byte leaves merging to bloom-filter coalescing. +# Dictionary-filter pushdown is off: each dictionary page is its own GET and is not part of the bloom-filter read. + +CUR_DIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd) +# shellcheck source=../shell_config.sh +. "$CUR_DIR"/../shell_config.sh + +URL="http://localhost:11111/test/${CLICKHOUSE_TEST_UNIQUE_NAME}.parquet" + +${CLICKHOUSE_LOCAL} -q " + INSERT INTO FUNCTION s3('${URL}', 'test', 'testtest', 'Parquet', 'k UInt32') + SELECT if(number % 1000 IN (0, 1), + if(number % 1000 = 0, toUInt32(0), toUInt32(1000000)), + toUInt32(intDiv(number, 1000) * 10000 + (number % 1000))) AS k + FROM numbers(50000) + SETTINGS + output_format_parquet_row_group_size = 1000, + output_format_parquet_write_bloom_filter = 1, + s3_truncate_on_insert = 1 +" + +# Every row group spans 0..1000000, so min/max cannot skip. Value 42 is in one row group. +s3_gets() +{ + local coalesce_max=$1 + local min_seek=$2 + ${CLICKHOUSE_LOCAL} --print-profile-events --profile-events-delay-ms=-1 -q " + SELECT count() + FROM s3('${URL}', 'test', 'testtest', 'Parquet', 'k UInt32') + WHERE k = 42 + SETTINGS + max_threads = 1, + enable_filesystem_cache = 0, + use_page_cache_for_object_storage = 0, + read_through_distributed_cache = 0, + max_download_buffer_size = 1, + remote_read_min_bytes_for_seek = ${min_seek}, + input_format_parquet_filter_push_down = 1, + input_format_parquet_bloom_filter_push_down = 1, + input_format_parquet_page_filter_push_down = 0, + input_format_parquet_use_offset_index = 0, + input_format_parquet_dictionary_filter_push_down = 0, + input_format_parquet_bloom_filter_coalesce_max_bytes = ${coalesce_max} + " 2>&1 | awk ' + $0 ~ /S3GetObject:/ { gets += $(NF-1) } + $1 == "1" { rows = 1 } + END { + if (rows != 1) exit 2 + print gets + 0 + } + ' +} + +# 1-byte seek keeps the ordinary prefetcher from merging ranges. Coalescing then does it. +SEPARATE=$(s3_gets 0 1) || { echo "separate GET count failed"; exit 1; } +CLUSTER=$(s3_gets 67108864 1) || { echo "coalesced GET count failed"; exit 1; } +# Seek larger than the file, file still under the coalesce limit: one read of the object. +WHOLE=$(s3_gets 67108864 4000000) || { echo "whole-file GET count failed"; exit 1; } + +if [ "$SEPARATE" -lt 40 ]; then + echo "expected one GET per row group, got $SEPARATE (cluster=$CLUSTER whole=$WHOLE)" + exit 1 +fi +if [ "$CLUSTER" -ge "$SEPARATE" ] || [ "$CLUSTER" -gt 15 ]; then + echo "coalescing did not cut S3 GETs: separate=$SEPARATE cluster=$CLUSTER whole=$WHOLE" + exit 1 +fi +if [ "$WHOLE" -gt 4 ]; then + echo "whole-file read used too many S3 GETs: whole=$WHOLE separate=$SEPARATE cluster=$CLUSTER" + exit 1 +fi + +echo OK diff --git a/tests/queries/0_stateless/05304_parquet_bloom_filter_coalesce_compatibility.reference b/tests/queries/0_stateless/05304_parquet_bloom_filter_coalesce_compatibility.reference new file mode 100644 index 0000000000000..7ba0a0a428a2c --- /dev/null +++ b/tests/queries/0_stateless/05304_parquet_bloom_filter_coalesce_compatibility.reference @@ -0,0 +1,5 @@ +67108864 +67108864 +0 +0 +0 diff --git a/tests/queries/0_stateless/05304_parquet_bloom_filter_coalesce_compatibility.sql b/tests/queries/0_stateless/05304_parquet_bloom_filter_coalesce_compatibility.sql new file mode 100644 index 0000000000000..d968fecb79196 --- /dev/null +++ b/tests/queries/0_stateless/05304_parquet_bloom_filter_coalesce_compatibility.sql @@ -0,0 +1,17 @@ +-- Tags: no-random-settings +-- `input_format_parquet_bloom_filter_coalesce_max_bytes` is new in 26.8.6 Antalya. +-- `compatibility` with 26.8, 26.7, or 26.6 must restore 0 (one read per bloom filter). + +SELECT getSetting('input_format_parquet_bloom_filter_coalesce_max_bytes'); + +SET compatibility = '26.8.6.20001.altinityantalya'; +SELECT getSetting('input_format_parquet_bloom_filter_coalesce_max_bytes'); + +SET compatibility = '26.8'; +SELECT getSetting('input_format_parquet_bloom_filter_coalesce_max_bytes'); + +SET compatibility = '26.7'; +SELECT getSetting('input_format_parquet_bloom_filter_coalesce_max_bytes'); + +SET compatibility = '26.6'; +SELECT getSetting('input_format_parquet_bloom_filter_coalesce_max_bytes'); diff --git a/tests/queries/0_stateless/05386_parquet_bloom_filter_dictionary_fallback.reference b/tests/queries/0_stateless/05386_parquet_bloom_filter_dictionary_fallback.reference new file mode 100644 index 0000000000000..d86bac9de59ab --- /dev/null +++ b/tests/queries/0_stateless/05386_parquet_bloom_filter_dictionary_fallback.reference @@ -0,0 +1 @@ +OK diff --git a/tests/queries/0_stateless/05386_parquet_bloom_filter_dictionary_fallback.sh b/tests/queries/0_stateless/05386_parquet_bloom_filter_dictionary_fallback.sh new file mode 100755 index 0000000000000..f7914049e14c9 --- /dev/null +++ b/tests/queries/0_stateless/05386_parquet_bloom_filter_dictionary_fallback.sh @@ -0,0 +1,80 @@ +#!/usr/bin/env bash +# Tags: no-fasttest + +# A dictionary-encoded chunk keeps its bloom filter as a fallback. Coalescing the header must +# not swallow that body. Min/max cannot skip: every row group contains 0 and 1000000. +# `file` reads go through mmap here, so the check is the random-read count, not bytes. + +CUR_DIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd) +# shellcheck source=../shell_config.sh +. "$CUR_DIR"/../shell_config.sh + +WORKING_DIR="${CLICKHOUSE_TMP}/${CLICKHOUSE_TEST_UNIQUE_NAME}" +mkdir -p "${WORKING_DIR}" +FILE_PATH="${WORKING_DIR}/bloom.parquet" + +${CLICKHOUSE_LOCAL} -q " + INSERT INTO FUNCTION file('${FILE_PATH}', Parquet) + SELECT if(number % 30000 = 0, toUInt32(0), + if(number % 30000 = 1, toUInt32(1000000), + toUInt32(intDiv(number, 30000) * 30000 + (number % 30000)))) AS k + FROM numbers(240000) + SETTINGS + output_format_parquet_row_group_size = 30000, + output_format_parquet_write_bloom_filter = 1 +" + +# Prints " ". Value 42 is in the first row group only. +read_stats() +{ + local coalesce_max=$1 + local dict_limit=$2 + ${CLICKHOUSE_LOCAL} --print-profile-events --profile-events-delay-ms=-1 -q " + SELECT count() + FROM file('${FILE_PATH}', Parquet) + WHERE k = 42 + SETTINGS + max_threads = 1, + use_cache_for_count_from_files = 0, + input_format_parquet_filter_push_down = 1, + input_format_parquet_bloom_filter_push_down = 1, + input_format_parquet_page_filter_push_down = 0, + input_format_parquet_use_offset_index = 0, + input_format_parquet_local_file_min_bytes_for_seek = 1, + input_format_parquet_dictionary_filter_push_down = ${dict_limit}, + input_format_parquet_bloom_filter_coalesce_max_bytes = ${coalesce_max} + " 2>&1 | awk ' + $0 ~ /ParquetPrefetcherReadRandomRead:/ { reads += $(NF-1) } + $0 ~ /ReadBufferFromFileDescriptorReadBytes:/ { bytes += $(NF-1) } + $1 == "1" { rows = 1 } + END { + if (rows != 1) exit 2 + print reads, bytes + 0 + } + ' +} + +# Dictionary filter on. Coalescing must not pull the fallback bloom bodies into the header read. +DICT_SEPARATE=$(read_stats 0 1048576) || { echo "dictionary separate read failed"; exit 1; } +DICT_CLUSTER=$(read_stats 67108864 1048576) || { echo "dictionary cluster read failed"; exit 1; } +# Dictionary filter off. The same bodies are the bloom read and must still coalesce. +BLOOM_SEPARATE=$(read_stats 0 0) || { echo "bloom separate read failed"; exit 1; } +BLOOM_CLUSTER=$(read_stats 67108864 0) || { echo "bloom cluster read failed"; exit 1; } + +DICT_SEPARATE_READS=${DICT_SEPARATE%% *} +DICT_CLUSTER_READS=${DICT_CLUSTER%% *} +BLOOM_SEPARATE_READS=${BLOOM_SEPARATE%% *} +BLOOM_CLUSTER_READS=${BLOOM_CLUSTER%% *} + +# The fallback bodies sit between the headers. Absorbing them turns those headers into one read. +if [ "$DICT_CLUSTER_READS" -lt "$DICT_SEPARATE_READS" ]; then + echo "dictionary fallback bloom bodies were read with the header: dict_cluster=$DICT_CLUSTER dict_separate=$DICT_SEPARATE bloom_cluster=$BLOOM_CLUSTER bloom_separate=$BLOOM_SEPARATE" + exit 1 +fi +if [ "$BLOOM_CLUSTER_READS" -ge "$BLOOM_SEPARATE_READS" ]; then + echo "bloom coalesce did not reduce reads: bloom_cluster=$BLOOM_CLUSTER bloom_separate=$BLOOM_SEPARATE dict_cluster=$DICT_CLUSTER dict_separate=$DICT_SEPARATE" + exit 1 +fi + +rm -rf "${WORKING_DIR}" +echo OK