From a6be4fe2f0f1557c624c549eab03c87fa691614a Mon Sep 17 00:00:00 2001
From: Anton Ivashkin <ianton@live.com>
Date: Thu, 8 Oct 2026 18:53:48 +0200
Subject: [PATCH 01/15] 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/reference/formats/Parquet/Parquet.mdx    |  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/reference/formats/Parquet/Parquet.mdx b/docs/reference/formats/Parquet/Parquet.mdx
index ef6cd249d41dc..eb7d18da2844b 100644
--- a/docs/reference/formats/Parquet/Parquet.mdx
+++ b/docs/reference/formats/Parquet/Parquet.mdx
@@ -175,6 +175,7 @@ To exchange data with Hadoop, you can use the [`HDFS table engine`](/reference/e
 | `input_format_parquet_filter_push_down`                                        | When reading Parquet files, skip whole row groups based on the WHERE/PREWHERE expressions and min/max statistics in the Parquet metadata.                                                                                          | `1`         |
 | `input_format_parquet_bloom_filter_push_down`                                  | When reading Parquet files, skip whole row groups based on the WHERE expressions and bloom filter in the Parquet metadata.                                                                                                          | `0`         |
 | `input_format_parquet_dictionary_filter_push_down`                             | When reading Parquet files (with reader v3), skip whole row groups based on the WHERE/PREWHERE expressions and the dictionary page contents, for equality and `IN` conditions, when all data pages of a column chunk are dictionary-encoded. The value is the maximum dictionary page size (in bytes) for which this optimization is applied; set to `0` to disable. Takes precedence over the bloom filter when both are available. | `1048576`   |
+| `input_format_parquet_bloom_filter_coalesce_max_bytes`                         | Max bytes of one read when loading Parquet bloom filters. Contiguous filters are read together. A file not larger than this, with many filters, is read once. 0 keeps one read per bloom filter.                                      | `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 12fad80b961bf..ebd43facc1360 100644
--- a/src/Core/FormatFactorySettings.h
+++ b/src/Core/FormatFactorySettings.h
@@ -208,6 +208,9 @@ 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. 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 6e44ec5efbc01..d491a2fe191d9 100644
--- a/src/Core/SettingsChangesHistory.cpp
+++ b/src/Core/SettingsChangesHistory.cpp
@@ -222,6 +222,7 @@ const VersionToSettingsChangesMap & getSettingsChangesHistory()
             {"object_storage_propagate_credentials_to_other_storages", false, false, "New setting, allowing to reuse the base storage's `S3` credentials for a secondary object storage, even if the target endpoint differs from the base one. Disabled by default. Ignored while `s3_allow_server_credentials_in_user_queries` is disabled."},
             {"datalake_ignore_unsupported_table_properties", false, false, "New setting: allow `CREATE TABLE` and `SHOW CREATE TABLE` in a `DataLakeCatalog` to omit unsupported table properties."},
             {"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."},
         });
         addSettingsChanges(settings_changes_history, "26.6",
diff --git a/src/Formats/FormatFactory.cpp b/src/Formats/FormatFactory.cpp
index 34c81d515589a..a06b597549ed6 100644
--- a/src/Formats/FormatFactory.cpp
+++ b/src/Formats/FormatFactory.cpp
@@ -227,6 +227,7 @@ FormatSettings getFormatSettings(const ContextPtr & context, const Settings & se
     format_settings.parquet.filter_push_down = settings[Setting::input_format_parquet_filter_push_down];
     format_settings.parquet.bloom_filter_push_down = settings[Setting::input_format_parquet_bloom_filter_push_down];
     format_settings.parquet.dictionary_filter_push_down = settings[Setting::input_format_parquet_dictionary_filter_push_down];
+    format_settings.parquet.bloom_filter_coalesce_max_bytes = settings[Setting::input_format_parquet_bloom_filter_coalesce_max_bytes];
     format_settings.parquet.page_filter_push_down = settings[Setting::input_format_parquet_page_filter_push_down];
     format_settings.parquet.spatial_filter_push_down = settings[Setting::input_format_parquet_spatial_filter_push_down];
     format_settings.parquet.use_offset_index = settings[Setting::input_format_parquet_use_offset_index];
diff --git a/src/Formats/FormatSettings.h b/src/Formats/FormatSettings.h
index dfd95af1fd981..1bbe5e14b6d06 100644
--- a/src/Formats/FormatSettings.h
+++ b/src/Formats/FormatSettings.h
@@ -370,6 +370,8 @@ struct FormatSettings
         bool filter_push_down = true;
         bool bloom_filter_push_down = true;
         size_t dictionary_filter_push_down = 1024 * 1024;
+        /// 0 disables coalescing bloom-filter reads. See `input_format_parquet_bloom_filter_coalesce_max_bytes`.
+        size_t bloom_filter_coalesce_max_bytes = 64 * 1024 * 1024;
         bool page_filter_push_down = true;
         bool use_offset_index = true;
 
diff --git a/src/Processors/Formats/Impl/Parquet/Prefetcher.cpp b/src/Processors/Formats/Impl/Parquet/Prefetcher.cpp
index c673a467b7279..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 e92a183b0d79b..42672e685eecc 100644
--- a/src/Processors/Formats/Impl/Parquet/Reader.cpp
+++ b/src/Processors/Formats/Impl/Parquet/Reader.cpp
@@ -1022,6 +1022,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
         || format_filter_info->rows_to_read
         || std::any_of(primitive_columns.begin(), primitive_columns.end(), [](const auto & c) { return !c.column_index_conditions.empty(); });
@@ -1105,13 +1134,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.
@@ -1190,7 +1219,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 452a0a102c85d..53a69eba5ffc2 100644
--- a/src/Processors/Formats/Impl/ParquetV3BlockInputFormat.cpp
+++ b/src/Processors/Formats/Impl/ParquetV3BlockInputFormat.cpp
@@ -621,6 +621,7 @@ To exchange data with Hadoop, you can use the [`HDFS table engine`](/reference/e
 | `input_format_parquet_filter_push_down`                                        | When reading Parquet files, skip whole row groups based on the WHERE/PREWHERE expressions and min/max statistics in the Parquet metadata.                                                                                          | `1`         |
 | `input_format_parquet_bloom_filter_push_down`                                  | When reading Parquet files, skip whole row groups based on the WHERE expressions and bloom filter in the Parquet metadata.                                                                                                          | `0`         |
 | `input_format_parquet_dictionary_filter_push_down`                             | When reading Parquet files (with reader v3), skip whole row groups based on the WHERE/PREWHERE expressions and the dictionary page contents, for equality and `IN` conditions, when all data pages of a column chunk are dictionary-encoded. The value is the maximum dictionary page size (in bytes) for which this optimization is applied; set to `0` to disable. Takes precedence over the bloom filter when both are available. | `1048576`   |
+| `input_format_parquet_bloom_filter_coalesce_max_bytes`                         | Max bytes of one read when loading Parquet bloom filters. Contiguous filters are read together. A file not larger than this, with many filters, is read once. 0 keeps one read per bloom filter.                                      | `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 2f854dedeb184bab04a8ec1a867e1af74619d51d Mon Sep 17 00:00:00 2001
From: Anton Ivashkin <ianton@live.com>
Date: Fri, 9 Oct 2026 18:38:15 +0200
Subject: [PATCH 02/15] 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 20e1c113ae7be..8ada36fef5167 100644
--- a/src/Processors/Formats/Impl/Parquet/ReadManager.cpp
+++ b/src/Processors/Formats/Impl/Parquet/ReadManager.cpp
@@ -628,10 +628,12 @@ void ReadManager::scheduleTasksIfNeeded(ReadStage stage_idx)
         if (row_group_idx != i)
             return false;
         const RowGroup & row_group = reader.row_groups[row_group_idx];
-        /// Must check stage first so that read_ptr is meaningful (we start advancing it in finishRowSubgroupStage).
+        /// Before column data, read_ptr is not advanced yet. This row group still has to run: a coalesced
+        /// prefetch is charged until every handle that pins it has finished, including handles of later
+        /// row groups. Blocking this row group would leave that charge stuck.
         /// Using acquire ordering to synchronize with the release (seq_cst) store in `finishRowGroupStage`.
         if (row_group.stage.load(std::memory_order_acquire) < ReadStage::ColumnData)
-            return false;
+            return true;
         return row_group.read_ptr.load() == row_group.delivery_ptr.load();
     };
 

