WeakHash32

dbms/src/Columns/ColumnAggregateFunction.cpp

@@ -10,6 +10,8 @@
 #include <Common/AlignedBuffer.h>
 #include <Common/typeid_cast.h>
 #include <Common/Arena.h>
+#include <Common/WeakHash.h>
+#include <Common/HashTable/Hash.h>
 
 #include <AggregateFunctions/AggregateFunctionMLMethod.h>
 
@@ -282,6 +284,25 @@ void ColumnAggregateFunction::updateHashWithValue(size_t n, SipHash & hash) cons
     hash.update(wbuf.str().c_str(), wbuf.str().size());
 }
 
+void ColumnAggregateFunction::updateWeakHash32(WeakHash32 & hash) const
+{
+    auto s = data.size();
+    if (hash.getData().size() != data.size())
+        throw Exception("Size of WeakHash32 does not match size of column: column size is " + std::to_string(s) +
+                        ", hash size is " + std::to_string(hash.getData().size()), ErrorCodes::LOGICAL_ERROR);
+
+    auto & hash_data = hash.getData();
+
+    std::vector<UInt8> v;
+    for (size_t i = 0; i < s; ++i)
+    {
+        WriteBufferFromVector<std::vector<UInt8>> wbuf(v);
+        func->serialize(data[i], wbuf);
+        wbuf.finalize();
+        hash_data[i] = ::updateWeakHash32(v.data(), v.size(), hash_data[i]);
+    }
+}
+
 /// The returned size is less than real size. The reason is that some parts of
 /// aggregate function data may be allocated on shared arenas. These arenas are
 /// used for several blocks, and also may be updated concurrently from other

dbms/src/Columns/ColumnAggregateFunction.h

@@ -160,6 +160,8 @@ class ColumnAggregateFunction final : public COWHelper<IColumn, ColumnAggregateF
 
     void updateHashWithValue(size_t n, SipHash & hash) const override;
 
+    void updateWeakHash32(WeakHash32 & hash) const override;
+
     size_t byteSize() const override;
 
     size_t allocatedBytes() const override;

dbms/src/Columns/ColumnArray.cpp

@@ -17,6 +17,8 @@
 #include <Common/SipHash.h>
 #include <Common/typeid_cast.h>
 #include <Common/assert_cast.h>
+#include <Common/WeakHash.h>
+#include <Common/HashTable/Hash.h>
 
 
 namespace DB
@@ -213,6 +215,36 @@ void ColumnArray::updateHashWithValue(size_t n, SipHash & hash) const
         getData().updateHashWithValue(offset + i, hash);
 }
 
