From 348489a2c0e9e77a0f26eb34d005ce4c0da004c8 Mon Sep 17 00:00:00 2001
From: Anton Ivashkin <ianton@live.com>
Date: Thu, 8 Oct 2026 18:53:48 +0200
Subject: [PATCH 1/7] Read Parquet bloom filters in one request when a file has
 many row groups.

Separate GETs for each filter dominate cold reads from object storage. `input_format_parquet_bloom_filter_coalesce_max_bytes` merges contiguous filters, and reads a small file once when that is cheaper.

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 docs/en/interfaces/formats/Parquet/Parquet.md |  1 +
 src/Core/FormatFactorySettings.h              |  3 +
 src/Core/SettingsChangesHistory.cpp           |  1 +
 src/Formats/FormatFactory.cpp                 |  1 +
 src/Formats/FormatSettings.h                  |  2 +
 .../Formats/Impl/Parquet/Prefetcher.cpp       | 32 ++++++--
 .../Formats/Impl/Parquet/Prefetcher.h         | 25 +++++-
 .../Formats/Impl/Parquet/Reader.cpp           | 35 ++++++++-
 .../Impl/ParquetV3BlockInputFormat.cpp        |  1 +
 ...97_parquet_bloom_filter_coalesce.reference |  1 +
 .../05297_parquet_bloom_filter_coalesce.sh    | 67 ++++++++++++++++
 ...298_parquet_bloom_filter_s3_gets.reference |  1 +
 .../05298_parquet_bloom_filter_s3_gets.sh     | 77 +++++++++++++++++++
 13 files changed, 235 insertions(+), 12 deletions(-)
 create mode 100644 tests/queries/0_stateless/05297_parquet_bloom_filter_coalesce.reference
 create mode 100755 tests/queries/0_stateless/05297_parquet_bloom_filter_coalesce.sh
 create mode 100644 tests/queries/0_stateless/05298_parquet_bloom_filter_s3_gets.reference
 create mode 100755 tests/queries/0_stateless/05298_parquet_bloom_filter_s3_gets.sh

diff --git a/docs/en/interfaces/formats/Parquet/Parquet.md b/docs/en/interfaces/formats/Parquet/Parquet.md
index db31fc0dab80e..9f6814bc53b25 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_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/src/Core/FormatFactorySettings.h b/src/Core/FormatFactorySettings.h
index 5e1375351a91e..f88afd3193034 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(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 9352f418f42c8..82528beceae8e 100644
--- a/src/Core/SettingsChangesHistory.cpp
+++ b/src/Core/SettingsChangesHistory.cpp
@@ -42,6 +42,7 @@ const VersionToSettingsChangesMap & getSettingsChangesHistory()
         addSettingsChanges(settings_changes_history, "26.6.2.20001.altinityantalya",
         {
             {"object_storage_cluster_fallback_to_local_if_empty", false, false, "New setting"},
+            {"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."},
             {"use_puffin_files_cache", false, true, "Enables cache of parsed Puffin file content such as deletion vectors."},
             {"data_lake_delete_data_on_drop", false, false, "New setting that unifies dropping of data lake data; the released `iceberg_delete_data_on_drop` is kept as an alias for it."},
             {"datalake_ignore_unsupported_table_properties", false, false, "New setting: allow `CREATE TABLE` and `SHOW CREATE TABLE` in a `DataLakeCatalog` to omit unsupported table properties."},
diff --git a/src/Formats/FormatFactory.cpp b/src/Formats/FormatFactory.cpp
index d6c30363c0d2a..711fd78431341 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.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 745898c0c751b..99522252164e6 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;
         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..506b835289a44 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);
@@ -179,11 +180,16 @@ 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);
+            /// A coalesced bloom-filter (or whole-file) read is much larger than the range that started it.
+            bool already_charged = false;
+            if (req->coalesce_group != 0 && task->length > memory_usage
+                && task->coalesce_memory_charged.compare_exchange_strong(already_charged, true, std::memory_order_relaxed))
+                memory_usage = task->length;
             handle->memory = MemoryUsageToken(memory_usage, diff);
         }
     }
