diff --git a/src/AggregateFunctions/AggregateFunctionGroupBitmap.cpp b/src/AggregateFunctions/AggregateFunctionGroupBitmap.cpp
index 25df0cb27bd..bf1d0af73ff 100644
--- a/src/AggregateFunctions/AggregateFunctionGroupBitmap.cpp
+++ b/src/AggregateFunctions/AggregateFunctionGroupBitmap.cpp
@@ -1,11 +1,10 @@
 #include <AggregateFunctions/AggregateFunctionFactory.h>
-#include <AggregateFunctions/Helpers.h>
 #include <AggregateFunctions/FactoryHelpers.h>
+#include <AggregateFunctions/Helpers.h>
 #include <DataTypes/DataTypeAggregateFunction.h>
 #include "registerAggregateFunctions.h"
 
-// TODO include this last because of a broken roaring header. See the comment
-// inside.
+// TODO include this last because of a broken roaring header. See the comment inside.
 #include <AggregateFunctions/AggregateFunctionGroupBitmap.h>
 
 namespace DB
@@ -17,46 +16,54 @@ namespace ErrorCodes
 
 namespace
 {
+    template <template <typename> class Data>
+    AggregateFunctionPtr createAggregateFunctionBitmap(const std::string & name, const DataTypes & argument_types, const Array & parameters)
+    {
+        assertNoParameters(name, parameters);
+        assertUnary(name, argument_types);
 
-template <template <typename> class Data>
-AggregateFunctionPtr createAggregateFunctionBitmap(const std::string & name, const DataTypes & argument_types, const Array & parameters)
-{
-    assertNoParameters(name, parameters);
-    assertUnary(name, argument_types);
+        if (!argument_types[0]->canBeUsedInBitOperations())
+            throw Exception(
+                "The type " + argument_types[0]->getName() + " of argument for aggregate function " + name
+                    + " is illegal, because it cannot be used in Bitmap operations",
+                ErrorCodes::ILLEGAL_TYPE_OF_ARGUMENT);
 
-    if (!argument_types[0]->canBeUsedInBitOperations())
-        throw Exception("The type " + argument_types[0]->getName() + " of argument for aggregate function " + name
-            + " is illegal, because it cannot be used in Bitmap operations",
-            ErrorCodes::ILLEGAL_TYPE_OF_ARGUMENT);
+        AggregateFunctionPtr res(createWithUnsignedIntegerType<AggregateFunctionBitmap, Data>(*argument_types[0], argument_types[0]));
 
-    AggregateFunctionPtr res(createWithUnsignedIntegerType<AggregateFunctionBitmap, Data>(*argument_types[0], argument_types[0]));
+        if (!res)
+            throw Exception(
+                "Illegal type " + argument_types[0]->getName() + " of argument for aggregate function " + name,
+                ErrorCodes::ILLEGAL_TYPE_OF_ARGUMENT);
 
-    if (!res)
-        throw Exception("Illegal type " + argument_types[0]->getName() + " of argument for aggregate function " + name, ErrorCodes::ILLEGAL_TYPE_OF_ARGUMENT);
+        return res;
+    }
 
-    return res;
-}
+    // Additional aggregate functions to manipulate bitmaps.
+    template <template <typename, typename> class AggregateFunctionTemplate>
+    AggregateFunctionPtr
+    createAggregateFunctionBitmapL2(const std::string & name, const DataTypes & argument_types, const Array & parameters)
+    {
+        assertNoParameters(name, parameters);
+        assertUnary(name, argument_types);
+        DataTypePtr argument_type_ptr = argument_types[0];
+        WhichDataType which(*argument_type_ptr);
+        if (which.idx != TypeIndex::AggregateFunction)
+            throw Exception(
+                "Illegal type " + argument_types[0]->getName() + " of argument for aggregate function " + name,
+                ErrorCodes::ILLEGAL_TYPE_OF_ARGUMENT);
 
-template <template <typename, typename> class AggregateFunctionTemplate>
-AggregateFunctionPtr createAggregateFunctionBitmapL2(const std::string & name, const DataTypes & argument_types, const Array & parameters)
-{
-    assertNoParameters(name, parameters);
-    assertUnary(name, argument_types);
-    DataTypePtr argument_type_ptr = argument_types[0];
-    WhichDataType which(*argument_type_ptr);
-    if (which.idx != TypeIndex::AggregateFunction)
-        throw Exception("Illegal type " + argument_types[0]->getName() + " of argument for aggregate function " + name, ErrorCodes::ILLEGAL_TYPE_OF_ARGUMENT);
-
-    const DataTypeAggregateFunction& datatype_aggfunc = dynamic_cast<const DataTypeAggregateFunction&>(*argument_type_ptr);
-    AggregateFunctionPtr aggfunc = datatype_aggfunc.getFunction();
-    argument_type_ptr = aggfunc->getArgumentTypes()[0];
-    AggregateFunctionPtr res(createWithUnsignedIntegerType<AggregateFunctionTemplate, AggregateFunctionGroupBitmapData>(*argument_type_ptr, argument_type_ptr));
-    if (!res)
-        throw Exception("Illegal type " + argument_types[0]->getName() + " of argument for aggregate function " + name, ErrorCodes::ILLEGAL_TYPE_OF_ARGUMENT);
-
-    return res;
-}
+        const DataTypeAggregateFunction & datatype_aggfunc = dynamic_cast<const DataTypeAggregateFunction &>(*argument_type_ptr);
+        AggregateFunctionPtr aggfunc = datatype_aggfunc.getFunction();
+        argument_type_ptr = aggfunc->getArgumentTypes()[0];
+        AggregateFunctionPtr res(createWithUnsignedIntegerType<AggregateFunctionTemplate, AggregateFunctionGroupBitmapData>(
+            *argument_type_ptr, argument_type_ptr));
+        if (!res)
+            throw Exception(
+                "Illegal type " + argument_types[0]->getName() + " of argument for aggregate function " + name,
+                ErrorCodes::ILLEGAL_TYPE_OF_ARGUMENT);
 
+        return res;
+    }
 }
 
 void registerAggregateFunctionsBitmap(AggregateFunctionFactory & factory)
diff --git a/src/AggregateFunctions/AggregateFunctionGroupBitmap.h b/src/AggregateFunctions/AggregateFunctionGroupBitmap.h
index e974cb753d3..315132cf05c 100644
--- a/src/AggregateFunctions/AggregateFunctionGroupBitmap.h
+++ b/src/AggregateFunctions/AggregateFunctionGroupBitmap.h
@@ -1,31 +1,26 @@
 #pragma once
 
-#include <Columns/ColumnVector.h>
-#include <Common/assert_cast.h>
 #include <AggregateFunctions/IAggregateFunction.h>
-#include <DataTypes/DataTypesNumber.h>
 #include <Columns/ColumnAggregateFunction.h>
