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

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



Discussion / hints:

Fix patch:
diff --git a/cpp/arcticdb/processing/clause.cpp b/cpp/arcticdb/processing/clause.cpp
index 20bc37d399..78da7ab2f0 100644
--- a/cpp/arcticdb/processing/clause.cpp
+++ b/cpp/arcticdb/processing/clause.cpp
@@ -407,6 +407,20 @@ Composite<EntityIds> AggregationClause::process(Composite<EntityIds>&& entity_id
                                                 // 11.01 seconds with caching
                                                 // Not worth worrying about right now
                                                 robin_hood::unordered_flat_map<RawType, size_t> offset_to_group;
+
+                                                const bool is_sparse = col.column_->is_sparse();
+                                                using optional_iter_type = std::optional<decltype(input_data.bit_vector()->first())>;
+                                                optional_iter_type iter = std::nullopt;
+                                                size_t previous_value_index = 0;
+                                                constexpr size_t missing_value_group_id = 0;
+
+                                                if (is_sparse)
+                                                {
+                                                    iter = std::make_optional(input_data.bit_vector()->first());
+                                                    // We use 0 for the missing value group id
+                                                    next_group_id++;
+                                                }
+
                                                 while (auto block = input_data.next<ScalarTagType<DataTypeTagType>>()) {
                                                     const auto row_count = block->row_count();
                                                     auto ptr = block->data();
@@ -429,6 +443,14 @@ Composite<EntityIds> AggregationClause::process(Composite<EntityIds>&& entity_id
                                                         } else {
                                                             val = *ptr;
                                                         }
+                                                        if (is_sparse) {
+                                                            for (size_t j = previous_value_index; j != *(iter.value()); ++j) {
+                                                                row_to_group.emplace_back(missing_value_group_id);
+                                                            }
+                                                            previous_value_index = *(iter.value()) + 1;
+                                                            ++(iter.value());
+                                                        }
+
                                                         if (auto it = hash_to_group->find(val); it == hash_to_group->end()) {
                                                             row_to_group.emplace_back(next_group_id);
                                                             auto group_id = next_group_id++;
@@ -439,6 +461,11 @@ Composite<EntityIds> AggregationClause::process(Composite<EntityIds>&& entity_id
                                                     }
                                                 }
 
+                                                // Marking all the last non-represented values as missing.
+                                                for (size_t i = row_to_group.size(); i <= size_t(col.column_->last_row()); ++i) {
+                                                    row_to_group.emplace_back(missing_value_group_id);
+                                                }
+
                                                 num_unique = next_group_id;
                                                 util::check(num_unique != 0, "Got zero unique values");
                                                 for (auto agg_data: folly::enumerate(aggregators_data)) {


Test patch:
diff --git a/cpp/arcticdb/processing/test/test_clause.cpp b/cpp/arcticdb/processing/test/test_clause.cpp
index 11aebe2bbe..d55155e8d1 100644
--- a/cpp/arcticdb/processing/test/test_clause.cpp
+++ b/cpp/arcticdb/processing/test/test_clause.cpp
@@ -253,6 +253,67 @@ TEST(Clause, AggregationSparseColumn)
     });
 }
 