@@ -328,9 +334,15 @@ 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.
+        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 ? std::max(coalesce_group_span_limit, bytes_per_read_task) : bytes_per_read_task;
+        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
+            (!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 +375,13 @@ 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 ? std::max(coalesce_group_span_limit, bytes_per_read_task) : bytes_per_read_task;
+        if ((initial_req->coalesce_group != 0 && !same_group) ||
+            end_offset + gap_limit <= r.start ||
+            initial_offset + span_limit <= r.end ||
+            (!same_group && !r.request->allow_incidental_read.load(std::memory_order_relaxed)))
             break;
 
         const auto s = r.request->state.load(std::memory_order_relaxed);
diff --git a/src/Processors/Formats/Impl/Parquet/Prefetcher.h b/src/Processors/Formats/Impl/Parquet/Prefetcher.h
index 135492aff741b..39521d0b2f2ae 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,8 @@ class Prefetcher
         std::atomic<State> state {State::Scheduled};
         /// How many RequestState-s in HasTask state point to this Task.
         std::atomic<size_t> refcount {};
+        /// The first prefetch of a coalesced group charges `length` bytes once.
+        std::atomic<bool> coalesce_memory_charged {false};
         /// Notified when the state changes from Running to Done or Exception.
         CompletionNotification completion;
         std::exception_ptr exception;
@@ -182,6 +202,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>();
 
diff --git a/src/Processors/Formats/Impl/Parquet/Reader.cpp b/src/Processors/Formats/Impl/Parquet/Reader.cpp
index 64e25a3015493..46ac545a5ec65 100644
--- a/src/Processors/Formats/Impl/Parquet/Reader.cpp
+++ b/src/Processors/Formats/Impl/Parquet/Reader.cpp
@@ -565,6 +565,35 @@ 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;
+                if (meta.__isset.bloom_filter_offset && meta.__isset.bloom_filter_length && meta.bloom_filter_length > 0)
+                    ++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;
@@ -618,13 +647,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.
@@ -703,7 +732,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 8436e55647f91..1c6f922d84a6f 100644
--- a/src/Processors/Formats/Impl/ParquetV3BlockInputFormat.cpp
+++ b/src/Processors/Formats/Impl/ParquetV3BlockInputFormat.cpp
@@ -571,6 +571,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..7d349831bfdbe
--- /dev/null
+++ b/tests/queries/0_stateless/05297_parquet_bloom_filter_coalesce.sh
@@ -0,0 +1,67 @@
+#!/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_count()
+{
+    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:/ { sum += $(NF-1) }
+        $1 == "1" { rows = 1 }
+        END {
+            if (rows != 1) exit 2
+            print sum
+        }
+    '
+}
+
+# min_bytes_for_seek = 1 disables the old gap coalescing, so each bloom filter is its own read.
+SEPARATE=$(read_count 0 1) || { echo "separate read failed"; exit 1; }
+CLUSTER=$(read_count 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_count 67108864 1000000000) || { echo "whole-file read failed"; exit 1; }
+
+if [ "$CLUSTER" -ge "$SEPARATE" ]; then
+    echo "cluster coalesce did not reduce reads: cluster=$CLUSTER separate=$SEPARATE"
+    exit 1
+fi
+if [ "$WHOLE" -gt 4 ]; then
+    echo "whole-file read used too many requests: whole=$WHOLE cluster=$CLUSTER 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

From 3ed1dafbb50ba9e69ef3a995932046a7860bb189 Mon Sep 17 00:00:00 2001
From: Anton Ivashkin <ianton@live.com>
Date: Fri, 9 Oct 2026 18:38:15 +0200
Subject: [PATCH 2/7] Keep a coalesced Parquet read charged until the last
 handle drops the buffer.

The bloom-filter header that starts the read is reset as soon as it is parsed, so charging that handle let the scheduler treat a live buffer as free.

Related: https://github.com/Altinity/ClickHouse/pull/2526
Co-authored-by: Cursor <cursoragent@cursor.com>
---
 .../Formats/Impl/Parquet/Prefetcher.cpp       | 43 +++++++++++++++----
 .../Formats/Impl/Parquet/Prefetcher.h         | 10 +++--
 .../Formats/Impl/Parquet/ReadManager.cpp      |  6 ++-
 3 files changed, 45 insertions(+), 14 deletions(-)

diff --git a/src/Processors/Formats/Impl/Parquet/Prefetcher.cpp b/src/Processors/Formats/Impl/Parquet/Prefetcher.cpp
index 506b835289a44..1bfc184f1cda3 100644
--- a/src/Processors/Formats/Impl/Parquet/Prefetcher.cpp
+++ b/src/Processors/Formats/Impl/Parquet/Prefetcher.cpp
@@ -185,12 +185,24 @@ void Prefetcher::startPrefetch(const std::vector<PrefetchHandle *> & requests_to
         if (!handle->memory)
         {
             size_t memory_usage = static_cast<size_t>(static_cast<double>(req->length) * task->memory_amplification);
-            /// A coalesced bloom-filter (or whole-file) read is much larger than the range that started it.
-            bool already_charged = false;
-            if (req->coalesce_group != 0 && task->length > memory_usage
-                && task->coalesce_memory_charged.compare_exchange_strong(already_charged, true, std::memory_order_relaxed))
-                memory_usage = task->length;
-            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);
         }
     }
 }
