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

Issue:
Filtering does not work on sparse columns
See also https://github.com/man-group/arcticdb-man/issues/82


Discussion / hints:

Fix patch:
diff --git a/cpp/arcticdb/CMakeLists.txt b/cpp/arcticdb/CMakeLists.txt
index 518731a014..7de4835f4d 100644
--- a/cpp/arcticdb/CMakeLists.txt
+++ b/cpp/arcticdb/CMakeLists.txt
@@ -89,13 +89,14 @@ else()
     message(STATUS "NEW_FOLLY_DEPS" ${NEW_FOLLY_DEPS})
     list(REMOVE_ITEM NEW_FOLLY_DEPS "/usr/local/opt/openssl/include" )
     set_target_properties(Folly::folly_deps PROPERTIES INTERFACE_INCLUDE_DIRECTORIES "${NEW_FOLLY_DEPS}")
-
-    add_library(xxHash STATIC IMPORTED)
-    set_target_properties(xxHash PROPERTIES IMPORTED_LOCATION ${xxHash_LIBRARY})
-    set_target_properties(xxHash PROPERTIES INTERFACE_INCLUDE_DIRECTORIES ${xxHash_INCLUDE_DIR})
-
 endif()
 
+# Have to statically link xxhash as inlining causes non-deterministic hashing, and cannot dynamically link
+# # since `XXH_state_t` which we rely on is only available when statically linking.
+add_library(xxHash STATIC IMPORTED)
+set_target_properties(xxHash PROPERTIES IMPORTED_LOCATION ${xxHash_LIBRARY})
+set_target_properties(xxHash PROPERTIES INTERFACE_INCLUDE_DIRECTORIES ${xxHash_INCLUDE_DIR})
+
 # RocksDB can be manually excluded with set(ARCTICDB_INCLUDE_ROCKSDB 0)
 # Default is to include on PyPI builds and exclude on conda builds
 if(${ARCTICDB_INCLUDE_ROCKSDB})
