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

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`      |

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.

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

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

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;

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,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;
 }

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