@@ -424,20 +436,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 = {};
@@ -623,7 +648,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 39521d0b2f2ae..d0734ff23708c 100644
--- a/src/Processors/Formats/Impl/Parquet/Prefetcher.h
+++ b/src/Processors/Formats/Impl/Parquet/Prefetcher.h
@@ -169,8 +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 {};
-        /// The first prefetch of a coalesced group charges `length` bytes once.
-        std::atomic<bool> coalesce_memory_charged {false};
+        /// 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;
@@ -227,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);
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();
     };
 

From 7d0a392ea36b4e574ebe23f38f0aa4f5565574f0 Mon Sep 17 00:00:00 2001
From: Anton Ivashkin <ianton@live.com>
Date: Fri, 9 Oct 2026 18:46:05 +0200
Subject: [PATCH 3/7] Count Parquet bloom filters that omit
 `bloom_filter_length`.

That field is optional. Filters that only store an offset were left out of coalescing, even though their length is already derived later.

Related: https://github.com/Altinity/ClickHouse/pull/2526
Co-authored-by: Cursor <cursoragent@cursor.com>
---
 src/Processors/Formats/Impl/Parquet/Reader.cpp | 5 ++++-
 1 file changed, 4 insertions(+), 1 deletion(-)

diff --git a/src/Processors/Formats/Impl/Parquet/Reader.cpp b/src/Processors/Formats/Impl/Parquet/Reader.cpp
index 421f4c0e903ed..7340671620d9a 100644
--- a/src/Processors/Formats/Impl/Parquet/Reader.cpp
+++ b/src/Processors/Formats/Impl/Parquet/Reader.cpp
@@ -588,7 +588,10 @@ void Reader::initializePrefetches()
                 if (!primitive_columns[column_idx].use_bloom_filter)
                     continue;
                 const auto & meta = row_group.columns[column_idx].meta->meta_data;
-                if (meta.__isset.bloom_filter_offset && meta.__isset.bloom_filter_length && meta.bloom_filter_length > 0)
+                /// `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;
             }
         }

From 0e8a14ae33f90e95e60e30a580822257d055f0a8 Mon Sep 17 00:00:00 2001
From: Anton Ivashkin <ianton@live.com>
Date: Thu, 15 Oct 2026 12:50:22 +0200
Subject: [PATCH 4/7] Skip dictionary-filtered chunks when counting Parquet
 bloom filters for a whole-file read.

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 src/Processors/Formats/Impl/Parquet/Reader.cpp | 18 ++++++++++++++++++
 1 file changed, 18 insertions(+)

diff --git a/src/Processors/Formats/Impl/Parquet/Reader.cpp b/src/Processors/Formats/Impl/Parquet/Reader.cpp
index 7340671620d9a..afa9f7bd55c31 100644
--- a/src/Processors/Formats/Impl/Parquet/Reader.cpp
+++ b/src/Processors/Formats/Impl/Parquet/Reader.cpp
@@ -588,6 +588,24 @@ void Reader::initializePrefetches()
                 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;