@@ -195,6 +196,7 @@ set(arcticdb_srcs
         column_store/block.hpp
         column_store/chunked_buffer.hpp
         column_store/column_data.hpp
+        column_store/column_data_random_accessor.hpp
         column_store/column.hpp
         column_store/column_utils.hpp
         column_store/memory_segment.hpp
@@ -530,6 +532,7 @@ endif()
 set (arcticdb_core_libraries
         pybind11::module # Transitively includes Python::Module or Python::Python as appropriate
         arcticdb_proto
+        xxHash::xxHash
         prometheus-cpp::push
         prometheus-cpp::pull
         unordered_dense::unordered_dense
@@ -557,11 +560,7 @@ endif()
 # on their origin (i.e. vendored source or packages on conda-forge).
 if(${ARCTICDB_USING_CONDA})
 
-    # We must use dynamic libraries, except for xxHash, which must be statically linked.
-    # since `XXH_state_t` which we rely on is only available when statically linking.
-    # For all builds except linux conda's, we inline the functions so that xxhash does not need linking at all.
     list (APPEND arcticdb_core_libraries
-        xxHash::xxHash
         pcre
         msgpack-c
         ${LMDB_LIBRARIES}
@@ -759,6 +758,7 @@ if(${TEST})
             codec/test/test_codec.cpp
             column_store/test/ingestion_stress_test.cpp
             column_store/test/test_column.cpp
+            column_store/test/test_column_data_random_accessor.cpp
             column_store/test/test_index_filtering.cpp
             column_store/test/test_memory_segment.cpp
             entity/test/test_atom_key.cpp
@@ -774,6 +774,7 @@ if(${TEST})
             processing/test/test_clause.cpp
             processing/test/test_component_manager.cpp
             processing/test/test_expression.cpp
+            processing/test/test_filter_and_project_sparse.cpp
             processing/test/test_has_valid_type_promotion.cpp
             processing/test/test_operation_dispatch.cpp
             processing/test/test_set_membership.cpp
@@ -919,6 +920,7 @@ if(${TEST})
             column_store/test/rapidcheck_column_store.cpp
             column_store/test/rapidcheck_chunked_buffer.cpp
             column_store/test/rapidcheck_column.cpp
+            column_store/test/rapidcheck_column_data_random_accessor.cpp
             column_store/test/rapidcheck_column_map.cpp
             column_store/test/test_chunked_buffer.cpp
             stream/test/stream_test_common.cpp
diff --git a/cpp/arcticdb/column_store/chunked_buffer.hpp b/cpp/arcticdb/column_store/chunked_buffer.hpp
index a83b1caee0..2cc35f6251 100644
--- a/cpp/arcticdb/column_store/chunked_buffer.hpp
+++ b/cpp/arcticdb/column_store/chunked_buffer.hpp
@@ -91,8 +91,12 @@ class ChunkedBufferImpl {
     ChunkedBufferImpl() = default;
 
     explicit ChunkedBufferImpl(size_t size) {
-        if(size > 0)
+        if(size > 0) {
+            if (size > DefaultBlockSize) {
+                handle_transition_to_irregular();
+            }
             add_block(std::max(size, DefaultBlockSize), 0UL);
+        }
     }
 
     ChunkedBufferImpl &operator=(ChunkedBufferImpl &&other) noexcept {
diff --git a/cpp/arcticdb/column_store/column.cpp b/cpp/arcticdb/column_store/column.cpp
index c8d991efc8..3fe55c4c1f 100644
--- a/cpp/arcticdb/column_store/column.cpp
+++ b/cpp/arcticdb/column_store/column.cpp
@@ -12,6 +12,69 @@
 
 namespace arcticdb {
 
+void initialise_output_column(const Column& input_column, Column& output_column) {
+    if (&input_column != &output_column) {
+        size_t output_physical_rows;
+        if (input_column.is_sparse()) {
+            auto output_sparse_map = input_column.sparse_map();
+            output_physical_rows = output_sparse_map.count();
+            output_column.set_sparse_map(std::move(output_sparse_map));
+        } else {
+            output_physical_rows = input_column.row_count();
+        }
+        output_column.allocate_data(output_physical_rows * get_type_size(output_column.type().data_type()));
+        output_column.set_row_data(input_column.last_row());
+    }
+}
+
+void initialise_output_column(const Column& left_input_column, const Column& right_input_column, Column& output_column) {
+    if (&left_input_column != &output_column && &right_input_column != &output_column) {
+        size_t output_physical_rows;
+        std::optional<size_t> output_last_row;
+        if (!left_input_column.is_sparse() && !right_input_column.is_sparse()) {
+            // Both dense. Could be different lengths if the data is semantically sparse, but happens to be dense in the first n rows
+            output_physical_rows = std::min(left_input_column.row_count(), right_input_column.row_count());
+            output_last_row = std::min(left_input_column.last_row(), right_input_column.last_row());
+        } else {
+            // At least one sparse
+            util::BitSet output_sparse_map;
+            if (left_input_column.is_sparse() && right_input_column.is_sparse()) {
+                output_sparse_map = (left_input_column.sparse_map() & right_input_column.sparse_map());
+            } else if (left_input_column.is_sparse() && !right_input_column.is_sparse()) {
+                output_sparse_map = left_input_column.sparse_map();
+                // If the sparse column had more logical rows than the dense column, truncate the sparse map to the length of the dense column
+                if (left_input_column.last_row() > right_input_column.last_row()) {
+                    output_sparse_map.resize(right_input_column.row_count());
+                }
+            } else if (!left_input_column.is_sparse() && right_input_column.is_sparse()) {
+                output_sparse_map = right_input_column.sparse_map();
+                // If the sparse column had more logical rows than the dense column, truncate the sparse map to the length of the dense column
+                if (left_input_column.last_row() < right_input_column.last_row()) {
+                    output_sparse_map.resize(left_input_column.row_count());
+                }
+            }
+            output_physical_rows = output_sparse_map.count();
+            // Find the index of the last set bit (if any) for the output_last_row
+            util::BitSetSizeType last_set_bit;
+            if (output_sparse_map.find_reverse(last_set_bit)) {
+                output_last_row = static_cast<size_t>(last_set_bit);
+            }
+            output_column.set_sparse_map(std::move(output_sparse_map));
+        }
+        output_column.allocate_data(output_physical_rows * get_type_size(output_column.type().data_type()));
+        if (output_last_row.has_value()) {
+            output_column.set_row_data(*output_last_row);
+        }
+    }
+}
+
+void initialise_output_bitset(const util::BitSet& input_bitset, bool sparse_missing_value_output, util::BitSet& output_bitset) {
+    if (sparse_missing_value_output) {
+        output_bitset = input_bitset;
+        output_bitset.flip();
+    }
+}
+
 // Column operators
 bool operator==(const Column& left, const Column& right) {
     if (left.row_count() != right.row_count())
@@ -714,4 +777,8 @@ std::optional<util::BitMagic>& Column::opt_sparse_map() {
     return sparse_map_;
 }
 
+std::optional<util::BitMagic> Column::opt_sparse_map() const {
+    return sparse_map_;
+}
+
 } //namespace arcticdb
diff --git a/cpp/arcticdb/column_store/column.hpp b/cpp/arcticdb/column_store/column.hpp
index 2b65fbc798..7d7a11cdbe 100644
--- a/cpp/arcticdb/column_store/column.hpp
+++ b/cpp/arcticdb/column_store/column.hpp
@@ -9,6 +9,7 @@
 
 #include <arcticdb/column_store/chunked_buffer.hpp>
 #include <arcticdb/column_store/column_data.hpp>
+#include <arcticdb/column_store/column_data_random_accessor.hpp>
 #include <arcticdb/entity/native_tensor.hpp>
 #include <arcticdb/entity/performance_tracing.hpp>
 #include <arcticdb/entity/types.hpp>
@@ -27,6 +28,7 @@
 #include <pybind11/pybind11.h>
 #include <pybind11/numpy.h>
 
+#include <concepts>
 #include <numeric>
 #include <optional>
 
@@ -77,6 +79,12 @@ class StringPool;
 template <typename T>
 JiveTable create_jive_table(const Column& col);
 
+void initialise_output_column(const Column& input_column, Column& output_column);
+
+void initialise_output_column(const Column& left_input_column, const Column& right_input_column, Column& output_column);
+
+void initialise_output_bitset(const util::BitSet& input_bitset, bool sparse_missing_value_output, util::BitSet& output_bitset);
+
 class Column {
 
 public:
@@ -247,6 +255,7 @@ class Column {
     [[nodiscard]] util::BitMagic& sparse_map();
     [[nodiscard]] const util::BitMagic& sparse_map() const;
     [[nodiscard]] std::optional<util::BitMagic>& opt_sparse_map();
+    [[nodiscard]] std::optional<util::BitMagic> opt_sparse_map() const;
 
     template<typename TagType>
     auto begin() const {
@@ -603,27 +612,28 @@ class Column {
         internal::check<ErrorCode::E_ASSERTION_FAILURE>(!is_sparse(),
                                                         "Column::search_sorted not supported with sparse columns");
         std::optional<size_t> res;
-        type().visit_tag([this, &res, val, from_right](auto type_desc_tag){
-            using TDT = decltype(type_desc_tag);
-            using RawType = typename TDT::DataTypeTag::raw_type;
-            if constexpr(std::is_same_v<T, RawType>) {
+        auto column_data = data();
+        details::visit_type(type().data_type(), [this, &res, &column_data, val, from_right](auto type_desc_tag) {
+            using type_info = ScalarTypeInfo<decltype(type_desc_tag)>;
+            auto accessor = random_accessor<typename type_info::TDT>(&column_data);
+            if constexpr(std::is_same_v<T, typename type_info::RawType>) {
                 int64_t low{0};
                 int64_t high{row_count() -1};
                 while (!res.has_value()) {
                     auto mid{low + (high - low) / 2};
-                    auto mid_value = *scalar_at<RawType>(mid);
+                    auto mid_value = accessor.at(mid);
                     if (val == mid_value) {
                         // At least one value in the column exactly matches the input val
                         // Search to the right/left for the last/first such value
                         if (from_right) {
-                            while (++mid <= high && val == *scalar_at<RawType>(mid)) {}
+                            while (++mid <= high && val == accessor.at(mid)) {}
                             res = mid;
                         } else {
-                            while (--mid >= low && val == *scalar_at<RawType>(mid)) {}
+                            while (--mid >= low && val == accessor.at(mid)) {}
                             res = mid + 1;
                         }
                     } else if (val > mid_value) {
-                        if (mid + 1 <= high && val >= *scalar_at<RawType>(mid + 1)) {
+                        if (mid + 1 <= high && val >= accessor.at(mid + 1)) {
                             // Narrow the search interval
                             low = mid + 1;
                         } else {
@@ -631,7 +641,7 @@ class Column {
                             res = mid + 1;
                         }
                     } else { // val < mid_value
-                        if (mid - 1 >= low && val <= *scalar_at<RawType>(mid + 1)) {
+                        if (mid - 1 >= low && val <= accessor.at(mid + 1)) {
                             // Narrow the search interval
                             high = mid - 1;
                         } else {
@@ -665,24 +675,37 @@ class Column {
     );
 
     template <
-        typename input_tdt,
-        typename functor,
-        typename = std::enable_if_t<std::is_invocable_r_v<void, functor, typename input_tdt::DataTypeTag::raw_type>>>
+            typename input_tdt,
+            typename functor>
+    requires std::is_invocable_r_v<void, functor, typename input_tdt::DataTypeTag::raw_type>
     static void for_each(const Column& input_column, functor&& f) {
         auto input_data = input_column.data();
         std::for_each(input_data.cbegin<input_tdt>(), input_data.cend<input_tdt>(), std::forward<functor>(f));
     }
 
+    template <
+            typename input_tdt,
+            typename functor>
+    requires std::is_invocable_r_v<void, functor, typename ColumnData::Enumeration<typename input_tdt::DataTypeTag::raw_type>>
+    static void for_each_enumerated(const Column& input_column, functor&& f) {
+        auto input_data = input_column.data();
+        if (input_column.is_sparse()) {
+            std::for_each(input_data.cbegin<input_tdt, IteratorType::ENUMERATED, IteratorDensity::SPARSE>(), input_data.cend<input_tdt, IteratorType::ENUMERATED, IteratorDensity::SPARSE>(),
+                          std::forward<functor>(f));
+        } else {
+            std::for_each(input_data.cbegin<input_tdt, IteratorType::ENUMERATED, IteratorDensity::DENSE>(), input_data.cend<input_tdt, IteratorType::ENUMERATED, IteratorDensity::DENSE>(),
+                          std::forward<functor>(f));
+        }
+    }
+
     template <
         typename input_tdt,
         typename output_tdt,
-        typename functor,
-        typename = std::enable_if_t<std::is_invocable_r_v<
-            typename output_tdt::DataTypeTag::raw_type,
-            functor,
-            typename input_tdt::DataTypeTag::raw_type>>>
+        typename functor>
+    requires std::is_invocable_r_v<typename output_tdt::DataTypeTag::raw_type, functor, typename input_tdt::DataTypeTag::raw_type>
     static void transform(const Column& input_column, Column& output_column, functor&& f) {
         auto input_data = input_column.data();
+        initialise_output_column(input_column, output_column);
         auto output_data = output_column.data();
         std::transform(
             input_data.cbegin<input_tdt>(),
@@ -693,74 +716,177 @@ class Column {
     }
 
     template<
-        typename left_input_tdt,
-        typename right_input_tdt,
-        typename output_tdt,
-        typename functor,
-        typename = std::enable_if_t<std::is_invocable_r_v<
+            typename left_input_tdt,
+            typename right_input_tdt,
+            typename output_tdt,
+            typename functor>
+    requires std::is_invocable_r_v<
             typename output_tdt::DataTypeTag::raw_type,
             functor,
             typename left_input_tdt::DataTypeTag::raw_type,
-            typename right_input_tdt::DataTypeTag::raw_type>>>
+            typename right_input_tdt::DataTypeTag::raw_type>
     static void transform(const Column& left_input_column,
                           const Column& right_input_column,
                           Column& output_column,
                           functor&& f) {
         auto left_input_data = left_input_column.data();
         auto right_input_data = right_input_column.data();
+        initialise_output_column(left_input_column, right_input_column, output_column);
         auto output_data = output_column.data();
-        std::transform(left_input_data.cbegin<left_input_tdt>(), left_input_data.cend<left_input_tdt>(), right_input_data.cbegin<right_input_tdt>(), output_data.begin<output_tdt>(), std::forward<functor>(f));
+        auto output_it = output_data.begin<output_tdt>();
+
+        if (!left_input_column.is_sparse() && !right_input_column.is_sparse()) {
+            // Both dense, use std::transform over the shorter column to avoid going out-of-bounds
+            if (left_input_column.row_count() <= right_input_column.row_count()) {
+                std::transform(left_input_data.cbegin<left_input_tdt>(),
+                               left_input_data.cend<left_input_tdt>(),
+                               right_input_data.cbegin<right_input_tdt>(),
+                               output_it,
+                               std::forward<functor>(f));
+            } else {
+                std::transform(right_input_data.cbegin<left_input_tdt>(),
+                               right_input_data.cend<left_input_tdt>(),
+                               left_input_data.cbegin<right_input_tdt>(),
+                               output_it,
+                               std::forward<functor>(f));
+            }
+        } else if (left_input_column.is_sparse() && right_input_column.is_sparse()) {
+            // Both sparse, only project rows in the intersection of on-bits from both sparse maps
+            auto left_accessor = random_accessor<left_input_tdt>(&left_input_data);
+            auto right_accessor = random_accessor<right_input_tdt>(&right_input_data);
+            // TODO: experiment with more efficient bitset traversal methods
+            // https://github.com/tlk00/BitMagic/tree/master/samples/bvsample25
+            auto end_bit = output_column.sparse_map().end();
+            for (auto set_bit = output_column.sparse_map().first(); set_bit < end_bit; ++set_bit) {
+                *output_it++ = f(left_accessor.at(*set_bit), right_accessor.at(*set_bit));
+            }
+        } else if (left_input_column.is_sparse() && !right_input_column.is_sparse()) {
+            // One sparse, one dense. Use the enumerating forward iterator over the sparse column as it is more efficient than random access
+            auto right_accessor = random_accessor<right_input_tdt>(&right_input_data);
+            const auto right_column_row_count = right_input_column.row_count();
+            for (auto left_it = left_input_data.cbegin<left_input_tdt, IteratorType::ENUMERATED, IteratorDensity::SPARSE>();
+                 left_it != left_input_data.cend<left_input_tdt, IteratorType::ENUMERATED, IteratorDensity::SPARSE>() && left_it->idx() < right_column_row_count;
+                 ++left_it) {
+                *output_it++ = f(left_it->value(), right_accessor.at(left_it->idx()));
+            }
+        } else if (!left_input_column.is_sparse() && right_input_column.is_sparse()) {
+            // One sparse, one dense. Use the enumerating forward iterator over the sparse column as it is more efficient than random access
+            auto left_accessor = random_accessor<left_input_tdt>(&left_input_data);
+            const auto left_column_row_count = left_input_column.row_count();
+            for (auto right_it = right_input_data.cbegin<right_input_tdt, IteratorType::ENUMERATED, IteratorDensity::SPARSE>();
+                 right_it != right_input_data.cend<right_input_tdt, IteratorType::ENUMERATED, IteratorDensity::SPARSE>() && right_it->idx() < left_column_row_count;
+                 ++right_it) {
+                *output_it++ = f(left_accessor.at(right_it->idx()), right_it->value());
+            }
+        }
     }
 
     template <
-        typename input_tdt,
-        typename functor,
-        typename = std::enable_if_t<std::is_invocable_r_v<bool, functor, typename input_tdt::DataTypeTag::raw_type>>>
+            typename input_tdt,
+            std::predicate<typename input_tdt::DataTypeTag::raw_type> functor>
     static void transform(const Column& input_column,
                           util::BitSet& output_bitset,
+                          bool sparse_missing_value_output,
                           functor&& f) {
-        auto input_data = input_column.data();
+        if (input_column.is_sparse()) {
+            initialise_output_bitset(input_column.sparse_map(), sparse_missing_value_output, output_bitset);
+        } else {
+            // This allows for empty/full result optimisations, technically bitsets are always dynamically sized
+            output_bitset.resize(input_column.row_count());
+        }
         util::BitSet::bulk_insert_iterator inserter(output_bitset);
-        auto pos = 0u;
-        std::for_each(
-            input_data.cbegin<input_tdt>(),
-            input_data.cend<input_tdt>(),
-            [&inserter, &pos, f = std::forward<functor>(f)](auto input_value) {
-            if (f(input_value)) {
-                inserter = pos;
+        Column::for_each_enumerated<input_tdt>(input_column, [&inserter, f = std::forward<functor>(f)](auto enumerated_it) {
+            if (f(enumerated_it.value())) {
+                inserter = enumerated_it.idx();
             }
-            ++pos;
         });
         inserter.flush();
     }
 
     template <
-        typename left_input_tdt,
-        typename right_input_tdt,
-        typename functor,
-        typename = std::enable_if_t<std::is_invocable_r_v<
-            bool,
-            functor,
-            typename left_input_tdt::DataTypeTag::raw_type,
-            typename right_input_tdt::DataTypeTag::raw_type>>>
+            typename left_input_tdt,
+            typename right_input_tdt,
+            std::relation<typename left_input_tdt::DataTypeTag::raw_type, typename right_input_tdt::DataTypeTag::raw_type> functor>
     static void transform(const Column& left_input_column,
                           const Column& right_input_column,
                           util::BitSet& output_bitset,
+                          bool sparse_missing_value_output,
                           functor&& f) {
         auto left_input_data = left_input_column.data();
-        auto right_it = right_input_column.data().cbegin<right_input_tdt>();
+        auto right_input_data = right_input_column.data();
         util::BitSet::bulk_insert_iterator inserter(output_bitset);
-        auto pos = 0u;
-        std::for_each(
-            left_input_data.cbegin<left_input_tdt>(),
-            left_input_data.cend<left_input_tdt>(),
-            [&right_it, &inserter, &pos, f = std::forward<functor>(f)](auto left_value) {
-            auto right_value = *right_it++;
-            if (f(left_value, right_value)) {
-                inserter = pos;
+
+        if (!left_input_column.is_sparse() && !right_input_column.is_sparse()) {
+            // Both dense, use std::for_each over the shorter column to avoid going out-of-bounds
+            auto rows = std::max(left_input_column.last_row(), right_input_column.last_row()) + 1;
+            output_bitset.resize(rows);
+            if (sparse_missing_value_output && left_input_column.row_count() != right_input_column.row_count()) {
+                // Dense columns of different lengths, and missing values should be on in the output bitset
+                output_bitset.set_range(std::min(left_input_column.last_row(), right_input_column.last_row()) + 1, rows - 1);
             }
-            ++pos;
-        });
+            auto pos = 0u;
+            if (left_input_column.row_count() <= right_input_column.row_count()) {
+                auto right_it = right_input_data.cbegin<right_input_tdt>();
+                std::for_each(
+                        left_input_data.cbegin<left_input_tdt>(),
+                        left_input_data.cend<left_input_tdt>(),
+                        [&right_it, &inserter, &pos, f = std::forward<functor>(f)](auto left_value) {
+                            if (f(left_value, *right_it++)) {
+                                inserter = pos;
+                            }
+                            ++pos;
+                        });
+            } else {
+                auto left_it = left_input_data.cbegin<left_input_tdt>();
+                std::for_each(
+                        right_input_data.cbegin<right_input_tdt>(),
+                        right_input_data.cend<right_input_tdt>(),
+                        [&left_it, &inserter, &pos, f = std::forward<functor>(f)](auto right_value) {
+                            if (f(*left_it++, right_value)) {
+                                inserter = pos;
+                            }
+                            ++pos;
+                        });
+            }
+        } else if (left_input_column.is_sparse() && right_input_column.is_sparse()) {
+            // Both sparse, only check the intersection of on-bits from both sparse maps
+            auto bits_to_check = left_input_column.sparse_map() & right_input_column.sparse_map();
+            initialise_output_bitset(bits_to_check, sparse_missing_value_output, output_bitset);
+            auto left_accessor = random_accessor<left_input_tdt>(&left_input_data);
+            auto right_accessor = random_accessor<right_input_tdt>(&right_input_data);
+            // TODO: experiment with more efficient bitset traversal methods
+            // https://github.com/tlk00/BitMagic/tree/master/samples/bvsample25
+            auto end_bit = bits_to_check.end();
+            for (auto set_bit = bits_to_check.first(); set_bit < end_bit; ++set_bit) {
+                if(f(left_accessor.at(*set_bit), right_accessor.at(*set_bit))) {
+                    inserter = *set_bit;
+                }
+            }
+        } else if (left_input_column.is_sparse() && !right_input_column.is_sparse()) {
+            // One sparse, one dense. Use the enumerating forward iterator over the sparse column as it is more efficient than random access
+            initialise_output_bitset(left_input_column.sparse_map(), sparse_missing_value_output, output_bitset);
+            auto right_accessor = random_accessor<right_input_tdt>(&right_input_data);
+            const auto right_column_row_count = right_input_column.row_count();
+            for (auto left_it = left_input_data.cbegin<left_input_tdt, IteratorType::ENUMERATED, IteratorDensity::SPARSE>();
+                 left_it != left_input_data.cend<left_input_tdt, IteratorType::ENUMERATED, IteratorDensity::SPARSE>() && left_it->idx() < right_column_row_count;
+                 ++left_it) {
+                if(f(left_it->value(), right_accessor.at(left_it->idx()))) {
+                    inserter = left_it->idx();
+                }
+            }
+        } else if (!left_input_column.is_sparse() && right_input_column.is_sparse()) {
+            // One sparse, one dense. Use the enumerating forward iterator over the sparse column as it is more efficient than random access
+            initialise_output_bitset(right_input_column.sparse_map(), sparse_missing_value_output, output_bitset);
+            auto left_accessor = random_accessor<left_input_tdt>(&left_input_data);
+            const auto left_column_row_count = left_input_column.row_count();
+            for (auto right_it = right_input_data.cbegin<right_input_tdt, IteratorType::ENUMERATED, IteratorDensity::SPARSE>();
+                 right_it != right_input_data.cend<right_input_tdt, IteratorType::ENUMERATED, IteratorDensity::SPARSE>() && right_it->idx() < left_column_row_count;
+                 ++right_it) {
+                if(f(left_accessor.at(right_it->idx()), right_it->value())) {
+                    inserter = right_it->idx();
+                }
+            }
+        }
         inserter.flush();
     }
 
diff --git a/cpp/arcticdb/column_store/column_data.hpp b/cpp/arcticdb/column_store/column_data.hpp
index f583c86850..bba7fa81c5 100644
--- a/cpp/arcticdb/column_store/column_data.hpp
+++ b/cpp/arcticdb/column_store/column_data.hpp
@@ -127,19 +127,77 @@ struct TypedBlockData {
     const MemBlock *block_;  // pointer to the parent memblock from which this was created from.
 };
 
+enum class IteratorType {
+    REGULAR,
+    ENUMERATED
+};
+
+enum class IteratorDensity {
+    DENSE,
+    SPARSE
+};
+
 struct ColumnData {
 /*
  * ColumnData is just a thin wrapper that helps in iteration over all the blocks in the column
  */
+public:
+
+    template<typename RawType>
+    struct Enumeration {
+        ssize_t idx_{0};
+        RawType* ptr_{nullptr};
+
+        inline ssize_t idx() const {
+            return idx_;
+        }
+
+        inline RawType& value() {
+            debug::check<ErrorCode::E_ASSERTION_FAILURE>(ptr_ != nullptr, "Dereferencing nullptr in enumerating ColumnDataIterator");
+            return *ptr_;
+        };
+
+        inline const RawType& value() const {
+            debug::check<ErrorCode::E_ASSERTION_FAILURE>(ptr_ != nullptr, "Dereferencing nullptr in enumerating ColumnDataIterator");
+            return *ptr_;
+        };
+    };
 
-  public:
-    template<typename TDT, bool constant>
-    class ColumnDataIterator: public boost::iterator_facade<
-            ColumnDataIterator<TDT, constant>,
-            std::conditional_t<constant, typename TDT::DataTypeTag::raw_type const, typename TDT::DataTypeTag::raw_type>,
-            boost::forward_traversal_tag
-            > {
-    using RawType = std::conditional_t<constant, const typename TDT::DataTypeTag::raw_type, typename TDT::DataTypeTag::raw_type>;
+    // Wrapper to keep code in iterator clean when differentiating between regular and enumerating iterators
+    template<typename RawType>
+    struct PointerWrapper {
+        RawType* ptr_{nullptr};
+    };
+
+    template <class T, IteratorType iterator_type>
+    using IteratorValueType_t = typename std::conditional_t<
+            iterator_type == IteratorType::ENUMERATED,
+            Enumeration<T>,
+            PointerWrapper<T>
+    >;
+
+    template <class T, IteratorType iterator_type, bool constant>
+    using IteratorReferenceType_t = typename std::conditional_t<
+        iterator_type == IteratorType::ENUMERATED,
+            std::conditional_t<constant, const Enumeration<T>, Enumeration<T>>,
+            std::conditional_t<constant, const T, T>
+    >;
+
+    template<typename TDT, IteratorType iterator_type, IteratorDensity iterator_density, bool constant>
+    class ColumnDataIterator;
+
+    template<typename TDT, IteratorType iterator_type, IteratorDensity iterator_density, bool constant>
+    using base_iterator_type = boost::iterator_facade<
+        ColumnDataIterator<TDT, iterator_type, iterator_density, constant>,
+        IteratorValueType_t<typename TDT::DataTypeTag::raw_type, iterator_type>,
+        boost::forward_traversal_tag,
+        IteratorReferenceType_t<typename TDT::DataTypeTag::raw_type, iterator_type, constant>&
+    >;
+
+    template<typename TDT, IteratorType iterator_type, IteratorDensity iterator_density, bool constant>
+    class ColumnDataIterator: public base_iterator_type<TDT, iterator_type, iterator_density, constant> {
+        using base_type = base_iterator_type<TDT, iterator_type, iterator_density, constant>;
+        using RawType = typename TDT::DataTypeTag::raw_type;
     public:
         ColumnDataIterator() = delete;
 
@@ -148,57 +206,71 @@ struct ColumnData {
         parent_(parent)
         {
             increment_block();
+            if constexpr (iterator_type == IteratorType::ENUMERATED && iterator_density == IteratorDensity::SPARSE) {
+                // idx_ default-constructs to 0, which is correct for dense case
+                data_.idx_ = parent_->bit_vector()->get_first();
+            }
         }
 
         // Used to construct [c]end iterators
-        explicit ColumnDataIterator(ColumnData* parent, RawType* end_ptr):
-                parent_(parent),
-                ptr_(end_ptr) {}
-
-        template <class OtherValue, bool OtherConst>
-        ColumnDataIterator(ColumnDataIterator<OtherValue, OtherConst> const& other):
-        parent_(other.parent_),
-        opt_block_(other.opt_block_),
-        remaining_values_in_block_(other.remaining_values_in_block_),
-        ptr_(other.ptr_) {}
+        explicit ColumnDataIterator(ColumnData* parent, typename TDT::DataTypeTag::raw_type* end_ptr):
+                parent_(parent) {
+            data_.ptr_ = end_ptr;
+        }
+
+        template <bool OtherConst>
+        explicit ColumnDataIterator(const ColumnDataIterator<TDT, iterator_type, iterator_density, OtherConst>& other):
+            parent_(other.parent_),
+            opt_block_(other.opt_block_),
+            remaining_values_in_block_(other.remaining_values_in_block_),
+            data_(other.data_)
+        {}
+
     private:
         friend class boost::iterator_core_access;
-        template <class, bool> friend class ColumnDataIterator;
 
         void increment() {
-            if (ARCTICDB_LIKELY(remaining_values_in_block_ > 0)) {
-                ++ptr_;
-                --remaining_values_in_block_;
-            } else {
+            ++data_.ptr_;
+            if (ARCTICDB_UNLIKELY(--remaining_values_in_block_ == 0)) {
                 increment_block();
             }
+            if constexpr (iterator_type == IteratorType::ENUMERATED) {
+                if constexpr (iterator_density == IteratorDensity::SPARSE) {
+                    data_.idx_ = parent_->bit_vector()->get_next(data_.idx_);
+                } else {
+                    ++data_.idx_;
+                }
+            }
         }
 
         void increment_block() {
             opt_block_ = parent_->next<TDT>();
             if(ARCTICDB_LIKELY(opt_block_.has_value())) {
                 remaining_values_in_block_ = opt_block_->row_count();
-                if constexpr(constant) {
-                    ptr_ = reinterpret_cast<RawType*>(opt_block_->data());
-                } else {
-                    ptr_ = const_cast<RawType*>(opt_block_->data());
-                }
+                data_.ptr_ = const_cast<typename TDT::DataTypeTag::raw_type*>(opt_block_->data());
             }
         }
 
-        template <typename OtherValue, bool OtherConst>
-        bool equal(ColumnDataIterator<OtherValue, OtherConst> const& other) const {
-            return parent_ == other.parent_ && ptr_ == other.ptr_;
+        template <bool OtherConst>
+        bool equal(const ColumnDataIterator<TDT, iterator_type, iterator_density, OtherConst>& other) const {
+            debug::check<ErrorCode::E_ASSERTION_FAILURE>(parent_ == other.parent_,
+                                                         "ColumnDataIterator::equal called with different parent ColumnData*");
+            return data_.ptr_ == other.data_.ptr_;
         }
 
-        RawType& dereference() const {
-            return *ptr_;
+        base_type::reference dereference() const {
+            if constexpr (iterator_type == IteratorType::ENUMERATED) {
+                return data_;
+            } else {
+                debug::check<ErrorCode::E_ASSERTION_FAILURE>(data_.ptr_ != nullptr, "Dereferencing nullptr in ColumnDataIterator");
+                return *data_.ptr_;
+            }
         }
 
         ColumnData* parent_{nullptr};
         std::optional<TypedBlockData<TDT>> opt_block_{std::nullopt};
         std::size_t remaining_values_in_block_{0};
-        RawType* ptr_{nullptr};
+        base_type::value_type data_;
     };
 
     ColumnData(
@@ -215,18 +287,18 @@ struct ColumnData {
 
     ARCTICDB_MOVE_COPY_DEFAULT(ColumnData)
 
-    template<typename TDT>
-    ColumnDataIterator<TDT, false> begin() {
-        return ColumnDataIterator<TDT, false>(this);
+    template<typename TDT, IteratorType iterator_type=IteratorType::REGULAR, IteratorDensity iterator_density=IteratorDensity::DENSE>
+    ColumnDataIterator<TDT, iterator_type, iterator_density, false> begin() {
+        return ColumnDataIterator<TDT, iterator_type, iterator_density, false>(this);
     }
 
-    template<typename TDT>
-    ColumnDataIterator<TDT, true> cbegin() {
-        return ColumnDataIterator<TDT, true>(this);
+    template<typename TDT, IteratorType iterator_type=IteratorType::REGULAR, IteratorDensity iterator_density=IteratorDensity::DENSE>
+    ColumnDataIterator<TDT, iterator_type, iterator_density, true> cbegin() {
+        return ColumnDataIterator<TDT, iterator_type, iterator_density, true>(this);
     }
 
-    template<typename TDT>
-    ColumnDataIterator<TDT, false> end() {
+    template<typename TDT, IteratorType iterator_type=IteratorType::REGULAR, IteratorDensity iterator_density=IteratorDensity::DENSE>
+    ColumnDataIterator<TDT, iterator_type, iterator_density, false> end() {
         using RawType = typename TDT::DataTypeTag::raw_type;
         RawType* end_ptr{nullptr};
         if(!data_->blocks().empty()) {
@@ -234,19 +306,19 @@ struct ColumnData {
             auto typed_block_data = next_typed_block<TDT>(block);
             end_ptr = typed_block_data.data() + typed_block_data.row_count();
         }
-        return ColumnDataIterator<TDT, false>(this, end_ptr);
+        return ColumnDataIterator<TDT, iterator_type, iterator_density, false>(this, end_ptr);
     }
 
-    template<typename TDT>
-    ColumnDataIterator<TDT, true> cend() {
+    template<typename TDT, IteratorType iterator_type=IteratorType::REGULAR, IteratorDensity iterator_density=IteratorDensity::DENSE>
+    ColumnDataIterator<TDT, iterator_type, iterator_density, true> cend() {
         using RawType = typename TDT::DataTypeTag::raw_type;
-        const RawType* end_ptr{nullptr};
+        RawType* end_ptr{nullptr};
         if(!data_->blocks().empty()) {
             auto block = data_->blocks().at(num_blocks() - 1);
             auto typed_block_data = next_typed_block<TDT>(block);
-            end_ptr = typed_block_data.data() + typed_block_data.row_count();
+            end_ptr = const_cast<RawType*>(typed_block_data.data() + typed_block_data.row_count());
         }
-        return ColumnDataIterator<TDT, true>(this, end_ptr);
+        return ColumnDataIterator<TDT, iterator_type, iterator_density, true>(this, end_ptr);
     }
 
     TypeDescriptor type() const {
diff --git a/cpp/arcticdb/column_store/column_data_random_accessor.hpp b/cpp/arcticdb/column_store/column_data_random_accessor.hpp
new file mode 100644
index 0000000000..ca211c34af
--- /dev/null
+++ b/cpp/arcticdb/column_store/column_data_random_accessor.hpp
@@ -0,0 +1,192 @@
+/* Copyright 2024 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.
+ */
+
+#pragma once
+
+#include <folly/Poly.h>
+
+#include <arcticdb/column_store/column_data.hpp>
+
+namespace arcticdb {
+
+// Not handling regular sized until case, only skips one if statement in block_and_offset, cannot avoid branching completely
+enum class BlockStructure {
+    SINGLE,
+    REGULAR,
+    IRREGULAR
+};
+
+template<typename TDT>
+struct IChunkedBufferRandomAccessor {
+    template<class Base>
+    struct Interface : Base {
+        typename TDT::DataTypeTag::raw_type at(size_t idx) const { return folly::poly_call<0>(*this, idx); };
+    };
+
+    template<class T>
+    using Members = folly::PolyMembers<&T::at>;
+};
+
+template<typename TDT>
+using ChunkedBufferRandomAccessor = folly::Poly<IChunkedBufferRandomAccessor<TDT>>;
+
+template<typename TDT>
+class ChunkedBufferSingleBlockAccessor {
+    using RawType = typename TDT::DataTypeTag::raw_type;
+public:
+    ChunkedBufferSingleBlockAccessor(ColumnData* parent){
+        auto typed_block = parent->next<TDT>();
+        // Cache the base pointer of the block, as typed_block->data() has an if statement for internal vs external
+        base_ptr_ = reinterpret_cast<const RawType *>(typed_block->data());
+    }
+
+    RawType at(size_t idx) const {
+        return *(base_ptr_ + idx);
+    }
+private:
+    const RawType* base_ptr_;
+};
+
+template<typename TDT>
+class ChunkedBufferRegularBlocksAccessor {
+    using RawType = typename TDT::DataTypeTag::raw_type;
+public:
+    ChunkedBufferRegularBlocksAccessor(ColumnData* parent){
+        // Cache the base pointers of each block, as typed_block->data() has an if statement for internal vs external
+        base_ptrs_.reserve(parent->num_blocks());
+        while (auto typed_block = parent->next<TDT>()) {
+            base_ptrs_.emplace_back(reinterpret_cast<const RawType *>(typed_block->data()));
+        }
+    }
+
+    RawType at(size_t idx) const {
+        // quot is the block index, rem is the offset within the block
+        auto div = std::div(static_cast<long long>(idx), values_per_block_);
+        debug::check<ErrorCode::E_ASSERTION_FAILURE>(div.quot < static_cast<long long>(base_ptrs_.size()),
+                                                     "ColumnData::at called with out of bounds index");
+        return *(base_ptrs_[div.quot] + div.rem);
+    }
+private:
+    static constexpr auto values_per_block_ = BufferSize / sizeof(RawType);
+    std::vector<const RawType*> base_ptrs_;
+};
+
+template<typename TDT>
+class ChunkedBufferIrregularBlocksAccessor {
+    using RawType = typename TDT::DataTypeTag::raw_type;
+public:
+    ChunkedBufferIrregularBlocksAccessor(ColumnData* parent):
+            parent_(parent){
+    }
+
+    RawType at(size_t idx) const {
+        auto pos_bytes = idx * sizeof(RawType);
+        auto block_and_offset = parent_->buffer().block_and_offset(pos_bytes);
+        auto ptr = block_and_offset.block_->data();
+        ptr += block_and_offset.offset_;
+        return *reinterpret_cast<const RawType *>(ptr);
+    }
+private:
+    ColumnData* parent_;
+};
+
+template<typename TDT>
+struct IColumnDataRandomAccessor {
+    template<class Base>
+    struct Interface : Base {
+        typename TDT::DataTypeTag::raw_type at(size_t idx) const { return folly::poly_call<0>(*this, idx); };
+    };
+
+    template<class T>
+    using Members = folly::PolyMembers<&T::at>;
+};
+
+template<typename TDT>
+using ColumnDataRandomAccessor = folly::Poly<IColumnDataRandomAccessor<TDT>>;
+
+template<typename TDT, BlockStructure block_structure>
+class ColumnDataRandomAccessorSparse {
+    using RawType = typename TDT::DataTypeTag::raw_type;
+public:
+    ColumnDataRandomAccessorSparse(ColumnData* parent):
+    parent_(parent) {
+        parent_->bit_vector()->build_rs_index(&(bit_index_));
+        if constexpr (block_structure == BlockStructure::SINGLE) {
+            chunked_buffer_random_accessor_ = ChunkedBufferSingleBlockAccessor<TDT>(parent);
+        } else if constexpr (block_structure == BlockStructure::REGULAR) {
+            chunked_buffer_random_accessor_ = ChunkedBufferRegularBlocksAccessor<TDT>(parent);
+        } else {
+            // Irregular
+            chunked_buffer_random_accessor_ = ChunkedBufferIrregularBlocksAccessor<TDT>(parent);
+        }
+    }
+
+    RawType at(size_t idx) const {
+        debug::check<ErrorCode::E_ASSERTION_FAILURE>(parent_->bit_vector(),
+                                                     "ColumnData::at called with sparse true, but bit_vector_ == nullptr");
+        debug::check<ErrorCode::E_ASSERTION_FAILURE>(parent_->bit_vector()->size() > idx,
+                                                     "ColumnData::at called with sparse true, but index is out of range");
+        debug::check<ErrorCode::E_ASSERTION_FAILURE>(parent_->bit_vector()->get_bit(idx),
+                                                     "ColumnData::at called with sparse true, but selected bit is false");
+        // This is the same as using rank_corrected, but we always require the idx bit to be true, so do the -1 ourselves for efficiency
+        auto physical_offset = parent_->bit_vector()->rank(idx, bit_index_) - 1;
+        return chunked_buffer_random_accessor_.at(physical_offset);
+    }
+
+private:
+    ChunkedBufferRandomAccessor<TDT> chunked_buffer_random_accessor_;
+    ColumnData* parent_;
+    util::BitIndex bit_index_;
+};
+
+template<typename TDT, BlockStructure block_structure>
+class ColumnDataRandomAccessorDense {
+    using RawType = typename TDT::DataTypeTag::raw_type;
+public:
+    ColumnDataRandomAccessorDense(ColumnData* parent) {
+        if constexpr (block_structure == BlockStructure::SINGLE) {
+            chunked_buffer_random_accessor_ = ChunkedBufferSingleBlockAccessor<TDT>(parent);
+        } else if constexpr (block_structure == BlockStructure::REGULAR) {
+            chunked_buffer_random_accessor_ = ChunkedBufferRegularBlocksAccessor<TDT>(parent);
+        } else {
+            // Irregular
+            chunked_buffer_random_accessor_ = ChunkedBufferIrregularBlocksAccessor<TDT>(parent);
+        }
+    }
+
+    RawType at(size_t idx) const {
+        return chunked_buffer_random_accessor_.at(idx);
+    }
+
+private:
+    ChunkedBufferRandomAccessor<TDT> chunked_buffer_random_accessor_;
+};
+
+template<typename TDT>
+ColumnDataRandomAccessor<TDT> random_accessor(ColumnData* parent) {
+    bool sparse = parent->bit_vector() != nullptr;
+    if (parent->buffer().num_blocks() == 1) {
+        if (sparse) {
+            return ColumnDataRandomAccessorSparse<TDT, BlockStructure::SINGLE>(parent);
+        } else {
+            return ColumnDataRandomAccessorDense<TDT, BlockStructure::SINGLE>(parent);
+        }
+    } else if (parent->buffer().is_regular_sized()) {
+        if (sparse) {
+            return ColumnDataRandomAccessorSparse<TDT, BlockStructure::REGULAR>(parent);
+        } else {
+            return ColumnDataRandomAccessorDense<TDT, BlockStructure::REGULAR>(parent);
+        }
+    } else {
+        if (sparse) {
+            return ColumnDataRandomAccessorSparse<TDT, BlockStructure::IRREGULAR>(parent);
+        } else {
+            return ColumnDataRandomAccessorDense<TDT, BlockStructure::IRREGULAR>(parent);
+        }
+    }
+}
+}
\ No newline at end of file
diff --git a/cpp/arcticdb/column_store/memory_segment_impl.cpp b/cpp/arcticdb/column_store/memory_segment_impl.cpp
index 50e403db9f..0bb0fbe763 100644
--- a/cpp/arcticdb/column_store/memory_segment_impl.cpp
+++ b/cpp/arcticdb/column_store/memory_segment_impl.cpp
@@ -172,7 +172,6 @@ std::shared_ptr<SegmentInMemoryImpl> SegmentInMemoryImpl::filter(const util::Bit
                                                    bool filter_down_stringpool,
                                                    bool validate) const {
     bool is_input_sparse = is_sparse();
-    util::check(is_input_sparse || row_count() == filter_bitset.size(), "Filter input sizes do not match: {} != {}", row_count(), filter_bitset.size());
     auto num_values = filter_bitset.count();
     if(num_values == 0)
         return std::shared_ptr<SegmentInMemoryImpl>{};
@@ -461,15 +460,15 @@ std::shared_ptr<SegmentInMemoryImpl> SegmentInMemoryImpl::truncate(
     }
 
     for (const auto&& [idx, column] : folly::enumerate(columns_)) {
-        const TypeDescriptor column_type = column->type();
         const Field& field = descriptor_->field(idx);
         std::shared_ptr<Column> truncated_column = Column::truncate(column, start_row, end_row);
-        column_type.visit_tag([&](auto tag) {
-            if constexpr (is_sequence_type(decltype(tag)::data_type())) {
-                Column::transform<decltype(tag), decltype(tag)>(
+        details::visit_type(column->type().data_type(), [&](auto col_tag) {
+            using type_info = ScalarTypeInfo<decltype(col_tag)>;
+            if constexpr (is_sequence_type(type_info::data_type)) {
+                Column::transform<typename type_info::TDT, typename type_info::TDT>(
                     *truncated_column,
                     *truncated_column,
-                    [this, &output](auto string_pool_offset) -> arcticdb::OffsetString::offset_t {
+                    [this, &output](auto string_pool_offset) -> typename type_info::RawType {
                         if (is_a_string(string_pool_offset)) {
                             const std::string_view string = get_string_from_pool(string_pool_offset, *string_pool_);
                             return output->string_pool().get(string).offset();
diff --git a/cpp/arcticdb/processing/aggregation.cpp b/cpp/arcticdb/processing/aggregation.cpp
index 45422844b0..b52dd4b971 100644
--- a/cpp/arcticdb/processing/aggregation.cpp
+++ b/cpp/arcticdb/processing/aggregation.cpp
@@ -13,8 +13,8 @@ namespace arcticdb
 {
 
 void MinMaxAggregatorData::aggregate(const ColumnWithStrings& input_column) {
-    details::visit_type(input_column.column_->type().data_type(), [&] (auto col_desc_tag) {
-        using type_info = ScalarTypeInfo<decltype(col_desc_tag)>;
+    details::visit_type(input_column.column_->type().data_type(), [&] (auto col_tag) {
+        using type_info = ScalarTypeInfo<decltype(col_tag)>;
         if constexpr(!is_sequence_type(type_info::data_type)) {
             Column::for_each<typename type_info::TDT>(*input_column.column_, [this](auto value) {
                 const auto& curr = static_cast<typename type_info::RawType>(value);
@@ -40,8 +40,8 @@ SegmentInMemory MinMaxAggregatorData::finalize(const std::vector<ColumnName>& ou
             output_column_names.size());
     SegmentInMemory seg;
     if (min_.has_value()) {
-        details::visit_type(min_->data_type_, [&output_column_names, &seg, that = this](auto type_desc_tag) {
-            using RawType = typename ScalarTypeInfo<decltype(type_desc_tag)>::RawType;
+        details::visit_type(min_->data_type_, [&output_column_names, &seg, that = this](auto col_tag) {
+            using RawType = typename ScalarTypeInfo<decltype(col_tag)>::RawType;
             auto min_col = std::make_shared<Column>(make_scalar_type(that->min_->data_type_), true);
             min_col->template push_back<RawType>(that->min_->get<RawType>());
 
@@ -167,38 +167,24 @@ void SumAggregatorData::aggregate(const std::optional<ColumnWithStrings>& input_
     if (!data_type_.has_value() || *data_type_ == DataType::EMPTYVAL) {
         data_type_ = DataType::FLOAT64;
     }
-    entity::details::visit_type(*data_type_, [&input_column, unique_values, &groups, this] (auto global_type_desc_tag) {
-        using GlobalInputType = decltype(global_type_desc_tag);
+    details::visit_type(*data_type_, [&input_column, unique_values, &groups, this] (auto global_tag) {
+        using GlobalInputType = decltype(global_tag);
         if constexpr(!is_sequence_type(GlobalInputType::DataTypeTag::data_type)) {
             using GlobalTypeDescriptorTag =  typename OutputType<GlobalInputType>::type;
             using GlobalRawType = typename GlobalTypeDescriptorTag::DataTypeTag::raw_type;
             aggregated_.resize(sizeof(GlobalRawType)* unique_values);
             auto out_ptr = reinterpret_cast<GlobalRawType*>(aggregated_.data());
             if (input_column.has_value()) {
-                entity::details::visit_type(input_column->column_->type().data_type(), [&input_column, &groups, &out_ptr] (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 col_data = input_column->column_->data();
-                        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) {
-                                    if constexpr (std::is_same_v<GlobalRawType, bool>) {
-                                        out_ptr[groups[deref(iter)]] |= GlobalRawType(*ptr);
-                                    } else {
-                                        out_ptr[groups[deref(iter)]] += GlobalRawType(*ptr);
-                                    }
-                                }
+                details::visit_type(input_column->column_->type().data_type(), [&input_column, &groups, &out_ptr] (auto col_tag) {
+                    using col_type_info = ScalarTypeInfo<decltype(col_tag)>;
+                    if constexpr(!is_sequence_type(col_type_info::data_type)) {
+                        Column::for_each_enumerated<typename col_type_info::TDT>(*input_column->column_, [&out_ptr, &groups](auto enumerating_it) {
+                            if constexpr (std::is_same_v<GlobalRawType, bool>) {
+                                out_ptr[groups[enumerating_it.idx()]] |= GlobalRawType(enumerating_it.value());
+                            } else {
+                                out_ptr[groups[enumerating_it.idx()]] += GlobalRawType(enumerating_it.value());
                             }
-                        };
-
-                        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");
                     }
@@ -211,8 +197,8 @@ void SumAggregatorData::aggregate(const std::optional<ColumnWithStrings>& input_
 SegmentInMemory SumAggregatorData::finalize(const ColumnName& output_column_name, bool, size_t unique_values) {
     SegmentInMemory res;
     if(!aggregated_.empty()) {
-        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;
+        details::visit_type(*data_type_, [that=this, &res, &output_column_name, unique_values] (auto col_tag) {
+            using RawType = typename decltype(col_tag)::DataTypeTag::raw_type;
             that->aggregated_.resize(sizeof(RawType)* unique_values);
             auto col = std::make_shared<Column>(make_scalar_type(that->data_type_.value()), unique_values, true, false);
             memcpy(col->ptr(), that->aggregated_.data(), that->aggregated_.size());
@@ -261,55 +247,42 @@ namespace
         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_, &input_column, unique_values, &groups] (auto global_type_desc_tag) {
-                using GlobalInputType = decltype(global_type_desc_tag);
+            details::visit_type(*data_type_, [&aggregated_, &input_column, unique_values, &groups] (auto global_tag) {
+                using GlobalInputType = decltype(global_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);
-                                                }
-                                            }
+                    details::visit_type(input_column->column_->type().data_type(), [&input_column, &groups, &out_ptr] (auto col_tag) {
+                        using col_type_info = ScalarTypeInfo<decltype(col_tag)>;
+                        if constexpr(!is_sequence_type(col_type_info::data_type)) {
+                            Column::for_each_enumerated<typename col_type_info::TDT>(*input_column->column_, [&groups, &out_ptr](auto enumerating_it) {
+                                auto& val = out_ptr[groups[enumerating_it.idx()]];
+                                if constexpr(std::is_floating_point_v<typename col_type_info::RawType>) {
+                                    const auto& curr = GlobalRawType(enumerating_it.value());
+                                    if (!val.written_ || std::isnan(static_cast<typename col_type_info::RawType>(val.value_))) {
+                                        val.value_ = curr;
+                                        val.written_ = true;
+                                    } else if (!std::isnan(static_cast<typename col_type_info::RawType>(curr))) {
+                                        if constexpr(T == Extremum::MAX) {
+                                            val.value_ = std::max(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;
+                                            val.value_ = std::min(val.value_, curr);
                                         }
                                     }
+                                } else {
+                                    if constexpr(T == Extremum::MAX) {
+                                        val.value_ = std::max(val.value_, GlobalRawType(enumerating_it.value()));
+                                    } else {
+                                        val.value_ = std::min(val.value_, GlobalRawType(enumerating_it.value()));
+                                    }
+                                    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");
                         }
@@ -330,8 +303,8 @@ namespace
         SegmentInMemory res;
         if(!aggregated_.empty()) {
             if(dynamic_schema) {
-                entity::details::visit_type(*data_type_, [&aggregated_, &res, &output_column_name, unique_values] (auto type_desc_tag) {
-                    using RawType = typename decltype(type_desc_tag)::DataTypeTag::raw_type;
+                details::visit_type(*data_type_, [&aggregated_, &res, &output_column_name, unique_values] (auto col_tag) {
+                    using RawType = typename decltype(col_tag)::DataTypeTag::raw_type;
                     using MaybeValueType = MaybeValue<RawType, T>;
                     auto prev_size = aggregated_.size() / sizeof(MaybeValueType);
                     auto new_size = sizeof(MaybeValueType) * unique_values;
@@ -347,8 +320,8 @@ namespace
                     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;
+                details::visit_type(*data_type_, [&aggregated_, &data_type_, &res, output_column_name, unique_values] (auto col_tag) {
+                    using RawType = typename decltype(col_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());
@@ -408,28 +381,17 @@ SegmentInMemory MinAggregatorData::finalize(const ColumnName& output_column_name
 
 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;
-
-            fractions_.resize(unique_values);
-
-            auto col_data = input_column->column_->data();
-            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));
+        fractions_.resize(unique_values);
+        details::visit_type(input_column->column_->type().data_type(), [&input_column, &groups, this] (auto col_tag) {
+            using col_type_info = ScalarTypeInfo<decltype(col_tag)>;
+            if constexpr(!is_sequence_type(col_type_info::data_type)) {
+                Column::for_each_enumerated<typename col_type_info::TDT>(*input_column->column_, [&groups, this](auto enumerating_it) {
+                    auto& fraction = fractions_[groups[enumerating_it.idx()]];
+                    fraction.numerator_ += double(enumerating_it.value());
+                    ++fraction.denominator_;
+                });
+            } else {
+                util::raise_rte("String aggregations not currently supported");
             }
         });
     }
@@ -462,39 +424,20 @@ double MeanAggregatorData::Fraction::to_double() const
 
 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) {
-            using TypeDescriptorTag =  decltype(type_desc_tag);
-            using RawType = typename TypeDescriptorTag::DataTypeTag::raw_type;
-
-            aggregated_.resize(unique_values);
-            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 {
-                    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;
-                        }
+        aggregated_.resize(unique_values);
+        details::visit_type(input_column->column_->type().data_type(), [&input_column, &groups, this] (auto col_tag) {
+            using col_type_info = ScalarTypeInfo<decltype(col_tag)>;
+            Column::for_each_enumerated<typename col_type_info::TDT>(*input_column->column_, [&groups, this](auto enumerating_it) {
+                if constexpr (is_floating_point_type(col_type_info::data_type)) {
+                    if (!std::isnan(static_cast<double>(enumerating_it.value()))) {
+                        auto& val = aggregated_[groups[enumerating_it.idx()]];
+                        ++val;
                     }
+                } else {
+                    auto& val = aggregated_[groups[enumerating_it.idx()]];
+                    ++val;
                 }
-            };
-            if (input_column->column_->is_sparse()) {
-                lambda(col_data.bit_vector()->first());
-            }
-            else {
-                lambda(std::size_t(0));
-            }
+            });
         });
     }
 }
@@ -522,15 +465,15 @@ void FirstAggregatorData::add_data_type(DataType data_type) {
 
 void FirstAggregatorData::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, this] (auto global_type_desc_tag) {
-            using GlobalInputType = decltype(global_type_desc_tag);
+        details::visit_type(*data_type_, [&input_column, unique_values, &groups, this] (auto global_tag) {
+            using GlobalInputType = decltype(global_tag);
             using GlobalTypeDescriptorTag =  typename OutputType<GlobalInputType>::type;
             using GlobalRawType = typename GlobalTypeDescriptorTag::DataTypeTag::raw_type;
             aggregated_.resize(sizeof(GlobalRawType)* unique_values);
             auto col_data = input_column->column_->data();
             auto out_ptr = reinterpret_cast<GlobalRawType*>(aggregated_.data());
-            entity::details::visit_type(input_column->column_->type().data_type(), [this, &groups, &out_ptr, &col_data] (auto type_desc_tag) {
-                using ColumnTagType = std::decay_t<decltype(type_desc_tag)>;
+            details::visit_type(input_column->column_->type().data_type(), [this, &groups, &out_ptr, &col_data] (auto col_tag) {
+                using ColumnTagType = std::decay_t<decltype(col_tag)>;
                 using ColumnType =  typename ColumnTagType::raw_type;
                 auto groups_pos = 0;
                 while (auto block = col_data.next<TypeDescriptorTag<ColumnTagType, DimensionTag<entity::Dimension::Dim0>>>()) {
@@ -559,8 +502,8 @@ void FirstAggregatorData::aggregate(const std::optional<ColumnWithStrings>& inpu
 SegmentInMemory FirstAggregatorData::finalize(const ColumnName& output_column_name, bool, size_t unique_values) {
     SegmentInMemory res;
     if(!aggregated_.empty()) {
-        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;
+        details::visit_type(*data_type_, [that=this, &res, &output_column_name, unique_values] (auto col_tag) {
+            using RawType = typename decltype(col_tag)::DataTypeTag::raw_type;
             that->aggregated_.resize(sizeof(RawType)* unique_values);
             auto col = std::make_shared<Column>(make_scalar_type(that->data_type_.value()), unique_values, true, false);
             memcpy(col->ptr(), that->aggregated_.data(), that->aggregated_.size());
@@ -581,15 +524,15 @@ void LastAggregatorData::add_data_type(DataType data_type) {
 
 void LastAggregatorData::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, this] (auto global_type_desc_tag) {
-            using GlobalInputType = decltype(global_type_desc_tag);
+        details::visit_type(*data_type_, [&input_column, unique_values, &groups, this] (auto global_tag) {
+            using GlobalInputType = decltype(global_tag);
             using GlobalTypeDescriptorTag =  typename OutputType<GlobalInputType>::type;
             using GlobalRawType = typename GlobalTypeDescriptorTag::DataTypeTag::raw_type;
             aggregated_.resize(sizeof(GlobalRawType)* unique_values);
             auto col_data = input_column->column_->data();
             auto out_ptr = reinterpret_cast<GlobalRawType*>(aggregated_.data());
-            entity::details::visit_type(input_column->column_->type().data_type(), [&groups, &out_ptr, &col_data, this] (auto type_desc_tag) {
-                using ColumnTagType = std::decay_t<decltype(type_desc_tag)>;
+            details::visit_type(input_column->column_->type().data_type(), [&groups, &out_ptr, &col_data, this] (auto col_tag) {
+                using ColumnTagType = std::decay_t<decltype(col_tag)>;
                 using ColumnType =  typename ColumnTagType::raw_type;
                 auto groups_pos = 0;
                 while (auto block = col_data.next<TypeDescriptorTag<ColumnTagType, DimensionTag<entity::Dimension::Dim0>>>()) {
@@ -616,8 +559,8 @@ void LastAggregatorData::aggregate(const std::optional<ColumnWithStrings>& input
 SegmentInMemory LastAggregatorData::finalize(const ColumnName& output_column_name, bool, size_t unique_values) {
     SegmentInMemory res;
     if(!aggregated_.empty()) {
-        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;
+        details::visit_type(*data_type_, [that=this, &res, &output_column_name, unique_values] (auto col_tag) {
+            using RawType = typename decltype(col_tag)::DataTypeTag::raw_type;
             that->aggregated_.resize(sizeof(RawType)* unique_values);
             auto col = std::make_shared<Column>(make_scalar_type(that->data_type_.value()), unique_values, true, false);
             memcpy(col->ptr(), that->aggregated_.data(), that->aggregated_.size());
diff --git a/cpp/arcticdb/processing/grouper.hpp b/cpp/arcticdb/processing/grouper.hpp
index f1f2e04fa5..701d9ea309 100644
--- a/cpp/arcticdb/processing/grouper.hpp
+++ b/cpp/arcticdb/processing/grouper.hpp
@@ -56,10 +56,10 @@ class HashingGroupers {
                 if (std::isnan(key)) {
                     return std::nullopt;
                 } else {
-                    return hash<RawType>(&key, 1);
+                    return hash<RawType>(&key);
                 }
             } else {
-                return hash<RawType>(&key, 1);
+                return hash<RawType>(&key);
             }
         }
     private:
diff --git a/cpp/arcticdb/processing/operation_dispatch.cpp b/cpp/arcticdb/processing/operation_dispatch.cpp
index f1c06429ab..3dd0558732 100644
--- a/cpp/arcticdb/processing/operation_dispatch.cpp
+++ b/cpp/arcticdb/processing/operation_dispatch.cpp
@@ -46,24 +46,13 @@ VariantData transform_to_bitset(const VariantData& data) {
                 util::raise_rte("ValueSet inputs cannot be input to boolean operations");
             },
             [&] (const ColumnWithStrings& column_with_strings) -> VariantData {
-                util::BitSet output_bitset(static_cast<util::BitSetSizeType>(column_with_strings.column_->row_count()));
-                column_with_strings.column_->type().visit_tag([&column_with_strings, &output_bitset] (auto column_desc_tag) {
-                    using ColumnDescriptorType = std::decay_t<decltype(column_desc_tag)>;
-                    using ColumnTagType =  typename ColumnDescriptorType::DataTypeTag;
-                    using ColumnType =  typename ColumnTagType::raw_type;
-                    if constexpr (is_bool_type(ColumnTagType::data_type)) {
-                        auto column_data = column_with_strings.column_->data();
-                        util::BitSet::bulk_insert_iterator inserter(output_bitset);
-                        auto pos = 0u;
-                        while (auto block = column_data.next<ColumnDescriptorType>()) {
-                            auto ptr = reinterpret_cast<const ColumnType*>(block->data());
-                            const auto row_count = block->row_count();
-                            for (auto i = 0u; i < row_count; ++i, ++pos) {
-                                if(*ptr++)
-                                    inserter = pos;
-                            }
-                        }
-                        inserter.flush();
+                util::BitSet output_bitset;
+                details::visit_type(column_with_strings.column_->type().data_type(), [&column_with_strings, &output_bitset](auto col_tag) {
+                    using type_info = ScalarTypeInfo<decltype(col_tag)>;
+                    if constexpr (is_bool_type(type_info::data_type)) {
+                        Column::transform<typename type_info::TDT>(*column_with_strings.column_, output_bitset, false, [](auto input_value) -> bool {
+                            return input_value;
+                        });
                     } else {
                         util::raise_rte("Cannot convert column of type {} to a bitset", column_with_strings.column_->type());
                     }
diff --git a/cpp/arcticdb/processing/operation_dispatch_binary.hpp b/cpp/arcticdb/processing/operation_dispatch_binary.hpp
index 6ba2c985fd..fa047931a5 100644
--- a/cpp/arcticdb/processing/operation_dispatch_binary.hpp
+++ b/cpp/arcticdb/processing/operation_dispatch_binary.hpp
@@ -51,72 +51,67 @@ VariantData binary_membership(const ColumnWithStrings& column_with_strings, Valu
             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");
         }
     }
-    util::BitSet output_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
+    // If the value set is empty, we can short-circuit
     if (value_set.empty()) {
-        if constexpr(std::is_same_v<Func, IsNotInOperator&&>)
-            output_bitset.set();
-    } else {
-        details::visit_type(column_with_strings.column_->type().data_type(),[&] (auto col_desc_tag) {
-            using col_type_info = ScalarTypeInfo<decltype(col_desc_tag)>;
-            details::visit_type(value_set.base_type().data_type(), [&] (auto val_set_desc_tag) {
-                using val_set_type_info = ScalarTypeInfo<decltype(val_set_desc_tag)>;
-                if constexpr(is_sequence_type(col_type_info::data_type) && is_sequence_type(val_set_type_info::data_type)) {
-                    std::shared_ptr<std::unordered_set<std::string>> typed_value_set;
-                    if constexpr(is_fixed_string_type(col_type_info::data_type)) {
-                        auto width = column_with_strings.get_fixed_width_string_size();
-                        if (width.has_value()) {
-                            typed_value_set = value_set.get_fixed_width_string_set(*width);
-                        }
-                    } else {
-                        typed_value_set = value_set.get_set<std::string>();
+        if constexpr(std::is_same_v<Func, IsNotInOperator&&>) {
+            return FullResult{};
+        } else {
+            return EmptyResult{};
+        }
+    }
+
+    util::BitSet output_bitset;
+    constexpr auto sparse_missing_value_output = std::is_same_v<Func, IsNotInOperator&&>;
+    details::visit_type(column_with_strings.column_->type().data_type(),[&] (auto col_tag) {
+        using col_type_info = ScalarTypeInfo<decltype(col_tag)>;
+        details::visit_type(value_set.base_type().data_type(), [&] (auto val_set_tag) {
+            using val_set_type_info = ScalarTypeInfo<decltype(val_set_tag)>;
+            if constexpr(is_sequence_type(col_type_info::data_type) && is_sequence_type(val_set_type_info::data_type)) {
+                std::shared_ptr<std::unordered_set<std::string>> typed_value_set;
+                if constexpr(is_fixed_string_type(col_type_info::data_type)) {
+                    auto width = column_with_strings.get_fixed_width_string_size();
+                    if (width.has_value()) {
+                        typed_value_set = value_set.get_fixed_width_string_set(*width);
                     }
-                    auto offset_set = column_with_strings.string_pool_->get_offsets_for_column(typed_value_set, *column_with_strings.column_);
-                    Column::transform<typename col_type_info::TDT>(*column_with_strings.column_, output_bitset, [&func, &offset_set](auto input_value) -> bool {
-                        auto offset = static_cast<entity::position_t>(input_value);
-                        return func(offset, offset_set);
-                    });
-                } else if constexpr (is_bool_type(col_type_info::data_type) && is_bool_type(val_set_type_info::data_type)) {
-                    util::raise_rte("Binary membership not implemented for bools");
-                } else if constexpr (is_numeric_type(col_type_info::data_type) && is_numeric_type(val_set_type_info::data_type)) {
-                    using WideType = typename type_arithmetic_promoted_type<typename col_type_info::RawType, typename val_set_type_info::RawType, std::remove_reference_t<Func>>::type;
-                    auto typed_value_set = value_set.get_set<WideType>();
-                    Column::transform<typename col_type_info::TDT>(*column_with_strings.column_, output_bitset, [&func, &typed_value_set](auto input_value) -> bool {
-                        if constexpr (MembershipOperator::needs_uint64_special_handling<typename col_type_info::RawType, typename val_set_type_info::RawType>) {
-                            // Avoid narrowing conversion on *input_it:
-                            return func(input_value, *typed_value_set, UInt64SpecialHandlingTag{});
-                        } else {
-                            return func(static_cast<WideType>(input_value), *typed_value_set);
-                        }
-                    });
                 } else {
-                    util::raise_rte("Cannot check membership of {} in set of {} (possible categorical?)",
-                                    column_with_strings.column_->type(), value_set.base_type());
+                    typed_value_set = value_set.get_set<std::string>();
                 }
-            });
+                auto offset_set = column_with_strings.string_pool_->get_offsets_for_column(typed_value_set, *column_with_strings.column_);
+                Column::transform<typename col_type_info::TDT>(
+                        *column_with_strings.column_,
+                        output_bitset,
+                        sparse_missing_value_output,
+                        [&func, &offset_set](auto input_value) -> bool {
+                    auto offset = static_cast<entity::position_t>(input_value);
+                    return func(offset, offset_set);
+                });
+            } else if constexpr (is_bool_type(col_type_info::data_type) && is_bool_type(val_set_type_info::data_type)) {
+                util::raise_rte("Binary membership not implemented for bools");
+            } else if constexpr (is_numeric_type(col_type_info::data_type) && is_numeric_type(val_set_type_info::data_type)) {
+                using WideType = typename type_arithmetic_promoted_type<typename col_type_info::RawType,typename val_set_type_info::RawType, std::remove_reference_t<Func>>::type;
+                auto typed_value_set = value_set.get_set<WideType>();
+                Column::transform<typename col_type_info::TDT>(
+                        *column_with_strings.column_,
+                        output_bitset,
+                        sparse_missing_value_output,
+                        [&func, &typed_value_set](auto input_value) -> bool {
+                    if constexpr (MembershipOperator::needs_uint64_special_handling<typename col_type_info::RawType, typename val_set_type_info::RawType>) {
+                        // Avoid narrowing conversion on *input_it:
+                        return func(input_value, *typed_value_set, UInt64SpecialHandlingTag{});
+                    } else {
+                        return func(static_cast<WideType>(input_value), *typed_value_set);
+                    }
+                });
+            } else {
+                util::raise_rte("Cannot check membership of {} in set of {} (possible categorical?)",
+                                column_with_strings.column_->type(), value_set.base_type());
+            }
         });
-    }
-
-    output_bitset.optimize();
-    auto pos = 0;
-
-    if(column_with_strings.column_->is_sparse()) {
-        const auto& sparse_map = column_with_strings.column_->opt_sparse_map().value();
-        util::BitSet replace(sparse_map.size());
-        auto it = sparse_map.first();
-        auto it_end = sparse_map.end();
-        while(it < it_end) {
-           replace.set(*it, output_bitset.test(pos++));
-            ++it;
-        }
-    }
+    });
 
-    log::version().debug("Filtered segment of size {} down to {} bits", output_bitset.size(), output_bitset.count());
+    log::version().debug("Filtered column of size {} down to {} bits", column_with_strings.column_->last_row() + 1, output_bitset.count());
 
     return {std::move(output_bitset)};
 }
@@ -141,26 +136,37 @@ VariantData binary_comparator(const ColumnWithStrings& left, const ColumnWithStr
     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());
-    util::BitSet output_bitset(static_cast<util::BitSetSizeType>(left.column_->row_count()));
+    util::BitSet output_bitset;
+    constexpr auto sparse_missing_value_output = std::is_same_v<Func, NotEqualsOperator&&>;
 
-    details::visit_type(left.column_->type().data_type(), [&](auto left_desc_tag) {
-        using left_type_info = ScalarTypeInfo<decltype(left_desc_tag)>;
-        details::visit_type(right.column_->type().data_type(), [&](auto right_desc_tag) {
-            using right_type_info = ScalarTypeInfo<decltype(right_desc_tag)>;
+    details::visit_type(left.column_->type().data_type(), [&](auto left_tag) {
+        using left_type_info = ScalarTypeInfo<decltype(left_tag)>;
+        details::visit_type(right.column_->type().data_type(), [&](auto right_tag) {
+            using right_type_info = ScalarTypeInfo<decltype(right_tag)>;
             if constexpr(is_sequence_type(left_type_info::data_type) && is_sequence_type(right_type_info::data_type)) {
                 bool strip_fixed_width_trailing_nulls{false};
                 // If one or both columns are fixed width strings, we need to strip trailing null characters to get intuitive results
                 if constexpr (is_fixed_string_type(left_type_info::data_type) || is_fixed_string_type(right_type_info::data_type)) {
                     strip_fixed_width_trailing_nulls = true;
                 }
-                Column::transform<typename left_type_info::TDT, typename right_type_info::TDT>(*left.column_, *right.column_, output_bitset, [&func, &left, &right, strip_fixed_width_trailing_nulls] (auto left_value, auto right_value) -> bool {
-                    return func(left.string_at_offset(left_value, strip_fixed_width_trailing_nulls), right.string_at_offset(right_value, strip_fixed_width_trailing_nulls));
+                Column::transform<typename left_type_info::TDT, typename right_type_info::TDT>(
+                        *left.column_,
+                        *right.column_,
+                        output_bitset,
+                        sparse_missing_value_output,
+                        [&func, &left, &right, strip_fixed_width_trailing_nulls] (auto left_value, auto right_value) -> bool {
+                    return func(left.string_at_offset(left_value, strip_fixed_width_trailing_nulls),
+                                right.string_at_offset(right_value, strip_fixed_width_trailing_nulls));
                 });
             } else if constexpr ((is_numeric_type(left_type_info::data_type) && is_numeric_type(right_type_info::data_type)) ||
                                  (is_bool_type(left_type_info::data_type) && is_bool_type(right_type_info::data_type))) {
                 using comp = typename arcticdb::Comparable<typename left_type_info::RawType, typename right_type_info::RawType>;
-                Column::transform<typename left_type_info::TDT, typename right_type_info::TDT>(*left.column_, *right.column_, output_bitset, [&func] (auto left_value, auto right_value) -> bool {
+                Column::transform<typename left_type_info::TDT, typename right_type_info::TDT>(
+                        *left.column_,
+                        *right.column_,
+                        output_bitset,
+                        sparse_missing_value_output,
+                        [&func] (auto left_value, auto right_value) -> bool {
                     return func(static_cast<typename comp::left_type>(left_value), static_cast<typename comp::right_type>(right_value));
                 });
             } else {
@@ -168,7 +174,7 @@ VariantData binary_comparator(const ColumnWithStrings& left, const ColumnWithStr
             }
         });
     });
-    ARCTICDB_DEBUG(log::version(), "Filtered segment of size {} down to {} bits", output_bitset.size(), output_bitset.count());
+    ARCTICDB_DEBUG(log::version(), "Filtered column of size {} down to {} bits", std::max(left.column_->last_row(), right.column_->last_row()) + 1, output_bitset.count());
 
     return VariantData{std::move(output_bitset)};
 }
@@ -178,12 +184,13 @@ VariantData binary_comparator(const ColumnWithStrings& column_with_strings, cons
     if (is_empty_type(column_with_strings.column_->type().data_type())) {
         return EmptyResult{};
     }
-    util::BitSet output_bitset(static_cast<util::BitSetSizeType>(column_with_strings.column_->row_count()));
+    util::BitSet output_bitset;
+    constexpr auto sparse_missing_value_output = std::is_same_v<Func, NotEqualsOperator&&>;
 
-    details::visit_type(column_with_strings.column_->type().data_type(), [&](auto col_desc_tag) {
-        using col_type_info = ScalarTypeInfo<decltype(col_desc_tag)>;
-        details::visit_type(val.type().data_type(), [&](auto val_desc_tag) {
-            using val_type_info = ScalarTypeInfo<decltype(val_desc_tag)>;
+    details::visit_type(column_with_strings.column_->type().data_type(), [&](auto col_tag) {
+        using col_type_info = ScalarTypeInfo<decltype(col_tag)>;
+        details::visit_type(val.type().data_type(), [&](auto val_tag) {
+            using val_type_info = ScalarTypeInfo<decltype(val_tag)>;
             if constexpr(is_sequence_type(col_type_info::data_type) && is_sequence_type(val_type_info::data_type)) {
                 std::optional<std::string> utf32_string;
                 std::string value_string;
@@ -199,7 +206,11 @@ VariantData binary_comparator(const ColumnWithStrings& column_with_strings, cons
                     value_string = std::string(*val.str_data(), val.len());
                 }
                 auto value_offset = column_with_strings.string_pool_->get_offset_for_column(value_string, *column_with_strings.column_);
-                Column::transform<typename col_type_info::TDT>(*column_with_strings.column_, output_bitset, [&func, value_offset](auto input_value) -> bool {
+                Column::transform<typename col_type_info::TDT>(
+                        *column_with_strings.column_,
+                        output_bitset,
+                        sparse_missing_value_output,
+                        [&func, value_offset](auto input_value) -> bool {
                     auto offset = static_cast<entity::position_t>(input_value);
                     if constexpr (arguments_reversed) {
                         return func(offset, value_offset);
@@ -213,7 +224,11 @@ VariantData binary_comparator(const ColumnWithStrings& column_with_strings, cons
                                                 typename arcticdb::Comparable<typename col_type_info::RawType, typename val_type_info::RawType>,
                                                 typename arcticdb::Comparable<typename val_type_info::RawType, typename col_type_info::RawType>>;
                 auto value = static_cast<typename comp::left_type>(*reinterpret_cast<const typename val_type_info::RawType *>(val.data_));
-                Column::transform<typename col_type_info::TDT>(*column_with_strings.column_, output_bitset, [&func, value](auto input_value) -> bool {
+                Column::transform<typename col_type_info::TDT>(
+                        *column_with_strings.column_,
+                        output_bitset,
+                        sparse_missing_value_output,
+                        [&func, value](auto input_value) -> bool {
                     if constexpr (arguments_reversed) {
                         return func(value, static_cast<typename comp::right_type>(input_value));
                     } else {
@@ -226,7 +241,7 @@ VariantData binary_comparator(const ColumnWithStrings& column_with_strings, cons
             }
         });
     });
-    ARCTICDB_DEBUG(log::version(), "Filtered segment of size {} down to {} bits", output_bitset.size(), output_bitset.count());
+    ARCTICDB_DEBUG(log::version(), "Filtered column of size {} down to {} bits", column_with_strings.column_->last_row() + 1, output_bitset.count());
 
     return VariantData{std::move(output_bitset)};
 }
@@ -262,14 +277,14 @@ template <typename Func>
 VariantData binary_operator(const Value& left, const Value& right, Func&& func) {
     auto output_value = std::make_unique<Value>();
 
-    details::visit_type(left.type().data_type(), [&](auto left_desc_tag) {
-        using left_type_info = ScalarTypeInfo<decltype(left_desc_tag)>;
+    details::visit_type(left.type().data_type(), [&](auto left_tag) {
+        using left_type_info = ScalarTypeInfo<decltype(left_tag)>;
         if constexpr(!is_numeric_type(left_type_info::data_type)) {
             util::raise_rte("Non-numeric type provided to binary operation: {}", left.type());
         }
         auto left_value = *reinterpret_cast<const typename left_type_info::RawType*>(left.data_);
-        details::visit_type(right.type().data_type(), [&](auto right_desc_tag) {
-            using right_type_info = ScalarTypeInfo<decltype(right_desc_tag)>;
+        details::visit_type(right.type().data_type(), [&](auto right_tag) {
+            using right_type_info = ScalarTypeInfo<decltype(right_tag)>;
             if constexpr(!is_numeric_type(right_type_info::data_type)) {
                 util::raise_rte("Non-numeric type provided to binary operation: {}", right.type());
             }
@@ -287,25 +302,26 @@ 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_column;
 
-    details::visit_type(left.type().data_type(), [&](auto left_desc_tag) {
-        using left_type_info = ScalarTypeInfo<decltype(left_desc_tag)>;
+    details::visit_type(left.type().data_type(), [&](auto left_tag) {
+        using left_type_info = ScalarTypeInfo<decltype(left_tag)>;
         if constexpr(!is_numeric_type(left_type_info::data_type)) {
             util::raise_rte("Non-numeric type provided to binary operation: {}", left.type());
         }
-        details::visit_type(right.type().data_type(), [&](auto right_desc_tag) {
-            using right_type_info = ScalarTypeInfo<decltype(right_desc_tag)>;
+        details::visit_type(right.type().data_type(), [&](auto right_tag) {
+            using right_type_info = ScalarTypeInfo<decltype(right_tag)>;
             if constexpr(!is_numeric_type(right_type_info::data_type)) {
                 util::raise_rte("Non-numeric type provided to binary operation: {}", right.type());
             }
             using TargetType = typename type_arithmetic_promoted_type<typename left_type_info::RawType, typename right_type_info::RawType, std::remove_reference_t<decltype(func)>>::type;
             constexpr auto output_data_type = data_type_from_raw_type<TargetType>();
-            output_column = std::make_unique<Column>(make_scalar_type(output_data_type), left.row_count(), true, false);
-            output_column->set_row_data(left.last_row());
+            output_column = std::make_unique<Column>(make_scalar_type(output_data_type), true);
             Column::transform<typename left_type_info::TDT, typename right_type_info::TDT, ScalarTagType<DataTypeTag<output_data_type>>>(
-                    left, right, *output_column, [&func] (auto left_value, auto right_value) -> TargetType {
+                    left,
+                    right,
+                    *output_column,
+                    [&func] (auto left_value, auto right_value) -> TargetType {
                         return func.apply(left_value, right_value);
                     });
         });
@@ -320,13 +336,13 @@ VariantData binary_operator(const Column& col, const Value& val, Func&& func) {
             "Empty column provided to binary operator");
     std::unique_ptr<Column> output_column;
 
-    details::visit_type(col.type().data_type(), [&](auto col_desc_tag) {
-        using col_type_info = ScalarTypeInfo<decltype(col_desc_tag)>;
+    details::visit_type(col.type().data_type(), [&](auto col_tag) {
+        using col_type_info = ScalarTypeInfo<decltype(col_tag)>;
         if constexpr(!is_numeric_type(col_type_info::data_type)) {
             util::raise_rte("Non-numeric type provided to binary operation: {}", col.type());
         }
-        details::visit_type(val.type().data_type(), [&](auto val_desc_tag) {
-            using val_type_info = ScalarTypeInfo<decltype(val_desc_tag)>;
+        details::visit_type(val.type().data_type(), [&](auto val_tag) {
+            using val_type_info = ScalarTypeInfo<decltype(val_tag)>;
             if constexpr(!is_numeric_type(val_type_info::data_type)) {
                 util::raise_rte("Non-numeric type provided to binary operation: {}", val.type());
             }
@@ -335,16 +351,20 @@ VariantData binary_operator(const Column& col, const Value& val, Func&& func) {
             using ReversedTargetType = typename type_arithmetic_promoted_type<typename val_type_info::RawType, typename col_type_info::RawType, std::remove_reference_t<decltype(func)>>::type;
             if constexpr(arguments_reversed) {
                 constexpr auto output_data_type = data_type_from_raw_type<ReversedTargetType>();
-                output_column = std::make_unique<Column>(make_scalar_type(output_data_type), col.row_count(), true, false);
-                output_column->set_row_data(col.last_row());
-                Column::transform<typename col_type_info ::TDT, ScalarTagType<DataTypeTag<output_data_type>>>(col, *output_column, [&func, raw_value](auto input_value) -> ReversedTargetType {
+                output_column = std::make_unique<Column>(make_scalar_type(output_data_type), true);
+                Column::transform<typename col_type_info::TDT, ScalarTagType<DataTypeTag<output_data_type>>>(
+                        col,
+                        *output_column,
+                        [&func, raw_value](auto input_value) -> ReversedTargetType {
                     return func.apply(raw_value, input_value);
                 });
             } else {
                 constexpr auto output_data_type = data_type_from_raw_type<TargetType>();
-                output_column = std::make_unique<Column>(make_scalar_type(output_data_type), col.row_count(), true, false);
-                output_column->set_row_data(col.last_row());
-                Column::transform<typename col_type_info ::TDT, ScalarTagType<DataTypeTag<output_data_type>>>(col, *output_column, [&func, raw_value](auto input_value) -> TargetType {
+                output_column = std::make_unique<Column>(make_scalar_type(output_data_type), true);
+                Column::transform<typename col_type_info::TDT, ScalarTagType<DataTypeTag<output_data_type>>>(
+                        col,
+                        *output_column,
+                        [&func, raw_value](auto input_value) -> TargetType {
                     return func.apply(input_value, raw_value);
                 });
             }
diff --git a/cpp/arcticdb/processing/operation_dispatch_unary.hpp b/cpp/arcticdb/processing/operation_dispatch_unary.hpp
index a95aa23dae..c8018d7ba6 100644
--- a/cpp/arcticdb/processing/operation_dispatch_unary.hpp
+++ b/cpp/arcticdb/processing/operation_dispatch_unary.hpp
@@ -41,8 +41,8 @@ VariantData unary_operator(const Value& val, Func&& func) {
     auto output = std::make_unique<Value>();
     output->data_type_ = val.data_type_;
 
-    details::visit_type(val.type().data_type(), [&](auto val_desc_tag) {
-        using type_info = ScalarTypeInfo<decltype(val_desc_tag)>;
+    details::visit_type(val.type().data_type(), [&](auto val_tag) {
+        using type_info = ScalarTypeInfo<decltype(val_tag)>;
         if constexpr (!is_numeric_type(type_info::data_type)) {
             util::raise_rte("Cannot perform arithmetic on {}", val.type());
         }
@@ -62,18 +62,21 @@ VariantData unary_operator(const Column& col, Func&& func) {
             "Empty column provided to unary operator");
     std::unique_ptr<Column> output_column;
 
-    details::visit_type(col.type().data_type(), [&](auto col_desc_tag) {
-        using type_info = ScalarTypeInfo<decltype(col_desc_tag)>;
-        if constexpr (!is_numeric_type(type_info::data_type)) {
+    details::visit_type(col.type().data_type(), [&](auto col_tag) {
+        using type_info = ScalarTypeInfo<decltype(col_tag)>;
+        if constexpr (is_numeric_type(type_info::data_type)) {
+            using TargetType = typename unary_arithmetic_promoted_type<typename type_info::RawType, std::remove_reference_t<Func>>::type;
+            constexpr auto output_data_type = data_type_from_raw_type<TargetType>();
+            output_column = std::make_unique<Column>(make_scalar_type(output_data_type), true);
+            Column::transform<typename type_info::TDT, ScalarTagType<DataTypeTag<output_data_type>>>(col,
+                                                                                                     *output_column,
+                                                                                                     [&func](auto input_value) -> TargetType {
+                                                                                                         return func.apply(
+                                                                                                                 input_value);
+                                                                                                     });
+        } else {
             util::raise_rte("Cannot perform arithmetic on {}", col.type());
         }
-        using TargetType = typename unary_arithmetic_promoted_type<typename type_info::RawType, std::remove_reference_t<Func>>::type;
-        constexpr auto output_data_type = data_type_from_raw_type<TargetType>();
-        output_column = std::make_unique<Column>(make_scalar_type(output_data_type), col.row_count(), true, false);
-        output_column->set_row_data(col.last_row());
-        Column::transform<typename type_info::TDT, ScalarTagType<DataTypeTag<output_data_type>>>(col, *output_column, [&func](auto input_value) {
-            return func.apply(input_value);
-        });
     });
     return {ColumnWithStrings(std::move(output_column))};
 }
@@ -108,10 +111,11 @@ VariantData unary_comparator(const Column& col, Func&& func) {
         }
     }
 
-    util::BitSet output_bitset(static_cast<util::BitSetSizeType>(col.row_count()));
-    details::visit_type(col.type().data_type(), [&](auto col_desc_tag) {
-        using type_info = ScalarTypeInfo<decltype(col_desc_tag)>;
-        Column::transform<typename type_info::TDT>(col, output_bitset, [&func](auto input_value) -> bool {
+    util::BitSet output_bitset;
+    constexpr auto sparse_missing_value_output = std::is_same_v<Func, IsNullOperator&&>;
+    details::visit_type(col.type().data_type(), [&](auto col_tag) {
+        using type_info = ScalarTypeInfo<decltype(col_tag)>;
+        Column::transform<typename type_info::TDT>(col, output_bitset, sparse_missing_value_output, [&func](auto input_value) -> bool {
             if constexpr (is_floating_point_type(type_info::data_type)) {
                 return func.apply(input_value);
             } else if constexpr (is_sequence_type(type_info::data_type)) {
diff --git a/cpp/arcticdb/processing/processing_unit.hpp b/cpp/arcticdb/processing/processing_unit.hpp
index 2e7bb1beb1..b9a499bd51 100644
--- a/cpp/arcticdb/processing/processing_unit.hpp
+++ b/cpp/arcticdb/processing/processing_unit.hpp
@@ -132,7 +132,6 @@ namespace arcticdb {
             Bucketizer& bucketizer) {
         schema::check<ErrorCode::E_UNSUPPORTED_COLUMN_TYPE>(!col.column_->is_sparse(),
                                                             "GroupBy not supported with sparse columns");
-        auto input_data = col.column_->data();
         // Mapping from row to bucket
         // std::nullopt only for Nones and NaNs in string/float columns
         std::vector<std::optional<uint8_t>> row_to_bucket;
@@ -141,25 +140,17 @@ namespace arcticdb {
         // Use to skip empty buckets, and presize columns in the output ProcessingUnit
         std::vector<uint64_t> bucket_counts(bucketizer.num_buckets(), 0);
 
-        using TypeDescriptorTag = typename Grouper::GrouperDescriptor;
-        using RawType = typename TypeDescriptorTag::DataTypeTag::raw_type;
-
-        while (auto block = input_data.next<TypeDescriptorTag>()) {
-            const auto row_count = block->row_count();
-            auto ptr = reinterpret_cast<const RawType*>(block->data());
-            for(auto i = 0u; i < row_count; ++i, ++ptr){
-                if constexpr(std::is_same_v<typename Grouper::GrouperDescriptor, TypeDescriptorTag>) {
-                    auto opt_group = grouper.group(*ptr, col.string_pool_);
-                    if (opt_group.has_value()) {
-                        auto bucket = bucketizer.bucket(*opt_group);
-                        row_to_bucket.emplace_back(bucket);
-                        ++bucket_counts[bucket];
-                    } else {
-                        row_to_bucket.emplace_back(std::nullopt);
-                    }
-                }
+        using TDT = typename Grouper::GrouperDescriptor;
+        Column::for_each<TDT>(*col.column_, [&] (auto val) {
+            auto opt_group = grouper.group(val, col.string_pool_);
+            if (opt_group.has_value()) {
+                auto bucket = bucketizer.bucket(*opt_group);
+                row_to_bucket.emplace_back(bucket);
+                ++bucket_counts[bucket];
+            } else {
+                row_to_bucket.emplace_back(std::nullopt);
             }
-        }
+        });
         return {std::move(row_to_bucket), std::move(bucket_counts)};
     }
 
diff --git a/cpp/arcticdb/util/hash.hpp b/cpp/arcticdb/util/hash.hpp
index d80794ea48..67fdce2b6a 100644
--- a/cpp/arcticdb/util/hash.hpp
+++ b/cpp/arcticdb/util/hash.hpp
@@ -12,23 +12,9 @@
 
 #include <folly/hash/Hash.h>
 
-// XXH_INLINE_ALL causes hashes to be non-deterministic with the linux conda build, needs investigating
-// https://github.com/man-group/ArcticDB/issues/1268
-#if defined(ARCTICDB_USING_CONDA) && defined(__linux__)
 #define XXH_STATIC_LINKING_ONLY
-#else
-// xxhash does some raw pointer manipulation that are safe, but compilers view as violating array bounds
-#pragma GCC diagnostic push
-#pragma GCC diagnostic ignored "-Warray-bounds"
-#define XXH_INLINE_ALL
-#endif
 #include <xxhash.h>
-#if defined(ARCTICDB_USING_CONDA) && defined(__linux__)
 #undef XXH_STATIC_LINKING_ONLY
-#else
-#undef XXH_INLINE_ALL
-#pragma GCC diagnostic pop
-#endif
 
 namespace arcticdb {
 
diff --git a/cpp/arcticdb/util/preconditions.hpp b/cpp/arcticdb/util/preconditions.hpp
index ad3d28996b..6a8704317e 100644
--- a/cpp/arcticdb/util/preconditions.hpp
+++ b/cpp/arcticdb/util/preconditions.hpp
@@ -68,6 +68,19 @@ namespace internal {
     constexpr auto raise = check<code>.raise;
 }
 
+namespace debug {
+#ifdef DEBUG_BUILD
+    template<ErrorCode code>
+    constexpr auto check = internal::check<code>;
+#else
+    template<ErrorCode code, typename...Args>
+    inline void check(ARCTICDB_UNUSED bool cond, ARCTICDB_UNUSED fmt::format_string<Args...> format, ARCTICDB_UNUSED Args&&...args) {}
+
+    template<ErrorCode code, typename FormatString, typename...Args>
+    inline void check(ARCTICDB_UNUSED bool cond, ARCTICDB_UNUSED FormatString format, ARCTICDB_UNUSED Args&&...args) {}
+#endif
+}
+
 namespace normalization {
 template<ErrorCode code>
 constexpr auto check = util::detail::Check<code, ErrorCategory::NORMALIZATION>{};


Test patch:
diff --git a/cpp/arcticdb/column_store/test/rapidcheck_column_data_random_accessor.cpp b/cpp/arcticdb/column_store/test/rapidcheck_column_data_random_accessor.cpp
new file mode 100644
index 0000000000..8f5c378e6a
--- /dev/null
+++ b/cpp/arcticdb/column_store/test/rapidcheck_column_data_random_accessor.cpp
@@ -0,0 +1,61 @@
+/* Copyright 2024 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/column_store/column.hpp>
+#include <arcticdb/column_store/column_data_random_accessor.hpp>
+#include <arcticdb/util/test/rapidcheck.hpp>
+
+// Tricky to construct columns using a ChunkedBuffer with the testing size of 64 bytes, so only use rapidcheck for
+// single block tests, and test regualr and irregular accessors in test_column_data_random_accessor.cpp
+RC_GTEST_PROP(ColumnDataRandomAccessor, DenseSingleBlock, (const std::vector<int64_t> &input)) {
+    using namespace arcticdb;
+    RC_PRE(input.size() > 0u);
+    auto n = input.size();
+    using TDT = TypeDescriptorTag<DataTypeTag<DataType::INT64>, DimensionTag<Dimension ::Dim0>>;
+    Column column(static_cast<TypeDescriptor>(TDT{}), n, true, false);
+    RC_ASSERT(column.num_blocks() == 1);
+    memcpy(column.ptr(), input.data(), n * sizeof(int64_t));
+
+    auto column_data = column.data();
+    auto accessor = random_accessor<TDT>(&column_data);
+    for (size_t idx = 0; idx < n; ++idx) {
+        RC_ASSERT(accessor.at(idx) == input[idx]);
+    }
+}
+
+RC_GTEST_PROP(ColumnDataRandomAccessor, SparseSingleBlock, (const std::vector<int64_t> &input)) {
+    using namespace arcticdb;
+    RC_PRE(input.size() > 0u);
+    auto n = input.size();
+    auto mask = *rc::gen::container<std::vector<bool>>(n, rc::gen::arbitrary<bool>());
+    // The last value in the bitset will always be true in a sparse column
+    mask[n - 1] = true;
+    auto on_bits = std::count(mask.begin(), mask.end(), true);
+    util::BitSet sparse_map;
+    for (size_t idx = 0; idx < mask.size(); ++idx) {
+        if (mask[idx]) {
+            sparse_map.set_bit(idx);
+        }
+    }
+
+    using TDT = TypeDescriptorTag<DataTypeTag<DataType::INT64>, DimensionTag<Dimension ::Dim0>>;
+    Column column(static_cast<TypeDescriptor>(TDT{}), on_bits, true, true);
+    RC_ASSERT(column.num_blocks() == 1);
+    memcpy(column.ptr(), input.data(), on_bits * sizeof(int64_t));
+    column.set_sparse_map(std::move(sparse_map));
+
+    auto column_data = column.data();
+    auto accessor = random_accessor<TDT>(&column_data);
+    size_t input_idx{0};
+    for (size_t idx = 0; idx < n; ++idx) {
+        if (mask[idx]) {
+            RC_ASSERT(accessor.at(idx) == input[input_idx++]);
+        }
+    }
+}
\ No newline at end of file
diff --git a/cpp/arcticdb/column_store/test/test_column_data_random_accessor.cpp b/cpp/arcticdb/column_store/test/test_column_data_random_accessor.cpp
new file mode 100644
index 0000000000..cf47e70cbb
--- /dev/null
+++ b/cpp/arcticdb/column_store/test/test_column_data_random_accessor.cpp
@@ -0,0 +1,107 @@
+/* Copyright 2024 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/column_store/column.hpp>
+#include <arcticdb/column_store/column_data_random_accessor.hpp>
+
+using namespace arcticdb;
+
+class ColumnDataRandomAccessorTest : public testing::Test {
+protected:
+    using TDT = TypeDescriptorTag<DataTypeTag<DataType::INT64>, DimensionTag<Dimension ::Dim0>>;
+    void SetUp() override {
+        input_data.resize(n);
+        std::iota(input_data.begin(), input_data.end(), 42);
+    }
+    // 3968 bytes == 496 int64s per block, so 3 blocks here
+    size_t n{1000};
+    std::vector<int64_t> input_data;
+    TypeDescriptor type_descriptor{static_cast<TypeDescriptor>(TDT{})};
+};
+
+// Note that dense and sparse single block accessors are tested in rapidcheck_column_data_random_accessor.cpp
+TEST_F(ColumnDataRandomAccessorTest, DenseRegularBlocks) {
+    Column column(type_descriptor, false);
+    for (auto& val: input_data) {
+        column.push_back<int64_t>(val);
+    }
+
+    ASSERT_TRUE(column.buffer().is_regular_sized());
+    ASSERT_EQ(column.num_blocks(), 3);
+
+    auto column_data = column.data();
+    auto accessor = random_accessor<TDT>(&column_data);
+    for (size_t idx = 0; idx < n; ++idx) {
+        ASSERT_EQ(accessor.at(idx), input_data[idx]);
+    }
+}
+
+TEST_F(ColumnDataRandomAccessorTest, SparseRegularBlocks) {
+    Column column(type_descriptor, true);
+    // Set every third value
+    for (size_t idx = 0; idx < n; ++idx) {
+        column.set_scalar<int64_t>(3 * idx, input_data[idx]);
+    }
+
+    ASSERT_TRUE(column.is_sparse());
+    ASSERT_TRUE(column.buffer().is_regular_sized());
+    ASSERT_EQ(column.num_blocks(), 3);
+
+    auto column_data = column.data();
+    auto accessor = random_accessor<TDT>(&column_data);
+    for (size_t idx = 0; idx < n; ++idx) {
+        ASSERT_EQ(accessor.at(3 * idx), input_data[idx]);
+    }
+}
+
+TEST_F(ColumnDataRandomAccessorTest, DenseIrregularBlocks) {
+    // Presize to 500 values forces irregular blocks
+    Column column(type_descriptor, n / 2, true, false);
+    ASSERT_EQ(column.num_blocks(), 1);
+    memcpy(column.ptr(), input_data.data(), (n / 2) * sizeof(int64_t));
+    column.set_row_data((n / 2) - 1);
+    for (size_t idx = n / 2; idx < n; ++idx) {
+        column.push_back<int64_t>(input_data[idx]);
+    }
+
+    ASSERT_FALSE(column.buffer().is_regular_sized());
+    ASSERT_EQ(column.num_blocks(), 3);
+
+    auto column_data = column.data();
+    auto accessor = random_accessor<TDT>(&column_data);
+    for (size_t idx = 0; idx < n; ++idx) {
+        ASSERT_EQ(accessor.at(idx), input_data[idx]);
+    }
+}
+
+TEST_F(ColumnDataRandomAccessorTest, SparseIrregularBlocks) {
+    // Presize to 500 values forces irregular blocks
+    Column column(type_descriptor, n / 2, true, true);
+    ASSERT_EQ(column.num_blocks(), 1);
+    memcpy(column.ptr(), input_data.data(), (n / 2) * sizeof(int64_t));
+    column.set_row_data((n / 2) - 1);
+    // Set every third value
+    for (size_t idx = n / 2; idx < n; ++idx) {
+        column.set_scalar<int64_t>(3 * idx, input_data[idx]);
+    }
+
+    ASSERT_TRUE(column.is_sparse());
+    ASSERT_FALSE(column.buffer().is_regular_sized());
+    ASSERT_EQ(column.num_blocks(), 3);
+
+    auto column_data = column.data();
+    auto accessor = random_accessor<TDT>(&column_data);
+    for (size_t idx = 0; idx < n; ++idx) {
+        if (idx < n / 2) {
+            ASSERT_EQ(accessor.at(idx), input_data[idx]);
+        } else {
+            ASSERT_EQ(accessor.at(3 * idx), input_data[idx]);
+        }
+    }
+}
\ No newline at end of file
diff --git a/cpp/arcticdb/processing/test/test_filter_and_project_sparse.cpp b/cpp/arcticdb/processing/test/test_filter_and_project_sparse.cpp
new file mode 100644
index 0000000000..32d35fd613
--- /dev/null
+++ b/cpp/arcticdb/processing/test/test_filter_and_project_sparse.cpp
@@ -0,0 +1,421 @@
+/* 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/processing_unit.hpp>
+#include <arcticdb/util/test/generators.hpp>
+
+using namespace arcticdb;
+
+class FilterProjectSparse : public testing::Test {
+protected:
+    void SetUp() override {
+        auto input_segment = generate_filter_and_project_testing_sparse_segment();
+        sparse_floats_1 = input_segment.column_ptr(input_segment.column_index("sparse_floats_1").value());
+        sparse_floats_2 = input_segment.column_ptr(input_segment.column_index("sparse_floats_2").value());
+        dense_floats_1 = input_segment.column_ptr(input_segment.column_index("dense_floats_1").value());
+        dense_floats_2 = input_segment.column_ptr(input_segment.column_index("dense_floats_2").value());
+        sparse_bools = input_segment.column_ptr(input_segment.column_index("sparse_bools").value());
+        proc_unit = ProcessingUnit(std::move(input_segment));
+        proc_unit.set_expression_context(expression_context);
+    }
+
+    std::shared_ptr<Column> unary_projection(std::string_view input_column_name, OperationType op) {
+        const std::string output_column("unary projection");
+        expression_context->root_node_name_ = ExpressionName(output_column);
+        auto expression_node = std::make_shared<ExpressionNode>(ColumnName(input_column_name), op);
+        expression_context->add_expression_node(output_column, expression_node);
+
+        auto variant_data = proc_unit.get(expression_context->root_node_name_);
+        return std::get<ColumnWithStrings>(variant_data).column_;
+    }
+
+    util::BitSet unary_filter(std::string_view input_column_name, OperationType op) {
+        const std::string root_node_name("unary filter");
+        expression_context->root_node_name_ = ExpressionName(root_node_name);
+        auto expression_node = std::make_shared<ExpressionNode>(ColumnName(input_column_name), op);
+        expression_context->add_expression_node(root_node_name, expression_node);
+
+        auto variant_data = proc_unit.get(expression_context->root_node_name_);
+        return std::get<util::BitSet>(variant_data);
+    }
+
+    util::BitSet binary_filter(std::string_view left_column_name,
+                               std::variant<std::string_view, double, std::unordered_set<double>> right_input,
+                               OperationType op) {
+        const std::string root_node_name("binary filter");
+        const std::string value_name("value");
+        const std::string value_set_name("value set");
+        expression_context->root_node_name_ = ExpressionName(root_node_name);
+        std::shared_ptr<ExpressionNode> expression_node;
+        util::variant_match(right_input,
+            [&](std::string_view right_column_name) {
+                expression_node = std::make_shared<ExpressionNode>(ColumnName(left_column_name), ColumnName(right_column_name), op);
+            },
+            [&](double value) {
+                expression_node = std::make_shared<ExpressionNode>(ColumnName(left_column_name), ValueName(value_name), op);
+                expression_context->add_value(value_name, std::make_shared<Value>(value, DataType::FLOAT64));
+            },
+            [&](std::unordered_set<double> value_set) {
+                expression_node = std::make_shared<ExpressionNode>(ColumnName(left_column_name), ValueSetName(value_set_name), op);
+                expression_context->add_value_set(value_set_name, std::make_shared<ValueSet>(std::make_shared<std::unordered_set<double>>(value_set)));
+            });
+        expression_context->add_expression_node(root_node_name, expression_node);
+
+        auto variant_data = proc_unit.get(expression_context->root_node_name_);
+        return std::get<util::BitSet>(variant_data);
+    }
+
+    std::shared_ptr<Column> binary_projection(std::string_view left_column_name,
+                                              std::variant<std::string_view, double> right_input,
+                                              OperationType op) {
+        const std::string output_column("binary filter");
+        const std::string value_name("value");
+        expression_context->root_node_name_ = ExpressionName(output_column);
+        std::shared_ptr<ExpressionNode> expression_node;
+        util::variant_match(right_input,
+                            [&](std::string_view right_column_name) {
+                                expression_node = std::make_shared<ExpressionNode>(ColumnName(left_column_name), ColumnName(right_column_name), op);
+                            },
+                            [&](double value) {
+                                expression_node = std::make_shared<ExpressionNode>(ColumnName(left_column_name), ValueName(value_name), op);
+                                expression_context->add_value(value_name, std::make_shared<Value>(value, DataType::FLOAT64));
+                            });
+        expression_context->add_expression_node(output_column, expression_node);
+
+        auto variant_data = proc_unit.get(expression_context->root_node_name_);
+        return std::get<ColumnWithStrings>(variant_data).column_;
+    }
+
+    ProcessingUnit proc_unit;
+    std::shared_ptr<ExpressionContext> expression_context{std::make_shared<ExpressionContext>()};
+    std::shared_ptr<Column> sparse_floats_1;
+    std::shared_ptr<Column> sparse_floats_2;
+    std::shared_ptr<Column> dense_floats_1;
+    std::shared_ptr<Column> dense_floats_2;
+    std::shared_ptr<Column> sparse_bools;
+};
+
+TEST_F(FilterProjectSparse, UnaryProjection) {
+    auto projected_column = unary_projection("sparse_floats_1", OperationType::NEG);
+
+    ASSERT_EQ(sparse_floats_1->last_row(), projected_column->last_row());
+    ASSERT_EQ(sparse_floats_1->row_count(), projected_column->row_count());
+    ASSERT_EQ(sparse_floats_1->opt_sparse_map(), projected_column->opt_sparse_map());
+
+    for (auto idx=0; idx< sparse_floats_1->row_count(); idx++) {
+        ASSERT_FLOAT_EQ(sparse_floats_1->reference_at<double>(idx), -projected_column->reference_at<double>(idx));
+    }
+}
+
+TEST_F(FilterProjectSparse, BoolColumnIdentity) {
+    auto bitset = unary_filter("sparse_bools", OperationType::IDENTITY);
+    for (auto idx = 0; idx <= sparse_bools->last_row(); idx++) {
+        auto opt_input_value = sparse_bools->scalar_at<bool>(idx);
+        if (opt_input_value.has_value() && *opt_input_value) {
+            ASSERT_TRUE(bitset.get_bit(idx));
+        } else {
+            ASSERT_FALSE(bitset.get_bit(idx));
+        }
+    }
+}
+
+TEST_F(FilterProjectSparse, BoolColumnNot) {
+    auto bitset = unary_filter("sparse_bools", OperationType::NOT);
+    for (auto idx = 0; idx <= sparse_bools->last_row(); idx++) {
+        auto opt_input_value = sparse_bools->scalar_at<bool>(idx);
+        if (opt_input_value.has_value() && *opt_input_value) {
+            ASSERT_FALSE(bitset.get_bit(idx));
+        } else {
+            ASSERT_TRUE(bitset.get_bit(idx));
+        }
+    }
+}
+
+TEST_F(FilterProjectSparse, IsNull) {
+    auto bitset = unary_filter("sparse_floats_2", OperationType::ISNULL);
+    for (auto idx = 0; idx <= sparse_floats_2->last_row(); idx++) {
+        auto opt_input_value = sparse_floats_2->scalar_at<double>(idx);
+        if (opt_input_value.has_value() && !std::isnan(*opt_input_value)) {
+            ASSERT_FALSE(bitset.get_bit(idx));
+        } else {
+            ASSERT_TRUE(bitset.get_bit(idx));
+        }
+    }
+}
+
+TEST_F(FilterProjectSparse, NotNull) {
+    auto bitset = unary_filter("sparse_floats_2", OperationType::NOTNULL);
+    for (auto idx = 0; idx <= sparse_floats_2->last_row(); idx++) {
+        auto opt_input_value = sparse_floats_2->scalar_at<double>(idx);
+        if (opt_input_value.has_value() && !std::isnan(*opt_input_value)) {
+            ASSERT_TRUE(bitset.get_bit(idx));
+        } else {
+            ASSERT_FALSE(bitset.get_bit(idx));
+        }
+    }
+}
+
+TEST_F(FilterProjectSparse, BinaryComparisonIsIn) {
+    auto bitset = binary_filter("sparse_floats_2", std::unordered_set<double>{5.0}, OperationType::ISIN);
+    for (auto idx = 0; idx <= sparse_floats_2->last_row(); idx++) {
+        auto opt_input_value = sparse_floats_2->scalar_at<double>(idx);
+        if (opt_input_value.has_value() && *opt_input_value == double(5.0)) {
+            ASSERT_TRUE(bitset.get_bit(idx));
+        } else {
+            ASSERT_FALSE(bitset.get_bit(idx));
+        }
+    }
+}
+
+TEST_F(FilterProjectSparse, BinaryComparisonIsNotIn) {
+    auto bitset = binary_filter("sparse_floats_2", std::unordered_set<double>{5.0}, OperationType::ISNOTIN);
+    for (auto idx = 0; idx <= sparse_floats_2->last_row(); idx++) {
+        auto opt_input_value = sparse_floats_2->scalar_at<double>(idx);
+        if (!opt_input_value.has_value() || *opt_input_value != double(5.0)) {
+            ASSERT_TRUE(bitset.get_bit(idx));
+        } else {
+            ASSERT_FALSE(bitset.get_bit(idx));
+        }
+    }
+}
+
+TEST_F(FilterProjectSparse, BinaryComparisonColValEquals) {
+    auto bitset = binary_filter("sparse_floats_2", double{5.0}, OperationType::EQ);
+    for (auto idx = 0; idx <= sparse_floats_2->last_row(); idx++) {
+        auto opt_input_value = sparse_floats_2->scalar_at<double>(idx);
+        if (opt_input_value.has_value() && *opt_input_value == double(5.0)) {
+            ASSERT_TRUE(bitset.get_bit(idx));
+        } else {
+            ASSERT_FALSE(bitset.get_bit(idx));
+        }
+    }
+}
+
+TEST_F(FilterProjectSparse, BinaryComparisonColValNotEqual) {
+    auto bitset = binary_filter("sparse_floats_2", double{5.0}, OperationType::NE);
+    for (auto idx = 0; idx <= sparse_floats_2->last_row(); idx++) {
+        auto opt_input_value = sparse_floats_2->scalar_at<double>(idx);
+        if (!opt_input_value.has_value() || *opt_input_value != double(5.0)) {
+            ASSERT_TRUE(bitset.get_bit(idx));
+        } else {
+            ASSERT_FALSE(bitset.get_bit(idx));
+        }
+    }
+}
+
+TEST_F(FilterProjectSparse, BinaryComparisonSparseColSparseCol) {
+    auto bitset = binary_filter("sparse_floats_1", std::string_view{"sparse_floats_2"}, OperationType::LT);
+    for (auto idx = 0; idx <= sparse_floats_2->last_row(); idx++) {
+        auto opt_left_value = sparse_floats_1->scalar_at<double>(idx);
+        auto opt_right_value = sparse_floats_2->scalar_at<double>(idx);
+        if (opt_left_value.has_value() && opt_right_value.has_value()) {
+            ASSERT_EQ(*opt_left_value < *opt_right_value, bitset.get_bit(idx));
+        } else {
+            ASSERT_FALSE(bitset.get_bit(idx));
+        }
+    }
+}
+
+TEST_F(FilterProjectSparse, BinaryComparisonSparseColSparseColNotEqual) {
+    auto bitset = binary_filter("sparse_floats_1", std::string_view{"sparse_floats_2"}, OperationType::NE);
+    for (auto idx = 0; idx <= sparse_floats_2->last_row(); idx++) {
+        auto opt_left_value = sparse_floats_1->scalar_at<double>(idx);
+        auto opt_right_value = sparse_floats_2->scalar_at<double>(idx);
+        if (opt_left_value.has_value() && opt_right_value.has_value()) {
+            ASSERT_EQ(*opt_left_value != *opt_right_value, bitset.get_bit(idx));
+        } else {
+            ASSERT_TRUE(bitset.get_bit(idx));
+        }
+    }
+}
+
+TEST_F(FilterProjectSparse, BinaryComparisonDenseColDenseCol) {
+    auto bitset = binary_filter("dense_floats_1", std::string_view{"dense_floats_2"}, OperationType::EQ);
+    for (auto idx = 0; idx <= dense_floats_2->last_row(); idx++) {
+        if (idx <= dense_floats_1->last_row()) {
+            auto opt_left_value = dense_floats_1->scalar_at<double>(idx);
+            auto opt_right_value = dense_floats_2->scalar_at<double>(idx);
+            if (opt_left_value.has_value() && opt_right_value.has_value()) {
+                ASSERT_EQ(*opt_left_value == *opt_right_value, bitset.get_bit(idx));
+            } else {
+                ASSERT_FALSE(bitset.get_bit(idx));
+            }
+        } else {
+            ASSERT_FALSE(bitset.get_bit(idx));
+        }
+    }
+}
+
+TEST_F(FilterProjectSparse, BinaryComparisonDenseColDenseColNotEqual) {
+    auto bitset = binary_filter("dense_floats_1", std::string_view{"dense_floats_2"}, OperationType::NE);
+    for (auto idx = 0; idx <= dense_floats_2->last_row(); idx++) {
+        if (idx <= dense_floats_1->last_row()) {
+            auto opt_left_value = dense_floats_1->scalar_at<double>(idx);
+            auto opt_right_value = dense_floats_2->scalar_at<double>(idx);
+            if (opt_left_value.has_value() && opt_right_value.has_value()) {
+                ASSERT_EQ(*opt_left_value != *opt_right_value, bitset.get_bit(idx));
+            } else {
+                ASSERT_TRUE(bitset.get_bit(idx));
+            }
+        } else {
+            ASSERT_TRUE(bitset.get_bit(idx));
+        }
+    }
+}
+
+TEST_F(FilterProjectSparse, BinaryComparisonSparseColShorterThanDenseCol) {
+    auto bitset = binary_filter("sparse_floats_1", std::string_view{"dense_floats_1"}, OperationType::GT);
+    for (auto idx = 0; idx <= dense_floats_1->last_row(); idx++) {
+        if (idx <= sparse_floats_1->last_row()) {
+            auto opt_left_value = sparse_floats_1->scalar_at<double>(idx);
+            auto opt_right_value = dense_floats_1->scalar_at<double>(idx);
+            if (opt_left_value.has_value() && opt_right_value.has_value()) {
+                ASSERT_EQ(*opt_left_value > *opt_right_value, bitset.get_bit(idx));
+            } else {
+                ASSERT_FALSE(bitset.get_bit(idx));
+            }
+        } else {
+            ASSERT_FALSE(bitset.get_bit(idx));
+        }
+    }
+}
+
+TEST_F(FilterProjectSparse, BinaryComparisonSparseColShorterThanDenseColNotEqual) {
+    auto bitset = binary_filter("sparse_floats_1", std::string_view{"dense_floats_1"}, OperationType::NE);
+    for (auto idx = 0; idx <= sparse_floats_1->last_row(); idx++) {
+        auto opt_left_value = sparse_floats_1->scalar_at<double>(idx);
+        auto opt_right_value = dense_floats_1->scalar_at<double>(idx);
+        if (opt_left_value.has_value() && opt_right_value.has_value()) {
+            ASSERT_EQ(*opt_left_value != *opt_right_value, bitset.get_bit(idx));
+        } else {
+            ASSERT_TRUE(bitset.get_bit(idx));
+        }
+    }
+}
+
+TEST_F(FilterProjectSparse, BinaryComparisonSparseColLongerThanDenseCol) {
+    auto bitset = binary_filter("dense_floats_1", std::string_view{"sparse_floats_1"}, OperationType::GT);
+    for (auto idx = 0; idx <= sparse_floats_1->last_row(); idx++) {
+        if (idx <= dense_floats_1->last_row()) {
+            auto opt_left_value = dense_floats_1->scalar_at<double>(idx);
+            auto opt_right_value = sparse_floats_1->scalar_at<double>(idx);
+            if (opt_left_value.has_value() && opt_right_value.has_value()) {
+                ASSERT_EQ(*opt_left_value > *opt_right_value, bitset.get_bit(idx));
+            } else {
+                ASSERT_FALSE(bitset.get_bit(idx));
+            }
+        } else {
+            ASSERT_FALSE(bitset.get_bit(idx));
+        }
+    }
+}
+
+TEST_F(FilterProjectSparse, BinaryComparisonSparseColLongerThanDenseColNotEqual) {
+    auto bitset = binary_filter("dense_floats_1", std::string_view{"sparse_floats_1"}, OperationType::NE);
+    for (auto idx = 0; idx <= sparse_floats_1->last_row(); idx++) {
+        auto opt_left_value = dense_floats_1->scalar_at<double>(idx);
+        auto opt_right_value = sparse_floats_1->scalar_at<double>(idx);
+        if (opt_left_value.has_value() && opt_right_value.has_value()) {
+            ASSERT_EQ(*opt_left_value != *opt_right_value, bitset.get_bit(idx));
+        } else {
+            ASSERT_TRUE(bitset.get_bit(idx));
+        }
+    }
+}
+
+TEST_F(FilterProjectSparse, BinaryArithmeticColVal) {
+    auto projected_column = binary_projection("sparse_floats_1", double{10.0}, OperationType::MUL);
+    ASSERT_EQ(sparse_floats_1->last_row(), projected_column->last_row());
+    ASSERT_EQ(sparse_floats_1->row_count(), projected_column->row_count());
+    ASSERT_EQ(sparse_floats_1->opt_sparse_map(), projected_column->opt_sparse_map());
+    for (auto idx=0; idx< sparse_floats_1->row_count(); idx++) {
+        ASSERT_FLOAT_EQ(10.0 * sparse_floats_1->reference_at<double>(idx), projected_column->reference_at<double>(idx));
+    }
+}
+
+TEST_F(FilterProjectSparse, BinaryArithmeticSparseColSparseCol) {
+    auto projected_column = binary_projection("sparse_floats_1", std::string_view{"sparse_floats_2"}, OperationType::MUL);
+    ASSERT_TRUE(projected_column->opt_sparse_map().has_value());
+    ASSERT_EQ(sparse_floats_1->sparse_map() & sparse_floats_2->sparse_map(), projected_column->sparse_map());
+    ASSERT_EQ(projected_column->row_count(), projected_column->sparse_map().count());
+
+    for (auto idx = 0; idx <= projected_column->last_row(); idx++) {
+        auto opt_left_value = sparse_floats_1->scalar_at<double>(idx);
+        auto opt_right_value = sparse_floats_2->scalar_at<double>(idx);
+        auto opt_projected_value = projected_column->scalar_at<double>(idx);
+        if (opt_left_value.has_value() && opt_right_value.has_value()) {
+            ASSERT_TRUE(opt_projected_value.has_value());
+            if (std::isnan(*opt_left_value * *opt_right_value)) {
+                ASSERT_TRUE(std::isnan(*opt_projected_value));
+            } else {
+                ASSERT_FLOAT_EQ(*opt_left_value * *opt_right_value, *projected_column->scalar_at<double>(idx));
+            }
+        } else {
+            ASSERT_FALSE(projected_column->has_value_at(idx));
+        }
+    }
+}
+
+TEST_F(FilterProjectSparse, BinaryArithmeticDenseColDenseCol) {
+    auto projected_column = binary_projection("dense_floats_1", std::string_view{"dense_floats_2"}, OperationType::MUL);
+    // dense_floats_1 has fewer values than dense_floats_2
+    ASSERT_EQ(dense_floats_1->last_row(), projected_column->last_row());
+    ASSERT_EQ(dense_floats_1->row_count(), projected_column->row_count());
+    ASSERT_FALSE(projected_column->opt_sparse_map().has_value());
+
+    for (auto idx=0; idx< dense_floats_1->last_row(); idx++) {
+        ASSERT_FLOAT_EQ(dense_floats_1->reference_at<double>(idx) * dense_floats_2->reference_at<double>(idx),
+                        projected_column->reference_at<double>(idx));
+    }
+}
+
+TEST_F(FilterProjectSparse, BinaryArithmeticSparseColShorterThanDenseCol) {
+    auto projected_column = binary_projection("sparse_floats_1", std::string_view{"dense_floats_1"}, OperationType::MUL);
+    ASSERT_TRUE(projected_column->opt_sparse_map().has_value());
+    ASSERT_EQ(*sparse_floats_1->opt_sparse_map(), *projected_column->opt_sparse_map());
+    ASSERT_EQ(projected_column->row_count(), projected_column->sparse_map().count());
+    for (auto idx = 0; idx <= projected_column->last_row(); idx++) {
+        auto opt_left_value = sparse_floats_1->scalar_at<double>(idx);
+        auto opt_right_value = dense_floats_1->scalar_at<double>(idx);
+        auto opt_projected_value = projected_column->scalar_at<double>(idx);
+        if (opt_left_value.has_value() && opt_right_value.has_value()) {
+            ASSERT_TRUE(opt_projected_value.has_value());
+            if (std::isnan(*opt_left_value * *opt_right_value)) {
+                ASSERT_TRUE(std::isnan(*opt_projected_value));
+            } else {
+                ASSERT_FLOAT_EQ(*opt_left_value * *opt_right_value, *projected_column->scalar_at<double>(idx));
+            }
+        } else {
+            ASSERT_FALSE(projected_column->has_value_at(idx));
+        }
+    }
+}
+
+TEST_F(FilterProjectSparse, BinaryArithmeticSparseColLongerThanDenseCol) {
+    auto projected_column = binary_projection("dense_floats_1", std::string_view{"sparse_floats_2"}, OperationType::MUL);
+    ASSERT_TRUE(projected_column->opt_sparse_map().has_value());
+    for (auto idx = 0; idx < dense_floats_1->row_count(); idx++) {
+        ASSERT_EQ(sparse_floats_2->sparse_map().get_bit(idx), projected_column->sparse_map().get_bit(idx));
+    }
+    ASSERT_EQ(projected_column->row_count(), projected_column->sparse_map().count());
+    for (auto idx = 0; idx <= projected_column->last_row(); idx++) {
+        auto opt_left_value = dense_floats_1->scalar_at<double>(idx);
+        auto opt_right_value = sparse_floats_2->scalar_at<double>(idx);
+        auto opt_projected_value = projected_column->scalar_at<double>(idx);
+        if (opt_left_value.has_value() && opt_right_value.has_value()) {
+            ASSERT_TRUE(opt_projected_value.has_value());
+            if (std::isnan(*opt_left_value * *opt_right_value)) {
+                ASSERT_TRUE(std::isnan(*opt_projected_value));
+            } else {
+                ASSERT_FLOAT_EQ(*opt_left_value * *opt_right_value, *projected_column->scalar_at<double>(idx));
+            }
+        } else {
+            ASSERT_FALSE(projected_column->has_value_at(idx));
+        }
+    }
+}
\ No newline at end of file
diff --git a/cpp/arcticdb/util/test/generators.hpp b/cpp/arcticdb/util/test/generators.hpp
index d313d548aa..c73c526842 100644
--- a/cpp/arcticdb/util/test/generators.hpp
+++ b/cpp/arcticdb/util/test/generators.hpp
@@ -131,6 +131,67 @@ inline Column generate_empty_column() {
     return column;
 }
 
+// Generate a segment in memory suitable for testing filters and projections with sparse columns:
+// * sparse_floats_1 - a sparse column of doubles with 5 rows including missing values and NaNs
+// * sparse_floats_2 - a sparse column of doubles with 10 rows including missing values and NaNs
+// * dense_floats_1 - a dense column of doubles with 7 rows
+// * dense_floats_2 - a dense column of doubles with 15 rows
+// * sparse_bools - a sparse column of bools with 4 rows including missing values
+// The columns deliberately have differing row counts, as the sparse map bitset in a sparse column is not padded with
+// zeros to the end of the SegmentInMemory, they just stop at the last row which contains a value
+// The two dense columns of differing lengths are to cover the corner case where a column is "sparse", but happens to
+// have the first n values populated, and so appears dense, but with fewer rows than the containing SegmentInMemory
+inline SegmentInMemory generate_filter_and_project_testing_sparse_segment() {
+    SegmentInMemory seg;
+    using FTDT = ScalarTagType<DataTypeTag<DataType::FLOAT64>>;
+    using BTDT = ScalarTagType<DataTypeTag<DataType::BOOL8>>;
+    auto sparse_floats_1 = std::make_shared<Column>(static_cast<TypeDescriptor>(FTDT{}), 0, false, true);
+    auto sparse_floats_2 = std::make_shared<Column>(static_cast<TypeDescriptor>(FTDT{}), 0, false, true);
+    auto dense_floats_1 = std::make_shared<Column>(static_cast<TypeDescriptor>(FTDT{}), 0, false, false);
+    auto dense_floats_2 = std::make_shared<Column>(static_cast<TypeDescriptor>(FTDT{}), 0, false, false);
+    auto sparse_bools = std::make_shared<Column>(static_cast<TypeDescriptor>(BTDT{}), 0, false, true);
+
+    constexpr auto nan = std::numeric_limits<double>::quiet_NaN();
+
+    // Row 0 - both sparse float columns have non-NaN values
+    sparse_floats_1->set_scalar<double>(0, 1.0);
+    sparse_floats_2->set_scalar<double>(0, 2.0);
+    // Row 1 - 1 non-NaN value, 1 missing
+    sparse_floats_1->set_scalar<double>(1, 3.0);
+    // Row 2 - 1 NaN value, 1 missing
+    sparse_floats_2->set_scalar<double>(2, nan);
+    // Row 3 - both missing
+    // Row 4 - 1 NaN value, 1 non-NaN value
+    sparse_floats_1->set_scalar<double>(4, 4.0);
+    sparse_floats_2->set_scalar<double>(4, nan);
+
+    // Add another 5 rows to sparse_floats_2 so that the 2 sparse float columns are of differing lengths
+    sparse_floats_2->set_scalar<double>(5, 5.0);
+    sparse_floats_2->set_scalar<double>(9, nan);
+
+    // Dense float column values are just 10*idx
+    for (auto idx = 0; idx < 7; idx++) {
+        dense_floats_1->set_scalar<double>(idx, 10.0 * static_cast<double>(idx));
+    }
+    for (auto idx = 0; idx < 15; idx++) {
+        dense_floats_2->set_scalar<double>(idx, 10.0 * static_cast<double>(idx));
+    }
+
+    // Sparse bool column goes missing, true, missing, false
+    sparse_bools->set_scalar<bool>(1, true);
+    sparse_bools->set_scalar<bool>(3, false);
+
+    seg.add_column(scalar_field(sparse_floats_1->type().data_type(), "sparse_floats_1"), sparse_floats_1);
+    seg.add_column(scalar_field(sparse_floats_2->type().data_type(), "sparse_floats_2"), sparse_floats_2);
+    seg.add_column(scalar_field(dense_floats_1->type().data_type(), "dense_floats_1"), dense_floats_1);
+    seg.add_column(scalar_field(dense_floats_2->type().data_type(), "dense_floats_2"), dense_floats_2);
+    seg.add_column(scalar_field(sparse_bools->type().data_type(), "sparse_bools"), sparse_bools);
+
+    // 1 less than the number of rows in the largest column
+    seg.set_row_id(14);
+    return seg;
+}
+
 // 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
diff --git a/python/tests/unit/arcticdb/version_store/test_aggregation.py b/python/tests/unit/arcticdb/version_store/test_aggregation.py
index 4bfb56edce..ed4dbdf674 100644
--- a/python/tests/unit/arcticdb/version_store/test_aggregation.py
+++ b/python/tests/unit/arcticdb/version_store/test_aggregation.py
@@ -428,7 +428,7 @@ def test_mean_aggregation_float(local_object_version_store):
     assert_frame_equal(res.data, df)
 
 
-def test_mean_aggregation_float_nan(lmdb_version_store_v2):
+def test_mean_aggregation_float_nan(lmdb_version_store):
     df = DataFrame(
         {
             "grouping_column": ["group_1", "group_1", "group_1", "group_2", "group_2"],
@@ -439,9 +439,9 @@ def test_mean_aggregation_float_nan(lmdb_version_store_v2):
     q = QueryBuilder()
     q = q.groupby("grouping_column").agg({"to_mean": "mean"})
     symbol = "test_aggregation"
-    lmdb_version_store_v2.write(symbol, df)
+    lmdb_version_store.write(symbol, df)
 
-    res = lmdb_version_store_v2.read(symbol, query_builder=q)
+    res = lmdb_version_store.read(symbol, query_builder=q)
 
     df = pd.DataFrame({"to_mean": [(1.1 + 1.4 + 2.5) / 3, np.nan]}, index=["group_1", "group_2"])
     df.index.rename("grouping_column", inplace=True)