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

Issue:
Research if specific handling of EmptyType (and other new types) is needed
**Is your feature request related to a problem? Please describe.**
Up until #646 ArcticDB supported only two types strings and numbers. Some functions assumed that if something is not strings it's a number and vice versa. With the appearance of new types (EmptyType, Decimal, Nullable Bolleans, etc...) we might need to change that and add specific handling for each type rather than
```c++
if(is_sequence_type()) {
// handle strings
} else {
// handle everything else the same way
}
```
The functions which do this are listed here:
```c++
VariantData binary_membership(const ColumnWithStrings& column_with_strings, ValueSet& value_set, Func&& func)
VariantData binary_comparator(const Value& val, const ColumnWithStrings& column_with_strings, Func&& func)
VariantData binary_comparator(const ColumnWithStrings& left, const ColumnWithStrings& right, Func&& func)
VariantData binary_comparator(const ColumnWithStrings& column_with_strings, const Value& val, Func&& func)
SegmentInMemoryImpl::visit_field(Callable &&c)
SegmentInMemoryImpl::operator==(const SegmentInMemoryImpl& left, const SegmentInMemoryImpl& right)
FrameSliceMap::FrameSliceMap(const std::shared_ptr<PipelineContext>& context, bool dynamic_schema)
std::shared_ptr<SegmentInMemoryImpl> filter(const util::BitSet& filter_bitset,
                                                       bool filter_down_stringpool=false,
                                                       bool validate=false) const {
Value(const Value& other)
void SumAggregatorData::aggregate(const std::optional<ColumnWithStrings>& input_column, const std::vector<size_t>& groups, size_t unique_values)
void MinMaxAggregatorData::aggregate(const ColumnWithStrings& input_column)
void MaxOrMinAggregatorData::aggregate(const std::optional<ColumnWithStrings>& input_column, const std::vector<size_t>& groups, size_t unique_values)
VariantData binary_membership(const ColumnWithStrings& column_with_strings, ValueSet& value_set, Func&& func) {
VariantData binary_comparator(const Value& val, const ColumnWithStrings& column_with_strings, Func&& func) {
VariantData binary_comparator(const ColumnWithStrings& left, const ColumnWithStrings& right, Func&& func) {
VariantData binary_comparator(const ColumnWithStrings& column_with_strings, const Value& val, Func&& func) {
inline StreamId stream_id_from_segment(const SegmentInMemory &seg, ssize_t row) {
folly::Future<arcticdb::entity::VariantKey> write_frame_data(const TestDataFrame &data_frame, WriterType &writer) {
```

**Describe the solution you'd like**
Research which of the functions need to change and add handling for the new types.

**Describe alternatives you've considered**

**Additional context**
Relates to: #646 and all other PRs adding new types.



Discussion / hints:

Fix patch:
diff --git a/cpp/arcticdb/CMakeLists.txt b/cpp/arcticdb/CMakeLists.txt
index 9aed3606b8..79c45b0825 100644
--- a/cpp/arcticdb/CMakeLists.txt
+++ b/cpp/arcticdb/CMakeLists.txt
@@ -712,6 +712,7 @@ if(${TEST})
             processing/test/test_clause.cpp
             processing/test/test_expression.cpp
             processing/test/test_has_valid_type_promotion.cpp
+            processing/test/test_operation_dispatch.cpp
             processing/test/test_set_membership.cpp
             processing/test/test_signed_unsigned_comparison.cpp
             processing/test/test_type_comparison.cpp
diff --git a/cpp/arcticdb/column_store/memory_segment_impl.hpp b/cpp/arcticdb/column_store/memory_segment_impl.hpp
index 027aa717fc..cff65561f6 100644
--- a/cpp/arcticdb/column_store/memory_segment_impl.hpp
+++ b/cpp/arcticdb/column_store/memory_segment_impl.hpp
@@ -102,8 +102,12 @@ class SegmentInMemoryImpl {
                 using RawType = typename DataTypeTag::raw_type;
                 if constexpr (is_sequence_type(DataTypeTag::data_type))
                     return c(that->parent_->string_at(that->row_id_, position_t(that->column_id_)), std::string_view{field.name()}, field.type());
-                else
+                else if constexpr (is_numeric_type(DataTypeTag::data_type) || is_bool_type(DataTypeTag::data_type))
                     return c(that->parent_->scalar_at<RawType>(that->row_id_, that->column_id_), std::string_view{field.name()}, field.type());
+                else if constexpr(is_empty_type(DataTypeTag::data_type))
+                    internal::raise<ErrorCode::E_ASSERTION_FAILURE>("visit_field does not support empty-type columns");
+                else
+                    internal::raise<ErrorCode::E_ASSERTION_FAILURE>("visit_field called with unexpected column type");
             });
         }
 
@@ -285,25 +289,30 @@ class SegmentInMemoryImpl {
         }
 
         template<class S>