+void ColumnArray::updateWeakHash32(WeakHash32 & hash) const
+{
+    auto s = offsets->size();
+    if (hash.getData().size() != s)
+        throw Exception("Size of WeakHash32 does not match size of column: column size is " + std::to_string(s) +
+                        ", hash size is " + std::to_string(hash.getData().size()), ErrorCodes::LOGICAL_ERROR);
+
+    WeakHash32 internal_hash(data->size());
+    data->updateWeakHash32(internal_hash);
+
+    Offset prev_offset = 0;
+    auto & offsets_data = getOffsets();
+    auto & hash_data = hash.getData();
+    auto & internal_hash_data = internal_hash.getData();
+
+    for (size_t i = 0; i < s; ++i)
+    {
+        /// This row improves hash a little bit according to integration tests.
+        /// It is the same as to use previous hash value as the first element of array.
+        hash_data[i] = intHashCRC32(hash_data[i]);
+
+        for (size_t row = prev_offset; row < offsets_data[i]; ++row)
+            /// It is probably not the best way to combine hashes.
+            /// But much better then xor which lead to similar hash for arrays like [1], [1, 1, 1], [1, 1, 1, 1, 1], ...
+            /// Much better implementation - to add offsets as an optional argument to updateWeakHash32.
+            hash_data[i] = intHashCRC32(internal_hash_data[row], hash_data[i]);
+
+        prev_offset = offsets_data[i];
+    }
+}
 
 void ColumnArray::insert(const Field & x)
 {

dbms/src/Columns/ColumnArray.h

@@ -61,6 +61,7 @@ class ColumnArray final : public COWHelper<IColumn, ColumnArray>
     StringRef serializeValueIntoArena(size_t n, Arena & arena, char const *& begin) const override;
     const char * deserializeAndInsertFromArena(const char * pos) override;
     void updateHashWithValue(size_t n, SipHash & hash) const override;
+    void updateWeakHash32(WeakHash32 & hash) const override;
     void insertRangeFrom(const IColumn & src, size_t start, size_t length) override;
     void insert(const Field & x) override;
     void insertFrom(const IColumn & src_, size_t n) override;

dbms/src/Columns/ColumnConst.cpp

@@ -4,6 +4,8 @@
 #include <Columns/ColumnsCommon.h>
 #include <Common/PODArray.h>
 #include <Common/typeid_cast.h>
+#include <Common/WeakHash.h>
+#include <Common/HashTable/Hash.h>
 
 
 namespace DB
@@ -12,6 +14,7 @@ namespace DB
 namespace ErrorCodes
 {
     extern const int SIZES_OF_COLUMNS_DOESNT_MATCH;
+    extern const int LOGICAL_ERROR;
 }
 
 ColumnConst::ColumnConst(const ColumnPtr & data_, size_t s_)
@@ -103,4 +106,18 @@ void ColumnConst::getPermutation(bool /*reverse*/, size_t /*limit*/, int /*nan_d
         res[i] = i;
 }
 
+void ColumnConst::updateWeakHash32(WeakHash32 & hash) const
+{
+    if (hash.getData().size() != s)
+        throw Exception("Size of WeakHash32 does not match size of column: column size is " + std::to_string(s) +
+                        ", hash size is " + std::to_string(hash.getData().size()), ErrorCodes::LOGICAL_ERROR);
+
+    WeakHash32 element_hash(1);
+    data->updateWeakHash32(element_hash);
+    size_t data_hash = element_hash.getData()[0];
+
+    for (auto & value : hash.getData())
+        value = intHashCRC32(data_hash, value);
+}
+
 }

dbms/src/Columns/ColumnConst.h

@@ -163,6 +163,8 @@ class ColumnConst final : public COWHelper<IColumn, ColumnConst>
         data->updateHashWithValue(0, hash);
     }
 
+    void updateWeakHash32(WeakHash32 & hash) const override;
+
     ColumnPtr filter(const Filter & filt, ssize_t result_size_hint) const override;
     ColumnPtr replicate(const Offsets & offsets) const override;
     ColumnPtr permute(const Permutation & perm, size_t limit) const override;

dbms/src/Columns/ColumnDecimal.cpp

@@ -2,6 +2,8 @@
 #include <Common/Arena.h>
 #include <Common/SipHash.h>
 #include <Common/assert_cast.h>
+#include <Common/WeakHash.h>
+#include <Common/HashTable/Hash.h>
 
 #include <common/unaligned.h>
 
@@ -22,6 +24,7 @@ namespace ErrorCodes
     extern const int PARAMETER_OUT_OF_BOUND;
     extern const int SIZES_OF_COLUMNS_DOESNT_MATCH;
     extern const int NOT_IMPLEMENTED;
+    extern const int LOGICAL_ERROR;
 }
 
 template <typename T>
@@ -65,6 +68,27 @@ void ColumnDecimal<T>::updateHashWithValue(size_t n, SipHash & hash) const
     hash.update(data[n]);
 }
 
+template <typename T>
+void ColumnDecimal<T>::updateWeakHash32(WeakHash32 & hash) const
+{
+    auto s = data.size();
+
+    if (hash.getData().size() != s)
+        throw Exception("Size of WeakHash32 does not match size of column: column size is " + std::to_string(s) +
+                        ", hash size is " + std::to_string(hash.getData().size()), ErrorCodes::LOGICAL_ERROR);
+
+    const T * begin = data.data();
+    const T * end = begin + s;
+    UInt32 * hash_data = hash.getData().data();
+
+    while (begin < end)
+    {
+        *hash_data = intHashCRC32(*begin, *hash_data);
+        ++begin;
+        ++hash_data;
+    }
+}
+
 template <typename T>
 void ColumnDecimal<T>::getPermutation(bool reverse, size_t limit, int , IColumn::Permutation & res) const
 {

dbms/src/Columns/ColumnDecimal.h

@@ -105,6 +105,7 @@ class ColumnDecimal final : public COWHelper<ColumnVectorHelper, ColumnDecimal<T
     StringRef serializeValueIntoArena(size_t n, Arena & arena, char const *& begin) const override;
     const char * deserializeAndInsertFromArena(const char * pos) override;
     void updateHashWithValue(size_t n, SipHash & hash) const override;
+    void updateWeakHash32(WeakHash32 & hash) const override;
     int compareAt(size_t n, size_t m, const IColumn & rhs_, int nan_direction_hint) const override;
     void getPermutation(bool reverse, size_t limit, int nan_direction_hint, IColumn::Permutation & res) const override;
 

dbms/src/Columns/ColumnFixedString.cpp

@@ -6,6 +6,8 @@
 #include <Common/memcpySmall.h>
 #include <Common/memcmpSmall.h>
 #include <Common/assert_cast.h>
+#include <Common/WeakHash.h>
+#include <Common/HashTable/Hash.h>
 
 #include <DataStreams/ColumnGathererStream.h>
 
@@ -25,6 +27,7 @@ namespace ErrorCodes
     extern const int SIZE_OF_FIXED_STRING_DOESNT_MATCH;
     extern const int SIZES_OF_COLUMNS_DOESNT_MATCH;
     extern const int PARAMETER_OUT_OF_BOUND;
+    extern const int LOGICAL_ERROR;
 }
 
 
@@ -101,6 +104,26 @@ void ColumnFixedString::updateHashWithValue(size_t index, SipHash & hash) const
     hash.update(reinterpret_cast<const char *>(&chars[n * index]), n);
 }
 
+void ColumnFixedString::updateWeakHash32(WeakHash32 & hash) const
+{
+    auto s = size();
+
+    if (hash.getData().size() != s)
+        throw Exception("Size of WeakHash32 does not match size of column: column size is " + std::to_string(s) +
+                        ", hash size is " + std::to_string(hash.getData().size()), ErrorCodes::LOGICAL_ERROR);
+
+    const UInt8 * pos = chars.data();
+    UInt32 * hash_data = hash.getData().data();
+
+    for (size_t row = 0; row < s; ++row)
+    {
+        *hash_data = ::updateWeakHash32(pos, n, *hash_data);
+
+        pos += n;
+        ++hash_data;
+    }
+}
+
 template <bool positive>
 struct ColumnFixedString::less
 {

dbms/src/Columns/ColumnFixedString.h

@@ -108,6 +108,8 @@ class ColumnFixedString final : public COWHelper<ColumnVectorHelper, ColumnFixed
 
     void updateHashWithValue(size_t index, SipHash & hash) const override;
 
+    void updateWeakHash32(WeakHash32 & hash) const override;
+
     int compareAt(size_t p1, size_t p2, const IColumn & rhs_, int /*nan_direction_hint*/) const override
     {
         const ColumnFixedString & rhs = assert_cast<const ColumnFixedString &>(rhs_);

dbms/src/Columns/ColumnFunction.h

@@ -101,6 +101,11 @@ class ColumnFunction final : public COWHelper<IColumn, ColumnFunction>
         throw Exception("updateHashWithValue is not implemented for " + getName(), ErrorCodes::NOT_IMPLEMENTED);
     }
 
+    void updateWeakHash32(WeakHash32 &) const override
+    {
+        throw Exception("updateWeakHash32 is not implemented for " + getName(), ErrorCodes::NOT_IMPLEMENTED);
+    }
+
     void popBack(size_t) override
     {
         throw Exception("popBack is not implemented for " + getName(), ErrorCodes::NOT_IMPLEMENTED);

dbms/src/Columns/ColumnLowCardinality.cpp

@@ -4,6 +4,7 @@
 #include <DataTypes/NumberTraits.h>
 #include <Common/HashTable/HashMap.h>
 #include <Common/assert_cast.h>
+#include <Common/WeakHash.h>
 
 
 namespace DB
@@ -241,6 +242,21 @@ const char * ColumnLowCardinality::deserializeAndInsertFromArena(const char * po
     return new_pos;
 }
 
+void ColumnLowCardinality::updateWeakHash32(WeakHash32 & hash) const
+{
+    auto s = size();
+
+    if (hash.getData().size() != s)
+        throw Exception("Size of WeakHash32 does not match size of column: column size is " + std::to_string(s) +
+                        ", hash size is " + std::to_string(hash.getData().size()), ErrorCodes::LOGICAL_ERROR);
+
+    auto & dict = getDictionary().getNestedColumn();
+    WeakHash32 dict_hash(dict->size());
+    dict->updateWeakHash32(dict_hash);
+
+    idx.updateWeakHash(hash, dict_hash);
+}
+
 void ColumnLowCardinality::gather(ColumnGathererStream & gatherer)
 {
     gatherer.gather(*this);
@@ -645,6 +661,24 @@ bool ColumnLowCardinality::Index::containsDefault() const
     return contains;
 }
 
+void ColumnLowCardinality::Index::updateWeakHash(WeakHash32 & hash, WeakHash32 & dict_hash) const
+{
+    auto & hash_data = hash.getData();
+    auto & dict_hash_data = dict_hash.getData();
+
+    auto update_weak_hash = [&](auto x)
+    {
+        using CurIndexType = decltype(x);
+        auto & data = getPositionsData<CurIndexType>();
+        auto size = data.size();
+
+        for (size_t i = 0; i < size; ++i)
+            hash_data[i] = intHashCRC32(dict_hash_data[data[i]], hash_data[i]);
+    };
+
+    callForType(std::move(update_weak_hash), size_of_type);
+}
+
 
 ColumnLowCardinality::Dictionary::Dictionary(MutableColumnPtr && column_unique_, bool is_shared)
     : column_unique(std::move(column_unique_)), shared(is_shared)

dbms/src/Columns/ColumnLowCardinality.h

@@ -90,6 +90,8 @@ class ColumnLowCardinality final : public COWHelper<IColumn, ColumnLowCardinalit
         return getDictionary().updateHashWithValue(getIndexes().getUInt(n), hash);
     }
 
+    void updateWeakHash32(WeakHash32 & hash) const override;
+
     ColumnPtr filter(const Filter & filt, ssize_t result_size_hint) const override
     {
         return ColumnLowCardinality::create(dictionary.getColumnUniquePtr(), getIndexes().filter(filt, result_size_hint));
@@ -230,6 +232,8 @@ class ColumnLowCardinality final : public COWHelper<IColumn, ColumnLowCardinalit
 
         bool containsDefault() const;
 
+        void updateWeakHash(WeakHash32 & hash, WeakHash32 & dict_hash) const;
+
     private:
         WrappedPtr positions;
         size_t size_of_type = 0;

dbms/src/Columns/ColumnNullable.cpp

@@ -3,6 +3,7 @@
 #include <Common/NaNUtils.h>
 #include <Common/typeid_cast.h>
 #include <Common/assert_cast.h>
+#include <Common/WeakHash.h>
 #include <Columns/ColumnNullable.h>
 #include <Columns/ColumnConst.h>
 #include <DataStreams/ColumnGathererStream.h>
@@ -42,6 +43,26 @@ void ColumnNullable::updateHashWithValue(size_t n, SipHash & hash) const
         getNestedColumn().updateHashWithValue(n, hash);
 }
 
+void ColumnNullable::updateWeakHash32(WeakHash32 & hash) const
+{
+    auto s = size();
+
+    if (hash.getData().size() != s)
+        throw Exception("Size of WeakHash32 does not match size of column: column size is " + std::to_string(s) +
+                        ", hash size is " + std::to_string(hash.getData().size()), ErrorCodes::LOGICAL_ERROR);
+
+    WeakHash32 old_hash = hash;
+    nested_column->updateWeakHash32(hash);
+
+    auto & null_map_data = getNullMapData();
+    auto & hash_data = hash.getData();
+    auto & old_hash_data = old_hash.getData();
+
+    /// Use old data for nulls.
+    for (size_t row = 0; row < s; ++row)
+        if (null_map_data[row])
+            hash_data[row] = old_hash_data[row];
+}
 
 MutableColumnPtr ColumnNullable::cloneResized(size_t new_size) const
 {

dbms/src/Columns/ColumnNullable.h

@@ -84,6 +84,7 @@ class ColumnNullable final : public COWHelper<IColumn, ColumnNullable>
     void protect() override;
     ColumnPtr replicate(const Offsets & replicate_offsets) const override;
     void updateHashWithValue(size_t n, SipHash & hash) const override;
+    void updateWeakHash32(WeakHash32 & hash) const override;
     void getExtremes(Field & min, Field & max) const override;
 
     MutableColumns scatter(ColumnIndex num_columns, const Selector & selector) const override