auto pass through hashagg

dbms/src/Columns/ColumnAggregateFunction.cpp

@@ -161,13 +161,11 @@ void ColumnAggregateFunction::updateHashWithValues(
 void ColumnAggregateFunction::updateWeakHash32(WeakHash32 & hash, const TiDB::TiDBCollatorPtr &, String &) const
 {
     auto s = data.size();
-    if (hash.getData().size() != data.size())
-        throw Exception(
-            fmt::format(
-                "Size of WeakHash32 does not match size of column: column size is {}, hash size is {}",
-                s,
-                hash.getData().size()),
-            ErrorCodes::LOGICAL_ERROR);
+    RUNTIME_CHECK_MSG(
+        hash.getData().size() == data.size(),
+        "Size of WeakHash32({}) does not match size of column({})",
+        hash.getData().size(),
+        s);
 
     auto & hash_data = hash.getData();
 
@@ -181,6 +179,32 @@ void ColumnAggregateFunction::updateWeakHash32(WeakHash32 & hash, const TiDB::Ti
     }
 }
 
+void ColumnAggregateFunction::updateWeakHash32(
+    WeakHash32 & hash,
+    const TiDB::TiDBCollatorPtr &,
+    String &,
+    const BlockSelectivePtr & selective_ptr) const
+{
+    const auto selective_rows = selective_ptr->size();
+    RUNTIME_CHECK_MSG(
+        hash.getData().size() == selective_rows,
+        "Size of WeakHash32({}) does not match size of selective column({})",
+        hash.getData().size(),
+        selective_rows);
+
+    UInt32 * hash_data = hash.getData().data();
+
+    std::vector<UInt8> v;
+    for (const auto & row : *selective_ptr)
+    {
+        WriteBufferFromVector<std::vector<UInt8>> wbuf(v);
+        func->serialize(data[row], wbuf);
+        wbuf.finalize();
+        *hash_data = ::updateWeakHash32(v.data(), v.size(), *hash_data);
+        hash_data++;
+    }
+}
+
 /// NOTE: Highly overestimates size of a column if it was produced in AggregatingBlockInputStream (it contains size of other columns)
 size_t ColumnAggregateFunction::byteSize() const
 {
@@ -419,9 +443,38 @@ MutableColumns ColumnAggregateFunction::scatter(IColumn::ColumnIndex num_columns
     return columns;
 }
 
-void ColumnAggregateFunction::scatterTo(
-    ScatterColumns & columns [[maybe_unused]],
-    const Selector & selector [[maybe_unused]]) const
+MutableColumns ColumnAggregateFunction::scatter(
+    IColumn::ColumnIndex num_columns,
+    const IColumn::Selector & selector,
+    const BlockSelectivePtr & selective) const
+{
+    /// Columns with scattered values will point to this column as the owner of values.
+    MutableColumns columns(num_columns);
+    for (auto & column : columns)
+        column = createView();
+
+    const auto & selective_rows = selective->size();
+
+    {
+        size_t reserve_size = 1.1 * selective_rows / num_columns; /// 1.1 is just a guess. Better to use n-sigma rule.
+
+        if (reserve_size > 1)
+            for (auto & column : columns)
+                column->reserve(reserve_size);
+    }
+
+    for (size_t i = 0; i < selective_rows; ++i)
+        static_cast<ColumnAggregateFunction &>(*columns[selector[i]]).data.push_back(data[(*selective)[i]]);
+
+    return columns;
+}
+
+void ColumnAggregateFunction::scatterTo(ScatterColumns &, const Selector &) const
+{
+    throw TiFlashException("ColumnAggregateFunction does not support scatterTo", Errors::Coprocessor::Unimplemented);
+}
+
+void ColumnAggregateFunction::scatterTo(ScatterColumns &, const Selector &, const BlockSelectivePtr &) const
 {
     throw TiFlashException("ColumnAggregateFunction does not support scatterTo", Errors::Coprocessor::Unimplemented);
 }

dbms/src/Columns/ColumnAggregateFunction.h

@@ -171,6 +171,8 @@ class ColumnAggregateFunction final : public COWPtrHelper<IColumn, ColumnAggrega
         const override;
 
     void updateWeakHash32(WeakHash32 & hash, const TiDB::TiDBCollatorPtr &, String &) const override;
+    void updateWeakHash32(WeakHash32 & hash, const TiDB::TiDBCollatorPtr &, String &, const BlockSelectivePtr &)
+        const override;
 
     size_t byteSize() const override;
 
@@ -189,8 +191,12 @@ class ColumnAggregateFunction final : public COWPtrHelper<IColumn, ColumnAggrega
     ColumnPtr replicateRange(size_t start_row, size_t end_row, const IColumn::Offsets & offsets) const override;
 
     MutableColumns scatter(ColumnIndex num_columns, const Selector & selector) const override;
+    MutableColumns scatter(ColumnIndex num_columns, const Selector & selector, const BlockSelectivePtr & selective)
+        const override;
 
     void scatterTo(ScatterColumns & columns, const Selector & selector) const override;
+    void scatterTo(ScatterColumns & columns, const Selector & selector, const BlockSelectivePtr & selective)
+        const override;
 
     void gather(ColumnGathererStream & gatherer_stream) override;
 

dbms/src/Columns/ColumnArray.cpp

@@ -254,11 +254,11 @@ void ColumnArray::updateWeakHash32(
     String & sort_key_container) 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);
+    RUNTIME_CHECK_MSG(
+        hash.getData().size() == s,
+        "Size of WeakHash32({}) does not match size of column({})",
+        hash.getData().size(),
+        s);
 
     WeakHash32 internal_hash(data->size());
     data->updateWeakHash32(internal_hash, collator, sort_key_container);
@@ -284,6 +284,44 @@ void ColumnArray::updateWeakHash32(
     }
 }
 