+TEST(Clause, AggregationSparseGroupby) {
+    using namespace arcticdb;
+    auto component_manager = std::make_shared<ComponentManager>();
+
+    AggregationClause aggregation("int_sparse_repeated_values", {{"sum_int", "sum"}, {"min_int", "min"}, {"max_int", "max"}, {"mean_int", "mean"}, {"count_int", "count"}});
+    aggregation.set_component_manager(component_manager);
+
+    size_t num_rows{100};
+    size_t unique_grouping_values{10};
+    // 1 more group because of missing values
+    size_t unique_groups{unique_grouping_values + 1};
+    auto proc_unit = ProcessingUnit{generate_sparse_groupby_testing_segment(num_rows, unique_grouping_values)};
+    auto entity_ids = Composite<EntityIds>(push_entities(component_manager, std::move(proc_unit)));
+
+    auto aggregated = gather_entities(component_manager, aggregation.process(std::move(entity_ids))).as_range();
+    ASSERT_EQ(1, aggregated.size());
+    ASSERT_TRUE(aggregated[0].segments_.has_value());
+    auto segments = aggregated[0].segments_.value();
+    ASSERT_EQ(1, segments.size());
+
+    using aggregation_test::check_column;
+    check_column<int64_t>(*segments[0], "sum_int", unique_groups, [unique_grouping_values](size_t idx) -> int64_t {
+        if (idx == 0) {
+            return 495;
+        } else {
+            return 450 - static_cast<int64_t>(idx % unique_grouping_values);
+        }
+    });
+    check_column<int64_t>(*segments[0], "min_int", unique_groups, [](size_t idx) -> int64_t {
+        if (idx == 0) {
+            return 0;
+        } else {
+            return idx;
+        }
+    });
+    check_column<int64_t>(*segments[0], "max_int", unique_groups, [unique_grouping_values](size_t idx) -> int64_t {
+        if (idx == 0) {
+            return 99;
+        }
+        else if (idx == 9) {
+            return 89;
+        } else {
+            return 90 + idx % unique_grouping_values;
+        }
+    });
+    check_column<double>(*segments[0], "mean_int", unique_groups, [unique_grouping_values](size_t idx) -> double {
+        if (idx == 0) {
+            return 49.5;
+        } else {
+            return (450 - static_cast<int64_t>(idx % unique_grouping_values)) / 9.;
+        }
+    });
+    check_column<uint64_t>(*segments[0], "count_int", unique_groups, [](size_t idx) -> uint64_t {
+        if (idx == 0) {
+            return 10;
+        } else {
+            return 9;
+        }
+    });
+}
+
 TEST(Clause, Passthrough) {
     using namespace arcticdb;
     auto component_manager = std::make_shared<ComponentManager>();
diff --git a/cpp/arcticdb/util/test/generators.hpp b/cpp/arcticdb/util/test/generators.hpp
index 706560fce8..4eb922b356 100644
--- a/cpp/arcticdb/util/test/generators.hpp
+++ b/cpp/arcticdb/util/test/generators.hpp
@@ -112,6 +112,18 @@ inline Column generate_int_column_repeated_values(size_t num_rows, size_t unique
     return column;
 }
 
+// Similar to generate_int_column_repeated_values, but with missing values where index % (unique_values + 1) == 0
+inline Column generate_int_column_sparse_repeated_values(size_t num_rows, size_t unique_values) {
+    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 % (unique_values + 1) != 0) {
+            column.set_scalar<int64_t>(static_cast<ssize_t>(idx), static_cast<int64_t>(idx % unique_values));
+        }
+    }
+    return column;
+}
+
 // Generates a Column of empty type
 inline Column generate_empty_column() {
     using TDT = TypeDescriptorTag<DataTypeTag<DataType::EMPTYVAL>, DimensionTag<Dimension ::Dim0>>;
@@ -164,6 +176,22 @@ inline SegmentInMemory generate_groupby_testing_sparse_segment(size_t num_rows,
     return seg;
 }
 
+inline SegmentInMemory generate_sparse_groupby_testing_segment(size_t num_rows, size_t unique_values)
+{
+    SegmentInMemory seg;
+    auto int_sparse_repeated_values_col = std::make_shared<Column>(generate_int_column_sparse_repeated_values(num_rows, unique_values));
+    int_sparse_repeated_values_col->mark_absent_rows(num_rows-1);
+    seg.add_column(scalar_field(int_sparse_repeated_values_col->type().data_type(), "int_sparse_repeated_values"), int_sparse_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 get_standard_timeseries_segment(const std::string& name, size_t num_rows = 10) {
     auto wrapper = SinkWrapper(name, {
         scalar_field(DataType::INT8, "int8"),