From 57636c8ba2294e9d1078afb0489b3244dd1928d7 Mon Sep 17 00:00:00 2001
From: Anton Ivashkin <ianton@live.com>
Date: Fri, 9 Oct 2026 18:46:05 +0200
Subject: [PATCH 03/15] 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 42672e685eecc..e521bc485d3e7 100644
--- a/src/Processors/Formats/Impl/Parquet/Reader.cpp
+++ b/src/Processors/Formats/Impl/Parquet/Reader.cpp
@@ -1037,7 +1037,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 32a3f575acdc4bb331c960d67256ab5c5d8027f5 Mon Sep 17 00:00:00 2001
From: Anton Ivashkin <ianton@live.com>
Date: Thu, 15 Oct 2026 12:50:22 +0200
Subject: [PATCH 04/15] 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 e521bc485d3e7..502c09942784f 100644
--- a/src/Processors/Formats/Impl/Parquet/Reader.cpp
+++ b/src/Processors/Formats/Impl/Parquet/Reader.cpp
@@ -1037,6 +1037,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 77c539357c531139605a7bea699d44318b01634a Mon Sep 17 00:00:00 2001
From: Anton Ivashkin <ianton@live.com>
Date: Thu, 15 Oct 2026 13:02:00 +0200
Subject: [PATCH 05/15] 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             |  7 ++++++-
 ...loom_filter_coalesce_compatibility.reference |  5 +++++
 ...quet_bloom_filter_coalesce_compatibility.sql | 17 +++++++++++++++++
 3 files changed, 28 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 d491a2fe191d9..f0d0563b757ae 100644
--- a/src/Core/SettingsChangesHistory.cpp
+++ b/src/Core/SettingsChangesHistory.cpp
@@ -42,6 +42,12 @@ const VersionToSettingsChangesMap & getSettingsChangesHistory()
         /// 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",
