Repository: man-group/ArcticDB (language: python)

Issue:
GroupBy and aggregation will not work with sparse  aggregation columns



Discussion / hints:

Fix patch:
diff --git a/.github/workflows/build_steps.yml b/.github/workflows/build_steps.yml
index f7fe44e86e..8f67931917 100644
--- a/.github/workflows/build_steps.yml
+++ b/.github/workflows/build_steps.yml
@@ -83,6 +83,8 @@ jobs:
       - name: Enable Windows compiler commands
         if: matrix.os == 'windows'
         uses: ilammy/msvc-dev-cmd@v1.12.1
+        with:
+          toolset: 14.37
 
       - name: Extra envs
         # This has to come after msvc-dev-cmd to overwrite the bad VCPKG_ROOT it sets
diff --git a/cpp/arcticdb/processing/aggregation.cpp b/cpp/arcticdb/processing/aggregation.cpp
index 5328a27445..ae75109efc 100644
--- a/cpp/arcticdb/processing/aggregation.cpp
+++ b/cpp/arcticdb/processing/aggregation.cpp
@@ -9,21 +9,8 @@
 
 #include <cmath>
 
-namespace arcticdb {
-
-void add_data_type_impl(DataType data_type, std::optional<DataType>& current_data_type) {
-    if (current_data_type.has_value()) {
-        auto common_type = has_valid_common_type(entity::TypeDescriptor(*current_data_type, 0),
-                                                 entity::TypeDescriptor(data_type, 0));
-        schema::check<ErrorCode::E_UNSUPPORTED_COLUMN_TYPE>(
-                common_type.has_value(),
-                "Cannot perform aggregation on column, incompatible types present: {} and {}",
-                entity::TypeDescriptor(*current_data_type, 0), entity::TypeDescriptor(data_type, 0));
-        current_data_type = common_type->data_type();
-    } else {
-        current_data_type = data_type;
-    }
-}
+namespace arcticdb
+{
 
 void MinMaxAggregatorData::aggregate(const ColumnWithStrings& input_column) {
     entity::details::visit_type(input_column.column_->type().data_type(), [&input_column, that=this] (auto type_desc_tag) {
@@ -74,6 +61,87 @@ SegmentInMemory MinMaxAggregatorData::finalize(const std::vector<ColumnName>& ou
     return seg;
 }
 
+namespace
+{
+    void add_data_type_impl(DataType data_type, std::optional<DataType>& current_data_type) {
+        if (current_data_type.has_value()) {
+            auto common_type = has_valid_common_type(entity::TypeDescriptor(*current_data_type, 0),
+                                                     entity::TypeDescriptor(data_type, 0));
+            schema::check<ErrorCode::E_UNSUPPORTED_COLUMN_TYPE>(
+                    common_type.has_value(),
+                    "Cannot perform aggregation on column, incompatible types present: {} and {}",
+                    entity::TypeDescriptor(*current_data_type, 0), entity::TypeDescriptor(data_type, 0));
+            current_data_type = common_type->data_type();
+        } else {
+            current_data_type = data_type;
+        }
+    }
+
+    inline util::BitMagic::enumerator::value_type deref(util::BitMagic::enumerator iter) {
+        return *iter;
+    }
+
+    inline std::size_t deref(std::size_t index) {
+        return index;
+    }
+
+    template<typename T, typename T2=void>
+    struct OutputType;
+
+    template <typename InputType>
+    struct OutputType <InputType, typename std::enable_if_t<is_floating_point_type(InputType::DataTypeTag::data_type)>> {
+        using type = ScalarTagType<DataTypeTag<DataType::FLOAT64>>;
+    };
+
+    template <typename InputType>
+    struct OutputType <InputType, typename std::enable_if_t<is_unsigned_type(InputType::DataTypeTag::data_type)>> {
+        using type = ScalarTagType<DataTypeTag<DataType::UINT64>>;
+    };
+
+    template <typename InputType>
+    struct OutputType<InputType, typename std::enable_if_t<is_signed_type(InputType::DataTypeTag::data_type) && is_integer_type(InputType::DataTypeTag::data_type)>> {
+        using type = ScalarTagType<DataTypeTag<DataType::INT64>>;
+    };
+
+    template<>
+    struct OutputType<DataTypeTag<DataType::BOOL8>, void> {
+        using type = ScalarTagType<DataTypeTag<DataType::BOOL8>>;
+    };
+
+    template<>
+    struct OutputType<DataTypeTag<DataType::NANOSECONDS_UTC64>, void> {
+        using type = ScalarTagType<DataTypeTag<DataType::NANOSECONDS_UTC64>>;
+    };
+
+    template<>
+    struct OutputType<DataTypeTag<DataType::EMPTYVAL>, void> {
+        using type = ScalarTagType<DataTypeTag<DataType::EMPTYVAL>>;
+    };
+}
+
+/**********************
+ * AggregatorDataBase *
+ **********************/
+
+AggregatorDataBase::AggregatorDataBase(const AggregatorDataBase&)
+{
+    log::version().warn("Copying potentially large buffer in AggregatorData");
+}
+
+AggregatorDataBase& AggregatorDataBase::operator=(const AggregatorDataBase&)
+{
+    log::version().warn("Copying potentially large buffer in AggregatorData");
+    return *this;
+}
+
+/*********************
+ * SumAggregatorData *
+ *********************/
+
+void SumAggregatorData::add_data_type(DataType data_type) {
+    add_data_type_impl(data_type, data_type_);
+}
+
 void SumAggregatorData::aggregate(const std::optional<ColumnWithStrings>& input_column, const std::vector<size_t>& groups, size_t unique_values) {
     // If data_type_ has no value, it means there is no data for this aggregation
     // For sums, we want this to display as zero rather than NaN
@@ -93,12 +161,20 @@ void SumAggregatorData::aggregate(const std::optional<ColumnWithStrings>& input_
                     using ColumnType =  typename ColumnTagType::raw_type;
                     if constexpr(!is_sequence_type(ColumnTagType::data_type)) {
                         auto col_data = input_column->column_->data();
-                        auto groups_pos = 0;
-                        while (auto block = col_data.next<TypeDescriptorTag<ColumnTagType, DimensionTag<entity::Dimension::Dim0>>>()) {
-                            auto ptr = reinterpret_cast<const ColumnType *>(block.value().data());
-                            for (auto i = 0u; i < block.value().row_count(); ++i, ++ptr) {
-                                out_ptr[groups[groups_pos++]] += GlobalRawType(*ptr);
+                        auto lambda = [&col_data, &out_ptr, &groups](auto iter) {
+                            while (auto block = col_data.next<TypeDescriptorTag<ColumnTagType, DimensionTag<entity::Dimension::Dim0>>>()) {
+                                auto ptr = reinterpret_cast<const ColumnType *>(block.value().data());
+                                for (auto i = 0u; i < block.value().row_count(); ++i, ++ptr, ++iter) {
+                                    out_ptr[groups[deref(iter)]] += GlobalRawType(*ptr);
+                                }
                             }
+                        };
+
+                        if (input_column->column_->is_sparse()) {
+                            lambda(col_data.bit_vector()->first());
+                        }
+                        else {
+                            lambda(std::size_t(0));
                         }
                     } else {
                         util::raise_rte("String aggregations not currently supported");
@@ -124,23 +200,213 @@ SegmentInMemory SumAggregatorData::finalize(const ColumnName& output_column_name
     return res;
 }
 
+/********************
+ * MinMaxAggregator *
+ ********************/
+
+namespace
+{
+    enum class Extremum
+    {
+        MAX,
+        MIN
+    };
+
+    template <typename T, Extremum E>
+    struct MaybeValue
+    {
+        bool written_ = false;
+        T value_ = init_value();
+
+    private:
+
+        static constexpr T init_value()
+        {
+            if constexpr (E == Extremum::MAX)
+                return std::numeric_limits<T>::lowest();
+            else
+                return std::numeric_limits<T>::max();
+        }
+    };
+
+    template <Extremum T>
+    inline void aggregate_impl(
+        const std::optional<ColumnWithStrings>& input_column,
+        const std::vector<size_t>& groups,
+        size_t unique_values,
+        std::vector<uint8_t>& aggregated_,
+        std::optional<DataType>& data_type_
+    ) {
+        if(data_type_.has_value() && *data_type_ != DataType::EMPTYVAL && input_column.has_value()) {
+            entity::details::visit_type(*data_type_, [&aggregated_, &data_type_, &input_column, unique_values, &groups] (auto global_type_desc_tag) {
+                using GlobalInputType = decltype(global_type_desc_tag);
+                if constexpr(!is_sequence_type(GlobalInputType::DataTypeTag::data_type)) {
+                    using GlobalTypeDescriptorTag =  typename OutputType<GlobalInputType>::type;
+                    using GlobalRawType = typename GlobalTypeDescriptorTag::DataTypeTag::raw_type;
+                    using MaybeValueType = MaybeValue<GlobalRawType, T>;
+                    auto prev_size = aggregated_.size() / sizeof(MaybeValueType);
+                    aggregated_.resize(sizeof(MaybeValueType) * unique_values);
+                    auto col_data = input_column->column_->data();
+                    auto out_ptr = reinterpret_cast<MaybeValueType*>(aggregated_.data());
+                    std::fill(out_ptr + prev_size, out_ptr + unique_values, MaybeValueType{});
+                    entity::details::visit_type(input_column->column_->type().data_type(), [&input_column, &groups, &out_ptr, &col_data] (auto type_desc_tag) {
+                        using ColumnTagType = std::decay_t<decltype(type_desc_tag)>;
+                        using ColumnType =  typename ColumnTagType::raw_type;
+                        if constexpr(!is_sequence_type(ColumnTagType::data_type)) {
+                            auto lambda = [&col_data, &out_ptr, &groups](auto iter) {
+                                while (auto block = col_data.next<TypeDescriptorTag<ColumnTagType, DimensionTag<entity::Dimension::Dim0>>>()) {
+                                    auto ptr = reinterpret_cast<const ColumnType *>(block.value().data());
+                                    for (auto i = 0u; i < block.value().row_count(); ++i, ++ptr, ++iter) {
+                                        auto& val = out_ptr[groups[deref(iter)]];
+                                        if constexpr(std::is_floating_point_v<ColumnType>) {
+                                            const auto& curr = GlobalRawType(*ptr);
+                                            if (!val.written_ || std::isnan(static_cast<ColumnType>(val.value_))) {
+                                                val.value_ = curr;
+                                                val.written_ = true;
+                                            } else if (!std::isnan(static_cast<ColumnType>(curr))) {
+                                                if constexpr(T == Extremum::MAX) {
+                                                    val.value_ = std::max(val.value_, curr);
+                                                } else {
+                                                    val.value_ = std::min(val.value_, curr);
+                                                }
+                                            }
+                                        } else {
+                                            if constexpr(T == Extremum::MAX) {
+                                                val.value_ = std::max(val.value_, GlobalRawType(*ptr));
+                                            } else {
+                                                val.value_ = std::min(val.value_, GlobalRawType(*ptr));
+                                            }
+                                            val.written_ = true;
+                                        }
+                                    }
+                                }
+                            };
+                            if (input_column->column_->is_sparse()) {
+                                lambda(col_data.bit_vector()->first());
+                            }
+                            else {
+                                lambda(std::size_t(0));
+                            }
+                        } else {
+                            util::raise_rte("String aggregations not currently supported");
+                        }
+                    });
+                }
+            });
+        }
+    }
+
+    template <Extremum T>
+    inline SegmentInMemory finalize_impl(
+            const ColumnName& output_column_name,
+            bool dynamic_schema,
+            size_t unique_values,
+            std::vector<uint8_t>& aggregated_,
+            std::optional<DataType>& data_type_
+    ) {
+        SegmentInMemory res;
+        if(!aggregated_.empty()) {
+            if(dynamic_schema) {
+                entity::details::visit_type(*data_type_, [&aggregated_, &data_type_, &res, &output_column_name, unique_values] (auto type_desc_tag) {
+                    using RawType = typename decltype(type_desc_tag)::DataTypeTag::raw_type;
+                    using MaybeValueType = MaybeValue<RawType, T>;
+                    auto prev_size = aggregated_.size() / sizeof(MaybeValueType);
+                    auto new_size = sizeof(MaybeValueType) * unique_values;
+                    aggregated_.resize(new_size);
+                    auto in_ptr =  reinterpret_cast<MaybeValueType*>(aggregated_.data());
+                    std::fill(in_ptr + prev_size, in_ptr + unique_values, MaybeValueType{});
+                    auto col = std::make_shared<Column>(make_scalar_type(DataType::FLOAT64), unique_values, true, false);
+                    auto out_ptr = reinterpret_cast<double*>(col->ptr());
+                    for(auto i = 0u; i < unique_values; ++i, ++in_ptr, ++out_ptr) {
+                        *out_ptr = in_ptr->written_ ? static_cast<double>(in_ptr->value_) : std::numeric_limits<double>::quiet_NaN();                }
+
+                    col->set_row_data(unique_values - 1);
+                    res.add_column(scalar_field(DataType::FLOAT64, output_column_name.value), col);
+                });
+            } else {
+                entity::details::visit_type(*data_type_, [&aggregated_, &data_type_, &res, output_column_name, unique_values] (auto type_desc_tag) {
+                    using RawType = typename decltype(type_desc_tag)::DataTypeTag::raw_type;
+                    auto col = std::make_shared<Column>(make_scalar_type(data_type_.value()), unique_values, true, false);
+                    const auto* in_ptr =  reinterpret_cast<const MaybeValue<RawType, T>*>(aggregated_.data());
+                    auto out_ptr = reinterpret_cast<RawType*>(col->ptr());
+                    for(auto i = 0u; i < unique_values; ++i, ++in_ptr, ++out_ptr) {
+                        *out_ptr = in_ptr->value_;
+                    }
+                    col->set_row_data(unique_values - 1);
+                    res.add_column(scalar_field(data_type_.value(), output_column_name.value), col);
+                });
+            }
+        }
+        return res;
+    }
+}
+
+/*********************
+ * MaxAggregatorData *
+ *********************/
+
+void MaxAggregatorData::add_data_type(DataType data_type)
+{
+    add_data_type_impl(data_type, data_type_);
+}
+
+void MaxAggregatorData::aggregate(const std::optional<ColumnWithStrings>& input_column, const std::vector<size_t>& groups, size_t unique_values)
+{
+    aggregate_impl<Extremum::MAX>(input_column, groups, unique_values, aggregated_, data_type_);
+}
+
+SegmentInMemory MaxAggregatorData::finalize(const ColumnName& output_column_name, bool dynamic_schema, size_t unique_values)
+{
+    return finalize_impl<Extremum::MAX>(output_column_name, dynamic_schema, unique_values, aggregated_, data_type_);
+}
+
+/*********************
+ * MinAggregatorData *
+ *********************/
+
+void MinAggregatorData::add_data_type(DataType data_type)
+{
+    add_data_type_impl(data_type, data_type_);
+}
+
+void MinAggregatorData::aggregate(const std::optional<ColumnWithStrings>& input_column, const std::vector<size_t>& groups, size_t unique_values)
+{
+    aggregate_impl<Extremum::MIN>(input_column, groups, unique_values, aggregated_, data_type_);
+}
+
+SegmentInMemory MinAggregatorData::finalize(const ColumnName& output_column_name, bool dynamic_schema, size_t unique_values)
+{
+    return finalize_impl<Extremum::MIN>(output_column_name, dynamic_schema, unique_values, aggregated_, data_type_);
+}
+
+/**********************
+ * MeanAggregatorData *
+ **********************/
+
 void MeanAggregatorData::aggregate(const std::optional<ColumnWithStrings>& input_column, const std::vector<size_t>& groups, size_t unique_values) {
     if(input_column.has_value()) {
         input_column->column_->type().visit_tag([&] (auto type_desc_tag) {
             using TypeDescriptorTag =  decltype(type_desc_tag);
             using RawType = typename TypeDescriptorTag::DataTypeTag::raw_type;
 
-            data_.fractions_.resize(unique_values);
+            fractions_.resize(unique_values);
 
             auto col_data = input_column->column_->data();
-            while (auto block = col_data.next<TypeDescriptorTag>()) {
-                auto ptr = reinterpret_cast<const RawType *>(block.value().data());
-                for (auto i = 0u; i < block.value().row_count(); ++i) {
-                    auto group = groups[i];
-                    auto& fraction = data_.fractions_[group];
-                    fraction.numerator_ += double(*ptr++);
-                    ++fraction.denominator_;
+            auto lambda = [this, &col_data, &groups](auto iter) {
+                while (auto block = col_data.next<TypeDescriptorTag>()) {
+                    auto ptr = reinterpret_cast<const RawType *>(block.value().data());
+                    for (auto i = 0u; i < block.value().row_count(); ++i, ++ptr, ++iter) {
+                        auto& fraction = fractions_[groups[deref(iter)]];
+                        fraction.numerator_ += double(*ptr);
+                        ++fraction.denominator_;
+                    }
                 }
+            };
+            if (input_column->column_->is_sparse()) {
+                lambda(col_data.bit_vector()->first());
+            }
+            else {
+                lambda(std::size_t(0));
             }
         });
     }
@@ -148,20 +414,29 @@ void MeanAggregatorData::aggregate(const std::optional<ColumnWithStrings>& input
 
 SegmentInMemory MeanAggregatorData::finalize(const ColumnName& output_column_name,  bool, size_t unique_values) {
     SegmentInMemory res;
-    if(!data_.fractions_.empty()) {
-        data_.fractions_.resize(unique_values);
-        auto pos = res.add_column(scalar_field(DataType::FLOAT64, output_column_name.value), data_.fractions_.size(), true);
+    if(!fractions_.empty()) {
+        fractions_.resize(unique_values);
+        auto pos = res.add_column(scalar_field(DataType::FLOAT64, output_column_name.value), fractions_.size(), true);
         auto& column = res.column(pos);
         auto ptr = reinterpret_cast<double*>(column.ptr());
-        column.set_row_data(data_.fractions_.size() - 1);
+        column.set_row_data(fractions_.size() - 1);
 
-        for (const auto& fraction : folly::enumerate(data_.fractions_)) {
-            ptr[fraction.index] = fraction->to_double();
+        for (auto idx = 0u; idx < fractions_.size(); ++idx) {
+            ptr[idx] = fractions_[idx].to_double();
         }
     }
     return res;
 }
 
+double MeanAggregatorData::Fraction::to_double() const
+{
+    return denominator_ == 0 ? std::numeric_limits<double>::quiet_NaN(): numerator_ / static_cast<double>(denominator_);
+}
+
+/***********************
+ * CountAggregatorData *
+ ***********************/
+
 void CountAggregatorData::aggregate(const std::optional<ColumnWithStrings>& input_column, const std::vector<size_t>& groups, size_t unique_values) {
     if(input_column.has_value()) {
         input_column->column_->type().visit_tag([&] (auto type_desc_tag) {
@@ -169,24 +444,33 @@ void CountAggregatorData::aggregate(const std::optional<ColumnWithStrings>& inpu
             using RawType = typename TypeDescriptorTag::DataTypeTag::raw_type;
 
             aggregated_.resize(unique_values);
-
-            if constexpr (is_floating_point_type(TypeDescriptorTag::DataTypeTag::data_type)) {
-                auto col_data = input_column->column_->data();
-                while (auto block = col_data.next<TypeDescriptorTag>()) {
-                    auto ptr = reinterpret_cast<const RawType *>(block.value().data());
-                    for (auto i = 0u; i < block.value().row_count(); ++i) {
-                        if (!std::isnan(static_cast<double>(ptr[i]))) {
-                            auto group = groups[i];
-                            auto& val = aggregated_[group];
-                            ++val;
+            auto col_data = input_column->column_->data();
+            auto lambda = [this, &col_data, &groups](auto iter) {
+                if constexpr (is_floating_point_type(TypeDescriptorTag::DataTypeTag::data_type)) {
+                    while (auto block = col_data.next<TypeDescriptorTag>()) {
+                        auto ptr = reinterpret_cast<const RawType *>(block.value().data());
+                        for (auto i = 0u; i < block.value().row_count(); ++i, ++ptr, ++iter) {
+                            if (!std::isnan(static_cast<double>(*ptr))) {
+                                auto& val = aggregated_[groups[deref(iter)]];
+                                ++val;
+                            }
                         }
                     }
                 }
-            } else {
-                for (auto group: groups) {
-                    auto& val = aggregated_[group];
-                    ++val;
+                else {
+                    while (auto block = col_data.next<TypeDescriptorTag>()) {
+                        for (auto i = 0u; i < block.value().row_count(); ++i, ++iter) {
+                            auto& val = aggregated_[groups[deref(iter)]];
+                            ++val;
+                        }
+                    }
                 }
+            };
+            if (input_column->column_->is_sparse()) {
+                lambda(col_data.bit_vector()->first());
+            }
+            else {
+                lambda(std::size_t(0));
             }
         });
     }
diff --git a/cpp/arcticdb/processing/aggregation.hpp b/cpp/arcticdb/processing/aggregation.hpp
index f46d8a8ac3..c5848cd59b 100644
--- a/cpp/arcticdb/processing/aggregation.hpp
+++ b/cpp/arcticdb/processing/aggregation.hpp
@@ -7,346 +7,168 @@
 
 #pragma once
 
-#include <arcticdb/processing/expression_node.hpp>
-#include <arcticdb/processing/processing_unit.hpp>
-#include <arcticdb/processing/bucketizer.hpp>
+#include <arcticdb/column_store/memory_segment.hpp>
 #include <arcticdb/entity/types.hpp>
 #include <arcticdb/entity/type_utils.hpp>
-#include <arcticdb/pipeline/value.hpp>
-#include <arcticdb/pipeline/value_set.hpp>
-#include <arcticdb/processing/aggregation_interface.hpp>
-
-#include <folly/Poly.h>
+#include <arcticdb/processing/expression_node.hpp>
 
 namespace arcticdb {
 
-template<typename T, typename T2=void>
-struct OutputType;
+class MinMaxAggregatorData
+{
+public:
 
-template <typename InputType>
-struct OutputType <InputType, typename std::enable_if_t<is_floating_point_type(InputType::DataTypeTag::data_type)>> {
-    using type = ScalarTagType<DataTypeTag<DataType::FLOAT64>>;
-};
-
-template <typename InputType>
-struct OutputType <InputType, typename std::enable_if_t<is_unsigned_type(InputType::DataTypeTag::data_type)>> {
-    using type = ScalarTagType<DataTypeTag<DataType::UINT64>>;
-};
+    MinMaxAggregatorData() = default;
+    ARCTICDB_MOVE_COPY_DEFAULT(MinMaxAggregatorData)
 
-template <typename InputType>
-struct OutputType<InputType, typename std::enable_if_t<is_signed_type(InputType::DataTypeTag::data_type) && is_integer_type(InputType::DataTypeTag::data_type)>> {
-    using type = ScalarTagType<DataTypeTag<DataType::INT64>>;
-};
+    void aggregate(const ColumnWithStrings& input_column);
+    SegmentInMemory finalize(const std::vector<ColumnName>& output_column_names) const;
 
-template<>
-struct OutputType<DataTypeTag<DataType::BOOL8>, void> {
-    using type = ScalarTagType<DataTypeTag<DataType::BOOL8>>;
-};
+private:
 
-template<>
-struct OutputType<DataTypeTag<DataType::NANOSECONDS_UTC64>, void> {
-    using type = ScalarTagType<DataTypeTag<DataType::NANOSECONDS_UTC64>>;
+    std::optional<Value> min_;
+    std::optional<Value> max_;
 };
 
-template<>
-struct OutputType<DataTypeTag<DataType::EMPTYVAL>, void> {
-    using type = ScalarTagType<DataTypeTag<DataType::EMPTYVAL>>;
-};
+class MinMaxAggregator
+{
+public:
 
-void add_data_type_impl(DataType data_type, std::optional<DataType>& current_data_type);
+    explicit MinMaxAggregator(ColumnName column_name, ColumnName output_column_name_min, ColumnName output_column_name_max)
+        : column_name_(std::move(column_name))
+        , output_column_name_min_(std::move(output_column_name_min))
+        , output_column_name_max_(std::move(output_column_name_max))
+    {}
+    ARCTICDB_MOVE_COPY_DEFAULT(MinMaxAggregator)
 
-struct MinMaxAggregatorData {
-    std::optional<Value> min_;
-    std::optional<Value> max_;
+    [[nodiscard]] ColumnName get_input_column_name() const { return column_name_; }
+    [[nodiscard]] std::vector<ColumnName> get_output_column_names() const { return {output_column_name_min_, output_column_name_max_}; }
+    [[nodiscard]] MinMaxAggregatorData get_aggregator_data() const { return MinMaxAggregatorData(); }
 
-    MinMaxAggregatorData() = default;
-    ARCTICDB_MOVE_COPY_DEFAULT(MinMaxAggregatorData)
+private:
 
-    void aggregate(const ColumnWithStrings& input_column);
-    SegmentInMemory finalize(const std::vector<ColumnName>& output_column_names) const;
-};
-
-struct MinMaxAggregator {
     ColumnName column_name_;
     ColumnName output_column_name_min_;
     ColumnName output_column_name_max_;
+};
 
-    explicit MinMaxAggregator(ColumnName column_name, ColumnName output_column_name_min, ColumnName output_column_name_max) :
-            column_name_(std::move(column_name)),
-            output_column_name_min_(std::move(output_column_name_min)),
-            output_column_name_max_(std::move(output_column_name_max)){
-    }
-    ARCTICDB_MOVE_COPY_DEFAULT(MinMaxAggregator)
+class AggregatorDataBase
+{
+public:
 
-    [[nodiscard]] ColumnName get_input_column_name() const { return column_name_; }
-    [[nodiscard]] std::vector<ColumnName> get_output_column_names() const { return {output_column_name_min_, output_column_name_max_}; }
-    [[nodiscard]] ColumnStatsAggregatorData get_aggregator_data() const { return MinMaxAggregatorData(); }
+    AggregatorDataBase() = default;
+    // Warn on copies as inheriting classes may embed large buffers
+    AggregatorDataBase(const AggregatorDataBase&);
+    AggregatorDataBase& operator=(const AggregatorDataBase&);
+
+    ARCTICDB_MOVE(AggregatorDataBase);
 };
 
-struct SumAggregatorData {
-    std::vector<uint8_t> aggregated_;
-    std::optional<DataType> data_type_;
+class SumAggregatorData : private AggregatorDataBase
+{
+public:
 
-    SumAggregatorData() = default;
-    // Warn on copies as aggregated_ could be a large buffer
-    SumAggregatorData(const SumAggregatorData& other):
-    aggregated_(other.aggregated_),
-    data_type_(other.data_type_) {
-        log::version().warn("Copying potentially large buffer in SumAggregatorData");
-    }
-    SumAggregatorData& operator=(const SumAggregatorData& other) {
-        aggregated_ = other.aggregated_;
-        data_type_ = other.data_type_;
-        log::version().warn("Copying potentially large buffer in SumAggregatorData");
-        return *this;
-    }
-    ARCTICDB_MOVE(SumAggregatorData)
-
-    void add_data_type(DataType data_type) {
-        add_data_type_impl(data_type, data_type_);
-    }
+    void add_data_type(DataType data_type);
     void aggregate(const std::optional<ColumnWithStrings>& input_column, const std::vector<size_t>& groups, size_t unique_values);
     SegmentInMemory finalize(const ColumnName& output_column_name,  bool dynamic_schema, size_t unique_values);
-};
 
-struct SumAggregator {
-    ColumnName input_column_name_;
-    ColumnName output_column_name_;
+private:
 
-    explicit SumAggregator(ColumnName input_column_name, ColumnName output_column_name) :
-        input_column_name_(std::move(input_column_name)),
-        output_column_name_(std::move(output_column_name)) {
-    }
-    ARCTICDB_MOVE_COPY_DEFAULT(SumAggregator)
-
-    [[nodiscard]] ColumnName get_input_column_name() const { return input_column_name_; }
-    [[nodiscard]] ColumnName get_output_column_name() const { return output_column_name_; }
-    [[nodiscard]] GroupingAggregatorData get_aggregator_data() const { return SumAggregatorData(); }
+    std::vector<uint8_t> aggregated_;
+    std::optional<DataType> data_type_;
 };
 
-template <typename T>
-struct MaybeValue {
-    bool written_ = false;
-    T value_ = std::numeric_limits<T>::lowest();
-};
+class MaxAggregatorData : private AggregatorDataBase
+{
+public:
 
-enum class Extremum {
-    MAX,
-    MIN
-};
+    void add_data_type(DataType data_type);
+    void aggregate(const std::optional<ColumnWithStrings>& input_column, const std::vector<size_t>& groups, size_t unique_values);
+    SegmentInMemory finalize(const ColumnName& output_column_name, bool dynamic_schema, size_t unique_values);
+
+private:
 
-template<Extremum T>
-struct MaxOrMinAggregatorData {
     std::vector<uint8_t> aggregated_;
     std::optional<DataType> data_type_;
-
-    MaxOrMinAggregatorData() = default;
-    // Warn on copies as aggregated_ could be a large buffer
-    MaxOrMinAggregatorData(const MaxOrMinAggregatorData& other):
-            aggregated_(other.aggregated_),
-            data_type_(other.data_type_) {
-        log::version().warn("Copying potentially large buffer in MaxOrMinAggregatorData");
-    }
-    MaxOrMinAggregatorData& operator=(const MaxOrMinAggregatorData& other) {
-        aggregated_ = other.aggregated_;
-        data_type_ = other.data_type_;
-        log::version().warn("Copying potentially large buffer in MaxOrMinAggregatorData");
-        return *this;
-    }
-    ARCTICDB_MOVE(MaxOrMinAggregatorData)
-
-    void add_data_type(DataType data_type) {
-        add_data_type_impl(data_type, data_type_);
-    }
-    void aggregate(const std::optional<ColumnWithStrings>& input_column, const std::vector<size_t>& groups, size_t unique_values) {
-        if(data_type_.has_value() && *data_type_ != DataType::EMPTYVAL && input_column.has_value()) {
-            entity::details::visit_type(*data_type_, [&input_column, unique_values, &groups, that=this] (auto global_type_desc_tag) {
-                using GlobalInputType = decltype(global_type_desc_tag);
-                if constexpr(!is_sequence_type(GlobalInputType::DataTypeTag::data_type)) {
-                    using GlobalTypeDescriptorTag =  typename OutputType<GlobalInputType>::type;
-                    using GlobalRawType = typename GlobalTypeDescriptorTag::DataTypeTag::raw_type;
-                    auto prev_size = that->aggregated_.size() / sizeof(MaybeValue<GlobalRawType>);
-                    that->aggregated_.resize(sizeof(MaybeValue<GlobalRawType>) * unique_values);
-                    auto col_data = input_column->column_->data();
-                    auto out_ptr = reinterpret_cast<MaybeValue<GlobalRawType>*>(that->aggregated_.data());
-                    std::fill(out_ptr + prev_size, out_ptr + unique_values, MaybeValue<GlobalRawType>{});
-                    entity::details::visit_type(input_column->column_->type().data_type(), [&groups, &out_ptr, &col_data] (auto type_desc_tag) {
-                        using ColumnTagType = std::decay_t<decltype(type_desc_tag)>;
-                        using ColumnType =  typename ColumnTagType::raw_type;
-                        if constexpr(!is_sequence_type(ColumnTagType::data_type)) {
-                            auto groups_pos = 0;
-                            while (auto block = col_data.next<TypeDescriptorTag<ColumnTagType, DimensionTag<entity::Dimension::Dim0>>>()) {
-                                auto ptr = reinterpret_cast<const ColumnType *>(block.value().data());
-                                for (auto i = 0u; i < block.value().row_count(); ++i, ++ptr) {
-                                    auto& val = out_ptr[groups[groups_pos]];
-                                    if constexpr(std::is_floating_point_v<ColumnType>) {
-                                        const auto& curr = GlobalRawType(*ptr);
-                                        if (!val.written_ || std::isnan(static_cast<ColumnType>(val.value_))) {
-                                            val.value_ = curr;
-                                            val.written_ = true;
-                                        } else if (!std::isnan(static_cast<ColumnType>(curr))) {
-                                            if constexpr(T == Extremum::MAX) {
-                                                val.value_ = std::max(val.value_, curr);
-                                            } else {
-                                                val.value_ = std::min(val.value_, curr);
-                                            }
-                                        }
-                                    } else {
-                                        if constexpr(T == Extremum::MAX) {
-                                            val.value_ = std::max(val.value_, GlobalRawType(*ptr));
-                                        } else {
-                                            // If unwritten, MaybeValue begins life with MIN_INT. That's always going to "win" when
-                                            // passed into `std::min`!
-                                            val.value_ = !val.written_ ? GlobalRawType(*ptr) : std::min(val.value_, GlobalRawType(*ptr));
-                                        }
-                                        val.written_ = true;
-                                    }
-                                    ++groups_pos;
-                                }
-                            }
-                        } else {
-                            util::raise_rte("String aggregations not currently supported");
-                        }
-                    });
-                }
-            });
-        }
-    }
-    [[nodiscard]] SegmentInMemory finalize(const ColumnName& output_column_name, bool dynamic_schema, size_t unique_values) {
-        SegmentInMemory res;
-        if(!aggregated_.empty()) {
-            if(dynamic_schema) {
-                entity::details::visit_type(*data_type_, [that=this, &res, &output_column_name, unique_values] (auto type_desc_tag) {
-                    using RawType = typename decltype(type_desc_tag)::DataTypeTag::raw_type;
-                    auto prev_size = that->aggregated_.size() / sizeof(MaybeValue<RawType>);
-                    auto new_size = sizeof(MaybeValue<RawType>) * unique_values;
-                    that->aggregated_.resize(new_size);
-                    auto in_ptr =  reinterpret_cast<MaybeValue<RawType>*>(that->aggregated_.data());
-                    std::fill(in_ptr + prev_size, in_ptr + unique_values, MaybeValue<RawType>{});
-                    auto col = std::make_shared<Column>(make_scalar_type(DataType::FLOAT64), unique_values, true, false);
-                    auto out_ptr = reinterpret_cast<double*>(col->ptr());
-                    for(auto i = 0u; i < unique_values; ++i, ++in_ptr, ++out_ptr) {
-                        *out_ptr = in_ptr->written_ ? static_cast<double>(in_ptr->value_) : std::numeric_limits<double>::quiet_NaN();                }
-
-                    col->set_row_data(unique_values - 1);
-                    res.add_column(scalar_field(DataType::FLOAT64, output_column_name.value), col);
-                });
-            } else {
-                entity::details::visit_type(*data_type_, [that=this, &res, output_column_name, unique_values] (auto type_desc_tag) {
-                    using RawType = typename decltype(type_desc_tag)::DataTypeTag::raw_type;
-                    auto col = std::make_shared<Column>(make_scalar_type(that->data_type_.value()), unique_values, true, false);
-                    const auto* in_ptr =  reinterpret_cast<const MaybeValue<RawType>*>(that->aggregated_.data());
-                    auto out_ptr = reinterpret_cast<RawType*>(col->ptr());
-                    for(auto i = 0u; i < unique_values; ++i, ++in_ptr, ++out_ptr) {
-                        *out_ptr = in_ptr->value_;
-                    }
-                    col->set_row_data(unique_values - 1);
-                    res.add_column(scalar_field(that->data_type_.value(), output_column_name.value), col);
-                });
-            }
-        }
-        return res;
-    }
 };
 
-struct MaxOrMinAggregator {
-    ColumnName input_column_name_;
-    ColumnName output_column_name_;
-    Extremum extremum_;
+class MinAggregatorData : private AggregatorDataBase
+{
+public:
 
-    explicit MaxOrMinAggregator(ColumnName input_column_name, ColumnName output_column_name, Extremum extremum) :
-        input_column_name_(std::move(input_column_name)),
-        output_column_name_(std::move(output_column_name)),
-        extremum_(extremum){
-    }
-    ARCTICDB_MOVE_COPY_DEFAULT(MaxOrMinAggregator)
+    void add_data_type(DataType data_type);
+    void aggregate(const std::optional<ColumnWithStrings>& input_column, const std::vector<size_t>& groups, size_t unique_values);
+    SegmentInMemory finalize(const ColumnName& output_column_name, bool dynamic_schema, size_t unique_values);
 
-    [[nodiscard]] ColumnName get_input_column_name() const { return input_column_name_; }
-    [[nodiscard]] ColumnName get_output_column_name() const { return output_column_name_; }
-    [[nodiscard]] GroupingAggregatorData get_aggregator_data() const {
-        switch(extremum_) {
-            case Extremum::MAX:
-                return MaxOrMinAggregatorData<Extremum::MAX>();
-            case Extremum::MIN:
-                return MaxOrMinAggregatorData<Extremum::MIN>();
-            default:
-                internal::raise<ErrorCode::E_ASSERTION_FAILURE>("Unexpected Extremum enum value in MaxOrMinAggregator");
-        }
-    }
-};
+private:
 
-struct MeanAggregatorData {
-    CountAndTotals data_;
+    std::vector<uint8_t> aggregated_;
+    std::optional<DataType> data_type_;
+};
 
-    MeanAggregatorData() = default;
-    // Warn on copies as data_ could be a large buffer
-    MeanAggregatorData(const MeanAggregatorData& other):
-            data_(other.data_) {
-        log::version().warn("Copying potentially large buffer in MeanAggregatorData");
-    }
-    MeanAggregatorData& operator=(const MeanAggregatorData& other) {
-        data_ = other.data_;
-        log::version().warn("Copying potentially large buffer in MeanAggregatorData");
-        return *this;
-    }
-    ARCTICDB_MOVE(MeanAggregatorData)
+class MeanAggregatorData : private AggregatorDataBase
+{
+public:
 
     // Mean values are always doubles so this is a no-op
-    void add_data_type(ARCTICDB_UNUSED DataType data_type) {}
+    void add_data_type(DataType) {}
     void aggregate(const std::optional<ColumnWithStrings>& input_column, const std::vector<size_t>& groups, size_t unique_values);
     SegmentInMemory finalize(const ColumnName& output_column_name,  bool dynamic_schema, size_t unique_values);
-};
 
-struct MeanAggregator {
-    ColumnName input_column_name_;
-    ColumnName output_column_name_;
+private:
 
-    explicit MeanAggregator(ColumnName input_column_name, ColumnName output_column_name) :
-            input_column_name_(std::move(input_column_name)),
-            output_column_name_(std::move(output_column_name)) {
-    }
-    ARCTICDB_MOVE_COPY_DEFAULT(MeanAggregator)
+    struct Fraction
+    {
+        double numerator_{0.0};
+        uint64_t denominator_{0};
 
-    [[nodiscard]] ColumnName get_input_column_name() const { return input_column_name_; }
-    [[nodiscard]] ColumnName get_output_column_name() const { return output_column_name_; }
-    [[nodiscard]] GroupingAggregatorData get_aggregator_data() const { return MeanAggregatorData(); }
-};
+        double to_double() const;
+    };
 
-struct CountAggregatorData {
-    std::vector<uint64_t> aggregated_;
+    std::vector<Fraction> fractions_;
+};
 
-    CountAggregatorData() = default;
-    // Warn on copies as aggregated_ could be a large buffer
-    CountAggregatorData(const CountAggregatorData& other):
-            aggregated_(other.aggregated_) {
-        log::version().warn("Copying potentially large buffer in CountAggregatorData");
-    }
-    CountAggregatorData& operator=(const CountAggregatorData& other) {
-        aggregated_ = other.aggregated_;
-        log::version().warn("Copying potentially large buffer in CountAggregatorData");
-        return *this;
-    }
-    ARCTICDB_MOVE(CountAggregatorData)
+class CountAggregatorData : private AggregatorDataBase
+{
+public:
 
     // Count values are always integers so this is a no-op
-    void add_data_type(ARCTICDB_UNUSED DataType data_type) {}
+    void add_data_type(DataType) {}
     void aggregate(const std::optional<ColumnWithStrings>& input_column, const std::vector<size_t>& groups, size_t unique_values);
     SegmentInMemory finalize(const ColumnName& output_column_name,  bool dynamic_schema, size_t unique_values);
+
+private:
+
+    std::vector<uint64_t> aggregated_;
 };
 
-struct CountAggregator {
-    ColumnName input_column_name_;
-    ColumnName output_column_name_;
+template <class AggregatorData>
+class GroupingAggregatorImpl
+{
+public:
 
-    explicit CountAggregator(ColumnName input_column_name, ColumnName output_column_name) :
-            input_column_name_(std::move(input_column_name)),
-            output_column_name_(std::move(output_column_name)) {
+    explicit GroupingAggregatorImpl(ColumnName input_column_name, ColumnName output_column_name)
+        : input_column_name_(std::move(input_column_name))
+        , output_column_name_(std::move(output_column_name))
+    {
     }
-    ARCTICDB_MOVE_COPY_DEFAULT(CountAggregator)
+
+    ARCTICDB_MOVE_COPY_DEFAULT(GroupingAggregatorImpl);
 
     [[nodiscard]] ColumnName get_input_column_name() const { return input_column_name_; }
     [[nodiscard]] ColumnName get_output_column_name() const { return output_column_name_; }
-    [[nodiscard]] GroupingAggregatorData get_aggregator_data() const { return CountAggregatorData(); }
+    [[nodiscard]] AggregatorData get_aggregator_data() const { return AggregatorData(); }
+
+private:
+
+    ColumnName input_column_name_;
+    ColumnName output_column_name_;
 };
 
+using SumAggregator = GroupingAggregatorImpl<SumAggregatorData>;
+using MinAggregator = GroupingAggregatorImpl<MinAggregatorData>;
+using MaxAggregator = GroupingAggregatorImpl<MaxAggregatorData>;
+using MeanAggregator = GroupingAggregatorImpl<MeanAggregatorData>;
+using CountAggregator = GroupingAggregatorImpl<CountAggregatorData>;
+
 } //namespace arcticdb
diff --git a/cpp/arcticdb/processing/aggregation_interface.hpp b/cpp/arcticdb/processing/aggregation_interface.hpp
index 8dd45b3409..a2dcd99fa6 100644
--- a/cpp/arcticdb/processing/aggregation_interface.hpp
+++ b/cpp/arcticdb/processing/aggregation_interface.hpp
@@ -73,17 +73,4 @@ struct IColumnStatsAggregator {
 
 using ColumnStatsAggregator = folly::Poly<IColumnStatsAggregator>;
 
-struct Fraction {
-    double numerator_{0.0};
-    uint64_t denominator_{0};
-
-    double to_double() const {
-        return denominator_ == 0 ? std::numeric_limits<double>::quiet_NaN(): numerator_ / static_cast<double>(denominator_);
-    }
-};
-
-struct CountAndTotals {
-    std::vector<Fraction> fractions_;
-};
-
-} //namespace arcticdb
\ No newline at end of file
+} //namespace arcticdb
diff --git a/cpp/arcticdb/processing/clause.cpp b/cpp/arcticdb/processing/clause.cpp
index 7a74a524b6..591756dc13 100644
--- a/cpp/arcticdb/processing/clause.cpp
+++ b/cpp/arcticdb/processing/clause.cpp
@@ -235,9 +235,9 @@ AggregationClause::AggregationClause(const std::string& grouping_column,
         } else if (aggregation_operator == "mean") {
             aggregators_.emplace_back(MeanAggregator(typed_column_name, typed_column_name));
         } else if (aggregation_operator == "max") {
-            aggregators_.emplace_back(MaxOrMinAggregator(typed_column_name, typed_column_name, Extremum::MAX));
+            aggregators_.emplace_back(MaxAggregator(typed_column_name, typed_column_name));
         } else if (aggregation_operator == "min") {
-            aggregators_.emplace_back(MaxOrMinAggregator(typed_column_name, typed_column_name, Extremum::MIN));
+            aggregators_.emplace_back(MinAggregator(typed_column_name, typed_column_name));
         } else if (aggregation_operator == "count") {
             aggregators_.emplace_back(CountAggregator(typed_column_name, typed_column_name));
         } else {
diff --git a/cpp/arcticdb/processing/clause.hpp b/cpp/arcticdb/processing/clause.hpp
index 821fdc5251..2caa0e24d2 100644
--- a/cpp/arcticdb/processing/clause.hpp
+++ b/cpp/arcticdb/processing/clause.hpp
@@ -16,6 +16,7 @@
 #include <arcticdb/processing/processing_unit.hpp>
 #include <arcticdb/entity/types.hpp>
 #include <arcticdb/processing/aggregation.hpp>
+#include <arcticdb/processing/aggregation_interface.hpp>
 #include <arcticdb/processing/grouper.hpp>
 #include <arcticdb/stream/aggregator.hpp>
 #include <arcticdb/util/movable_priority_queue.hpp>


Test patch:
diff --git a/cpp/arcticdb/processing/test/test_clause.cpp b/cpp/arcticdb/processing/test/test_clause.cpp
index 87a30dfcc3..1c83d8a3f9 100644
--- a/cpp/arcticdb/processing/test/test_clause.cpp
+++ b/cpp/arcticdb/processing/test/test_clause.cpp
@@ -102,7 +102,7 @@ TEST(Clause, PartitionString) {
 TEST(Clause, PartitionEmptyColumn) {
     using namespace arcticdb;
 
-    auto seg = generate_groupby_testing_segment(100, 10);
+    auto seg = generate_groupby_testing_empty_segment(100, 10);
     ProcessingUnit processing_unit(std::move(seg), pipelines::FrameSlice{});
     Composite<ProcessingUnit> comp;
     comp.push_back(std::move(processing_unit));
@@ -118,7 +118,7 @@ TEST(Clause, AggregationEmptyColumn) {
     std::shared_ptr<Store> empty_store;
     size_t num_rows{100};
     size_t unique_grouping_values{10};
-    auto seg = generate_groupby_testing_segment(num_rows, unique_grouping_values);
+    auto seg = generate_groupby_testing_empty_segment(num_rows, unique_grouping_values);
     ProcessingUnit processing_unit(std::move(seg), pipelines::FrameSlice{});
     Composite<ProcessingUnit> comp;
     comp.push_back(std::move(processing_unit));
@@ -145,6 +145,122 @@ TEST(Clause, AggregationEmptyColumn) {
     ASSERT_FALSE(slice_and_keys[0].segment_->column_index("empty_count").has_value());
 }
 
+namespace aggregation_test
+{
+    template <class T, class F>
+    void check_column(arcticdb::SliceAndKey sk, std::string_view column_name, std::size_t ugv, F&& f)
+    {
+        auto column_index = sk.segment_->column_index(column_name);
+        ASSERT_TRUE(column_index.has_value());
+        auto& column = sk.segment_->column(*column_index);
+        DataType dt = arcticdb::data_type_from_raw_type<T>();
+        ASSERT_EQ(dt, column.type().data_type());
+        for(std::size_t idx = 0u; idx < ugv; ++idx)
+        {
+            if constexpr (std::is_floating_point_v<T>)
+            {
+                T val = column.scalar_at<T>(idx).value();
+                if (std::isnan(val))
+                {
+                    ASSERT_TRUE(std::isnan(f(idx)));
+                }
+                else
+                {
+                    ASSERT_EQ(f(idx), val);
+                }
+            }
+            else
+            {
+                ASSERT_EQ(f(idx), column.scalar_at<T>(idx));
+            }
+        }
+    }
+}
+
+TEST(Clause, AggregationColumn)
+{
+    using namespace arcticdb;
+
+    std::shared_ptr<Store> empty_store;
+    size_t num_rows{100};
+    size_t unique_grouping_values{10};
+    auto seg = generate_groupby_testing_segment(num_rows, unique_grouping_values);
+    ProcessingUnit processing_unit(std::move(seg), pipelines::FrameSlice{});
+    Composite<ProcessingUnit> comp;
+    comp.push_back(std::move(processing_unit));
+    AggregationClause aggregation("int_repeated_values", {{"sum_int", "sum"}, {"min_int", "min"}, {"max_int", "max"}, {"mean_int", "mean"}, {"count_int", "count"}});
+
+    auto aggregated = aggregation.process(std::shared_ptr<Store>(), std::move(comp)).as_range();
+    ASSERT_EQ(1, aggregated.size());
+    auto slice_and_keys = aggregated[0].data();
+    ASSERT_EQ(1, slice_and_keys.size());
+
+    using aggregation_test::check_column;
+    check_column<int64_t>(slice_and_keys[0], "sum_int", unique_grouping_values, [](size_t idx) { return 450 + 10*idx; });
+    check_column<int64_t>(slice_and_keys[0], "min_int", unique_grouping_values, [](size_t idx) { return idx; });
+    check_column<int64_t>(slice_and_keys[0], "max_int", unique_grouping_values, [](size_t idx) { return 90+idx; });
+    check_column<double>(slice_and_keys[0], "mean_int", unique_grouping_values, [](size_t idx) { return double(45+idx); });
+    check_column<uint64_t>(slice_and_keys[0], "count_int", unique_grouping_values, [](size_t) { return 10; });
+}
+
+TEST(Clause, AggregationSparseColumn)
+{
+    using namespace arcticdb;
+
+    std::shared_ptr<Store> empty_store;
+    size_t num_rows{100};
+    size_t unique_grouping_values{10};
+    auto seg = generate_groupby_testing_sparse_segment(num_rows, unique_grouping_values);
+    ProcessingUnit processing_unit(std::move(seg), pipelines::FrameSlice{});
+    Composite<ProcessingUnit> comp;
+    comp.push_back(std::move(processing_unit));
+    AggregationClause aggregation("int_repeated_values", {{"sum_int", "sum"}, {"min_int", "min"}, {"max_int", "max"}, {"mean_int", "mean"}, {"count_int", "count"}});
+
+    auto aggregated = aggregation.process(std::shared_ptr<Store>(), std::move(comp)).as_range();
+    ASSERT_EQ(1, aggregated.size());
+    auto slice_and_keys = aggregated[0].data();
+    ASSERT_EQ(1, slice_and_keys.size());
+
+    using aggregation_test::check_column;
+    check_column<int64_t>(slice_and_keys[0], "sum_int", unique_grouping_values, [](size_t idx) {
+        if (idx%2 == 0) {
+            return 450 + 10*static_cast<int64_t>(idx);
+        } else {
+            return int64_t(0);
+        }
+    });
+    check_column<int64_t>(slice_and_keys[0], "min_int", unique_grouping_values, [](size_t idx) {
+        if (idx%2 == 0) {
+            return static_cast<int64_t>(idx);
+        } else {
+            return std::numeric_limits<int64_t>::max();
+        }
+    });
+    check_column<int64_t>(slice_and_keys[0], "max_int", unique_grouping_values, [](size_t idx) {
+        if (idx%2 == 0) {
+            return 90 + static_cast<int64_t>(idx);
+        } else {
+            return std::numeric_limits<int64_t>::lowest();
+        }
+    });
+    check_column<double>(slice_and_keys[0], "mean_int", unique_grouping_values, [](size_t idx) {
+        if (idx%2 == 0) {
+            return double(45 + idx);
+        }
+        else {
+            return std::numeric_limits<double>::quiet_NaN();
+        }
+    });
+    check_column<uint64_t>(slice_and_keys[0], "count_int", unique_grouping_values, [](size_t idx) {
+        if (idx%2 == 0) {
+            return uint64_t(10);
+        }
+        else {
+            return uint64_t(0);
+        }
+    });
+}
+
 TEST(Clause, Passthrough) {
     using namespace arcticdb;
     auto seg = get_standard_timeseries_segment("passthrough");
diff --git a/cpp/arcticdb/util/test/generators.hpp b/cpp/arcticdb/util/test/generators.hpp
index 2e9f9e0afa..831a9e0d56 100644
--- a/cpp/arcticdb/util/test/generators.hpp
+++ b/cpp/arcticdb/util/test/generators.hpp
@@ -89,6 +89,19 @@ inline Column generate_int_column(size_t num_rows) {
     return column;
 }
 
+// Generates an int64_t Column where the value in one row out of two is equal to the row index
+inline Column generate_int_sparse_column(size_t num_rows) {
+    using TDT = TypeDescriptorTag<DataTypeTag<DataType::INT64>, DimensionTag<Dimension ::Dim0>>;
+    Column column(static_cast<TypeDescriptor>(TDT{}), 0, false, true);
+    for(size_t idx = 0; idx < num_rows; ++idx) {
+        if (idx%2 == 0)
+        {
+            column.set_scalar<int64_t>(static_cast<ssize_t>(idx), static_cast<int64_t>(idx));
+        }
+    }
+    return column;
+}
+
 // Generates an int64_t Column where the value in each row is equal to the row index modulo the number of unique values
 inline Column generate_int_column_repeated_values(size_t num_rows, size_t unique_values) {
     using TDT = TypeDescriptorTag<DataTypeTag<DataType::INT64>, DimensionTag<Dimension ::Dim0>>;
@@ -109,7 +122,7 @@ inline Column generate_empty_column() {
 // Generate a segment in memory suitable for testing groupby's empty type column behaviour with 5 columns:
 // * int_repeating_values - an int64_t column with unique_values repeating values
 // * empty_<agg> - an empty column for each supported aggregation
-inline SegmentInMemory generate_groupby_testing_segment(size_t num_rows, size_t unique_values) {
+inline SegmentInMemory generate_groupby_testing_empty_segment(size_t num_rows, size_t unique_values) {
     SegmentInMemory seg;
     auto int_repeated_values_col = std::make_shared<Column>(generate_int_column_repeated_values(num_rows, unique_values));
     seg.add_column(scalar_field(int_repeated_values_col->type().data_type(), "int_repeated_values"), int_repeated_values_col);
@@ -121,6 +134,36 @@ inline SegmentInMemory generate_groupby_testing_segment(size_t num_rows, size_t
     return seg;
 }
 
+inline SegmentInMemory generate_groupby_testing_segment(size_t num_rows, size_t unique_values)
+{
+    SegmentInMemory seg;
+    auto int_repeated_values_col = std::make_shared<Column>(generate_int_column_repeated_values(num_rows, unique_values));
+    seg.add_column(scalar_field(int_repeated_values_col->type().data_type(), "int_repeated_values"), int_repeated_values_col);
+    std::vector<std::string> col_names = { "sum_int", "min_int", "max_int", "mean_int", "count_int" };
+    for (const auto& name: col_names)
+    {
+        auto col = std::make_shared<Column>(generate_int_column(num_rows));
+        seg.add_column(scalar_field(col->type().data_type(), name), col);
+    }
+    seg.set_row_id(num_rows - 1);
+    return seg;
+}
+
+inline SegmentInMemory generate_groupby_testing_sparse_segment(size_t num_rows, size_t unique_values)
+{
+    SegmentInMemory seg;
+    auto int_repeated_values_col = std::make_shared<Column>(generate_int_column_repeated_values(num_rows, unique_values));
+    seg.add_column(scalar_field(int_repeated_values_col->type().data_type(), "int_repeated_values"), int_repeated_values_col);
+    std::vector<std::string> col_names = { "sum_int", "min_int", "max_int", "mean_int", "count_int" };
+    for (const auto& name: col_names)
+    {
+        auto col = std::make_shared<Column>(generate_int_sparse_column(num_rows));
+        seg.add_column(scalar_field(col->type().data_type(), name), col);
+    }
+    seg.set_row_id(num_rows - 1);
+    return seg;
+}
+
 inline SegmentInMemory get_standard_timeseries_segment(const std::string& name, size_t num_rows = 10) {
     auto wrapper = SinkWrapper(name, {
         scalar_field(DataType::INT8, "int8"),
@@ -330,4 +373,4 @@ using TestSegmentAggregatorNoSegment = SegmentAggregator<TimeseriesIndex, FixedS
 using SegmentSinkWrapper = SegmentSinkWrapperImpl<TestSegmentAggregatorNoSegment>;
 
 
-} //namespace arcticdb
\ No newline at end of file
+} //namespace arcticdb