Antalya 26.8: Backport of Replace HashSet<UInt64> for parquet dictionary filter pruning with st…

src/Processors/Formats/Impl/Parquet/Reader.cpp

@@ -15,7 +15,6 @@
 #include <DataTypes/DataTypeNullable.h>
 #include <Common/FieldAccurateComparison.h>
 #include <Common/checkStackSize.h>
-#include <Common/HashTable/HashSet.h>
 #include <base/arithmeticOverflow.h>
 #include <Formats/FormatFilterInfo.h>
 #include <Interpreters/castColumn.h>
@@ -1691,10 +1690,54 @@ bool Reader::columnChunkCanUseDictionaryFilter(const parq::ColumnChunk & column_
     return has_dictionary_data_page;
 }
 
+/// The value set of one column chunk's dictionary, prepared for lookups: the hashes of all
+/// dictionary values, sorted for binary search. `default_value_hash` stands for the values the
+/// dictionary does not hold: nulls decoded as the type's default under `input_format_null_as_default`
+/// (see `hashDictionaryValues`). It is kept out of `hashes` so the vector stays exactly the
+/// allocation `parquetTryHashColumn` made, which the pruning-memory reservation accounts for;
+/// appending would regrow the vector geometrically past the reserved amount.
+struct DictionaryValueHashes
+{
+    std::vector<UInt64> hashes;
+    std::optional<UInt64> default_value_hash;
+
+    /// Whether any of `probes` is among the dictionary's values.
+    ///
+    /// For a sorted probe sequence - which is what `KeyCondition::prepareBloomFilterData` produces -
+    /// this is an intersection of two sorted sequences rather than a sequence of independent binary
+    /// searches: every probe continues where the previous one stopped. A handful of query constants
+    /// then costs a binary search each, while a large probe set degrades to a linear merge of the two
+    /// sequences instead of `probes.size()` full-height searches over the whole vector. That matters
+    /// because a probe set is not always small: `prepareBloomFilterCondition` deliberately keeps
+    /// hashing `IN` sets that are over the bloom filter's cap, so that the exact dictionary filter can
+    /// still prune them, which means `findAnyHash` can be called with thousands of probes for one
+    /// column chunk. An out-of-order probe merely restarts the window, so the result does not depend
+    /// on the probes being sorted.
+    bool containsAny(const std::vector<UInt64> & probes) const
+    {
+        auto it = hashes.begin();
+        UInt64 previous_probe = 0;
+        for (UInt64 probe : probes)
+        {
+            if (probe == default_value_hash)
+                return true;
+            if (probe < previous_probe)
+                it = hashes.begin();
+            previous_probe = probe;
+            it = std::lower_bound(it, hashes.end(), probe);
+            /// `it` at the end means no dictionary value is >= this probe: every later probe that is
+            /// not smaller searches an empty range, and a smaller one restarts from the beginning.
+            if (it != hashes.end() && *it == probe)
+                return true;
+        }
+        return false;
+    }
+};
+
 /// Hash all values of an already-decoded dictionary the same way query constants are hashed for
 /// bloom filters, so the two can be compared. Returns nullopt if the values can't be hashed (in
 /// which case the dictionary can't be used for filtering).