+        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",
         {
             {"allow_experimental_ai_functions", false, false, "The setting is obsolete, AI functions are beta now and enabled by default."},
             {"ai_function_max_retries", 0, 1, "Retry a transient API error once by default, so a single 429 or 5xx from the provider does not fail the query."},
@@ -222,7 +228,6 @@ const VersionToSettingsChangesMap & getSettingsChangesHistory()
             {"object_storage_propagate_credentials_to_other_storages", false, false, "New setting, allowing to reuse the base storage's `S3` credentials for a secondary object storage, even if the target endpoint differs from the base one. Disabled by default. Ignored while `s3_allow_server_credentials_in_user_queries` is disabled."},
             {"datalake_ignore_unsupported_table_properties", false, false, "New setting: allow `CREATE TABLE` and `SHOW CREATE TABLE` in a `DataLakeCatalog` to omit unsupported table properties."},
             {"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."},
         });
         addSettingsChanges(settings_changes_history, "26.6",
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 9226b751c3ce80d7f36a02e20538404ef5534775 Mon Sep 17 00:00:00 2001
From: Anton Ivashkin <ianton@live.com>
Date: Thu, 15 Oct 2026 13:06:16 +0200
Subject: [PATCH 06/15] 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 a54aa46849062751232610a68a3277551ab60c89 Mon Sep 17 00:00:00 2001
From: Anton Ivashkin <ianton@live.com>
Date: Fri, 16 Oct 2026 14:39:49 +0200
Subject: [PATCH 07/15] 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

From bd0f81927be5afb41c00090b212c6add11c6ab1a Mon Sep 17 00:00:00 2001
From: Anton Ivashkin <ianton@live.com>
Date: Fri, 16 Oct 2026 15:43:55 +0200
Subject: [PATCH 08/15] Record
 `input_format_parquet_bloom_filter_coalesce_max_bytes` in 26.8.6 so older
 `compatibility` values restore separate reads, and count only the bloom
 ranges `Reader` registers.

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 src/Core/SettingsChangesHistory.cpp           |  6 ++--
 .../Formats/Impl/Parquet/Reader.cpp           | 30 +++++++------------
 ...et_bloom_filter_coalesce_compatibility.sql | 10 +++----
 3 files changed, 17 insertions(+), 29 deletions(-)

diff --git a/src/Core/SettingsChangesHistory.cpp b/src/Core/SettingsChangesHistory.cpp
index f0d0563b757ae..fb395477545c2 100644
--- a/src/Core/SettingsChangesHistory.cpp
+++ b/src/Core/SettingsChangesHistory.cpp
@@ -41,13 +41,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.8",
-        addSettingsChanges(settings_changes_history, "26.6.8.20001.altinityantalya",
+        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.6.4.20001.altinityantalya",
+        addSettingsChanges(settings_changes_history, "26.8",
         {
             {"allow_experimental_ai_functions", false, false, "The setting is obsolete, AI functions are beta now and enabled by default."},
             {"ai_function_max_retries", 0, 1, "Retry a transient API error once by default, so a single 429 or 5xx from the provider does not fail the query."},
diff --git a/src/Processors/Formats/Impl/Parquet/Reader.cpp b/src/Processors/Formats/Impl/Parquet/Reader.cpp
index 502c09942784f..b47dbaefd76b9 100644
--- a/src/Processors/Formats/Impl/Parquet/Reader.cpp
+++ b/src/Processors/Formats/Impl/Parquet/Reader.cpp
@@ -1025,7 +1025,7 @@ 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)
+    if (coalesce_max > 0 && options.format.parquet.bloom_filter_push_down)
     {
         size_t bloom_count = 0;
         for (const RowGroup & row_group : row_groups)
@@ -1034,27 +1034,17 @@ void Reader::initializePrefetches()
                 continue;
             for (size_t column_idx = 0; column_idx < primitive_columns.size(); ++column_idx)
             {
-                if (!primitive_columns[column_idx].use_bloom_filter)
+                /// Same gates as the bloom-filter registration below. A dictionary-filtered chunk still
+                /// reads its bloom filter as a fallback, so it counts. A chunk that will not register
+                /// a range must not turn on a whole-file read.
+                const auto & column_info = primitive_columns[column_idx];
+                if (!column_info.use_bloom_filter || column_info.bloom_filter_hashes.empty())
                     continue;
                 const auto & meta = row_group.columns[column_idx].meta->meta_data;
-                /// 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;
-                }
+                const bool nullable = column_info.levels.back().def > 0;
+                const bool can_be_null = !meta.statistics.__isset.null_count || meta.statistics.null_count != 0;
+                if (nullable && can_be_null && !column_info.output_nullable)
+                    continue;
                 /// `bloom_filter_length` is optional (Parquet 2.10). An offset without it is still a
                 /// filter; the hard-way path below derives the length and registers that range.
                 const bool length_missing_or_positive = !meta.__isset.bloom_filter_length || meta.bloom_filter_length > 0;
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
index 3b56c3a992cdb..d968fecb79196 100644
--- a/tests/queries/0_stateless/05304_parquet_bloom_filter_coalesce_compatibility.sql
+++ b/tests/queries/0_stateless/05304_parquet_bloom_filter_coalesce_compatibility.sql
@@ -1,16 +1,16 @@
 -- 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).
+-- `input_format_parquet_bloom_filter_coalesce_max_bytes` is new in 26.8.6 Antalya.
+-- `compatibility` with 26.8, 26.7, or 26.6 must restore 0 (one read per bloom filter).
 
 SELECT getSetting('input_format_parquet_bloom_filter_coalesce_max_bytes');
 
-SET compatibility = '26.6.8.20001.altinityantalya';
+SET compatibility = '26.8.6.20001.altinityantalya';
 SELECT getSetting('input_format_parquet_bloom_filter_coalesce_max_bytes');
 
-SET compatibility = '26.6.4.20001.altinityantalya';
+SET compatibility = '26.8';
 SELECT getSetting('input_format_parquet_bloom_filter_coalesce_max_bytes');
 
-SET compatibility = '26.6.2.20001.altinityantalya';
+SET compatibility = '26.7';
 SELECT getSetting('input_format_parquet_bloom_filter_coalesce_max_bytes');
 
 SET compatibility = '26.6';

From 8b8061d20800db9cbedb3f631e227323e7a3bbed Mon Sep 17 00:00:00 2001
From: Anton Ivashkin <ianton@live.com>
Date: Wed, 21 Oct 2026 12:41:27 +0200
Subject: [PATCH 09/15] Leave the generated Parquet page unchanged. The
 `input_format_parquet_bloom_filter_coalesce_max_bytes` row stays in the
 `Documentation` source.

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 docs/reference/formats/Parquet/Parquet.mdx | 1 -
 1 file changed, 1 deletion(-)

diff --git a/docs/reference/formats/Parquet/Parquet.mdx b/docs/reference/formats/Parquet/Parquet.mdx
index eb7d18da2844b..ef6cd249d41dc 100644
--- a/docs/reference/formats/Parquet/Parquet.mdx
+++ b/docs/reference/formats/Parquet/Parquet.mdx
@@ -175,7 +175,6 @@ To exchange data with Hadoop, you can use the [`HDFS table engine`](/reference/e
 | `input_format_parquet_filter_push_down`                                        | When reading Parquet files, skip whole row groups based on the WHERE/PREWHERE expressions and min/max statistics in the Parquet metadata.                                                                                          | `1`         |
 | `input_format_parquet_bloom_filter_push_down`                                  | When reading Parquet files, skip whole row groups based on the WHERE expressions and bloom filter in the Parquet metadata.                                                                                                          | `0`         |
 | `input_format_parquet_dictionary_filter_push_down`                             | When reading Parquet files (with reader v3), skip whole row groups based on the WHERE/PREWHERE expressions and the dictionary page contents, for equality and `IN` conditions, when all data pages of a column chunk are dictionary-encoded. The value is the maximum dictionary page size (in bytes) for which this optimization is applied; set to `0` to disable. Takes precedence over the bloom filter when both are available. | `1048576`   |
-| `input_format_parquet_bloom_filter_coalesce_max_bytes`                         | Max bytes of one read when loading Parquet bloom filters. Contiguous filters are read together. A file not larger than this, with many filters, is read once. 0 keeps one read per bloom filter.                                      | `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`         |

From 24138136ea14cbd22191f94bf86c053ef266fad9 Mon Sep 17 00:00:00 2001
From: Anton Ivashkin <ianton@live.com>
Date: Wed, 21 Oct 2026 12:44:26 +0200
Subject: [PATCH 10/15] Keep a Parquet bloom-filter read out of a neighboring
 column read, whichever range starts the coalescing scan.

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

diff --git a/src/Processors/Formats/Impl/Parquet/Prefetcher.cpp b/src/Processors/Formats/Impl/Parquet/Prefetcher.cpp
index 693f2ada97a61..b9ab46abb202d 100644
--- a/src/Processors/Formats/Impl/Parquet/Prefetcher.cpp
+++ b/src/Processors/Formats/Impl/Parquet/Prefetcher.cpp
@@ -351,7 +351,7 @@ void Prefetcher::pickRangesAndCreateTaskIfNotExists(RequestState * initial_req,
     {
         const RangeState & r = ranges[idx - 1];
         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.
+        /// 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.
@@ -360,7 +360,9 @@ void Prefetcher::pickRangesAndCreateTaskIfNotExists(RequestState * initial_req,
         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) ||
+        const bool group_boundary = initial_req->coalesce_group != r.request->coalesce_group
+            && (initial_req->coalesce_group != 0 || r.request->coalesce_group != 0);
+        if (group_boundary ||
             r.end + gap_limit <= start_offset || // short gap
             r.start + span_limit <= initial_offset || // task not too big
             end_past_cap ||
@@ -403,7 +405,9 @@ void Prefetcher::pickRangesAndCreateTaskIfNotExists(RequestState * initial_req,
         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) ||
+        const bool group_boundary = initial_req->coalesce_group != r.request->coalesce_group
+            && (initial_req->coalesce_group != 0 || r.request->coalesce_group != 0);
+        if (group_boundary ||
             end_offset + gap_limit <= r.start ||
             (!same_group && initial_offset + span_limit <= r.end) ||
             end_past_cap ||

From 0eb108c35a3586f37ace5676b66c3e63eb315aad Mon Sep 17 00:00:00 2001
From: Anton Ivashkin <ianton@live.com>
Date: Wed, 21 Oct 2026 13:27:43 +0200
Subject: [PATCH 11/15] Keep back-to-back Parquet bloom filters in one read
 when `min_bytes_for_seek` is 0.

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

diff --git a/src/Processors/Formats/Impl/Parquet/Prefetcher.cpp b/src/Processors/Formats/Impl/Parquet/Prefetcher.cpp
index b9ab46abb202d..34b7ff7200112 100644
--- a/src/Processors/Formats/Impl/Parquet/Prefetcher.cpp
+++ b/src/Processors/Formats/Impl/Parquet/Prefetcher.cpp
@@ -362,8 +362,10 @@ void Prefetcher::pickRangesAndCreateTaskIfNotExists(RequestState * initial_req,
         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 ||
-            r.end + gap_limit <= start_offset || // short gap
+            (positive_gap && 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
@@ -407,8 +409,10 @@ void Prefetcher::pickRangesAndCreateTaskIfNotExists(RequestState * initial_req,
         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 ||
-            end_offset + gap_limit <= r.start ||
+            (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)))
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 cf8385c850e1d..3ce66a6cd5f9f 100755
--- a/tests/queries/0_stateless/05297_parquet_bloom_filter_coalesce.sh
+++ b/tests/queries/0_stateless/05297_parquet_bloom_filter_coalesce.sh
@@ -52,6 +52,8 @@ read_stats()
 # min_bytes_for_seek = 1 disables the old gap coalescing, so each bloom filter is its own read.
 SEPARATE=$(read_stats 0 1) || { echo "separate read failed"; exit 1; }
 CLUSTER=$(read_stats 67108864 1) || { echo "cluster read failed"; exit 1; }
+# Seek of 0 used to treat a zero-byte gap as a split. ClickHouse writes the filters back-to-back.
+TOUCH=$(read_stats 67108864 0) || { echo "zero-seek coalesce failed"; exit 1; }
 # Enough filters that the seek budget covers the file: one read of the whole object.
 WHOLE=$(read_stats 67108864 1000000000) || { echo "whole-file read failed"; exit 1; }
 # Cap of 1 byte, seek of 1 MiB: bytes_per_read_task is 4 MiB and would swallow this file if the cap were raised.
@@ -62,6 +64,7 @@ NARROW=$(read_stats 512 1) || { echo "narrow read failed"; exit 1; }
 SEPARATE_READS=${SEPARATE%% *}
 SEPARATE_BYTES=${SEPARATE##* }
 CLUSTER_READS=${CLUSTER%% *}
+TOUCH_READS=${TOUCH%% *}
 WHOLE_READS=${WHOLE%% *}
 CAPPED_READS=${CAPPED%% *}
 NARROW_READS=${NARROW%% *}
@@ -71,6 +74,10 @@ if [ "$CLUSTER_READS" -ge "$SEPARATE_READS" ]; then
     echo "cluster coalesce did not reduce reads: cluster=$CLUSTER separate=$SEPARATE"
     exit 1
 fi
+if [ "$TOUCH_READS" -ge "$SEPARATE_READS" ]; then
+    echo "zero seek split back-to-back filters: touch=$TOUCH separate=$SEPARATE cluster=$CLUSTER"
+    exit 1
+fi
 if [ "$WHOLE_READS" -gt 4 ]; then
     echo "whole-file read used too many requests: whole=$WHOLE cluster=$CLUSTER separate=$SEPARATE"
     exit 1

From b5be89948e04710ef5e8cfcb2fe62cd39eefffad Mon Sep 17 00:00:00 2001
From: Anton Ivashkin <ianton@live.com>
Date: Wed, 21 Oct 2026 13:52:41 +0200
Subject: [PATCH 12/15] Keep a max Parquet bloom coalesce cap from splitting a
 filter, and skip a filter for a row group that produces no rows.

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

diff --git a/src/Core/SettingsChangesHistory.cpp b/src/Core/SettingsChangesHistory.cpp
index db630336f9480..c07524d09ee57 100644
--- a/src/Core/SettingsChangesHistory.cpp
+++ b/src/Core/SettingsChangesHistory.cpp
@@ -44,7 +44,7 @@ const VersionToSettingsChangesMap & getSettingsChangesHistory()
         addSettingsChanges(settings_changes_history, "26.8.6.20001.altinityantalya",
         {
             {"input_format_parquet_bloom_filter_coalesce_max_bytes", 0, 64 * 1024 * 1024, "Read contiguous Parquet bloom filters, or a small file that has many of them, in one request."},
-        });         
+        });
         addSettingsChanges(settings_changes_history, "26.10",
         {
             {"azure_validate_etag_on_read", false, true, "New setting to detect concurrent in-place overwrites of Azure blobs during a read by pinning every `GET` to the listed `ETag` with `If-Match` and validating the `ETag` of the response, like `s3_validate_etag_on_read` does for S3. `compatibility` with versions before 26.10 restores the previous behavior (no validation)."},
diff --git a/src/Processors/Formats/Impl/Parquet/Prefetcher.cpp b/src/Processors/Formats/Impl/Parquet/Prefetcher.cpp
index 34b7ff7200112..32b2f978b37e8 100644
--- a/src/Processors/Formats/Impl/Parquet/Prefetcher.cpp
+++ b/src/Processors/Formats/Impl/Parquet/Prefetcher.cpp
@@ -366,7 +366,8 @@ void Prefetcher::pickRangesAndCreateTaskIfNotExists(RequestState * initial_req,
         const bool positive_gap = same_group ? r.end < start_offset : true;
         if (group_boundary ||
             (positive_gap && r.end + gap_limit <= start_offset) || // short gap
-            r.start + span_limit <= initial_offset || // task not too big
+            /// `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;
diff --git a/src/Processors/Formats/Impl/Parquet/Reader.cpp b/src/Processors/Formats/Impl/Parquet/Reader.cpp
index b1891ff43e75d..1cbbe4b2e94e7 100644
--- a/src/Processors/Formats/Impl/Parquet/Reader.cpp
+++ b/src/Processors/Formats/Impl/Parquet/Reader.cpp
@@ -1138,7 +1138,10 @@ void Reader::initializePrefetches()
                 || column.meta->meta_data.statistics.null_count != 0;
             bool bloom_filter_null_safe = !nullable || !can_be_null || primitive_columns[column_idx].output_nullable;
 
-            if (options.format.parquet.bloom_filter_push_down &&
+            /// A row group outside the requested subset produces no rows. Do not register its filter:
+            /// the count above skips it, and a registered filter would still be read.
+            if (row_group.need_to_process &&
+                options.format.parquet.bloom_filter_push_down &&
                 primitive_columns[column_idx].use_bloom_filter &&
                 !primitive_columns[column_idx].bloom_filter_hashes.empty() &&
                 bloom_filter_null_safe &&
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 3ce66a6cd5f9f..eed6c2b7df127 100755
--- a/tests/queries/0_stateless/05297_parquet_bloom_filter_coalesce.sh
+++ b/tests/queries/0_stateless/05297_parquet_bloom_filter_coalesce.sh
@@ -54,6 +54,8 @@ SEPARATE=$(read_stats 0 1) || { echo "separate read failed"; exit 1; }
 CLUSTER=$(read_stats 67108864 1) || { echo "cluster read failed"; exit 1; }
 # Seek of 0 used to treat a zero-byte gap as a split. ClickHouse writes the filters back-to-back.
 TOUCH=$(read_stats 67108864 0) || { echo "zero-seek coalesce failed"; exit 1; }
+# UInt64 max used to wrap the left-scan cap and split a filter header from its body.
+UNLIMITED=$(read_stats 18446744073709551615 0) || { echo "unlimited coalesce failed"; exit 1; }
 # Enough filters that the seek budget covers the file: one read of the whole object.
 WHOLE=$(read_stats 67108864 1000000000) || { echo "whole-file read failed"; exit 1; }
 # Cap of 1 byte, seek of 1 MiB: bytes_per_read_task is 4 MiB and would swallow this file if the cap were raised.
@@ -65,6 +67,7 @@ SEPARATE_READS=${SEPARATE%% *}
 SEPARATE_BYTES=${SEPARATE##* }
 CLUSTER_READS=${CLUSTER%% *}
 TOUCH_READS=${TOUCH%% *}
+UNLIMITED_READS=${UNLIMITED%% *}
 WHOLE_READS=${WHOLE%% *}
 CAPPED_READS=${CAPPED%% *}
 NARROW_READS=${NARROW%% *}
@@ -78,6 +81,10 @@ if [ "$TOUCH_READS" -ge "$SEPARATE_READS" ]; then
     echo "zero seek split back-to-back filters: touch=$TOUCH separate=$SEPARATE cluster=$CLUSTER"
     exit 1
 fi
+if [ "$UNLIMITED_READS" -ge "$SEPARATE_READS" ]; then
+    echo "max coalesce cap split the bloom header from its body: unlimited=$UNLIMITED separate=$SEPARATE touch=$TOUCH"
+    exit 1
+fi
 if [ "$WHOLE_READS" -gt 4 ]; then
     echo "whole-file read used too many requests: whole=$WHOLE cluster=$CLUSTER separate=$SEPARATE"
     exit 1

From ac5724ff5d08b65b1f3efdb0db77eb4fdfd49ab5 Mon Sep 17 00:00:00 2001
From: Anton Ivashkin <ianton@live.com>
Date: Wed, 21 Oct 2026 15:56:57 +0200
Subject: [PATCH 13/15] Measure Parquet bloom-filter S3 GETs without a
 dictionary-page read per row group.

https: //altinity-build-artifacts.s3.amazonaws.com/json.html?PR=2588&sha=6e16e1d87e4b679d422fd7bd4b4f35df4e0bd4be&name_0=PR&name_1=Stateless%20tests%20%28amd_asan_ubsan%2C%20distributed%20plan%2C%20parallel%2C%20selected%20tests%29
https: //github.com/Altinity/ClickHouse/pull/2588
Co-authored-by: Cursor <cursoragent@cursor.com>
---
 tests/queries/0_stateless/05298_parquet_bloom_filter_s3_gets.sh | 2 ++
 1 file changed, 2 insertions(+)

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
index 69a9ba6ccc97d..e6320a3a6860b 100755
--- a/tests/queries/0_stateless/05298_parquet_bloom_filter_s3_gets.sh
+++ b/tests/queries/0_stateless/05298_parquet_bloom_filter_s3_gets.sh
@@ -5,6 +5,7 @@
 # A small object is otherwise prefetched in one request, which hides the per-range GETs.
 # max_download_buffer_size keeps that prefetch off. remote_read_min_bytes_for_seek is what the
 # Parquet reader uses as the gap it will still merge, so 1 byte leaves merging to bloom-filter coalescing.
+# Dictionary-filter pushdown is off: each dictionary page is its own GET and is not part of the bloom-filter read.
 
 CUR_DIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
 # shellcheck source=../shell_config.sh
@@ -44,6 +45,7 @@ s3_gets()
             input_format_parquet_bloom_filter_push_down = 1,
             input_format_parquet_page_filter_push_down = 0,
             input_format_parquet_use_offset_index = 0,
+            input_format_parquet_dictionary_filter_push_down = 0,
             input_format_parquet_bloom_filter_coalesce_max_bytes = ${coalesce_max}
     " 2>&1 | awk '
         $0 ~ /S3GetObject:/ { gets += $(NF-1) }

From bc53758fa3634f5ecf76cf56bf7963a3d0ba9e94 Mon Sep 17 00:00:00 2001
From: Anton Ivashkin <ianton@live.com>
Date: Thu, 22 Oct 2026 13:32:48 +0200
Subject: [PATCH 14/15] Keep a dictionary-filter fallback Parquet bloom body
 out of the header read.

The header and the body were in one coalesce group, so the header read downloaded the body before the dictionary filter was tried.

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 .../Formats/Impl/Parquet/Reader.cpp           | 11 ++-
 .../05297_parquet_bloom_filter_coalesce.sh    |  2 +
 ...bloom_filter_dictionary_fallback.reference |  1 +
 ...arquet_bloom_filter_dictionary_fallback.sh | 80 +++++++++++++++++++
 4 files changed, 92 insertions(+), 2 deletions(-)
 create mode 100644 tests/queries/0_stateless/05386_parquet_bloom_filter_dictionary_fallback.reference
 create mode 100755 tests/queries/0_stateless/05386_parquet_bloom_filter_dictionary_fallback.sh

diff --git a/src/Processors/Formats/Impl/Parquet/Reader.cpp b/src/Processors/Formats/Impl/Parquet/Reader.cpp
index 1cbbe4b2e94e7..c71cc729f5a5c 100644
--- a/src/Processors/Formats/Impl/Parquet/Reader.cpp
+++ b/src/Processors/Formats/Impl/Parquet/Reader.cpp
@@ -1156,9 +1156,14 @@ void Reader::initializePrefetches()
                     size_t len = size_t(column.meta->meta_data.bloom_filter_length);
                     max_header_length = std::min(max_header_length, len);
                     column.bloom_filter_data_bytes = len;
+                    /// The body is only a fallback when the dictionary filter can still rule the chunk out.
+                    /// Leave it out of the bloom coalesce group: the header is read first and would
+                    /// otherwise download the body before that filter is tried. A whole-file read still
+                    /// assigns every range to the group.
+                    const uint8_t bloom_data_group = column.use_dictionary_filter ? uint8_t{0} : bloom_group;
                     column.bloom_filter_data_prefetch = prefetcher.registerRange(
                         size_t(column.meta->meta_data.bloom_filter_offset),
-                        len, /*likely_to_be_used=*/ false, bloom_group);
+                        len, /*likely_to_be_used=*/ false, bloom_data_group);
                 }
                 /// bloom_filter_header_prefetch and bloom_filter_data_prefetch overlap, that's ok.
                 column.use_bloom_filter = true;
@@ -1242,8 +1247,10 @@ void Reader::initializePrefetches()
                 size_t end = it == all_offsets.end() ? prefetcher.getFileSize() : *it;
 
                 column.bloom_filter_data_bytes = end - offset;
+                /// Same group rule as the length-known path above.
+                const uint8_t bloom_data_group = column.use_dictionary_filter ? uint8_t{0} : bloom_group;
                 column.bloom_filter_data_prefetch = prefetcher.registerRange(
-                    offset, end - offset, /*likely_to_be_used=*/ false, bloom_group);
+                    offset, end - offset, /*likely_to_be_used=*/ false, bloom_data_group);
             }
         }
     }
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 eed6c2b7df127..2437c1a21368b 100755
--- a/tests/queries/0_stateless/05297_parquet_bloom_filter_coalesce.sh
+++ b/tests/queries/0_stateless/05297_parquet_bloom_filter_coalesce.sh
@@ -2,6 +2,7 @@
 # Tags: no-fasttest
 
 # Many row groups, bloom filters flushed as one block. A selective WHERE must not issue one read per row group.
+# Dictionary-filter pushdown is off: a dictionary-eligible bloom body is a fallback and stays out of this read.
 
 CUR_DIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
 # shellcheck source=../shell_config.sh
@@ -36,6 +37,7 @@ read_stats()
             input_format_parquet_bloom_filter_push_down = 1,
             input_format_parquet_page_filter_push_down = 0,
             input_format_parquet_use_offset_index = 0,
+            input_format_parquet_dictionary_filter_push_down = 0,
             input_format_parquet_local_file_min_bytes_for_seek = ${min_seek},
             input_format_parquet_bloom_filter_coalesce_max_bytes = ${coalesce_max}
     " 2>&1 | awk '
diff --git a/tests/queries/0_stateless/05386_parquet_bloom_filter_dictionary_fallback.reference b/tests/queries/0_stateless/05386_parquet_bloom_filter_dictionary_fallback.reference
new file mode 100644
index 0000000000000..d86bac9de59ab
--- /dev/null
+++ b/tests/queries/0_stateless/05386_parquet_bloom_filter_dictionary_fallback.reference
@@ -0,0 +1 @@
+OK
diff --git a/tests/queries/0_stateless/05386_parquet_bloom_filter_dictionary_fallback.sh b/tests/queries/0_stateless/05386_parquet_bloom_filter_dictionary_fallback.sh
new file mode 100755
index 0000000000000..f7914049e14c9
--- /dev/null
+++ b/tests/queries/0_stateless/05386_parquet_bloom_filter_dictionary_fallback.sh
@@ -0,0 +1,80 @@
+#!/usr/bin/env bash
+# Tags: no-fasttest
+
+# A dictionary-encoded chunk keeps its bloom filter as a fallback. Coalescing the header must
+# not swallow that body. Min/max cannot skip: every row group contains 0 and 1000000.
+# `file` reads go through mmap here, so the check is the random-read count, not bytes.
+
+CUR_DIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
+# shellcheck source=../shell_config.sh
+. "$CUR_DIR"/../shell_config.sh
+
+WORKING_DIR="${CLICKHOUSE_TMP}/${CLICKHOUSE_TEST_UNIQUE_NAME}"
+mkdir -p "${WORKING_DIR}"
+FILE_PATH="${WORKING_DIR}/bloom.parquet"
+
+${CLICKHOUSE_LOCAL} -q "
+    INSERT INTO FUNCTION file('${FILE_PATH}', Parquet)
+    SELECT if(number % 30000 = 0, toUInt32(0),
+              if(number % 30000 = 1, toUInt32(1000000),
+                 toUInt32(intDiv(number, 30000) * 30000 + (number % 30000)))) AS k
+    FROM numbers(240000)
+    SETTINGS
+        output_format_parquet_row_group_size = 30000,
+        output_format_parquet_write_bloom_filter = 1
+"
+
+# Prints "<random reads> <bytes read>". Value 42 is in the first row group only.
+read_stats()
+{
+    local coalesce_max=$1
+    local dict_limit=$2
+    ${CLICKHOUSE_LOCAL} --print-profile-events --profile-events-delay-ms=-1 -q "
+        SELECT count()
+        FROM file('${FILE_PATH}', Parquet)
+        WHERE k = 42
+        SETTINGS
+            max_threads = 1,
+            use_cache_for_count_from_files = 0,
+            input_format_parquet_filter_push_down = 1,
+            input_format_parquet_bloom_filter_push_down = 1,
+            input_format_parquet_page_filter_push_down = 0,
+            input_format_parquet_use_offset_index = 0,
+            input_format_parquet_local_file_min_bytes_for_seek = 1,
+            input_format_parquet_dictionary_filter_push_down = ${dict_limit},
+            input_format_parquet_bloom_filter_coalesce_max_bytes = ${coalesce_max}
+    " 2>&1 | awk '
+        $0 ~ /ParquetPrefetcherReadRandomRead:/ { reads += $(NF-1) }
+        $0 ~ /ReadBufferFromFileDescriptorReadBytes:/ { bytes += $(NF-1) }
+        $1 == "1" { rows = 1 }
+        END {
+            if (rows != 1) exit 2
+            print reads, bytes + 0
+        }
+    '
+}
+
+# Dictionary filter on. Coalescing must not pull the fallback bloom bodies into the header read.
+DICT_SEPARATE=$(read_stats 0 1048576) || { echo "dictionary separate read failed"; exit 1; }
+DICT_CLUSTER=$(read_stats 67108864 1048576) || { echo "dictionary cluster read failed"; exit 1; }
+# Dictionary filter off. The same bodies are the bloom read and must still coalesce.
+BLOOM_SEPARATE=$(read_stats 0 0) || { echo "bloom separate read failed"; exit 1; }
+BLOOM_CLUSTER=$(read_stats 67108864 0) || { echo "bloom cluster read failed"; exit 1; }
+
+DICT_SEPARATE_READS=${DICT_SEPARATE%% *}
+DICT_CLUSTER_READS=${DICT_CLUSTER%% *}
+BLOOM_SEPARATE_READS=${BLOOM_SEPARATE%% *}
+BLOOM_CLUSTER_READS=${BLOOM_CLUSTER%% *}
+
+# The fallback bodies sit between the headers. Absorbing them turns those headers into one read.
+if [ "$DICT_CLUSTER_READS" -lt "$DICT_SEPARATE_READS" ]; then
+    echo "dictionary fallback bloom bodies were read with the header: dict_cluster=$DICT_CLUSTER dict_separate=$DICT_SEPARATE bloom_cluster=$BLOOM_CLUSTER bloom_separate=$BLOOM_SEPARATE"
+    exit 1
+fi
+if [ "$BLOOM_CLUSTER_READS" -ge "$BLOOM_SEPARATE_READS" ]; then
+    echo "bloom coalesce did not reduce reads: bloom_cluster=$BLOOM_CLUSTER bloom_separate=$BLOOM_SEPARATE dict_cluster=$DICT_CLUSTER dict_separate=$DICT_SEPARATE"
+    exit 1
+fi
+
+rm -rf "${WORKING_DIR}"
+echo OK

From 89a7aee60e31a5687cc3047387e1a36529001173 Mon Sep 17 00:00:00 2001
From: Anton Ivashkin <ianton@live.com>
Date: Thu, 22 Oct 2026 14:18:22 +0200
Subject: [PATCH 15/15] Document how
 `input_format_parquet_bloom_filter_coalesce_max_bytes` differs from
 `remote_read_min_bytes_for_seek`.

The seek threshold can merge small adjacent filters, but it also applies to other Parquet ranges and has no bloom-only cap.

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 src/Core/FormatFactorySettings.h                  | 13 ++++++++++++-
 .../Formats/Impl/ParquetV3BlockInputFormat.cpp    | 15 ++++++++++++++-
 2 files changed, 26 insertions(+), 2 deletions(-)

diff --git a/src/Core/FormatFactorySettings.h b/src/Core/FormatFactorySettings.h
index e1c5dc76c971c..c9ee5a524f977 100644
--- a/src/Core/FormatFactorySettings.h
+++ b/src/Core/FormatFactorySettings.h
@@ -210,7 +210,18 @@ When reading Parquet files, skip whole row groups based on the WHERE expressions
 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. 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.
+Max bytes of one read when loading Parquet bloom filters.
+
+ClickHouse writes those filters back-to-back, after the row-group data, so nothing sits between neighboring filters. Filters in that block are read together, up to this many bytes. If the file itself is not larger than this value and it has enough filters that separate reads would cost more than one read of the file, the whole file is read once and later column reads reuse that buffer. `0` keeps one read per bloom filter.
+
+This is separate from `remote_read_min_bytes_for_seek` (and from `input_format_parquet_local_file_min_bytes_for_seek` for a local file):
+
+- `remote_read_min_bytes_for_seek` merges a neighboring range only when that range is shorter than the seek threshold. A bloom filter longer than the threshold stays its own read.
+- Raising that threshold far enough to cover a large bloom filter also applies to dictionary pages, indexes, and column data, so unrelated bytes are read with the filters.
+- An ordinary coalesced read may grow to four times the seek threshold. This setting is its own cap, and it applies only to bloom-filter ranges.
+- Enough filters in a file no larger than this setting select one read of the whole file. The seek threshold has no such rule.
+
+A bloom body kept only as a fallback for the dictionary filter is not part of this read. Its header is still read.
 )", 0) \
     DECLARE(Bool, input_format_parquet_enable_json_parsing, true, R"(
 When reading Parquet files, parse JSON columns as ClickHouse JSON Column.
diff --git a/src/Processors/Formats/Impl/ParquetV3BlockInputFormat.cpp b/src/Processors/Formats/Impl/ParquetV3BlockInputFormat.cpp
index 53a69eba5ffc2..2f719f00dd70f 100644
--- a/src/Processors/Formats/Impl/ParquetV3BlockInputFormat.cpp
+++ b/src/Processors/Formats/Impl/ParquetV3BlockInputFormat.cpp
@@ -621,7 +621,7 @@ To exchange data with Hadoop, you can use the [`HDFS table engine`](/reference/e
 | `input_format_parquet_filter_push_down`                                        | When reading Parquet files, skip whole row groups based on the WHERE/PREWHERE expressions and min/max statistics in the Parquet metadata.                                                                                          | `1`         |
 | `input_format_parquet_bloom_filter_push_down`                                  | When reading Parquet files, skip whole row groups based on the WHERE expressions and bloom filter in the Parquet metadata.                                                                                                          | `0`         |
 | `input_format_parquet_dictionary_filter_push_down`                             | When reading Parquet files (with reader v3), skip whole row groups based on the WHERE/PREWHERE expressions and the dictionary page contents, for equality and `IN` conditions, when all data pages of a column chunk are dictionary-encoded. The value is the maximum dictionary page size (in bytes) for which this optimization is applied; set to `0` to disable. Takes precedence over the bloom filter when both are available. | `1048576`   |
-| `input_format_parquet_bloom_filter_coalesce_max_bytes`                         | Max bytes of one read when loading Parquet bloom filters. Contiguous filters are read together. A file not larger than this, with many filters, is read once. 0 keeps one read per bloom filter.                                      | `67108864`  |
+| `input_format_parquet_bloom_filter_coalesce_max_bytes`                         | Max bytes of one read when loading Parquet bloom filters. Contiguous filters are read together. A file not larger than this, with many filters, is read once. `0` keeps one read per bloom filter. Separate from `remote_read_min_bytes_for_seek`; see [Bloom filter reads](#bloom-filter-reads). | `67108864`  |
 | `input_format_parquet_allow_missing_columns`                                   | Allow missing columns while reading Parquet input formats                                                                                                                                                                          | `1`         |
 | `input_format_parquet_local_file_min_bytes_for_seek`                           | Min bytes required for local read (file) to do seek, instead of read with ignore in Parquet input format                                                                                                                          | `8192`      |
 | `input_format_parquet_enable_row_group_prefetch`                               | Enable row group prefetching during parquet parsing. Currently, only single-threaded parsing can prefetch.                                                                                                                          | `1`         |
@@ -643,6 +643,19 @@ To exchange data with Hadoop, you can use the [`HDFS table engine`](/reference/e
 | `output_format_parquet_geometadata`                                            | Write GeoParquet `geo` metadata into the Parquet file footer and encode top-level ClickHouse geo columns ([`Point`](/reference/data-types/geo#point), [`MultiPoint`](/reference/data-types/geo#multipoint), [`LineString`](/reference/data-types/geo#linestring), [`Polygon`](/reference/data-types/geo#polygon), [`MultiLineString`](/reference/data-types/geo#multilinestring), [`MultiPolygon`](/reference/data-types/geo#multipolygon)) as WKB. If `0`, those columns are written using their native underlying representation (e.g. `Point` as `Tuple(Float64, Float64)`) and no GeoParquet metadata is emitted.                                                                                                                                                                          | `1`         |
 | `input_format_parquet_import_nested`                                           | Obsolete setting, does nothing.                                                                                                                                                                                                   | `0`         |
 | `input_format_parquet_local_time_as_utc` | true | Determines the data type used by schema inference for Parquet timestamps with isAdjustedToUTC=false. If true: DateTime64(..., 'UTC'), if false: DateTime64(...). Neither behavior is fully correct as ClickHouse doesn't have a data type for local wall-clock time. Counterintuitively, 'true' is probably the less incorrect option, because formatting the 'UTC' timestamp as String will produce representation of the correct local time. |
+
+## Bloom filter reads {#bloom-filter-reads}
+
+`input_format_parquet_bloom_filter_coalesce_max_bytes` is the maximum size of one read that loads Parquet bloom filters. ClickHouse writes those filters back-to-back, after the row-group data, so nothing sits between neighboring filters. Filters in that block are read together, up to this many bytes. If the file itself is not larger than this value and it has enough filters that separate reads would cost more than one read of the file, the whole file is read once and later column reads reuse that buffer. `0` keeps one read per bloom filter.
+
+`remote_read_min_bytes_for_seek` can already merge the small filters, because they are adjacent and shorter than a large seek threshold. It is a different control, and so is `input_format_parquet_local_file_min_bytes_for_seek` for a local file:
+
+- `remote_read_min_bytes_for_seek` merges a neighboring range only when that range is shorter than the seek threshold. A bloom filter longer than the threshold stays its own read.
+- Raising that threshold far enough to cover a large bloom filter also applies to dictionary pages, indexes, and column data, so unrelated bytes are read with the filters.
+- An ordinary coalesced read may grow to four times the seek threshold. `input_format_parquet_bloom_filter_coalesce_max_bytes` is its own cap, and it applies only to bloom-filter ranges.
+- Enough filters in a file no larger than this setting select one read of the whole file. The seek threshold has no such rule.
+
+A bloom body kept only as a fallback for the dictionary filter is not part of this read. Its header is still read.
 )DOCS_MD"});
 }