+void ColumnArray::updateWeakHash32(
+    WeakHash32 & hash,
+    const TiDB::TiDBCollatorPtr & collator,
+    String & sort_key_container,
+    const BlockSelectivePtr & selective_ptr) const
+{
+    const auto selective_rows = selective_ptr->size();
+    RUNTIME_CHECK_MSG(
+        hash.getData().size() == selective_rows,
+        "Size of WeakHash32({}) does not match size of column({})",
+        hash.getData().size(),
+        selective_rows);
+
+    WeakHash32 internal_hash(data->size());
+    data->updateWeakHash32(internal_hash, collator, sort_key_container);
+
+    Offset prev_offset = 0;
+    const auto & offsets_data = getOffsets();
+    UInt32 * hash_data = hash.getData().data();
+    auto & internal_hash_data = internal_hash.getData();
+
+    for (const auto & i : *selective_ptr)
+    {
+        /// 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 = intHashCRC32(*hash_data);
+
+        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 = intHashCRC32(internal_hash_data[row], *hash_data);
+
+        prev_offset = offsets_data[i];
+        ++hash_data;
+    }
+}
+
 void ColumnArray::insert(const Field & x)
 {
     const auto & array = DB::get<const Array &>(x);

dbms/src/Columns/ColumnArray.h

@@ -85,6 +85,8 @@ class ColumnArray final : public COWPtrHelper<IColumn, ColumnArray>
     void updateHashWithValues(IColumn::HashValues & hash_values, const TiDB::TiDBCollatorPtr &, String &)
         const override;
     void updateWeakHash32(WeakHash32 & hash, const TiDB::TiDBCollatorPtr &, String &) const override;
+    void updateWeakHash32(WeakHash32 & hash, const TiDB::TiDBCollatorPtr &, String &, const BlockSelectivePtr &)
+        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;
@@ -144,10 +146,20 @@ class ColumnArray final : public COWPtrHelper<IColumn, ColumnArray>
     {
         return scatterImpl<ColumnArray>(num_columns, selector);
     }
+    MutableColumns scatter(ColumnIndex num_columns, const Selector & selector, const BlockSelectivePtr & selective)
+        const override
+    {
+        return scatterImpl<ColumnArray>(num_columns, selector, selective);
+    }
     void scatterTo(ScatterColumns & columns, const Selector & selector) const override
     {
         scatterToImpl<ColumnArray>(columns, selector);
     }
+    void scatterTo(ScatterColumns & columns, const Selector & selector, const BlockSelectivePtr & selective)
+        const override
+    {
+        scatterToImpl<ColumnArray>(columns, selector, selective);
+    }
     void gather(ColumnGathererStream & gatherer_stream) override;
 
     void forEachSubcolumn(ColumnCallback callback) override

dbms/src/Columns/ColumnConst.cpp

@@ -85,11 +85,32 @@ ColumnPtr ColumnConst::permute(const Permutation & perm, size_t limit) const
 
 MutableColumns ColumnConst::scatter(ColumnIndex num_columns, const Selector & selector) const
 {
-    if (s != selector.size())
-        throw Exception(
-            fmt::format("Size of selector ({}) doesn't match size of column ({})", selector.size(), s),
-            ErrorCodes::SIZES_OF_COLUMNS_DOESNT_MATCH);
+    RUNTIME_CHECK_MSG(
+        s == selector.size(),
+        "Size of selector ({}) doesn't match size of column ({})",
+        selector.size(),
+        s);
+
+    return scatterImplForColumnConst(num_columns, selector);
+}
 
+MutableColumns ColumnConst::scatter(
+    ColumnIndex num_columns,
+    const Selector & selector,
+    const BlockSelectivePtr & selective) const
+{
+    const auto selective_rows = selective->size();
+    RUNTIME_CHECK_MSG(
+        selective_rows == selector.size(),
+        "Size of selector ({}) doesn't match size of selective column ({})",
+        selector.size(),
+        selective_rows);
+
+    return scatterImplForColumnConst(num_columns, selector);
+}
+
+MutableColumns ColumnConst::scatterImplForColumnConst(ColumnIndex num_columns, const Selector & selector) const
+{
     std::vector<size_t> counts = countColumnsSizeInSelector(num_columns, selector);
 
     MutableColumns res(num_columns);
@@ -101,11 +122,28 @@ MutableColumns ColumnConst::scatter(ColumnIndex num_columns, const Selector & se
 
 void ColumnConst::scatterTo(ScatterColumns & columns, const Selector & selector) const
 {
-    if (s != selector.size())
-        throw Exception(
-            fmt::format("Size of selector ({}) doesn't match size of column ({})", selector.size(), s),
-            ErrorCodes::SIZES_OF_COLUMNS_DOESNT_MATCH);
+    RUNTIME_CHECK_MSG(
+        s == selector.size(),
+        "Size of selector ({}) doesn't match size of column ({})",
+        selector.size(),
+        s);
+    scatterToImplForColumnConst(columns, selector);
+}
+
+void ColumnConst::scatterTo(ScatterColumns & columns, const Selector & selector, const BlockSelectivePtr & selective)
+    const
+{
+    const auto selective_rows = selective->size();
+    RUNTIME_CHECK_MSG(
+        selective_rows == selector.size(),
+        "Size of selector ({}) doesn't match size of column ({})",
+        selector.size(),
+        selective_rows);
+    scatterToImplForColumnConst(columns, selector);
+}
 
+void ColumnConst::scatterToImplForColumnConst(ScatterColumns & columns, const Selector & selector) const
+{
     ColumnIndex num_columns = columns.size();
     std::vector<size_t> counts = countColumnsSizeInSelector(num_columns, selector);
 
@@ -126,20 +164,40 @@ void ColumnConst::updateWeakHash32(
     const TiDB::TiDBCollatorPtr & collator,
     String & sort_key_container) const
 {
-    if (hash.getData().size() != s)
-        throw Exception(
-            fmt::format(
-                "Size of WeakHash32 does not match size of column: column size is {}, hash size is {}",
-                s,
-                hash.getData().size()),
-            ErrorCodes::LOGICAL_ERROR);
+    RUNTIME_CHECK_MSG(
+        hash.getData().size() != s,
+        "Size of WeakHash32({}) does not match size of column({})",
+        hash.getData().size(),
+        s);
+    updateWeakHash32Impl(hash, collator, sort_key_container);
+}
+
+void ColumnConst::updateWeakHash32(
+    WeakHash32 & hash,
+    const TiDB::TiDBCollatorPtr & collator,
+    String & sort_key_container,
+    const BlockSelectivePtr & selective_ptr) const
+{
+    const auto selective_rows = selective_ptr->size();
+    RUNTIME_CHECK_MSG(
+        hash.getData().size() != selective_rows,
+        "Size of WeakHash32({}) does not match size of selective column({})",
+        hash.getData().size(),
+        selective_rows);
+
+    updateWeakHash32Impl(hash, collator, sort_key_container);
+}
 
+void ColumnConst::updateWeakHash32Impl(
+    WeakHash32 & hash,
+    const TiDB::TiDBCollatorPtr & collator,
+    String & sort_key_container) const
+{
     WeakHash32 element_hash(1);
     data->updateWeakHash32(element_hash, collator, sort_key_container);
     size_t data_hash = element_hash.getData()[0];
 
     for (auto & value : hash.getData())
         value = intHashCRC32(data_hash, value);
 }
-
 } // namespace DB

dbms/src/Columns/ColumnConst.h

@@ -133,6 +133,9 @@ class ColumnConst final : public COWPtrHelper<IColumn, ColumnConst>
     }
 
     void updateWeakHash32(WeakHash32 & hash, const TiDB::TiDBCollatorPtr &, String &) const override;
+    void updateWeakHash32(WeakHash32 & hash, const TiDB::TiDBCollatorPtr &, String &, const BlockSelectivePtr &)
+        const override;
+    void updateWeakHash32Impl(WeakHash32 & hash, const TiDB::TiDBCollatorPtr &, String &) const;
 
     ColumnPtr filter(const Filter & filt, ssize_t result_size_hint) const override;
     ColumnPtr replicateRange(size_t start_row, size_t end_row, const IColumn::Offsets & offsets) const override;
@@ -151,8 +154,14 @@ class ColumnConst final : public COWPtrHelper<IColumn, ColumnConst>
     }
 
     MutableColumns scatter(ColumnIndex num_columns, const Selector & selector) const override;
+    MutableColumns scatter(ColumnIndex num_columns, const Selector & selector, const BlockSelectivePtr & selective)
+        const override;
+    MutableColumns scatterImplForColumnConst(ColumnIndex num_columns, const Selector & selector) const;
 
     void scatterTo(ScatterColumns & columns, const Selector & selector) const override;
+    void scatterTo(ScatterColumns & columns, const Selector & selector, const BlockSelectivePtr & selective)
+        const override;
+    void scatterToImplForColumnConst(ScatterColumns & columns, const Selector & selector) const;
     void gather(ColumnGathererStream &) override
     {
         throw Exception("Cannot gather into constant column " + getName(), ErrorCodes::NOT_IMPLEMENTED);