-static std::optional<HashSet<UInt64>> hashDictionaryValues(
+static std::optional<DictionaryValueHashes> hashDictionaryValues(
     const parq::FileMetaData & file_metadata, const ReadOptions & options,
     Reader::ColumnChunk & column, const Reader::PrimitiveColumnInfo & column_info,
     const PruningMemoryReservation & reservation, size_t & held_pruning_bytes)
@@ -1707,8 +1750,8 @@ static std::optional<HashSet<UInt64>> hashDictionaryValues(
     /// on-disk dictionary page (`dictionary_filter_limit_bytes`, 1 MiB by default), not the decoded
     /// value set we are about to build here. A highly compressible dictionary can stay under that
     /// limit yet decode to many times more, and constructing the value set below (a materialized
-    /// column of all values, a vector of hashes, and a `HashSet` of them) then allocates that much
-    /// transient memory - potentially for several row groups in parallel during pruning. Charge it to
+    /// column of all values and a vector of their hashes) then allocates that much transient
+    /// memory - potentially for several row groups in parallel during pruning. Charge it to
     /// the shared pruning-stage budget (`reservation`): the reader's memory high watermark minus what
     /// the pruning stage already holds - the decoded dictionaries charged in `ReadManager::runTask`,
     /// plus the value sets already reserved by other dictionary lookups, whether earlier in this same
@@ -1725,7 +1768,7 @@ static std::optional<HashSet<UInt64>> hashDictionaryValues(
     /// `estimated_value_set_bytes` must be an upper bound on the peak transient memory allocated below,
     /// so that once the reservation succeeds the value set is guaranteed to stay within budget while it
     /// is built. The `hashes` vector (allocated at exactly `count` capacity by `parquetTryHashColumn`, so
-    /// exactly `count * sizeof(UInt64)`) and the resulting `value_hashes` HashSet are always built; the
+    /// exactly `count * sizeof(UInt64)`) is always built and sorted in place; the
     /// hashing itself allocates nothing on top - `parquetTryHashColumn` hashes string values in place
     /// from the column's buffers rather than copying each into a `Field` scratch string, and every other
     /// hashable type is stored inline in `Field`. When
@@ -1738,21 +1781,6 @@ static std::optional<HashSet<UInt64>> hashDictionaryValues(
     /// back by `count` understates a mixed-length string dictionary by up to `count` bytes. The
     /// `Mode::Column` path hashes the already-decoded (and already-charged) column in place, so it
     /// needs none of the materialization terms.
-    ///
-    /// The `HashSet` term cannot be approximated per value: `HashSet::reserve` picks a power-of-two
-    /// buffer with a maximum fill factor of 0.5 (`HashTableGrowerWithPrecalculation::set`), so the
-    /// table holds up to ~4 cells per inserted hash and never fewer than its initial 256 cells.
-    /// Compute the buffer size with the set's own growth rule so the reservation matches what
-    /// `reserve` really allocates, for the full insert cardinality: all `count` dictionary hashes
-    /// plus the one extra default-value hash possibly added under `input_format_null_as_default`
-    /// below (reserving for it up front also guarantees that insert never triggers a rehash past the
-    /// reservation). Add the set's initial constructor-allocated buffer, which can transiently
-    /// coexist with the resized one inside `realloc`.
-    using ValueHashSet = HashSet<UInt64>;
-    ValueHashSet::grower_type value_set_grower;
-    value_set_grower.set(count + 1);
-    size_t value_set_buffer_bytes =
-        (value_set_grower.bufSize() + ValueHashSet::grower_type::initial_count) * sizeof(ValueHashSet::cell_type);
     size_t per_value_bytes = sizeof(UInt64);    /// `hashes` vector
     size_t materialized_payload_bytes = 0;
     if (column.dictionary.mode != Dictionary::Mode::Column)
@@ -1767,12 +1795,12 @@ static std::optional<HashSet<UInt64>> hashDictionaryValues(
             sizeof(UInt64)      /// materialized `ColumnString` offsets (0 for fixed types; counted conservatively)
             + sizeof(UInt32);   /// identity `indexes`
     }
-    size_t estimated_value_set_bytes = count * per_value_bytes + materialized_payload_bytes + value_set_buffer_bytes;
+    size_t estimated_value_set_bytes = count * per_value_bytes + materialized_payload_bytes;
     /// Reserve the peak footprint before allocating anything. If it does not fit, skip pruning.
     if (!reservation.tryReserve(estimated_value_set_bytes))
         return std::nullopt;
     /// Release the whole reservation on every early-out below; only the successful path keeps the
-    /// persistent part reserved (reduced to the actual `HashSet` footprint) and hands it to the caller
+    /// persistent part reserved (reduced to the actual `hashes` buffer) and hands it to the caller
     /// via `held_pruning_bytes` (see `DictionaryLookup`, which releases it when the value set is freed).
     bool committed = false;
     SCOPE_EXIT({ if (!committed) reservation.release(estimated_value_set_bytes); });
@@ -1803,12 +1831,15 @@ static std::optional<HashSet<UInt64>> hashDictionaryValues(
     }
     if (!hashes.has_value())
         return std::nullopt;
-    ValueHashSet value_hashes;
-    /// +1 for the possible extra default-value hash below: when `hashes->size()` lands exactly on the
-    /// table's maximum fill, that late insert would otherwise rehash past the reserved buffer.
-    value_hashes.reserve(hashes->size() + 1);
-    for (UInt64 h : *hashes)
-        value_hashes.insert(h);
+
+    DictionaryValueHashes value_hashes;
+    value_hashes.hashes = std::move(*hashes);
+    hashes.reset();
+    /// Sort once so every lookup is a binary search. Sorting in place needs no extra memory, unlike
+    /// a hash table of the values, whose buffer (a power-of-two sized to a maximum fill factor of
+    /// 0.5) would hold up to ~4 cells per value on top of this vector - several times the footprint
+    /// for a value set that is built once per column chunk and probed a handful of times.
+    std::sort(value_hashes.hashes.begin(), value_hashes.hashes.end());
 
     /// The dictionary holds only the non-null values of the column chunk, so we must account for how
     /// nulls are read into the output, mirroring the conservative null handling of the min/max path in
@@ -1828,7 +1859,7 @@ static std::optional<HashSet<UInt64>> hashDictionaryValues(
             /// If the default value can't be hashed, we can't rule out a match.
             if (!default_hash.has_value())
                 return std::nullopt;
-            value_hashes.insert(*default_hash);
+            value_hashes.default_value_hash = default_hash;
         }
         else
         {
@@ -1839,23 +1870,23 @@ static std::optional<HashSet<UInt64>> hashDictionaryValues(
         }
     }
 
-    /// Free the transient `hashes` vector (the `indexes`/`values` allocations were already freed by
-    /// leaving their scope above) before releasing the reservation that covers it: otherwise a
-    /// concurrent pruning task could observe the released budget as free while `hashes` is still
-    /// allocated here, transiently exceeding the watermark.
-    hashes.reset();
-
     /// The value set is kept alive (in its `DictionaryLookup`) until this whole row-group filter
-    /// evaluation finishes, so keep its persistent footprint - the real `HashSet` buffer - reserved
+    /// evaluation finishes, so keep its persistent footprint - the sorted `hashes` buffer - reserved
     /// against the shared budget and hand the amount to the caller to release when the value set is
-    /// freed. The transient `indexes`/`values`/`hashes` allocations were already freed above, so
-    /// release that part of the reservation now: a second dictionary-filtered column, or another row
-    /// group pruning in parallel, then sees only the persistent part still held, keeping the combined
-    /// footprint within the watermark without over-reserving for the transients. The estimate above
-    /// follows the set's own growth rule, so it never under-predicts the buffer; the grow branch is a
-    /// defensive backstop (mirroring `decodeDictionaryPage`) in case the two ever drift apart, falling
-    /// back to a full scan if the correction no longer fits the budget.
-    size_t persistent_bytes = value_hashes.getBufferSizeInBytes();
+    /// freed. The transient `indexes`/`values` allocations were already freed by leaving their scope
+    /// above, so release that part of the reservation now: a second dictionary-filtered column, or
+    /// another row group pruning in parallel, then sees only the persistent part still held, keeping
+    /// the combined footprint within the watermark without over-reserving for the transients. The
+    /// estimate above covers the buffer exactly (`parquetTryHashColumn` reserves exactly `count`
+    /// elements); the grow branch is a defensive backstop (mirroring `decodeDictionaryPage`) in case
+    /// the two ever drift apart, falling back to a full scan if the correction no longer fits the
+    /// budget.
+    /// The vector must be exactly what `parquetTryHashColumn` allocated: the default value hash above
+    /// is kept in a separate field precisely so that nothing is appended here, which would grow the
+    /// vector geometrically past the reservation (and, for a dictionary whose reservation is nothing
+    /// but the vector - the `Mode::Column` path - past the whole budget).
+    chassert(value_hashes.hashes.size() == count);
+    size_t persistent_bytes = value_hashes.hashes.capacity() * sizeof(UInt64);
     if (persistent_bytes > estimated_value_set_bytes)
     {
         if (!reservation.tryReserve(persistent_bytes - estimated_value_set_bytes))
@@ -1884,7 +1915,7 @@ struct Reader::DictionaryLookup : public KeyCondition::BloomFilter
     size_t reserved_bytes = 0;
 
     bool computed = false;
-    std::optional<HashSet<UInt64>> value_hashes;
+    std::optional<DictionaryValueHashes> value_hashes;
 
     /// Bloom filter of the same column chunk, kept as a fallback for when the exact dictionary value set
     /// can't be built within the pruning memory budget (see `hashDictionaryValues`). Without it such a
@@ -1918,10 +1949,7 @@ bool Reader::DictionaryLookup::findAnyHash(const std::vector<uint64_t> & hashes)
     /// fall back to the column chunk's bloom filter if it has one; otherwise we can't rule out a match.
     if (!value_hashes.has_value())
         return bloom_fallback ? bloom_fallback->findAnyHash(hashes) : true;
-    for (UInt64 h : hashes)
-        if (value_hashes->contains(h))
-            return true;
-    return false;
+    return value_hashes->containsAny(hashes);
 }
 
 bool Reader::applyBloomAndDictionaryFilters(RowGroup & row_group, PruningMemoryReservation reservation)

src/Processors/Formats/Impl/Parquet/parquetBloomFilterHash.cpp

@@ -3,6 +3,7 @@
 #include <Columns/ColumnFixedString.h>
 #include <Columns/ColumnNullable.h>
 #include <Columns/ColumnString.h>
+#include <Columns/ColumnsNumber.h>
 
 #if USE_PARQUET
 
@@ -151,6 +152,36 @@ std::optional<uint64_t> tryHashInt(const Field & field, const std::shared_ptr<co
     return std::nullopt;
 }
 
+template <typename ParquetPhysicalType, typename T>
+static bool tryHashIntegerColumnTyped(const IColumn & column, std::vector<uint64_t> & hashes)
+{
+    const auto * typed = checkAndGetColumn<ColumnVector<T>>(&column);
+    if (!typed)
+        return false;
+    parquet::XxHasher hasher;
+    for (T value : typed->getData())
+        hashes.emplace_back(hasher.Hash(static_cast<ParquetPhysicalType>(value)));
+    return true;
+}
+
+/// Hash a whole integer column the way `tryHashInt` hashes one query constant. Going through the
+/// native data array must produce the same digests as going through `Field`: the `Field` path
+/// widens the value to `Int64`/`UInt64` (sign- or zero-extending by the value's own signedness)
+/// and then narrows to the physical type, which keeps exactly the low bits - the same result as
+/// `static_cast`ing the native value to the physical type directly.
+template <typename ParquetPhysicalType>
+static bool tryHashIntegerColumn(const IColumn & column, std::vector<uint64_t> & hashes)
+{
+    return tryHashIntegerColumnTyped<ParquetPhysicalType, Int8>(column, hashes)
+        || tryHashIntegerColumnTyped<ParquetPhysicalType, Int16>(column, hashes)
+        || tryHashIntegerColumnTyped<ParquetPhysicalType, Int32>(column, hashes)
+        || tryHashIntegerColumnTyped<ParquetPhysicalType, Int64>(column, hashes)
+        || tryHashIntegerColumnTyped<ParquetPhysicalType, UInt8>(column, hashes)
+        || tryHashIntegerColumnTyped<ParquetPhysicalType, UInt16>(column, hashes)
+        || tryHashIntegerColumnTyped<ParquetPhysicalType, UInt32>(column, hashes)
+        || tryHashIntegerColumnTyped<ParquetPhysicalType, UInt64>(column, hashes);
+}
+
 std::optional<uint64_t> parquetTryHashField(const Field & field, const parquet::ColumnDescriptor * parquet_column_descriptor)
 {
     const auto physical_type = parquet_column_descriptor->physical_type();
@@ -214,6 +245,25 @@ std::optional<std::vector<uint64_t>> parquetTryHashColumn(const IColumn * data_c
         return hashes;
     }
 
+    /// Hash integer columns directly from their data arrays. The generic `Field` path below spends
+    /// most of its time constructing a `Field` and re-dispatching on the type for every value; for
+    /// dictionaries of hundreds of thousands of entries per row group that dominates the whole
+    /// pruning stage. The type check is the same one `tryHashInt` applies per value, hoisted out of
+    /// the loop; a column the dispatch below does not recognize (e.g. `IPv4`) falls through to the
+    /// generic path unchanged.
+    if (physical_type == parquet::Type::type::INT32 || physical_type == parquet::Type::type::INT64)
+    {
+        if (isParquetIntegerTypeSupportedForBloomFilters(
+                parquet_column_descriptor->logical_type(), parquet_column_descriptor->converted_type()))
+        {
+            bool done = physical_type == parquet::Type::type::INT32
+                ? tryHashIntegerColumn<int32_t>(*column, hashes)
+                : tryHashIntegerColumn<int64_t>(*column, hashes);
+            if (done)
+                return hashes;
+        }
+    }
+
     for (size_t i = 0u; i < column->size(); i++)
     {
         Field f;

src/Storages/MergeTree/KeyCondition.cpp

@@ -6788,6 +6788,15 @@ void KeyCondition::prepareBloomFilterData(std::function<std::optional<uint64_t>(
                     continue;
                 }
 
+                /// Sort and deduplicate the probe hashes once, here, instead of leaving that to every
+                /// lookup on every granule/row group: a filter can then intersect them with its own
+                /// sorted value set in one merge pass (see the Parquet `DictionaryLookup`), and the
+                /// duplicates that an `IN` set can produce - different values whose hashes collide, or
+                /// a tuple element repeated across set rows - are probed once instead of many times.
+                std::sort(hashes_for_column.begin(), hashes_for_column.end());
+                hashes_for_column.erase(
+                    std::unique(hashes_for_column.begin(), hashes_for_column.end()), hashes_for_column.end());
+
                 hashes.emplace_back(hashes_for_column);
 
                 key_columns_for_element.push_back(indexes_mapping[i].key_index);

src/Storages/MergeTree/KeyCondition.h

@@ -105,6 +105,10 @@ class KeyCondition
     {
         virtual ~BloomFilter() = default;
 
+        /// `hashes` are the hashes of the query constants of one atom for one column. They are sorted
+        /// and deduplicated (see `prepareBloomFilterData`), which lets an implementation with a sorted
+        /// value set intersect the two sequences in one pass instead of searching for each hash
+        /// separately. Returns true if any of them may be present.
         virtual bool findAnyHash(const std::vector<uint64_t> & hashes) = 0;
     };