-            std::optional<S> scalar_at(std::size_t col) const {
+        std::optional<S> scalar_at(std::size_t col) const {
             parent_->check_magic();
             const auto& type_desc = parent_->column_descriptor(col).type();
             std::optional<S> val;
-            type_desc.visit_tag([that=this, col, &val](auto impl) {
+            type_desc.visit_tag([this, col, &val](auto impl) {
                 using T = std::decay_t<decltype(impl)>;
                 using RawType = typename T::DataTypeTag::raw_type;
                 if constexpr (T::DimensionTag::value == Dimension::Dim0) {
                     if constexpr (is_sequence_type(T::DataTypeTag::data_type)) {
                         // test only for now
-                        throw std::runtime_error("string type not implemented");
+                        internal::raise<ErrorCode::E_ASSERTION_FAILURE>("string type not implemented");
+                    } else if constexpr(is_numeric_type(T::DataTypeTag::data_type) || is_bool_type(T::DataTypeTag::data_type)) {
+                        if constexpr(std::is_same_v<RawType, S>) {
+                            val = parent_->scalar_at<RawType>(row_id_, col);
+                        } else {
+                            internal::raise<ErrorCode::E_ASSERTION_FAILURE>("Type mismatch in scalar access");
+                        }
+                    } else if constexpr(is_empty_type(T::DataTypeTag::data_type)) {
+                        internal::raise<ErrorCode::E_ASSERTION_FAILURE>("scalar_at not supported with empty-type columns");
                     } else {
-                        if constexpr(std::is_same_v<RawType, S>)
-                        val = that->parent_->scalar_at<RawType>(that->row_id_, col);
-                        else
-                            util::raise_rte("Type mismatch in scalar access");
+                        internal::raise<ErrorCode::E_ASSERTION_FAILURE>("Unexpected data type in scalar access");
                     }
                 } else {
-                    throw std::runtime_error("Scalar method called on multidimensional column");
+                    internal::raise<ErrorCode::E_ASSERTION_FAILURE>("Scalar method called on multidimensional column");
                 }
             });
             return val;
@@ -646,6 +655,8 @@ class SegmentInMemoryImpl {
     template<typename T>
     std::optional<T> scalar_at(position_t row, position_t col) const {
         util::check_arg(size_t(row) < row_count(), "Segment index {} out of bounds in scalar", row);
+        internal::check<ErrorCode::E_ASSERTION_FAILURE>(!is_empty_type(column(col).type().data_type()),
+                                                        "scalar_at called with empty-type column");
         return column(col).scalar_at<T>(row);
     }
 
@@ -832,7 +843,7 @@ class SegmentInMemoryImpl {
             return false;
 
         for(auto col = 0u; col < left.columns_.size(); ++col) {
-            if(is_sequence_type(left.column(col).type().data_type())) {
+            if (is_sequence_type(left.column(col).type().data_type())) {
                 const auto& left_col = left.column(col);
                 const auto& right_col = right.column(col);
                 if(left_col.type() != right_col.type())
@@ -844,10 +855,14 @@ class SegmentInMemoryImpl {
                 for(auto row = 0u; row < left_col.row_count(); ++row)
                     if(left.string_at(row, col) != right.string_at(row, col))
                         return false;
-            }
-            else {
+            } else if (is_numeric_type(left.column(col).type().data_type()) || is_bool_type(left.column(col).type().data_type())) {
                 if (left.column(col) != right.column(col))
                     return false;
+            } else if (is_empty_type(left.column(col).type().data_type())) {
+                if (!is_empty_type(right.column(col).type().data_type()))
+                    return false;
+            } else {
+                internal::raise<ErrorCode::E_ASSERTION_FAILURE>("Unexpected data type in SegmentInMemory equality check");
             }
         }
         return true;
@@ -954,7 +969,8 @@ class SegmentInMemoryImpl {
         for(const auto& column : folly::enumerate(columns())) {
             (*column)->type().visit_tag([&] (auto type_desc_tag){
                 using TypeDescriptorTag =  decltype(type_desc_tag);
-                using RawType = typename TypeDescriptorTag::DataTypeTag::raw_type;
+                using DataTypeTag = typename TypeDescriptorTag::DataTypeTag;
+                using RawType = typename DataTypeTag::raw_type;
                 const util::BitSet* final_bitset;
                 util::BitSet bitset_including_sparse;
 
@@ -1007,7 +1023,7 @@ class SegmentInMemoryImpl {
                             }
                             auto offset = sparse_map.value().rank(*bitset_iter, *sparse_idx) - row_count_so_far - 1;
                             auto value = *(input_ptr + offset);
-                            if constexpr(is_sequence_type(TypeDescriptorTag::DataTypeTag::data_type)) {
+                            if constexpr(is_sequence_type(DataTypeTag::data_type)) {
                                 if (filter_down_stringpool) {
                                     if (auto it = input_to_output_offsets.find(value);
                                     it != input_to_output_offsets.end()) {
@@ -1021,8 +1037,12 @@ class SegmentInMemoryImpl {
                                 } else {
                                     *output_ptr = value;
                                 }
-                            } else {
+                            } else if constexpr(is_numeric_type(DataTypeTag::data_type) || is_bool_type(DataTypeTag::data_type)){
                                 *output_ptr = value;
+                            } else if constexpr(is_empty_type(DataTypeTag::data_type)) {
+                                internal::raise<ErrorCode::E_ASSERTION_FAILURE>("Unexpected block in empty type column in SegmentInMemoryImpl::filter");
+                            } else {
+                                internal::raise<ErrorCode::E_ASSERTION_FAILURE>("Unexpected column type in SegmentInMemoryImpl::filter");
                             }
                             output_ptr++;
                             output_col.opt_sparse_map().value()[pos_output++] = true;
@@ -1037,7 +1057,7 @@ class SegmentInMemoryImpl {
                                 break;
 
                             auto value = *(input_ptr + offset);
-                            if constexpr(is_sequence_type(TypeDescriptorTag::DataTypeTag::data_type)) {
+                            if constexpr(is_sequence_type(DataTypeTag::data_type)) {
                                 if (filter_down_stringpool) {
                                     if (auto it = input_to_output_offsets.find(value);
                                     it != input_to_output_offsets.end()) {
@@ -1051,8 +1071,12 @@ class SegmentInMemoryImpl {
                                 } else {
                                     *output_ptr = value;
                                 }
-                            } else {
+                            } else if constexpr(is_numeric_type(DataTypeTag::data_type) || is_bool_type(DataTypeTag::data_type)){
                                 *output_ptr = value;
+                            } else if constexpr(is_empty_type(DataTypeTag::data_type)) {
+                                internal::raise<ErrorCode::E_ASSERTION_FAILURE>("Unexpected block in empty type column in SegmentInMemoryImpl::filter");
+                            } else {
+                                internal::raise<ErrorCode::E_ASSERTION_FAILURE>("Unexpected column type in SegmentInMemoryImpl::filter");
                             }
 
                             ++output_ptr;
diff --git a/cpp/arcticdb/pipeline/value_set.cpp b/cpp/arcticdb/pipeline/value_set.cpp
index 2885d4194e..e3a515f1dd 100644
--- a/cpp/arcticdb/pipeline/value_set.cpp
+++ b/cpp/arcticdb/pipeline/value_set.cpp
@@ -64,6 +64,47 @@ namespace arcticdb {
         });
     }
 
+    ValueSet::ValueSet(NumericSetType&& value_set):
+    numeric_base_set_(std::move(value_set)) {
+        util::variant_match(
+                numeric_base_set_,
+                [this](const std::shared_ptr<std::unordered_set<uint8_t>>&) {
+                    base_type_ = make_scalar_type(combine_data_type(entity::ValueType::UINT, entity::SizeBits::S8));
+                },
+                [this](const std::shared_ptr<std::unordered_set<uint16_t>>&) {
+                    base_type_ = make_scalar_type(combine_data_type(entity::ValueType::UINT, entity::SizeBits::S16));
+                },
+                [this](const std::shared_ptr<std::unordered_set<uint32_t>>&) {
+                    base_type_ = make_scalar_type(combine_data_type(entity::ValueType::UINT, entity::SizeBits::S32));
+                },
+                [this](const std::shared_ptr<std::unordered_set<uint64_t>>&) {
+                    base_type_ = make_scalar_type(combine_data_type(entity::ValueType::UINT, entity::SizeBits::S64));
+                },
+                [this](const std::shared_ptr<std::unordered_set<int8_t>>&) {
+                    base_type_ = make_scalar_type(combine_data_type(entity::ValueType::INT, entity::SizeBits::S8));
+                },
+                [this](const std::shared_ptr<std::unordered_set<int16_t>>&) {
+                    base_type_ = make_scalar_type(combine_data_type(entity::ValueType::INT, entity::SizeBits::S16));
+                },
+                [this](const std::shared_ptr<std::unordered_set<int32_t>>&) {
+                    base_type_ = make_scalar_type(combine_data_type(entity::ValueType::INT, entity::SizeBits::S32));
+                },
+                [this](const std::shared_ptr<std::unordered_set<int64_t>>&) {
+                    base_type_ = make_scalar_type(combine_data_type(entity::ValueType::INT, entity::SizeBits::S64));
+                },
+                [this](const std::shared_ptr<std::unordered_set<float>>&) {
+                    base_type_ = make_scalar_type(combine_data_type(entity::ValueType::FLOAT, entity::SizeBits::S32));
+                },
+                [this](const std::shared_ptr<std::unordered_set<double>>&) {
+                    base_type_ = make_scalar_type(combine_data_type(entity::ValueType::FLOAT, entity::SizeBits::S64));
+                }
+                );
+        util::variant_match(numeric_base_set_,
+                            [&](const auto& numeric_base_set) {
+                                empty_ = numeric_base_set->empty();
+                            });
+    }
+
     bool ValueSet::empty() const {
         return empty_;
     }
diff --git a/cpp/arcticdb/pipeline/value_set.hpp b/cpp/arcticdb/pipeline/value_set.hpp
index 2ff486849e..d3ac407843 100644
--- a/cpp/arcticdb/pipeline/value_set.hpp
+++ b/cpp/arcticdb/pipeline/value_set.hpp
@@ -25,10 +25,23 @@ namespace arcticdb {
 
 namespace py = pybind11;
 
+using NumericSetType = std::variant<
+        std::shared_ptr<std::unordered_set<uint8_t>>,
+        std::shared_ptr<std::unordered_set<uint16_t>>,
+        std::shared_ptr<std::unordered_set<uint32_t>>,
+        std::shared_ptr<std::unordered_set<uint64_t>>,
+        std::shared_ptr<std::unordered_set<int8_t>>,
+        std::shared_ptr<std::unordered_set<int16_t>>,
+        std::shared_ptr<std::unordered_set<int32_t>>,
+        std::shared_ptr<std::unordered_set<int64_t>>,
+        std::shared_ptr<std::unordered_set<float>>,
+        std::shared_ptr<std::unordered_set<double>>>;
+
 class ValueSet {
 public:
     explicit ValueSet(std::vector<std::string>&& value_list);
     explicit ValueSet(py::array value_list);
+    explicit ValueSet(NumericSetType&& value_set);
 
     bool empty() const;
 
@@ -42,18 +55,6 @@ class ValueSet {
     std::shared_ptr<std::unordered_set<std::string>> get_fixed_width_string_set(size_t width);
 
 private:
-    using NumericSetType = std::variant<
-            std::shared_ptr<std::unordered_set<uint8_t>>,
-            std::shared_ptr<std::unordered_set<uint16_t>>,
-            std::shared_ptr<std::unordered_set<uint32_t>>,
-            std::shared_ptr<std::unordered_set<uint64_t>>,
-            std::shared_ptr<std::unordered_set<int8_t>>,
-            std::shared_ptr<std::unordered_set<int16_t>>,
-            std::shared_ptr<std::unordered_set<int32_t>>,
-            std::shared_ptr<std::unordered_set<int64_t>>,
-            std::shared_ptr<std::unordered_set<float>>,
-            std::shared_ptr<std::unordered_set<double>>>;
-
     bool empty_;
     entity::TypeDescriptor base_type_;
     NumericSetType numeric_base_set_;
diff --git a/cpp/arcticdb/processing/aggregation.cpp b/cpp/arcticdb/processing/aggregation.cpp
index 41813cb238..3146847506 100644
--- a/cpp/arcticdb/processing/aggregation.cpp
+++ b/cpp/arcticdb/processing/aggregation.cpp
@@ -77,7 +77,9 @@ SegmentInMemory MinMaxAggregatorData::finalize(const std::vector<ColumnName>& ou
 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
-    data_type_ = data_type_.value_or(DataType::FLOAT64);
+    if (!data_type_.has_value() || *data_type_ == DataType::EMPTYVAL) {
+        data_type_ = DataType::FLOAT64;
+    }
     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)) {
diff --git a/cpp/arcticdb/processing/aggregation.hpp b/cpp/arcticdb/processing/aggregation.hpp
index 636bbe530c..e721aca19d 100644
--- a/cpp/arcticdb/processing/aggregation.hpp
+++ b/cpp/arcticdb/processing/aggregation.hpp
@@ -159,7 +159,7 @@ struct MaxOrMinAggregatorData {
         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() && input_column.has_value()) {
+        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)) {
diff --git a/cpp/arcticdb/processing/clause.cpp b/cpp/arcticdb/processing/clause.cpp
index 81c5d0f856..b828cf27cc 100644
--- a/cpp/arcticdb/processing/clause.cpp
+++ b/cpp/arcticdb/processing/clause.cpp
@@ -339,7 +339,11 @@ Composite<ProcessingUnit> AggregationClause::process(std::shared_ptr<Store> stor
                                                     auto input_column = proc_.get(input_column_name, store);
                                                     std::optional<ColumnWithStrings> opt_input_column;
                                                     if (std::holds_alternative<ColumnWithStrings>(input_column)) {
-                                                        opt_input_column.emplace(std::get<ColumnWithStrings>(input_column));
+                                                        auto column_with_strings = std::get<ColumnWithStrings>(input_column);
+                                                        // Empty columns don't contribute to aggregations
+                                                        if (!is_empty_type(column_with_strings.column_->type().data_type())) {
+                                                            opt_input_column.emplace(std::move(column_with_strings));
+                                                        }
                                                     }
                                                     agg_data->aggregate(opt_input_column, row_to_group, num_unique);
                                                 }
diff --git a/cpp/arcticdb/processing/operation_dispatch_binary.hpp b/cpp/arcticdb/processing/operation_dispatch_binary.hpp
index a1b9e6602c..2aad279ece 100644
--- a/cpp/arcticdb/processing/operation_dispatch_binary.hpp
+++ b/cpp/arcticdb/processing/operation_dispatch_binary.hpp
@@ -46,6 +46,15 @@ VariantData visit_binary_boolean(const VariantData& left, const VariantData& rig
 
 template <typename Func>
 VariantData binary_membership(const ColumnWithStrings& column_with_strings, ValueSet& value_set, Func&& func) {
+    if (is_empty_type(column_with_strings.column_->type().data_type())) {
+        if constexpr(std::is_same_v<Func, IsInOperator&&>) {
+            return EmptyResult{};
+        } else if constexpr(std::is_same_v<Func, IsNotInOperator&&>) {
+            return FullResult{};
+        } else {
+            internal::raise<ErrorCode::E_ASSERTION_FAILURE>("Unexpected operator passed to binary_membership");
+        }
+    }
     auto output = std::make_shared<util::BitSet>(column_with_strings.column_->row_count());
     // If the value set is empty, then for IsInOperator, we are done
     // For the IsNotInOperator, set all bits to 1
@@ -155,6 +164,9 @@ VariantData visit_binary_membership(const VariantData &left, const VariantData &
 
 template <typename Func>
 VariantData binary_comparator(const Value& val, const ColumnWithStrings& column_with_strings, Func&& func) {
+    if (is_empty_type(column_with_strings.column_->type().data_type())) {
+        return EmptyResult{};
+    }
     auto output = std::make_shared<util::BitSet>(static_cast<util::BitSetSizeType>(column_with_strings.column_->row_count()));
 
     column_with_strings.column_->type().visit_tag([&val, &column_with_strings, &func, &output] (auto column_desc_tag) {
@@ -193,7 +205,8 @@ VariantData binary_comparator(const Value& val, const ColumnWithStrings& column_
                     }
                 }
                 inserter.flush();
-            } else if constexpr (is_numeric_type(ColumnTagType::data_type) && is_numeric_type(DataTypeTag::data_type)) {
+            } else if constexpr ((is_numeric_type(ColumnTagType::data_type) && is_numeric_type(DataTypeTag::data_type)) ||
+                                 (is_bool_type(ColumnTagType::data_type) && is_bool_type(DataTypeTag::data_type))) {
                 using RawType =  typename decltype(value_desc_tag)::DataTypeTag::raw_type;
                 using comp = typename arcticdb::Comparable<RawType, ColumnType>;
                 auto value = static_cast<typename comp::left_type>(*reinterpret_cast<const RawType*>(val.data_));
@@ -222,6 +235,9 @@ VariantData binary_comparator(const Value& val, const ColumnWithStrings& column_
 
 template <typename Func>
 VariantData binary_comparator(const ColumnWithStrings& left, const ColumnWithStrings& right, Func&& func) {
+    if (is_empty_type(left.column_->type().data_type()) || is_empty_type(right.column_->type().data_type())) {
+        return EmptyResult{};
+    }
     util::check(left.column_->row_count() == right.column_->row_count(), "Columns with different row counts ({} and {}) in binary comparator", left.column_->row_count(), right.column_->row_count());
     auto output = std::make_shared<util::BitSet>(static_cast<util::BitSetSizeType>(left.column_->row_count()));
 
@@ -255,7 +271,8 @@ VariantData binary_comparator(const ColumnWithStrings& left, const ColumnWithStr
                     }
                 }
                 inserter.flush();
-            } else if constexpr (is_numeric_type(LeftTagType::data_type) && is_numeric_type(RightTagType::data_type)) {
+            } else if constexpr ((is_numeric_type(LeftTagType::data_type) && is_numeric_type(RightTagType::data_type)) ||
+                                 (is_bool_type(LeftTagType::data_type) && is_bool_type(RightTagType::data_type))) {
                 using comp = typename arcticdb::Comparable<LeftType, RightType>;
 
                 auto left_column_data = left.column_->data();
@@ -286,6 +303,9 @@ VariantData binary_comparator(const ColumnWithStrings& left, const ColumnWithStr
 
 template <typename Func>
 VariantData binary_comparator(const ColumnWithStrings& column_with_strings, const Value& val, Func&& func) {
+    if (is_empty_type(column_with_strings.column_->type().data_type())) {
+        return EmptyResult{};
+    }
     auto output = std::make_shared<util::BitSet>(static_cast<util::BitSetSizeType>(column_with_strings.column_->row_count()));
 
     column_with_strings.column_->type().visit_tag([&val, &column_with_strings, &func, &output] (auto column_desc_tag) {
@@ -326,18 +346,18 @@ VariantData binary_comparator(const ColumnWithStrings& column_with_strings, cons
                 inserter.flush();
             } else if constexpr ((is_numeric_type(ColumnTagType::data_type) && is_numeric_type(DataTypeTag::data_type)) ||
                                  (is_bool_type(ColumnTagType::data_type) && is_bool_type(DataTypeTag::data_type))) {
-                using RawType =  typename decltype(value_desc_tag)::DataTypeTag::raw_type;
+                using RawType = typename decltype(value_desc_tag)::DataTypeTag::raw_type;
                 using comp = typename arcticdb::Comparable<RawType, ColumnType>;
-                auto value = static_cast<typename comp::left_type>(*reinterpret_cast<const RawType*>(val.data_));
+                auto value = static_cast<typename comp::left_type>(*reinterpret_cast<const RawType *>(val.data_));
                 auto column_data = column_with_strings.column_->data();
 
                 util::BitSet::bulk_insert_iterator inserter(*output);
                 auto pos = 0u;
                 while (auto block = column_data.next<ColumnDescriptorType>()) {
-                    auto ptr = reinterpret_cast<const ColumnType*>(block.value().data());
+                    auto ptr = reinterpret_cast<const ColumnType *>(block.value().data());
                     const auto row_count = block.value().row_count();
                     for (auto i = 0u; i < row_count; ++i, ++pos) {
-                        if(func(static_cast<typename comp::right_type>(*ptr++), value))
+                        if (func(static_cast<typename comp::right_type>(*ptr++), value))
                             inserter = pos;
                     }
                 }
@@ -409,6 +429,9 @@ VariantData binary_operator(const Value& left, const Value& right, Func&& func)
 
 template <typename Func>
 VariantData binary_operator(const Value& val, const Column& col, Func&& func) {
+    schema::check<ErrorCode::E_UNSUPPORTED_COLUMN_TYPE>(
+            !is_empty_type(col.type().data_type()),
+            "Empty column provided to binary operator");
     std::unique_ptr<Column> output;
 
     details::visit_type(col.type().data_type(), [&](auto right_desc_tag) {
@@ -450,6 +473,9 @@ VariantData binary_operator(const Value& val, const Column& col, Func&& func) {
 
 template <typename Func>
 VariantData binary_operator(const Column& left, const Column& right, Func&& func) {
+    schema::check<ErrorCode::E_UNSUPPORTED_COLUMN_TYPE>(
+            !is_empty_type(left.type().data_type()) && !is_empty_type(right.type().data_type()),
+            "Empty column provided to binary operator");
     util::check(left.row_count() == right.row_count(), "Columns with different row counts ({} and {}) in binary operator", left.row_count(), right.row_count());
     std::unique_ptr<Column> output;
 
@@ -496,6 +522,9 @@ VariantData binary_operator(const Column& left, const Column& right, Func&& func
 
 template <typename Func>
 VariantData binary_operator(const Column& col, const Value& val, Func&& func) {
+    schema::check<ErrorCode::E_UNSUPPORTED_COLUMN_TYPE>(
+            !is_empty_type(col.type().data_type()),
+            "Empty column provided to binary operator");
     std::unique_ptr<Column> output;
 
     details::visit_type(col.type().data_type(), [&](auto left_desc_tag) {
diff --git a/cpp/arcticdb/processing/operation_dispatch_unary.hpp b/cpp/arcticdb/processing/operation_dispatch_unary.hpp
index eeefa907b3..4059cc4b89 100644
--- a/cpp/arcticdb/processing/operation_dispatch_unary.hpp
+++ b/cpp/arcticdb/processing/operation_dispatch_unary.hpp
@@ -59,6 +59,9 @@ VariantData unary_operator(const Value& val, Func&& func) {
 
 template <typename Func>
 VariantData unary_operator(const Column& col, Func&& func) {
+    schema::check<ErrorCode::E_UNSUPPORTED_COLUMN_TYPE>(
+            !is_empty_type(col.type().data_type()),
+            "Empty column provided to unary operator");
     std::unique_ptr<Column> output;
 
     details::visit_type(col.type().data_type(), [&](auto col_desc_tag) {
diff --git a/cpp/arcticdb/processing/processing_unit.hpp b/cpp/arcticdb/processing/processing_unit.hpp
index 50e3e90b66..97507be100 100644
--- a/cpp/arcticdb/processing/processing_unit.hpp
+++ b/cpp/arcticdb/processing/processing_unit.hpp
@@ -191,34 +191,41 @@ namespace arcticdb {
             auto partitioning_column = std::get<ColumnWithStrings>(get_result);
             partitioning_column.column_->type().visit_tag([&output, &input, &partitioning_column, &store](auto type_desc_tag) {
                 using TypeDescriptorTag = decltype(type_desc_tag);
+                using DescriptorType = std::decay_t<TypeDescriptorTag>;
+                using TagType =  typename DescriptorType::DataTypeTag;
                 using ResolvedGrouperType = typename GrouperType::template Grouper<TypeDescriptorTag>;
 
-                ResolvedGrouperType grouper;
-                auto num_buckets = ConfigsMap::instance()->get_int("Partition.NumBuckets", async::TaskScheduler::instance()->cpu_thread_count());
-                if (num_buckets > std::numeric_limits<uint8_t>::max()) {
-                    log::version().warn("GroupBy partitioning buckets capped at {} (received {})",
-                                        std::numeric_limits<uint8_t>::max(),
-                                        num_buckets);
-                    num_buckets = std::numeric_limits<uint8_t>::max();
-                }
-                std::vector<ProcessingUnit> procs{static_cast<size_t>(num_buckets)};
-                BucketizerType bucketizer(num_buckets);
-                auto [row_to_bucket, bucket_counts] = get_buckets(partitioning_column, grouper, bucketizer);
-                for (auto& seg_slice_and_key : input.data()) {
-                    const SegmentInMemory& seg = seg_slice_and_key.segment(store);
-                    auto new_segs = partition_segment(seg, row_to_bucket, bucket_counts);
-                    for (auto&& new_seg: folly::enumerate(new_segs)) {
-                        if (bucket_counts.at(new_seg.index) > 0) {
-                            pipelines::FrameSlice new_slice{seg_slice_and_key.slice()};
-                            new_slice.adjust_rows(new_seg->row_count());
-                            procs.at(new_seg.index).data().emplace_back(pipelines::SliceAndKey{std::move(*new_seg), std::move(new_slice)});
+                // Partitioning on an empty column should return an empty composite
+                if constexpr(!is_empty_type(TagType::data_type)) {
+                    ResolvedGrouperType grouper;
+                    auto num_buckets = ConfigsMap::instance()->get_int("Partition.NumBuckets",
+                                                                       async::TaskScheduler::instance()->cpu_thread_count());
+                    if (num_buckets > std::numeric_limits<uint8_t>::max()) {
+                        log::version().warn("GroupBy partitioning buckets capped at {} (received {})",
+                                            std::numeric_limits<uint8_t>::max(),
+                                            num_buckets);
+                        num_buckets = std::numeric_limits<uint8_t>::max();
+                    }
+                    std::vector<ProcessingUnit> procs{static_cast<size_t>(num_buckets)};
+                    BucketizerType bucketizer(num_buckets);
+                    auto [row_to_bucket, bucket_counts] = get_buckets(partitioning_column, grouper, bucketizer);
+                    for (auto &seg_slice_and_key: input.data()) {
+                        const SegmentInMemory &seg = seg_slice_and_key.segment(store);
+                        auto new_segs = partition_segment(seg, row_to_bucket, bucket_counts);
+                        for (auto &&new_seg: folly::enumerate(new_segs)) {
+                            if (bucket_counts.at(new_seg.index) > 0) {
+                                pipelines::FrameSlice new_slice{seg_slice_and_key.slice()};
+                                new_slice.adjust_rows(new_seg->row_count());
+                                procs.at(new_seg.index).data().emplace_back(
+                                        pipelines::SliceAndKey{std::move(*new_seg), std::move(new_slice)});
+                            }
                         }
                     }
-                }
-                for (auto&& proc: folly::enumerate(procs)) {
-                    if (bucket_counts.at(proc.index) > 0) {
-                        proc->set_bucket(proc.index);
-                        output.push_back(std::move(*proc));
+                    for (auto &&proc: folly::enumerate(procs)) {
+                        if (bucket_counts.at(proc.index) > 0) {
+                            proc->set_bucket(proc.index);
+                            output.push_back(std::move(*proc));
+                        }
                     }
                 }
             });
diff --git a/cpp/arcticdb/version/local_versioned_engine.cpp b/cpp/arcticdb/version/local_versioned_engine.cpp
index 14f94201dc..07c30621c7 100644
--- a/cpp/arcticdb/version/local_versioned_engine.cpp
+++ b/cpp/arcticdb/version/local_versioned_engine.cpp
@@ -15,7 +15,6 @@
 #include <arcticdb/stream/stream_sink.hpp>
 #include <arcticdb/util/preconditions.hpp>
 #include <arcticdb/util/storage_lock.hpp>
-#include <arcticdb/util/test/generators.hpp>
 #include <arcticdb/entity/metrics.hpp>
 #include <arcticdb/version/version_tasks.hpp>
 #include <arcticdb/pipeline/index_utils.hpp>


Test patch:
diff --git a/cpp/arcticdb/column_store/test/test_memory_segment.cpp b/cpp/arcticdb/column_store/test/test_memory_segment.cpp
index abf8da526b..9cc4d98c23 100644
--- a/cpp/arcticdb/column_store/test/test_memory_segment.cpp
+++ b/cpp/arcticdb/column_store/test/test_memory_segment.cpp
@@ -11,6 +11,7 @@
 #include <arcticdb/util/test/test_utils.hpp>
 #include <arcticdb/stream/test/stream_test_common.hpp>
 #include <arcticdb/util/test/generators.hpp>
+#include <arcticdb/util/bitset.hpp>
 
 #include <folly/container/Enumerate.h>
 
@@ -125,6 +126,24 @@ TEST(MemSegment, IterateAndGetValues) {
     }
 }
 
+TEST(MemSegment, IterateWithEmptyTypeColumn) {
+    SegmentInMemory seg;
+    size_t num_rows = 10;
+    auto int_column = std::make_shared<Column>(generate_int_column(num_rows));
+    seg.add_column(scalar_field(int_column->type().data_type(), "int_column"), int_column);
+
+    auto empty_column = std::make_shared<Column>(generate_empty_column());
+    seg.add_column(scalar_field(empty_column->type().data_type(), "empty_column"), empty_column);
+    seg.set_row_id(num_rows - 1);
+    for (auto&& [idx, row]: folly::enumerate(seg)) {
+        ASSERT_EQ(static_cast<int64_t>(idx), row.scalar_at<int64_t>(0));
+        // Exception should be thrown regardless of the type requested for empty type columns
+        EXPECT_THROW(row.scalar_at<int64_t>(1).has_value(), InternalException);
+        EXPECT_THROW(row.scalar_at<float>(1).has_value(), InternalException);
+        EXPECT_THROW(row.scalar_at<uint8_t>(1).has_value(), InternalException);
+    }
+}
+
 TEST(MemSegment, CopyViaIterator) {
     auto frame_wrapper = get_test_timeseries_frame("test_get_values", 100, 0);
     auto& source =frame_wrapper.segment_;
@@ -210,6 +229,20 @@ TEST(MemSegment, CloneAndCompare) {
     ASSERT_EQ(equal, true);
 }
 
+TEST(MemSegment, CloneAndCompareWithEmptyTypeColumn) {
+    SegmentInMemory seg;
+    size_t num_rows = 10;
+    auto int_column = std::make_shared<Column>(generate_int_column(num_rows));
+    seg.add_column(scalar_field(int_column->type().data_type(), "int_column"), int_column);
+
+    auto empty_column = std::make_shared<Column>(generate_empty_column());
+    seg.add_column(scalar_field(empty_column->type().data_type(), "empty_column"), empty_column);
+    seg.set_row_id(num_rows - 1);
+    auto copied = seg.clone();
+    bool equal = seg == copied;
+    ASSERT_TRUE(equal);
+}
+
 TEST(MemSegment, SplitSegment) {
     auto segment = get_standard_timeseries_segment("test_clone", 100);
     auto split_segs = segment.split(10);
@@ -425,3 +458,30 @@ TEST(MemSegment, Append) {
     }
 }
 
+TEST(MemSegment, Filter) {
+    SegmentInMemory seg;
+    size_t num_rows = 10;
+    auto int_column = std::make_shared<Column>(generate_int_column(num_rows));
+    seg.add_column(scalar_field(int_column->type().data_type(), "int_column"), int_column);
+
+    auto empty_column = std::make_shared<Column>(generate_empty_column());
+    seg.add_column(scalar_field(empty_column->type().data_type(), "empty_column"), empty_column);
+    seg.set_row_id(num_rows - 1);
+
+    util::BitSet filter_bitset(num_rows);
+    std::vector<size_t> retained_rows{0, 4, num_rows - 1};
+    for (auto retained_row: retained_rows) {
+        filter_bitset.set_bit(retained_row);
+    }
+
+    auto filtered_seg = seg.filter(filter_bitset);
+
+    for (auto&& [idx, row]: folly::enumerate(filtered_seg)) {
+        ASSERT_EQ(static_cast<int64_t>(retained_rows[idx]), row.scalar_at<int64_t>(0));
+        // Exception should be thrown regardless of the type requested for empty type columns
+        EXPECT_THROW(row.scalar_at<int64_t>(1).has_value(), InternalException);
+        EXPECT_THROW(row.scalar_at<float>(1).has_value(), InternalException);
+        EXPECT_THROW(row.scalar_at<uint8_t>(1).has_value(), InternalException);
+    }
+}
+
diff --git a/cpp/arcticdb/processing/test/test_clause.cpp b/cpp/arcticdb/processing/test/test_clause.cpp
index 29e5abafae..40857360d3 100644
--- a/cpp/arcticdb/processing/test/test_clause.cpp
+++ b/cpp/arcticdb/processing/test/test_clause.cpp
@@ -99,6 +99,51 @@ TEST(Clause, PartitionString) {
     }
 }
 
+TEST(Clause, PartitionEmptyColumn) {
+    using namespace arcticdb;
+
+    auto seg = generate_groupby_testing_segment(100, 10);
+    ProcessingUnit processing_unit(std::move(seg), pipelines::FrameSlice{});
+    Composite<ProcessingUnit> comp;
+    comp.push_back(std::move(processing_unit));
+    PartitionClause<arcticdb::grouping::HashingGroupers, arcticdb::grouping::ModuloBucketizer> partition{"empty_sum"};
+
+    auto partitioned = partition.process(std::shared_ptr<Store>(), std::move(comp));
+    ASSERT_TRUE(partitioned.empty());
+}
+
+TEST(Clause, AggregationEmptyColumn) {
+    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", {{"empty_sum", "sum"}, {"empty_min", "min"}, {"empty_max", "max"}, {"empty_mean", "mean"}});
+
+    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());
+
+    // Sum aggregations should produce a float64 column full of zeros
+    auto sum_column_index = slice_and_keys[0].segment_->column_index("empty_sum");
+    ASSERT_TRUE(sum_column_index.has_value());
+    auto& sum_column = slice_and_keys[0].segment_->column(*sum_column_index);
+    ASSERT_EQ(DataType::FLOAT64, sum_column.type().data_type());
+    for (size_t idx = 0; idx < unique_grouping_values; idx++) {
+        ASSERT_EQ(double(0), sum_column.scalar_at<double>(idx));
+    }
+
+    // Min, max, and mean aggregations should not be present in the output segment
+    ASSERT_FALSE(slice_and_keys[0].segment_->column_index("empty_min").has_value());
+    ASSERT_FALSE(slice_and_keys[0].segment_->column_index("empty_max").has_value());
+    ASSERT_FALSE(slice_and_keys[0].segment_->column_index("empty_mean").has_value());
+}
+
 TEST(Clause, Passthrough) {
     using namespace arcticdb;
     auto seg = get_standard_timeseries_segment("passthrough");
diff --git a/cpp/arcticdb/processing/test/test_operation_dispatch.cpp b/cpp/arcticdb/processing/test/test_operation_dispatch.cpp
new file mode 100644
index 0000000000..4e9d851c6b
--- /dev/null
+++ b/cpp/arcticdb/processing/test/test_operation_dispatch.cpp
@@ -0,0 +1,139 @@
+/* Copyright 2023 Man Group Operations Limited
+ *
+ * Use of this software is governed by the Business Source License 1.1 included in the file licenses/BSL.txt.
+ *
+ * As of the Change Date specified in that file, in accordance with the Business Source License, use of this software will be governed by the Apache License, version 2.0.
+ */
+
+#include <gtest/gtest.h>
+
+#include <arcticdb/processing/expression_node.hpp>
+#include <arcticdb/processing/operation_dispatch_binary.hpp>
+#include <arcticdb/processing/operation_dispatch_unary.hpp>
+#include <arcticdb/pipeline/value.hpp>
+#include <arcticdb/pipeline/value_set.hpp>
+#include <arcticdb/util/test/generators.hpp>
+
+TEST(OperationDispatch, unary_operator) {
+    using namespace arcticdb;
+    size_t num_rows = 100;
+    auto int_column = ColumnWithStrings(std::make_unique<Column>(generate_int_column(num_rows)));
+    auto empty_column = ColumnWithStrings(std::make_unique<Column>(generate_empty_column()));
+
+    // int col
+    auto variant_data = visit_unary_operator(int_column, NegOperator{});
+    ASSERT_TRUE(std::holds_alternative<ColumnWithStrings>(variant_data));
+    auto results_column = std::get<ColumnWithStrings>(variant_data).column_;
+    for (size_t idx = 0; idx < num_rows; idx++) {
+        ASSERT_EQ(-idx, results_column->scalar_at<int64_t>(idx));
+    }
+    // empty col
+    EXPECT_THROW(visit_unary_operator(empty_column, NegOperator{}), SchemaException);
+}
+
+TEST(OperationDispatch, binary_operator) {
+    using namespace arcticdb;
+    size_t num_rows = 100;
+    auto int_column = ColumnWithStrings(std::make_unique<Column>(generate_int_column(num_rows)));
+    auto empty_column = ColumnWithStrings(std::make_unique<Column>(generate_empty_column()));
+    auto value = std::make_shared<Value>(static_cast<int64_t>(50), DataType::INT64);
+
+    // int col + int col
+    auto variant_data_0 = visit_binary_operator(int_column, int_column, PlusOperator{});
+    ASSERT_TRUE(std::holds_alternative<ColumnWithStrings>(variant_data_0));
+    auto results_column_0 = std::get<ColumnWithStrings>(variant_data_0).column_;
+    for (size_t idx = 0; idx < num_rows; idx++) {
+        ASSERT_EQ(idx + idx, results_column_0->scalar_at<int64_t>(idx));
+    }
+    // int col + val
+    auto variant_data_1 = visit_binary_operator(int_column, value, PlusOperator{});
+    ASSERT_TRUE(std::holds_alternative<ColumnWithStrings>(variant_data_1));
+    auto results_column_1 = std::get<ColumnWithStrings>(variant_data_1).column_;
+    for (size_t idx = 0; idx < num_rows; idx++) {
+        ASSERT_EQ(idx + 50, results_column_1->scalar_at<int64_t>(idx));
+    }
+    // val + int col
+    auto variant_data_2 = visit_binary_operator(value, int_column, PlusOperator{});
+    ASSERT_TRUE(std::holds_alternative<ColumnWithStrings>(variant_data_2));
+    auto results_column_2 = std::get<ColumnWithStrings>(variant_data_2).column_;
+    ASSERT_TRUE(*results_column_1 == *results_column_2);
+    // val + val
+    auto variant_data_3 = visit_binary_operator(value, value, PlusOperator{});
+    ASSERT_TRUE(std::holds_alternative<std::shared_ptr<Value>>(variant_data_3));
+    auto results_value = std::get<std::shared_ptr<Value>>(variant_data_3)->get<int64_t>();
+    ASSERT_EQ(results_value, 100);
+    // int col + empty col
+    EXPECT_THROW(visit_binary_operator(int_column, empty_column, PlusOperator{}), SchemaException);
+    // empty col + int col
+    EXPECT_THROW(visit_binary_operator(empty_column, int_column, PlusOperator{}), SchemaException);
+    // empty col + empty col
+    EXPECT_THROW(visit_binary_operator(empty_column, empty_column, PlusOperator{}), SchemaException);
+    // empty col + val
+    EXPECT_THROW(visit_binary_operator(empty_column, value, PlusOperator{}), SchemaException);
+    // val + empty col
+    EXPECT_THROW(visit_binary_operator(value, empty_column, PlusOperator{}), SchemaException);
+}
+
+TEST(OperationDispatch, binary_comparator) {
+    using namespace arcticdb;
+    size_t num_rows = 100;
+    auto int_column = ColumnWithStrings(std::make_unique<Column>(generate_int_column(num_rows)));
+    auto empty_column = ColumnWithStrings(std::make_unique<Column>(generate_empty_column()));
+    auto value = std::make_shared<Value>(static_cast<int64_t>(50), DataType::INT64);
+
+    // int col < int col
+    ASSERT_TRUE(std::holds_alternative<EmptyResult>(visit_binary_comparator(int_column, int_column, LessThanOperator{})));
+    // int col < val
+    auto variant_data_0 = visit_binary_comparator(int_column, value, LessThanOperator{});
+    ASSERT_TRUE(std::holds_alternative<std::shared_ptr<util::BitSet>>(variant_data_0));
+    auto results_bitset_0 = std::get<std::shared_ptr<util::BitSet>>(variant_data_0);
+    for (size_t idx = 0; idx < num_rows; idx++) {
+        ASSERT_EQ(idx < 50, results_bitset_0->get_bit(idx));
+    }
+    // val < int col
+    auto variant_data_1 = visit_binary_comparator(value, int_column, LessThanOperator{});
+    ASSERT_TRUE(std::holds_alternative<std::shared_ptr<util::BitSet>>(variant_data_1));
+    auto results_bitset_1 = std::get<std::shared_ptr<util::BitSet>>(variant_data_1);
+    for (size_t idx = 0; idx < num_rows; idx++) {
+        ASSERT_EQ(50 < idx, results_bitset_1->get_bit(idx));
+    }
+    // val < val not supported, should be handled at expression evaluation time
+    // int col < empty col
+    ASSERT_TRUE(std::holds_alternative<EmptyResult>(visit_binary_comparator(int_column, empty_column, LessThanOperator{})));
+    // empty col < int col
+    ASSERT_TRUE(std::holds_alternative<EmptyResult>(visit_binary_comparator(empty_column, int_column, LessThanOperator{})));
+    // empty col < empty col
+    ASSERT_TRUE(std::holds_alternative<EmptyResult>(visit_binary_comparator(empty_column, empty_column, LessThanOperator{})));
+    // empty col < val
+    ASSERT_TRUE(std::holds_alternative<EmptyResult>(visit_binary_comparator(empty_column, value, LessThanOperator{})));
+    // val < empty col
+    ASSERT_TRUE(std::holds_alternative<EmptyResult>(visit_binary_comparator(value, empty_column, LessThanOperator{})));
+}
+
+TEST(OperationDispatch, binary_membership) {
+    using namespace arcticdb;
+    size_t num_rows = 100;
+    auto int_column = ColumnWithStrings(std::make_unique<Column>(generate_int_column(num_rows)));
+    auto empty_column = ColumnWithStrings(std::make_unique<Column>(generate_empty_column()));
+    std::unordered_set<int64_t> raw_set{0, 23, 82, static_cast<int64_t>(num_rows) - 1, 1000000};
+    auto value_set = std::make_shared<ValueSet>(std::make_shared<std::unordered_set<int64_t>>(raw_set));
+
+    // int col isin set
+    auto variant_data_0 = visit_binary_membership(int_column, value_set, IsInOperator{});
+    ASSERT_TRUE(std::holds_alternative<std::shared_ptr<util::BitSet>>(variant_data_0));
+    auto results_bitset_0 = std::get<std::shared_ptr<util::BitSet>>(variant_data_0);
+    for (size_t idx = 0; idx < num_rows; idx++) {
+        ASSERT_EQ(raw_set.count(static_cast<int64_t>(idx)) > 0, results_bitset_0->get_bit(idx));
+    }
+    // int col isnotin set
+    auto variant_data_1 = visit_binary_membership(int_column, value_set, IsNotInOperator{});
+    ASSERT_TRUE(std::holds_alternative<std::shared_ptr<util::BitSet>>(variant_data_0));
+    auto results_bitset_1 = std::get<std::shared_ptr<util::BitSet>>(variant_data_1);
+    for (size_t idx = 0; idx < num_rows; idx++) {
+        ASSERT_EQ(raw_set.count(static_cast<int64_t>(idx)) == 0, results_bitset_1->get_bit(idx));
+    }
+    // empty col isin set
+    ASSERT_TRUE(std::holds_alternative<EmptyResult>(visit_binary_membership(empty_column, value_set, IsInOperator{})));
+    // empty col isnotin set
+    ASSERT_TRUE(std::holds_alternative<FullResult>(visit_binary_membership(empty_column, value_set, IsNotInOperator{})));
+}
\ No newline at end of file
diff --git a/cpp/arcticdb/util/test/generators.hpp b/cpp/arcticdb/util/test/generators.hpp
index 53f4e773a0..a527ff46cb 100644
--- a/cpp/arcticdb/util/test/generators.hpp
+++ b/cpp/arcticdb/util/test/generators.hpp
@@ -79,6 +79,48 @@ using SinkWrapper = SinkWrapperImpl<TestAggregator>;
 using TestSparseAggregator = Aggregator<TimeseriesIndex, FixedSchema, stream::NeverSegmentPolicy, SparseColumnPolicy>;
 using SparseSinkWrapper = SinkWrapperImpl<TestSparseAggregator>;
 
+// Generates an int64_t Column where the value in each row is equal to the row index
+inline Column generate_int_column(size_t num_rows) {
+    using TDT = TypeDescriptorTag<DataTypeTag<DataType::INT64>, DimensionTag<Dimension ::Dim0>>;
+    Column column(static_cast<TypeDescriptor>(TDT{}), 0, false, false);
+    for(size_t idx = 0; idx < num_rows; ++idx) {
+        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>>;
+    Column column(static_cast<TypeDescriptor>(TDT{}), 0, false, false);
+    for(size_t idx = 0; idx < num_rows; ++idx) {
+        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>>;
+    Column column(static_cast<TypeDescriptor>(TDT{}), 0, false, false);
+    return 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) {
+    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);
+    seg.add_column(scalar_field(DataType::EMPTYVAL, "empty_sum"), std::make_shared<Column>(generate_empty_column()));
+    seg.add_column(scalar_field(DataType::EMPTYVAL, "empty_min"), std::make_shared<Column>(generate_empty_column()));
+    seg.add_column(scalar_field(DataType::EMPTYVAL, "empty_max"), std::make_shared<Column>(generate_empty_column()));
+    seg.add_column(scalar_field(DataType::EMPTYVAL, "empty_mean"), std::make_shared<Column>(generate_empty_column()));
+    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"),