Read Parquet bloom filters in one request when a file has many row groups- #2526

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.

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)."},

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];

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;
 

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<PrefetchHandle *> & 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<size_t>(static_cast<double>(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;
 }

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 {State::Scheduled};
         /// How many RequestState-s in HasTask state point to this Task.
         std::atomic<size_t> 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<uint8_t> 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<ShutdownHelper> shutdown = std::make_shared<ShutdownHelper>();
 
@@ -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<std::mutex> 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);

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();
     };