From ffea12bbc6bd0f01e47845d3da515b26227d6035 Mon Sep 17 00:00:00 2001
From: Anton Ivashkin <ianton@live.com>
Date: Thu, 15 Oct 2026 13:02:00 +0200
Subject: [PATCH 5/7] Record Parquet bloom-filter coalescing as a 26.6.8
 setting so older `compatibility` values restore separate reads.

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 src/Core/SettingsChangesHistory.cpp             |  6 +++++-
 ...loom_filter_coalesce_compatibility.reference |  5 +++++
 ...quet_bloom_filter_coalesce_compatibility.sql | 17 +++++++++++++++++
 3 files changed, 27 insertions(+), 1 deletion(-)
 create mode 100644 tests/queries/0_stateless/05304_parquet_bloom_filter_coalesce_compatibility.reference
 create mode 100644 tests/queries/0_stateless/05304_parquet_bloom_filter_coalesce_compatibility.sql

diff --git a/src/Core/SettingsChangesHistory.cpp b/src/Core/SettingsChangesHistory.cpp
index 63c91b9b1dfd4..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."},
@@ -47,7 +52,6 @@ const VersionToSettingsChangesMap & getSettingsChangesHistory()
         addSettingsChanges(settings_changes_history, "26.6.2.20001.altinityantalya",
         {
             {"object_storage_cluster_fallback_to_local_if_empty", false, false, "New setting"},
-            {"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."},
             {"use_puffin_files_cache", false, true, "Enables cache of parsed Puffin file content such as deletion vectors."},
             {"data_lake_delete_data_on_drop", false, false, "New setting that unifies dropping of data lake data; the released `iceberg_delete_data_on_drop` is kept as an alias for it."},
             {"datalake_ignore_unsupported_table_properties", false, false, "New setting: allow `CREATE TABLE` and `SHOW CREATE TABLE` in a `DataLakeCatalog` to omit unsupported table properties."},
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');

From cb03b8fcf9be9dbfb9dd7ecb3e3818adc44d0f63 Mon Sep 17 00:00:00 2001
From: Anton Ivashkin <ianton@live.com>
Date: Thu, 15 Oct 2026 13:06:16 +0200
Subject: [PATCH 6/7] Honor
 `input_format_parquet_bloom_filter_coalesce_max_bytes` when it is below
 `bytes_per_read_task`.

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 src/Processors/Formats/Impl/Parquet/Prefetcher.cpp          | 5 +++--
 .../0_stateless/05297_parquet_bloom_filter_coalesce.sh      | 6 ++++++
 2 files changed, 9 insertions(+), 2 deletions(-)

diff --git a/src/Processors/Formats/Impl/Parquet/Prefetcher.cpp b/src/Processors/Formats/Impl/Parquet/Prefetcher.cpp
index 1bfc184f1cda3..8f43119e7a25d 100644
--- a/src/Processors/Formats/Impl/Parquet/Prefetcher.cpp
+++ b/src/Processors/Formats/Impl/Parquet/Prefetcher.cpp
@@ -349,8 +349,9 @@ void Prefetcher::pickRangesAndCreateTaskIfNotExists(RequestState * initial_req,
         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`.
         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 ? std::max(coalesce_group_span_limit, bytes_per_read_task) : bytes_per_read_task;
+        const size_t span_limit = same_group ? coalesce_group_span_limit : bytes_per_read_task;
         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
@@ -389,7 +390,7 @@ void Prefetcher::pickRangesAndCreateTaskIfNotExists(RequestState * initial_req,
         const RangeState & r = ranges[end_idx];
         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 ? std::max(coalesce_group_span_limit, bytes_per_read_task) : bytes_per_read_task;
+        const size_t span_limit = same_group ? coalesce_group_span_limit : bytes_per_read_task;
         if ((initial_req->coalesce_group != 0 && !same_group) ||
             end_offset + gap_limit <= r.start ||
             initial_offset + span_limit <= r.end ||
diff --git a/tests/queries/0_stateless/05297_parquet_bloom_filter_coalesce.sh b/tests/queries/0_stateless/05297_parquet_bloom_filter_coalesce.sh
index 7d349831bfdbe..d05ba981c5e02 100755
--- a/tests/queries/0_stateless/05297_parquet_bloom_filter_coalesce.sh
+++ b/tests/queries/0_stateless/05297_parquet_bloom_filter_coalesce.sh
@@ -53,6 +53,8 @@ SEPARATE=$(read_count 0 1) || { echo "separate read failed"; exit 1; }
 CLUSTER=$(read_count 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_count 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_count 1 1000000) || { echo "capped read failed"; exit 1; }
 
 if [ "$CLUSTER" -ge "$SEPARATE" ]; then
     echo "cluster coalesce did not reduce reads: cluster=$CLUSTER separate=$SEPARATE"
@@ -62,6 +64,10 @@ if [ "$WHOLE" -gt 4 ]; then
     echo "whole-file read used too many requests: whole=$WHOLE cluster=$CLUSTER separate=$SEPARATE"
     exit 1
 fi
+if [ "$CAPPED" -le 20 ]; then
+    echo "small coalesce cap was raised: capped=$CAPPED separate=$SEPARATE cluster=$CLUSTER whole=$WHOLE"
+    exit 1
+fi
 
 rm -rf "${WORKING_DIR}"
 echo OK

From eb454b5c14c2e4f29a9970abd56423b89549b055 Mon Sep 17 00:00:00 2001
From: Anton Ivashkin <ianton@live.com>
Date: Fri, 16 Oct 2026 14:39:49 +0200
Subject: [PATCH 7/7] Cap a coalesced Parquet bloom read by the whole task so a
 bloom body and the filters after it stay inside
 `input_format_parquet_bloom_filter_coalesce_max_bytes`.

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 .../Formats/Impl/Parquet/Prefetcher.cpp       | 17 +++++++-
 .../05297_parquet_bloom_filter_coalesce.sh    | 40 ++++++++++++++-----
 2 files changed, 45 insertions(+), 12 deletions(-)

diff --git a/src/Processors/Formats/Impl/Parquet/Prefetcher.cpp b/src/Processors/Formats/Impl/Parquet/Prefetcher.cpp
index 8f43119e7a25d..693f2ada97a61 100644
--- a/src/Processors/Formats/Impl/Parquet/Prefetcher.cpp
+++ b/src/Processors/Formats/Impl/Parquet/Prefetcher.cpp
@@ -149,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)
     {
@@ -350,11 +354,16 @@ void Prefetcher::pickRangesAndCreateTaskIfNotExists(RequestState * initial_req,
         /// 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;
 
@@ -391,9 +400,13 @@ void Prefetcher::pickRangesAndCreateTaskIfNotExists(RequestState * initial_req,
         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 ||
-            initial_offset + span_limit <= r.end ||
+            (!same_group && initial_offset + span_limit <= r.end) ||
+            end_past_cap ||
             (!same_group && !r.request->allow_incidental_read.load(std::memory_order_relaxed)))
             break;
 
diff --git a/tests/queries/0_stateless/05297_parquet_bloom_filter_coalesce.sh b/tests/queries/0_stateless/05297_parquet_bloom_filter_coalesce.sh
index d05ba981c5e02..cf8385c850e1d 100755
--- a/tests/queries/0_stateless/05297_parquet_bloom_filter_coalesce.sh
+++ b/tests/queries/0_stateless/05297_parquet_bloom_filter_coalesce.sh
@@ -21,7 +21,7 @@ ${CLICKHOUSE_LOCAL} -q "
 "
 
 # Every row group has min 0 and max 1000000, so min/max cannot skip. Value 42 exists in one row group.
-read_count()
+read_stats()
 {
     local coalesce_max=$1
     local min_seek=$2
@@ -39,35 +39,55 @@ read_count()
             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:/ { sum += $(NF-1) }
+        $0 ~ /ParquetPrefetcherReadRandomRead:/ { reads += $(NF-1) }
+        $0 ~ /ReadBufferFromFileDescriptorReadBytes:/ { bytes += $(NF-1) }
         $1 == "1" { rows = 1 }
         END {
             if (rows != 1) exit 2
-            print sum
+            print reads, bytes + 0
         }
     '
 }
 
 # min_bytes_for_seek = 1 disables the old gap coalescing, so each bloom filter is its own read.
-SEPARATE=$(read_count 0 1) || { echo "separate read failed"; exit 1; }
-CLUSTER=$(read_count 67108864 1) || { echo "cluster read failed"; exit 1; }
+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_count 67108864 1000000000) || { echo "whole-file read failed"; exit 1; }
+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_count 1 1000000) || { echo "capped read failed"; exit 1; }
+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; }
 
-if [ "$CLUSTER" -ge "$SEPARATE" ]; then
+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" -gt 4 ]; then
+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" -le 20 ]; then
+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