+#include <Columns/ColumnVector.h>
+#include <DataTypes/DataTypesNumber.h>
+#include <Common/assert_cast.h>
 
 // TODO include this last because of a broken roaring header. See the comment inside.
 #include <AggregateFunctions/AggregateFunctionGroupBitmapData.h>
 
 namespace DB
 {
-
 /// Counts bitmap operation on numbers.
 template <typename T, typename Data>
 class AggregateFunctionBitmap final : public IAggregateFunctionDataHelper<Data, AggregateFunctionBitmap<T, Data>>
 {
 public:
-    AggregateFunctionBitmap(const DataTypePtr & type)
-            : IAggregateFunctionDataHelper<Data, AggregateFunctionBitmap<T, Data>>({type}, {}) {}
+    AggregateFunctionBitmap(const DataTypePtr & type) : IAggregateFunctionDataHelper<Data, AggregateFunctionBitmap<T, Data>>({type}, {}) { }
 
     String getName() const override { return Data::name(); }
 
-    DataTypePtr getReturnType() const override
-    {
-        return std::make_shared<DataTypeNumber<T>>();
-    }
+    DataTypePtr getReturnType() const override { return std::make_shared<DataTypeNumber<T>>(); }
 
     void add(AggregateDataPtr place, const IColumn ** columns, size_t row_num, Arena *) const override
     {
@@ -37,15 +32,9 @@ public:
         this->data(place).rbs.merge(this->data(rhs).rbs);
     }
 
-    void serialize(ConstAggregateDataPtr place, WriteBuffer & buf) const override
-    {
-        this->data(place).rbs.write(buf);
-    }
+    void serialize(ConstAggregateDataPtr place, WriteBuffer & buf) const override { this->data(place).rbs.write(buf); }
 
-    void deserialize(AggregateDataPtr place, ReadBuffer & buf, Arena *) const override
-    {
-        this->data(place).rbs.read(buf);
-    }
+    void deserialize(AggregateDataPtr place, ReadBuffer & buf, Arena *) const override { this->data(place).rbs.read(buf); }
 
     void insertResultInto(AggregateDataPtr place, IColumn & to, Arena *) const override
     {
@@ -59,23 +48,22 @@ class AggregateFunctionBitmapL2 final : public IAggregateFunctionDataHelper<Data
 {
 public:
     AggregateFunctionBitmapL2(const DataTypePtr & type)
-            : IAggregateFunctionDataHelper<Data, AggregateFunctionBitmapL2<T, Data, Policy>>({type}, {}){}
+        : IAggregateFunctionDataHelper<Data, AggregateFunctionBitmapL2<T, Data, Policy>>({type}, {})
+    {
+    }
 
     String getName() const override { return Data::name(); }
 
-    DataTypePtr getReturnType() const override
-    {
-        return std::make_shared<DataTypeNumber<T>>();
-    }
+    DataTypePtr getReturnType() const override { return std::make_shared<DataTypeNumber<T>>(); }
 
     void add(AggregateDataPtr place, const IColumn ** columns, size_t row_num, Arena *) const override
     {
         Data & data_lhs = this->data(place);
         const Data & data_rhs = this->data(assert_cast<const ColumnAggregateFunction &>(*columns[0]).getData()[row_num]);
-        if (!data_lhs.doneFirst)
+        if (!data_lhs.init)
         {
-            data_lhs.doneFirst = true;
-            data_lhs.rbs.rb_or(data_rhs.rbs);
+            data_lhs.init = true;
+            data_lhs.rbs.merge(data_rhs.rbs);
         }
         else
         {
@@ -88,13 +76,13 @@ public:
         Data & data_lhs = this->data(place);
         const Data & data_rhs = this->data(rhs);
 
-        if (!data_rhs.doneFirst)
+        if (!data_rhs.init)
             return;
 
-        if (!data_lhs.doneFirst)
+        if (!data_lhs.init)
         {
-            data_lhs.doneFirst = true;
-            data_lhs.rbs.rb_or(data_rhs.rbs);
+            data_lhs.init = true;
+            data_lhs.rbs.merge(data_rhs.rbs);
         }
         else
         {
@@ -102,15 +90,9 @@ public:
         }
     }
 
-    void serialize(ConstAggregateDataPtr place, WriteBuffer & buf) const override
-    {
-        this->data(place).rbs.write(buf);
-    }
+    void serialize(ConstAggregateDataPtr place, WriteBuffer & buf) const override { this->data(place).rbs.write(buf); }
 
-    void deserialize(AggregateDataPtr place, ReadBuffer & buf, Arena *) const override
-    {
-        this->data(place).rbs.read(buf);
-    }
+    void deserialize(AggregateDataPtr place, ReadBuffer & buf, Arena *) const override { this->data(place).rbs.read(buf); }
 
     void insertResultInto(AggregateDataPtr place, IColumn & to, Arena *) const override
     {
@@ -122,39 +104,30 @@ template <typename Data>
 class BitmapAndPolicy
 {
 public:
-    static void apply(Data& lhs, const Data& rhs)
-    {
-        lhs.rbs.rb_and(rhs.rbs);
-    }
+    static void apply(Data & lhs, const Data & rhs) { lhs.rbs.rb_and(rhs.rbs); }
 };
 
 template <typename Data>
 class BitmapOrPolicy
 {
 public:
-    static void apply(Data& lhs, const Data& rhs)
-    {
-        lhs.rbs.rb_or(rhs.rbs);
-    }
+    static void apply(Data & lhs, const Data & rhs) { lhs.rbs.rb_or(rhs.rbs); }
 };
 
 template <typename Data>
 class BitmapXorPolicy
 {
 public:
-    static void apply(Data& lhs, const Data& rhs)
-    {
-        lhs.rbs.rb_xor(rhs.rbs);
-    }
+    static void apply(Data & lhs, const Data & rhs) { lhs.rbs.rb_xor(rhs.rbs); }
 };
 
 template <typename T, typename Data>
-using AggregateFunctionBitmapL2And = AggregateFunctionBitmapL2<T, Data, BitmapAndPolicy<Data> >;
+using AggregateFunctionBitmapL2And = AggregateFunctionBitmapL2<T, Data, BitmapAndPolicy<Data>>;
 
 template <typename T, typename Data>
-using AggregateFunctionBitmapL2Or = AggregateFunctionBitmapL2<T, Data, BitmapOrPolicy<Data> >;
+using AggregateFunctionBitmapL2Or = AggregateFunctionBitmapL2<T, Data, BitmapOrPolicy<Data>>;
 
 template <typename T, typename Data>
-using AggregateFunctionBitmapL2Xor = AggregateFunctionBitmapL2<T, Data, BitmapXorPolicy<Data> >;
+using AggregateFunctionBitmapL2Xor = AggregateFunctionBitmapL2<T, Data, BitmapXorPolicy<Data>>;
 
 }
diff --git a/src/AggregateFunctions/AggregateFunctionGroupBitmapData.h b/src/AggregateFunctions/AggregateFunctionGroupBitmapData.h
index 000edd64a0f..b26e47f0b63 100644
--- a/src/AggregateFunctions/AggregateFunctionGroupBitmapData.h
+++ b/src/AggregateFunctions/AggregateFunctionGroupBitmapData.h
@@ -1,915 +1,635 @@
-#pragma once
-
-#include <algorithm>
-#include <IO/ReadHelpers.h>
-#include <IO/WriteHelpers.h>
-#include <boost/noncopyable.hpp>
-#include <Common/HashTable/SmallTable.h>
-#include <Common/PODArray.h>
-
-// Include this header last, because it is an auto-generated dump of questionable
-// garbage that breaks the build (e.g. it changes _POSIX_C_SOURCE).
-// TODO: find out what it is. On github, they have proper interface headers like
-// this one: https://github.com/RoaringBitmap/CRoaring/blob/master/include/roaring/roaring.h
-#include <roaring/roaring.h>
-#include "roaring.hh"
-#include <roaring64map.hh>
-
-namespace DB
-{
-/**
-  * For a small number of values - an array of fixed size "on the stack".
-  * For large, roaring_bitmap_t is allocated.
-  * For a description of the roaring_bitmap_t, see: https://github.com/RoaringBitmap/CRoaring
-  */
-template <typename T, UInt8 small_set_size>
-class RoaringBitmapWithSmallSet : private boost::noncopyable
-{
-private:
-    using Small = SmallSet<T, small_set_size>;
-    using ValueBuffer = std::vector<T>;
-    bool bUInt64 = sizeof(T) >= 8;
-
-    Small small;
-    roaring_bitmap_t * rb = nullptr;
-    std::shared_ptr<Roaring64Map> rb64 = nullptr;
-
-    void toLarge()
-    {
-        if (isUInt64())
-            rb64 = std::make_shared<Roaring64Map>();
-        else
-            rb = roaring_bitmap_create();
-
-        for (const auto & x : small)
-        {
-            if (isUInt64())
-                rb64->add(static_cast<UInt64>(x.getValue()));
-            else
-                roaring_bitmap_add(rb, x.getValue());
-        }
-    }
-
-public:
-    bool isLarge() const { return rb != nullptr || rb64 != nullptr; }
-
-    bool isSmall() const { return rb == nullptr && rb64 == nullptr; }
-
-    bool isUInt64() const { return bUInt64; }
-
-    ~RoaringBitmapWithSmallSet()
-    {
-        if (isLarge() && !isUInt64())
-            roaring_bitmap_free(rb);
-    }
-
-    void add(T value)
-    {
-        if (isSmall())
-        {
-            if (small.find(value) == small.end())
-            {
-                if (!small.full())
-                    small.insert(value);
-                else
-                {
-                    toLarge();
-                    if (isUInt64())
-                        rb64->add(static_cast<UInt64>(value));
-                    else
-                        roaring_bitmap_add(rb, value);
-                }
-            }
-        }
-        else
-        {
-            if (isUInt64())
-                rb64->add(static_cast<UInt64>(value));
-            else
-                roaring_bitmap_add(rb, value);
-        }
-    }
-
-    UInt64 size() const
-    {
-        if (isSmall())
-        {
-            return small.size();
-        }
-        else
-        {
-            if (isUInt64())
-                return rb64->cardinality();
-            else
-                return roaring_bitmap_get_cardinality(rb);
-        }
-    }
-
-    void merge(const RoaringBitmapWithSmallSet & r1)
-    {
-        if (r1.isLarge())
-        {
-            if (isSmall())
-                toLarge();
-
-            if (isUInt64())
-                *rb64 |= *r1.rb64;
-            else
-                roaring_bitmap_or_inplace(rb, r1.rb);
-        }
-        else
-        {
-            for (const auto & x : r1.small)
-                add(x.getValue());
-        }
-    }
-
-    void read(DB::ReadBuffer & in)
-    {
-        // Container type:  0: SmallSet,  1: RoaringBitmap32,  2: RoaringBitmap64
-        UInt8 containerType = 0;
-        readBinary(containerType, in);
-
-        if (0 == containerType)
-        {
-            small.read(in);
-        }
-        else if (1 == containerType)
-        {
-            std::string s;
-            readStringBinary(s, in);
-            rb = roaring_bitmap_portable_deserialize(s.c_str());
-            // It has been persisted in the bitmap and does not need to merge from small
-            //            for (const auto & x : small) // merge from small
-            //                roaring_bitmap_add(rb, x.getValue());
-        }
-        else
-        {
-            std::string s;
-            readStringBinary(s, in);
-            rb64 = std::make_shared<Roaring64Map>(Roaring64Map::read(s.c_str()));
-        }
-    }
-
-    void write(DB::WriteBuffer & out) const
-    {
-        // Container type:  0: SmallSet,  1: RoaringBitmap32,  2: RoaringBitmap64
-        UInt8 containerType = 0;
-        if (isLarge())
-        {
-            if (isUInt64())
-                containerType = 2;
-            else
-                containerType = 1;
-        }
-
-        writeBinary(containerType, out);
-
-        if (isLarge())
-        {
-            if (isUInt64())
-            {
-                uint32_t expectedsize = rb64->getSizeInBytes();
-                std::string s(expectedsize, 0);
-                rb64->write(const_cast<char *>(s.data()));
-                writeStringBinary(s, out);
-            }
-            else
-            {
-                uint32_t expectedsize = roaring_bitmap_portable_size_in_bytes(rb);
-                std::string s(expectedsize, 0);
-                roaring_bitmap_portable_serialize(rb, const_cast<char *>(s.data()));
-                writeStringBinary(s, out);
-            }
-        }
-        else
-            small.write(out);
-    }
-
-    roaring_bitmap_t * getRb() const { return rb; }
-
-    std::shared_ptr<Roaring64Map> getRb64() const { return rb64; }
-
-    Small & getSmall() const { return small; }
-
-    /**
-     * Get a new roaring_bitmap_t from elements of small
-     */
-    roaring_bitmap_t * getNewRbFromSmall() const
-    {
-        roaring_bitmap_t * smallRb = roaring_bitmap_create();
-        for (const auto & x : small)
-            roaring_bitmap_add(smallRb, x.getValue());
-        return smallRb;
-    }
-
-    /**
-     * Get a new Roaring64Map from elements of small
-     */
-    std::shared_ptr<Roaring64Map> getNewRb64FromSmall() const
-    {
-        std::shared_ptr<Roaring64Map> smallRb64 = std::make_shared<Roaring64Map>();
-        for (const auto & x : small)
-            smallRb64->add(static_cast<UInt64>(x.getValue()));
-        return smallRb64;
-    }
-
-    /**
-     * Computes the intersection between two bitmaps
-     */
-    void rb_and(const RoaringBitmapWithSmallSet & r1)
-    {
-        ValueBuffer buffer;
-        if (isSmall() && r1.isSmall())
-        {
-            // intersect
-            for (const auto & x : small)
-                if (r1.small.find(x.getValue()) != r1.small.end())
-                    buffer.push_back(x.getValue());
-
-            // Clear out the original values
-            small.clear();
-
-            for (const auto & value : buffer)
-                small.insert(value);
-
-            buffer.clear();
-        }
-        else if (isSmall() && r1.isLarge())
-        {
-            for (const auto & x : small)
-            {
-                if (isUInt64())
-                {
-                    if (rb64->contains(static_cast<UInt64>(x.getValue())))
-                        buffer.push_back(x.getValue());
-                }
-                else
-                {
-                    if (roaring_bitmap_contains(r1.rb, x.getValue()))
-                        buffer.push_back(x.getValue());
-                }
-            }
-
-            // Clear out the original values
-            small.clear();
-
-            for (const auto & value : buffer)
-                small.insert(value);
-
-            buffer.clear();
-        }
-        else
-        {
-            if (isUInt64())
-            {
-                std::shared_ptr<Roaring64Map> newRb64 = r1.isSmall() ? r1.getNewRb64FromSmall() : r1.getRb64();
-                *rb64 &= *newRb64;
-            }
-            else
-            {
-                roaring_bitmap_t * rb1 = r1.isSmall() ? r1.getNewRbFromSmall() : r1.getRb();
-                roaring_bitmap_and_inplace(rb, rb1);
-                if (r1.isSmall())
-                    roaring_bitmap_free(rb1);
-            }
-        }
-    }
-
-    /**
-     * Computes the union between two bitmaps.
-     */
-    void rb_or(const RoaringBitmapWithSmallSet & r1) { merge(r1); }
-
-    /**
-     * Computes the symmetric difference (xor) between two bitmaps.
-     */
-    void rb_xor(const RoaringBitmapWithSmallSet & r1)
-    {
-        if (isSmall())
-            toLarge();
-
-        if (isUInt64())
-        {
-            std::shared_ptr<Roaring64Map> newRb64 = r1.isSmall() ? r1.getNewRb64FromSmall() : r1.getRb64();
-            *rb64 ^= *newRb64;
-        }
-        else
-        {
-            roaring_bitmap_t * rb1 = r1.isSmall() ? r1.getNewRbFromSmall() : r1.getRb();
-            roaring_bitmap_xor_inplace(rb, rb1);
-            if (r1.isSmall())
-                roaring_bitmap_free(rb1);
-        }
-    }
-
-    /**
-     * Computes the difference (andnot) between two bitmaps
-     */
-    void rb_andnot(const RoaringBitmapWithSmallSet & r1)
-    {
-        ValueBuffer buffer;
-        if (isSmall() && r1.isSmall())
-        {
-            // subtract
-            for (const auto & x : small)
-                if (r1.small.find(x.getValue()) == r1.small.end())
-                    buffer.push_back(x.getValue());
-
-            // Clear out the original values
-            small.clear();
-
-            for (const auto & value : buffer)
-                small.insert(value);
-
-            buffer.clear();
-        }
-        else if (isSmall() && r1.isLarge())
-        {
-            for (const auto & x : small)
-            {
-                if (isUInt64())
-                {
-                    if (!rb64->contains(static_cast<UInt64>(x.getValue())))
-                        buffer.push_back(x.getValue());
-                }
-                else
-                {
-                    if (!roaring_bitmap_contains(r1.rb, x.getValue()))
-                        buffer.push_back(x.getValue());
-                }
-            }
-
-            // Clear out the original values
-            small.clear();
-
-            for (const auto & value : buffer)
-                small.insert(value);
-
-            buffer.clear();
-        }
-        else
-        {
-            if (isUInt64())
-            {
-                std::shared_ptr<Roaring64Map> newRb64 = r1.isSmall() ? r1.getNewRb64FromSmall() : r1.getRb64();
-                *rb64 -= *newRb64;
-            }
-            else
-            {
-                roaring_bitmap_t * rb1 = r1.isSmall() ? r1.getNewRbFromSmall() : r1.getRb();
-                roaring_bitmap_andnot_inplace(rb, rb1);
-                if (r1.isSmall())
-                    roaring_bitmap_free(rb1);
-            }
-        }
-    }
-
-    /**
-     * Computes the cardinality of the intersection between two bitmaps.
-     */
-    UInt64 rb_and_cardinality(const RoaringBitmapWithSmallSet & r1) const
-    {
-        UInt64 retSize = 0;
-        if (isSmall() && r1.isSmall())
-        {
-            for (const auto & x : small)
-                if (r1.small.find(x.getValue()) != r1.small.end())
-                    ++retSize;
-        }
-        else if (isSmall() && r1.isLarge())
-        {
-            for (const auto & x : small)
-            {
-                if (isUInt64())
-                {
-                    if (rb64->contains(static_cast<UInt64>(x.getValue())))
-                        ++retSize;
-                }
-                else
-                {
-                    if (roaring_bitmap_contains(r1.rb, x.getValue()))
-                        ++retSize;
-                }
-            }
-        }
-        else
-        {
-            if (isUInt64())
-            {
-                std::shared_ptr<Roaring64Map> newRb64 = r1.isSmall() ? r1.getNewRb64FromSmall() : r1.getRb64();
-                retSize = (*rb64 & *newRb64).cardinality();
-            }
-            else
-            {
-                roaring_bitmap_t * rb1 = r1.isSmall() ? r1.getNewRbFromSmall() : r1.getRb();
-                retSize = roaring_bitmap_and_cardinality(rb, rb1);
-                if (r1.isSmall())
-                    roaring_bitmap_free(rb1);
-            }
-        }
-        return retSize;
-    }
-
-    /**
-     * Computes the cardinality of the union between two bitmaps.
-    */
-    UInt64 rb_or_cardinality(const RoaringBitmapWithSmallSet & r1) const
-    {
-        UInt64 c1 = size();
-        UInt64 c2 = r1.size();
-        UInt64 inter = rb_and_cardinality(r1);
-        return c1 + c2 - inter;
-    }
-
-    /**
-     * Computes the cardinality of the symmetric difference (andnot) between two bitmaps.
-    */
-    UInt64 rb_xor_cardinality(const RoaringBitmapWithSmallSet & r1) const
-    {
-        UInt64 c1 = size();
-        UInt64 c2 = r1.size();
-        UInt64 inter = rb_and_cardinality(r1);
-        return c1 + c2 - 2 * inter;
-    }
-
-    /**
-     * Computes the cardinality of the difference (andnot) between two bitmaps.
-     */
-    UInt64 rb_andnot_cardinality(const RoaringBitmapWithSmallSet & r1) const
-    {
-        UInt64 c1 = size();
-        UInt64 inter = rb_and_cardinality(r1);
-        return c1 - inter;
-    }
-
-    /**
-     * Return 1 if the two bitmaps contain the same elements.
-     */
-    UInt8 rb_equals(const RoaringBitmapWithSmallSet & r1)
-    {
-        if (isSmall())
-            toLarge();
-
-        UInt8 is_true = 0;
-        if (isUInt64())
-        {
-            std::shared_ptr<Roaring64Map> newRb64 = r1.isSmall() ? r1.getNewRb64FromSmall() : r1.getRb64();
-            is_true = *rb64 == *newRb64;
-        }
-        else
-        {
-            roaring_bitmap_t * rb1 = r1.isSmall() ? r1.getNewRbFromSmall() : r1.getRb();
-            is_true = roaring_bitmap_equals(rb, rb1);
-            if (r1.isSmall())
-                roaring_bitmap_free(rb1);
-        }
-
-        return is_true;
-    }
-
-    /**
-     * Check whether two bitmaps intersect.
-     * Intersection with an empty set is always 0 (consistent with hasAny).
-     */
-    UInt8 rb_intersect(const RoaringBitmapWithSmallSet & r1) const
-    {
-        if (isSmall())
-        {
-            if (r1.isSmall())
-            {
-                for (const auto & x : r1.small)
-                    if (small.find(x.getValue()) != small.end())
-                        return 1;
-            }
-            else
-            {
-                for (const auto & x : small)
-                {
-                    if (isUInt64())
-                    {
-                        if (r1.rb64->contains(static_cast<UInt64>(x.getValue())))
-                            return 1;
-                    }
-                    else
-                    {
-                        if (roaring_bitmap_contains(r1.rb, x.getValue()))
-                            return 1;
-                    }
-                }
-            }
-        }
-        else if (r1.isSmall())
-        {
-            for (const auto & x : r1.small)
-            {
-                if (isUInt64())
-                {
-                    if (rb64->contains(static_cast<UInt64>(x.getValue())))
-                        return 1;
-                }
-                else
-                {
-                    if (roaring_bitmap_contains(rb, x.getValue()))
-                        return 1;
-                }
-            }
-        }
-        else
-        {
-            if (isUInt64())
-            {
-                if ((*rb64 & *r1.rb64).cardinality() > 0)
-                    return 1;
-            }
-            else
-            {
-                if (roaring_bitmap_intersect(rb, r1.rb))
-                    return 1;
-            }
-        }
-
-        return 0;
-    }
-
-    /**
-     * Check whether the argument is the subset of this set.
-     * Empty set is a subset of any other set (consistent with hasAll).
-     */
-    UInt8 rb_is_subset(const RoaringBitmapWithSmallSet & r1) const
-    {
-        if (isSmall())
-        {
-            if (r1.isSmall())
-            {
-                for (const auto & x : r1.small)
-                    if (small.find(x.getValue()) == small.end())
-                        return 0;
-            }
-            else
-            {
-                UInt64 r1_size = r1.size();
-
-                if (r1_size > small.size())
-                    return 0; // A bigger set can not be a subset of ours.
-
-                // This is a rare case with a small number of elements on
-                // both sides: r1 was promoted to large for some reason and
-                // it is still not larger than our small set.
-                // If r1 is our subset then our size must be equal to
-                // r1_size + number of not found elements, if this sum becomes
-                // greater then r1 is not a subset.
-                for (const auto & x : small)
-                {
-                    if (isUInt64())
-                    {
-                        if (!r1.rb64->contains(static_cast<UInt64>(x.getValue())) && ++r1_size > small.size())
-                            return 0;
-                    }
-                    else
-                    {
-                        if (!roaring_bitmap_contains(r1.rb, x.getValue()) && ++r1_size > small.size())
-                            return 0;
-                    }
-                }
-            }
-        }
-        else if (r1.isSmall())
-        {
-            for (const auto & x : r1.small)
-            {
-                if (isUInt64())
-                {
-                    if (!rb64->contains(static_cast<UInt64>(x.getValue())))
-                        return 0;
-                }
-                else
-                {
-                    if (!roaring_bitmap_contains(rb, x.getValue()))
-                        return 0;
-                }
-            }
-        }
-        else
-        {
-            if (isUInt64())
-            {
-                if (!r1.rb64->isSubset(*rb64))
-                    return 0;
-            }
-            else
-            {
-                if (!roaring_bitmap_is_subset(r1.rb, rb))
-                    return 0;
-            }
-        }
-        return 1;
-    }
-
-    /**
-     * Check whether this bitmap contains the argument.
-     */
-    UInt8 rb_contains(const UInt64 x) const
-    {
-        if (isSmall())
-        {
-            return small.find(x) != small.end();
-        }
-        else
-        {
-            if (isUInt64())
-            {
-                return rb64->contains(x);
-            }
-            else
-            {
-                return roaring_bitmap_contains(rb, x);
-            }
-        }
-    }
-
-    /**
-     * Remove value
-     */
-    void rb_remove(UInt64 offsetid)
-    {
-        if (isSmall())
-            toLarge();
-
-        if (isUInt64())
-            rb64->remove(offsetid);
-        else
-            roaring_bitmap_remove(rb, offsetid);
-    }
-
-    /**
-     * compute (in place) the negation of the roaring bitmap within a specified
-     * interval: [range_start, range_end). The number of negated values is
-     * range_end - range_start.
-     * Areas outside the range are passed through unchanged.
-     */
-    void rb_flip(UInt64 offsetstart, UInt64 offsetend)
-    {
-        if (isSmall())
-            toLarge();
-        if (isUInt64())
-            rb64->flip(offsetstart, offsetend);
-        else
-            roaring_bitmap_flip_inplace(rb, offsetstart, offsetend);
-    }
-
-    /**
-     * returns the number of integers that are smaller or equal to offsetid.
-     */
-    UInt64 rb_rank(UInt64 offsetid)
-    {
-        if (isSmall())
-            toLarge();
-
-        if (isUInt64())
-            return rb64->rank(offsetid);
-        else
-            return roaring_bitmap_rank(rb, offsetid);
-    }
-
-    /**
-     * Convert elements to integer array, return number of elements
-     */
-    template <typename Element>
-    UInt64 rb_to_array(PaddedPODArray<Element> & res_data) const
-    {
-        UInt64 count = 0;
-        if (isSmall())
-        {
-            for (const auto & x : small)
-            {
-                res_data.emplace_back(x.getValue());
-                count++;
-            }
-        }
-        else
-        {
-            if (isUInt64())
-            {
-                for (Roaring64Map::const_iterator iterator = rb64->begin(); iterator != rb64->end(); iterator++)
-                {
-                    res_data.emplace_back(*iterator);
-                    count++;
-                }
-            }
-            else
-            {
-                roaring_uint32_iterator_t iterator;
-                roaring_init_iterator(rb, &iterator);
-                while (iterator.has_value)
-                {
-                    res_data.emplace_back(iterator.current_value);
-                    roaring_advance_uint32_iterator(&iterator);
-                    count++;
-                }
-            }
-        }
-        return count;
-    }
-
-    /**
-     * Return new set with specified range (not include the range_end)
-     */
-    UInt64 rb_range(UInt64 range_start, UInt64 range_end, RoaringBitmapWithSmallSet & r1) const
-    {
-        UInt64 count = 0;
-        if (range_start >= range_end)
-            return count;
-        if (isSmall())
-        {
-            for (const auto & x : small)
-            {
-                T val = x.getValue();
-                if (UInt32(val) >= range_start && UInt32(val) < range_end)
-                {
-                    r1.add(val);
-                    ++count;
-                }
-            }
-        }
-        else
-        {
-            if (isUInt64())
-            {
-                for (Roaring64Map::const_iterator iterator = rb64->begin(); iterator != rb64->end(); iterator++)
-                {
-                    if (*iterator < range_start)
-                        continue;
-
-                    if (*iterator < range_end)
-                    {
-                        r1.add(*iterator);
-                        ++count;
-                    }
-                    else
-                        break;
-                }
-            }
-            else
-            {
-                roaring_uint32_iterator_t iterator;
-                roaring_init_iterator(rb, &iterator);
-                roaring_move_uint32_iterator_equalorlarger(&iterator, range_start);
-                while (iterator.has_value && UInt32(iterator.current_value) < range_end)
-                {
-                    r1.add(iterator.current_value);
-                    roaring_advance_uint32_iterator(&iterator);
-                    ++count;
-                }
-            }
-        }
-        return count;
-    }
-
-    /**
-     * Return new set of the smallest `limit` values in set which is no less than `range_start`.
-     */
-    UInt64 rb_limit(UInt64 range_start, UInt64 limit, RoaringBitmapWithSmallSet & r1) const
-    {
-        UInt64 count = 0;
-        if (isSmall())
-        {
-            std::vector<T> answer;
-            for (const auto & x : small)
-            {
-                T val = x.getValue();
-                if (UInt32(val) >= range_start)
-                {
-                    answer.push_back(val);
-                }
-            }
-            sort(answer.begin(), answer.end());
-            if (limit > answer.size())
-                limit = answer.size();
-            for (size_t i = 0; i < limit; ++i)
-                r1.add(answer[i]);
-            count = UInt64(limit);
-        }
-        else
-        {
-            if (isUInt64())
-            {
-                for (Roaring64Map::const_iterator iterator = rb64->begin(); iterator != rb64->end(); iterator++)
-                {
-                    if (*iterator < range_start)
-                        continue;
-
-                    if (count < limit)
-                    {
-                        r1.add(*iterator);
-                        ++count;
-                    }
-                    else
-                        break;
-                }
-            }
-            else
-            {
-                roaring_uint32_iterator_t iterator;
-                roaring_init_iterator(rb, &iterator);
-                roaring_move_uint32_iterator_equalorlarger(&iterator, range_start);
-                while (UInt32(count) < limit && iterator.has_value)
-                {
-                    r1.add(iterator.current_value);
-                    roaring_advance_uint32_iterator(&iterator);
-                    ++count;
-                }
-            }
-        }
-        return count;
-    }
-
-    UInt64 rb_min() const
-    {
-        UInt64 min_val = UINT32_MAX;
-        if (isSmall())
-        {
-            for (const auto & x : small)
-            {
-                T val = x.getValue();
-                if (UInt64(val) < min_val)
-                {
-                    min_val = UInt64(val);
-                }
-            }
-        }
-        else
-        {
-            if (isUInt64())
-                min_val = rb64->minimum();
-            else
-                min_val = UInt64(roaring_bitmap_minimum(rb));
-        }
-        return min_val;
-    }
-
-    UInt64 rb_max() const
-    {
-        UInt64 max_val = 0;
-        if (isSmall())
-        {
-            for (const auto & x : small)
-            {
-                T val = x.getValue();
-                if (UInt64(val) > max_val)
-                {
-                    max_val = UInt64(val);
-                }
-            }
-        }
-        else
-        {
-            if (isUInt64())
-                max_val = rb64->maximum();
-            else
-                max_val = UInt64(roaring_bitmap_maximum(rb));
-        }
-        return max_val;
-    }
-
-    /**
-     * Replace value
-     */
-    void rb_replace(const UInt32 * from_vals, const UInt32 * to_vals, size_t num)
-    {
-        if (isSmall())
-            toLarge();
-        for (size_t i = 0; i < num; ++i)
-        {
-            if (from_vals[i] == to_vals[i])
-                continue;
-
-            if (isUInt64())
-            {
-                bool changed = rb64->removeChecked(from_vals[i]);
-                if (changed)
-                    rb64->add(to_vals[i]);
-            }
-            else
-            {
-                bool changed = roaring_bitmap_remove_checked(rb, from_vals[i]);
-                if (changed)
-                    roaring_bitmap_add(rb, to_vals[i]);
-            }
-        }
-    }
-
-private:
- 
-};
-
-template <typename T>
-struct AggregateFunctionGroupBitmapData
-{
-    bool doneFirst = false;
-    RoaringBitmapWithSmallSet<T, 32> rbs;
-    static const char * name() { return "groupBitmap"; }
-};
-
-
-}
+#pragma once
+
+#include <algorithm>
+#include <IO/ReadHelpers.h>
+#include <IO/WriteHelpers.h>
+#include <boost/noncopyable.hpp>
+#include <Common/HashTable/SmallTable.h>
+#include <Common/PODArray.h>
+
+// Include this header last, because it is an auto-generated dump of questionable
+// garbage that breaks the build (e.g. it changes _POSIX_C_SOURCE).
+// TODO: find out what it is. On github, they have proper interface headers like
+// this one: https://github.com/RoaringBitmap/CRoaring/blob/master/include/roaring/roaring.h
+#include <roaring.hh>
+#include <roaring64map.hh>
+
+namespace DB
+{
+/**
+  * For a small number of values - an array of fixed size "on the stack".
+  * For large, roaring bitmap is allocated.
+  * For a description of the roaring_bitmap_t, see: https://github.com/RoaringBitmap/CRoaring
+  */
+template <typename T, UInt8 small_set_size>
+class RoaringBitmapWithSmallSet : private boost::noncopyable
+{
+private:
+    using Small = SmallSet<T, small_set_size>;
+    using ValueBuffer = std::vector<T>;
+    Small small;
+    using RoaringBitmap = std::conditional_t<sizeof(T) >= 8, Roaring64Map, Roaring>;
+    using Value = std::conditional_t<sizeof(T) >= 8, UInt64, UInt32>;
+    std::shared_ptr<RoaringBitmap> rb64 = nullptr;
+
+    void toLarge()
+    {
+        rb64 = std::make_shared<RoaringBitmap>();
+        for (const auto & x : small)
+            rb64->add(static_cast<Value>(x.getValue()));
+        small.clear();
+    }
+
+public:
+    bool isLarge() const { return rb64 != nullptr; }
+
+    bool isSmall() const { return rb64 == nullptr; }
+
+    void add(T value)
+    {
+        if (isSmall())
+        {
+            if (small.find(value) == small.end())
+            {
+                if (!small.full())
+                    small.insert(value);
+                else
+                {
+                    toLarge();
+                    rb64->add(static_cast<Value>(value));
+                }
+            }
+        }
+        else
+        {
+            rb64->add(static_cast<Value>(value));
+        }
+    }
+
+    UInt64 size() const
+    {
+        if (isSmall())
+            return small.size();
+        else
+            return rb64->cardinality();
+    }
+
+    void merge(const RoaringBitmapWithSmallSet & r1)
+    {
+        if (r1.isLarge())
+        {
+            if (isSmall())
+                toLarge();
+
+            *rb64 |= *r1.rb64;
+        }
+        else
+        {
+            for (const auto & x : r1.small)
+                add(x.getValue());
+        }
+    }
+
+    void read(DB::ReadBuffer & in)
+    {
+        // Container type:  0: SmallSet,  1: RoaringBitmap
+        UInt8 containerType = 0;
+        readBinary(containerType, in);
+
+        if (0 == containerType)
+        {
+            small.read(in);
+        }
+        else if (1 == containerType)
+        {
+            size_t size;
+            readVarUInt(size, in);
+            std::unique_ptr<char[]> buf(new char[size]);
+            in.readStrict(buf.get(), size);
+            rb64 = std::make_shared<RoaringBitmap>(RoaringBitmap::read(buf.get()));
+        }
+    }
+
+    void write(DB::WriteBuffer & out) const
+    {
+        // Container type:  0: SmallSet,  1: RoaringBitmap
+        UInt8 containerType = isLarge() ? 1 : 0;
+        writeBinary(containerType, out);
+        if (0 == containerType)
+        {
+            small.write(out);
+        }
+        else if (1 == containerType)
+        {
+            auto size = rb64->getSizeInBytes();
+            writeVarUInt(size, out);
+            std::unique_ptr<char[]> buf(new char[size]);
+            rb64->write(buf.get());
+            out.write(buf.get(), size);
+        }
+    }
+
+    /**
+     * Get a new RoaringBitmap from elements of small
+     */
+    std::shared_ptr<RoaringBitmap> getNewRb64FromSmall() const
+    {
+        std::shared_ptr<RoaringBitmap> smallRb64 = std::make_shared<RoaringBitmap>();
+        for (const auto & x : small)
+            smallRb64->add(static_cast<Value>(x.getValue()));
+        return smallRb64;
+    }
+
+    /**
+     * Computes the intersection between two bitmaps
+     */
+    void rb_and(const RoaringBitmapWithSmallSet & r1)
+    {
+        ValueBuffer buffer;
+        if (isSmall() && r1.isSmall())
+        {
+            // intersect
+            for (const auto & x : small)
+                if (r1.small.find(x.getValue()) != r1.small.end())
+                    buffer.push_back(x.getValue());
+
+            // Clear out the original values
+            small.clear();
+
+            for (const auto & value : buffer)
+                small.insert(value);
+
+            buffer.clear();
+        }
+        else if (isSmall() && r1.isLarge())
+        {
+            for (const auto & x : small)
+            {
+                if (rb64->contains(static_cast<Value>(x.getValue())))
+                    buffer.push_back(x.getValue());
+            }
+
+            // Clear out the original values
+            small.clear();
+
+            for (const auto & value : buffer)
+                small.insert(value);
+
+            buffer.clear();
+        }
+        else
+        {
+            std::shared_ptr<RoaringBitmap> newRb64 = r1.isSmall() ? r1.getNewRb64FromSmall() : r1.rb64;
+            *rb64 &= *newRb64;
+        }
+    }
+
+    /**
+     * Computes the union between two bitmaps.
+     */
+    void rb_or(const RoaringBitmapWithSmallSet & r1) { merge(r1); }
+
+    /**
+     * Computes the symmetric difference (xor) between two bitmaps.
+     */
+    void rb_xor(const RoaringBitmapWithSmallSet & r1)
+    {
+        if (isSmall())
+            toLarge();
+
+        std::shared_ptr<RoaringBitmap> newRb64 = r1.isSmall() ? r1.getNewRb64FromSmall() : r1.rb64;
+        *rb64 ^= *newRb64;
+    }
+
+    /**
+     * Computes the difference (andnot) between two bitmaps
+     */
+    void rb_andnot(const RoaringBitmapWithSmallSet & r1)
+    {
+        ValueBuffer buffer;
+        if (isSmall() && r1.isSmall())
+        {
+            // subtract
+            for (const auto & x : small)
+                if (r1.small.find(x.getValue()) == r1.small.end())
+                    buffer.push_back(x.getValue());
+
+            // Clear out the original values
+            small.clear();
+
+            for (const auto & value : buffer)
+                small.insert(value);
+
+            buffer.clear();
+        }
+        else if (isSmall() && r1.isLarge())
+        {
+            for (const auto & x : small)
+            {
+                if (!rb64->contains(static_cast<Value>(x.getValue())))
+                    buffer.push_back(x.getValue());
+            }
+
+            // Clear out the original values
+            small.clear();
+
+            for (const auto & value : buffer)
+                small.insert(value);
+
+            buffer.clear();
+        }
+        else
+        {
+            std::shared_ptr<RoaringBitmap> newRb64 = r1.isSmall() ? r1.getNewRb64FromSmall() : r1.rb64;
+            *rb64 -= *newRb64;
+        }
+    }
+
+    /**
+     * Computes the cardinality of the intersection between two bitmaps.
+     */
+    UInt64 rb_and_cardinality(const RoaringBitmapWithSmallSet & r1) const
+    {
+        UInt64 ret = 0;
+        if (isSmall() && r1.isSmall())
+        {
+            for (const auto & x : small)
+                if (r1.small.find(x.getValue()) != r1.small.end())
+                    ++ret;
+        }
+        else if (isSmall() && r1.isLarge())
+        {
+            for (const auto & x : small)
+            {
+                if (rb64->contains(static_cast<Value>(x.getValue())))
+                    ++ret;
+            }
+        }
+        else
+        {
+            std::shared_ptr<RoaringBitmap> newRb64 = r1.isSmall() ? r1.getNewRb64FromSmall() : r1.rb64;
+            ret = (*rb64 & *newRb64).cardinality();
+        }
+        return ret;
+    }
+
+    /**
+     * Computes the cardinality of the union between two bitmaps.
+    */
+    UInt64 rb_or_cardinality(const RoaringBitmapWithSmallSet & r1) const
+    {
+        UInt64 c1 = size();
+        UInt64 c2 = r1.size();
+        UInt64 inter = rb_and_cardinality(r1);
+        return c1 + c2 - inter;
+    }
+
+    /**
+     * Computes the cardinality of the symmetric difference (andnot) between two bitmaps.
+    */
+    UInt64 rb_xor_cardinality(const RoaringBitmapWithSmallSet & r1) const
+    {
+        UInt64 c1 = size();
+        UInt64 c2 = r1.size();
+        UInt64 inter = rb_and_cardinality(r1);
+        return c1 + c2 - 2 * inter;
+    }
+
+    /**
+     * Computes the cardinality of the difference (andnot) between two bitmaps.
+     */
+    UInt64 rb_andnot_cardinality(const RoaringBitmapWithSmallSet & r1) const
+    {
+        UInt64 c1 = size();
+        UInt64 inter = rb_and_cardinality(r1);
+        return c1 - inter;
+    }
+
+    /**
+     * Return 1 if the two bitmaps contain the same elements.
+     */
+    UInt8 rb_equals(const RoaringBitmapWithSmallSet & r1)
+    {
+        if (isSmall())
+            toLarge();
+
+        std::shared_ptr<RoaringBitmap> newRb64 = r1.isSmall() ? r1.getNewRb64FromSmall() : r1.rb64;
+        return *rb64 == *newRb64;
+    }
+
+    /**
+     * Check whether two bitmaps intersect.
+     * Intersection with an empty set is always 0 (consistent with hasAny).
+     */
+    UInt8 rb_intersect(const RoaringBitmapWithSmallSet & r1) const
+    {
+        if (isSmall())
+        {
+            if (r1.isSmall())
+            {
+                for (const auto & x : r1.small)
+                    if (small.find(x.getValue()) != small.end())
+                        return 1;
+            }
+            else
+            {
+                for (const auto & x : small)
+                {
+                    if (r1.rb64->contains(static_cast<Value>(x.getValue())))
+                        return 1;
+                }
+            }
+        }
+        else if (r1.isSmall())
+        {
+            for (const auto & x : r1.small)
+            {
+                if (rb64->contains(static_cast<Value>(x.getValue())))
+                    return 1;
+            }
+        }
+        else
+        {
+            if ((*rb64 & *r1.rb64).cardinality() > 0)
+                return 1;
+        }
+
+        return 0;
+    }
+
+    /**
+     * Check whether the argument is the subset of this set.
+     * Empty set is a subset of any other set (consistent with hasAll).
+     */
+    UInt8 rb_is_subset(const RoaringBitmapWithSmallSet & r1) const
+    {
+        if (isSmall())
+        {
+            if (r1.isSmall())
+            {
+                for (const auto & x : r1.small)
+                    if (small.find(x.getValue()) == small.end())
+                        return 0;
+            }
+            else
+            {
+                UInt64 r1_size = r1.size();
+
+                // A bigger set can not be a subset of ours.
+                if (r1_size > small.size())
+                    return 0;
+
+                // This is a rare case with a small number of elements on
+                // both sides: r1 was promoted to large for some reason and
+                // it is still not larger than our small set.
+                // If r1 is our subset then our size must be equal to
+                // r1_size + number of not found elements, if this sum becomes
+                // greater then r1 is not a subset.
+                for (const auto & x : small)
+                {
+                    if (!r1.rb64->contains(static_cast<Value>(x.getValue())) && ++r1_size > small.size())
+                        return 0;
+                }
+            }
+        }
+        else if (r1.isSmall())
+        {
+            for (const auto & x : r1.small)
+            {
+                if (!rb64->contains(static_cast<Value>(x.getValue())))
+                    return 0;
+            }
+        }
+        else
+        {
+            if (!r1.rb64->isSubset(*rb64))
+                return 0;
+        }
+        return 1;
+    }
+
+    /**
+     * Check whether this bitmap contains the argument.
+     */
+    UInt8 rb_contains(UInt64 x) const
+    {
+        if (isSmall())
+            return small.find(x) != small.end();
+        else
+            return rb64->contains(x);
+    }
+
+    /**
+     * Remove value
+     */
+    void rb_remove(UInt64 x)
+    {
+        if (isSmall())
+            toLarge();
+
+        rb64->remove(x);
+    }
+
+    /**
+     * compute (in place) the negation of the roaring bitmap within a specified
+     * interval: [range_start, range_end). The number of negated values is
+     * range_end - range_start.
+     * Areas outside the range are passed through unchanged.
+     */
+    void rb_flip(UInt64 begin, UInt64 end)
+    {
+        if (isSmall())
+            toLarge();
+
+        rb64->flip(begin, end);
+    }
+
+    /**
+     * returns the number of integers that are smaller or equal to offsetid.
+     */
+    UInt64 rb_rank(UInt64 x)
+    {
+        if (isSmall())
+            toLarge();
+
+        return rb64->rank(x);
+    }
+
+    /**
+     * Convert elements to integer array, return number of elements
+     */
+    template <typename Element>
+    UInt64 rb_to_array(PaddedPODArray<Element> & res) const
+    {
+        UInt64 count = 0;
+        if (isSmall())
+        {
+            for (const auto & x : small)
+            {
+                res.emplace_back(x.getValue());
+                ++count;
+            }
+        }
+        else
+        {
+            for (auto it = rb64->begin(); it != rb64->end(); ++it)
+            {
+                res.emplace_back(*it);
+                ++count;
+            }
+        }
+        return count;
+    }
+
+    /**
+     * Return new set with specified range (not include the range_end)
+     */
+    UInt64 rb_range(UInt64 range_start, UInt64 range_end, RoaringBitmapWithSmallSet & r1) const
+    {
+        UInt64 count = 0;
+        if (range_start >= range_end)
+            return count;
+        if (isSmall())
+        {
+            for (const auto & x : small)
+            {
+                T val = x.getValue();
+                if (UInt32(val) >= range_start && UInt32(val) < range_end)
+                {
+                    r1.add(val);
+                    ++count;
+                }
+            }
+        }
+        else
+        {
+            for (auto it = rb64->begin(); it != rb64->end(); ++it)
+            {
+                if (*it < range_start)
+                    continue;
+
+                if (*it < range_end)
+                {
+                    r1.add(*it);
+                    ++count;
+                }
+                else
+                    break;
+            }
+        }
+        return count;
+    }
+
+    /**
+     * Return new set of the smallest `limit` values in set which is no less than `range_start`.
+     */
+    UInt64 rb_limit(UInt64 range_start, UInt64 limit, RoaringBitmapWithSmallSet & r1) const
+    {
+        if (limit == 0)
+            return 0;
+
+        if (isSmall())
+        {
+            std::vector<T> answer;
+            for (const auto & x : small)
+            {
+                T val = x.getValue();
+                if (UInt32(val) >= range_start)
+                {
+                    answer.push_back(val);
+                }
+            }
+            if (limit < answer.size())
+            {
+                std::nth_element(answer.begin(), answer.begin() + limit, answer.end());
+                answer.resize(limit);
+            }
+
+            for (const auto & elem : answer)
+                r1.add(elem);
+            return answer.size();
+        }
+        else
+        {
+            UInt64 count = 0;
+            for (auto it = rb64->begin(); it != rb64->end(); ++it)
+            {
+                if (*it < range_start)
+                    continue;
+
+                if (count < limit)
+                {
+                    r1.add(*it);
+                    ++count;
+                }
+                else
+                    break;
+            }
+            return count;
+        }
+    }
+
+    UInt64 rb_min() const
+    {
+        if (isSmall())
+        {
+            auto min_val = std::numeric_limits<std::make_unsigned_t<T>>::max();
+            for (const auto & x : small)
+            {
+                auto val = x.getValue();
+                if (val < min_val)
+                    min_val = val;
+            }
+            return min_val;
+        }
+        else
+            return rb64->minimum();
+    }
+
+    UInt64 rb_max() const
+    {
+        if (isSmall())
+        {
+            auto max_val = std::numeric_limits<std::make_unsigned_t<T>>::min();
+            for (const auto & x : small)
+            {
+                auto val = x.getValue();
+                if (val < max_val)
+                    max_val = val;
+            }
+            return max_val;
+        }
+        else
+            return rb64->maximum();
+    }
+
+    /**
+     * Replace value
+     */
+    void rb_replace(const UInt32 * from_vals, const UInt32 * to_vals, size_t num)
+    {
+        if (isSmall())
+            toLarge();
+
+        for (size_t i = 0; i < num; ++i)
+        {
+            if (from_vals[i] == to_vals[i])
+                continue;
+            bool changed = rb64->removeChecked(from_vals[i]);
+            if (changed)
+                rb64->add(to_vals[i]);
+        }
+    }
+};
+
+template <typename T>
+struct AggregateFunctionGroupBitmapData
+{
+    // If false, all bitmap operations will be treated as merge to initialize the state
+    bool init = false;
+    RoaringBitmapWithSmallSet<T, 32> rbs;
+    static const char * name() { return "groupBitmap"; }
+};
+
+
+}