diff --git a/docs/en/interfaces/formats/Parquet/Parquet.md b/docs/en/interfaces/formats/Parquet/Parquet.md index 8f8bd703adf4d..2407fe00492ac 100644 --- a/docs/en/interfaces/formats/Parquet/Parquet.md +++ b/docs/en/interfaces/formats/Parquet/Parquet.md @@ -154,6 +154,7 @@ To exchange data with Hadoop, you can use the [`HDFS table engine`](/engines/tab | `input_format_parquet_preserve_order` | Avoid reordering rows when reading from Parquet files. Usually makes it much slower. | `0` | | `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_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. | `67108864` | | `input_format_parquet_footer_read_size` | Size in bytes of the initial tail read that fetches the Parquet footer when opening a file with reader v3. `0` (default) sizes the read adaptively to 1% of the file, clamped to [128 KiB, 2 MiB]. A non-zero value forces a fixed initial read size. | `0` | | `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` | diff --git a/src/Core/FormatFactorySettings.h b/src/Core/FormatFactorySettings.h index 4e6c1cca18287..1fd02457caa40 100644 --- a/src/Core/FormatFactorySettings.h +++ b/src/Core/FormatFactorySettings.h @@ -191,6 +191,9 @@ When reading Parquet files, skip whole row groups based on the WHERE/PREWHERE ex )", 0) \ DECLARE(Bool, input_format_parquet_bloom_filter_push_down, true, R"( When reading Parquet files, skip whole row groups based on the WHERE expressions and bloom filter in the Parquet metadata. +)", 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. Contiguous bloom filters are read together up to this size. If the file itself is not larger than this and it has enough bloom filters that separate reads would cost more than reading the file, the whole file is read once. 0 keeps a separate read per bloom filter. )", 0) \ DECLARE(UInt64, input_format_parquet_footer_read_size, 0, R"( Size (in bytes) of the initial tail read that fetches the Parquet footer (`FileMetaData`) when opening a file with reader v3. `0` (default) sizes the read adaptively to the file - 1% of the file size, clamped to `[128 KiB, 2 MiB]` - so it usually covers the whole footer in a single read even for wide files with hundreds of columns. Set a non-zero value to force a fixed initial read size instead of the adaptive one; this is useful for very high-latency object storage where reading a larger tail up front avoids a second round trip. The value is clamped to the file size. diff --git a/src/Core/SettingsChangesHistory.cpp b/src/Core/SettingsChangesHistory.cpp index 4dff6c74d38ee..3fe88e4af38e2 100644 --- a/src/Core/SettingsChangesHistory.cpp +++ b/src/Core/SettingsChangesHistory.cpp @@ -39,6 +39,11 @@ 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.6.8.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.6.4.20001.altinityantalya", { {"input_format_parquet_footer_read_size", 65536, 0, "New setting to override the adaptive Parquet footer initial read size with a fixed number of bytes; 0 keeps the adaptive behavior. Before this setting the reader always read a fixed 64 KiB tail."}, diff --git a/src/Formats/FormatFactory.cpp b/src/Formats/FormatFactory.cpp index b6dbd487b8541..b1a729c1cbae2 100644 --- a/src/Formats/FormatFactory.cpp +++ b/src/Formats/FormatFactory.cpp @@ -217,6 +217,7 @@ FormatSettings getFormatSettings(const ContextPtr & context, const Settings & se format_settings.parquet.preserve_order = settings[Setting::input_format_parquet_preserve_order]; 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.bloom_filter_coalesce_max_bytes = settings[Setting::input_format_parquet_bloom_filter_coalesce_max_bytes]; format_settings.parquet.footer_read_size = settings[Setting::input_format_parquet_footer_read_size]; 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]; diff --git a/src/Formats/FormatSettings.h b/src/Formats/FormatSettings.h index f30f3b6211248..cf869bacd841e 100644 --- a/src/Formats/FormatSettings.h +++ b/src/Formats/FormatSettings.h @@ -347,6 +347,8 @@ struct FormatSettings bool case_insensitive_column_matching = false; bool filter_push_down = true; bool bloom_filter_push_down = true; + /// 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; /// `0` sizes the initial footer read adaptively (1% of the file, clamped to [128 KiB, 2 MiB]). size_t footer_read_size = 0; bool page_filter_push_down = true; diff --git a/src/Processors/Formats/Impl/Parquet/Prefetcher.cpp b/src/Processors/Formats/Impl/Parquet/Prefetcher.cpp index c673a467b7279..693f2ada97a61 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,21 @@ 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; + /// Bloom-filter groups stay inside the group, so a filter is not merged with the column data next to it. + /// `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; + if ((initial_req->coalesce_group != 0 && !same_group) || + r.end + gap_limit <= start_offset || // short gap + 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 +397,17 @@ 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; + if ((initial_req->coalesce_group != 0 && !same_group) || + 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 +450,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 +662,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 17bda9f374475..047cbcb15b0ef 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 61839a51b60d3..afa9f7bd55c31 100644 --- a/src/Processors/Formats/Impl/Parquet/Reader.cpp +++ b/src/Processors/Formats/Impl/Parquet/Reader.cpp @@ -573,6 +573,56 @@ 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) + { + 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) + { + if (!primitive_columns[column_idx].use_bloom_filter) + continue; + const auto & meta = row_group.columns[column_idx].meta->meta_data; + /// Same predicate as `use_dictionary_filter` below. That flag is set only while + /// registering ranges, which is too late: a dictionary-filtered chunk does not read + /// its bloom filter and must not turn on a whole-file read. + if (meta.__isset.dictionary_page_offset && + primitive_columns[column_idx].used_by_key_condition && + size_t(meta.data_page_offset) - size_t(meta.dictionary_page_offset) < options.dictionary_filter_limit_bytes && + meta.__isset.encoding_stats) + { + bool all_pages_are_dictionary_encoded = true; + for (const parq::PageEncodingStats & s : meta.encoding_stats) + all_pages_are_dictionary_encoded &= + (s.page_type != parq::PageType::DATA_PAGE && s.page_type != parq::PageType::DATA_PAGE_V2) || + s.encoding == parq::Encoding::PLAIN_DICTIONARY || + s.encoding == parq::Encoding::RLE_DICTIONARY || + s.count == 0; + if (all_pages_are_dictionary_encoded) + 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 || std::any_of(primitive_columns.begin(), primitive_columns.end(), [](const auto & c) { return c.column_index_condition; }); bool need_to_find_bloom_filter_lengths_the_hard_way = false; @@ -626,13 +676,13 @@ void Reader::initializePrefetches() column.bloom_filter_data_bytes = len; 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_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. @@ -711,7 +761,7 @@ void Reader::initializePrefetches() column.bloom_filter_data_bytes = end - offset; column.bloom_filter_data_prefetch = prefetcher.registerRange( - offset, end - offset, /*likely_to_be_used=*/ false); + offset, end - offset, /*likely_to_be_used=*/ false, bloom_group); } } } diff --git a/src/Processors/Formats/Impl/ParquetV3BlockInputFormat.cpp b/src/Processors/Formats/Impl/ParquetV3BlockInputFormat.cpp index 65387f4a900b0..087017cac17b6 100644 --- a/src/Processors/Formats/Impl/ParquetV3BlockInputFormat.cpp +++ b/src/Processors/Formats/Impl/ParquetV3BlockInputFormat.cpp @@ -570,6 +570,7 @@ To exchange data with Hadoop, you can use the [`HDFS table engine`](/engines/tab | `input_format_parquet_preserve_order` | Avoid reordering rows when reading from Parquet files. Usually makes it much slower. | `0` | | `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_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. | `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` | 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..cf8385c850e1d --- /dev/null +++ b/tests/queries/0_stateless/05297_parquet_bloom_filter_coalesce.sh @@ -0,0 +1,93 @@ +#!/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. + +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_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; } +# 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%% *} +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 [ "$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..69a9ba6ccc97d --- /dev/null +++ b/tests/queries/0_stateless/05298_parquet_bloom_filter_s3_gets.sh @@ -0,0 +1,77 @@ +#!/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. + +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_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..3b56c3a992cdb --- /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.6.8 Antalya. +-- `compatibility` with 26.6.4, 26.6.2, or 26.6 must restore 0 (one read per bloom filter). + +SELECT getSetting('input_format_parquet_bloom_filter_coalesce_max_bytes'); + +SET compatibility = '26.6.8.20001.altinityantalya'; +SELECT getSetting('input_format_parquet_bloom_filter_coalesce_max_bytes'); + +SET compatibility = '26.6.4.20001.altinityantalya'; +SELECT getSetting('input_format_parquet_bloom_filter_coalesce_max_bytes'); + +SET compatibility = '26.6.2.20001.altinityantalya'; +SELECT getSetting('input_format_parquet_bloom_filter_coalesce_max_bytes'); + +SET compatibility = '26.6'; +SELECT getSetting('input_format_parquet_bloom_filter_coalesce_max_bytes');