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

Issue:
Fix returning sparse column to Python layer after processing pipeline
https://github.com/man-group/ArcticDB/pull/1399 adds the ability to filter and project on sparse columns, tested in C++, and earlier merges allow grouping and aggregating on sparse columns too.
However, `copy_frame_data_to_buffer` (the final stage of the processing pipeline) does not handle sparse columns correctly, unlike direct reads, which uses `decode_or_expand_impl`, which checks for the presence of a sparse map, and calls `expand_dense_buffer_using_bitmap` in this case.
We should be able to add tests for this from the Python layer using the `sparsify_floats` kwarg to `write` to generate sparse data.
Before merging, testing should be performed on a real library, see https://github.com/man-group/arcticdb-man/issues/82 for details.


Discussion / hints:

Fix patch:
diff --git a/cpp/arcticdb/codec/codec.cpp b/cpp/arcticdb/codec/codec.cpp
index 2b91a53b17..f19391d7e5 100644
--- a/cpp/arcticdb/codec/codec.cpp
+++ b/cpp/arcticdb/codec/codec.cpp
@@ -327,7 +327,7 @@ ssize_t calculate_last_row(const Column& col) {
             last_row = static_cast<ssize_t>(last_set_bit);
         }
     } else {
-        last_row = static_cast<ssize_t>(col.row_count());
+        last_row = static_cast<ssize_t>(col.row_count()) - 1;
     }
     return last_row;
 }
@@ -381,7 +381,7 @@ void decode_v2(
         util::check_magic<StringPoolMagic>(data);
         decode_string_pool(hdr, data, begin, end, res);
 
-        res.set_row_data(static_cast<ssize_t>(start_row + seg_row_count-1));
+        res.set_row_data(static_cast<ssize_t>(start_row + seg_row_count));
         res.set_compacted(segment.header().compacted());
     }
 }
@@ -434,7 +434,7 @@ void decode_v1(
         }
 
         decode_string_pool(hdr, data, begin, end, res);
-        res.set_row_data(static_cast<ssize_t>(start_row + seg_row_count-1));
+        res.set_row_data(static_cast<ssize_t>(start_row + seg_row_count));
         res.set_compacted(segment.header().compacted());
     }
 }
diff --git a/cpp/arcticdb/column_store/column.cpp b/cpp/arcticdb/column_store/column.cpp
index 1c7a718871..9a25ba33a5 100644
--- a/cpp/arcticdb/column_store/column.cpp
+++ b/cpp/arcticdb/column_store/column.cpp
@@ -61,7 +61,9 @@ void initialise_output_column(const Column& left_input_column, const Column& rig
             }
             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_physical_rows > 0) {
+            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);
         }
@@ -394,7 +396,7 @@ void Column::set_row_data(size_t row_id) {
     last_logical_row_ = row_id;
     const auto last_stored_row = row_count() - 1;
     if(sparse_map_) {
-        last_physical_row_ = sparse_map_->count() - 1;
+        last_physical_row_ = static_cast<ssize_t>(sparse_map_->count()) - 1;
     } else if (last_logical_row_ != last_stored_row) {
         last_physical_row_ = last_stored_row;
         backfill_sparse_map(last_stored_row);
diff --git a/cpp/arcticdb/column_store/column.hpp b/cpp/arcticdb/column_store/column.hpp
index f86aa88ef1..bd82df5703 100644
--- a/cpp/arcticdb/column_store/column.hpp
+++ b/cpp/arcticdb/column_store/column.hpp
@@ -250,7 +250,11 @@ class Column {
 
     void backfill_sparse_map(ssize_t to_row) {
         ARCTICDB_TRACE(log::version(), "Backfilling sparse map to position {}", to_row);
-        sparse_map().set_range(0, bv_size(to_row), true);
+        // Initialise the optional to an empty bitset if it has not been created yet
+        auto& bitset = sparse_map();
+        if (to_row >= 0) {
+            bitset.set_range(0, bv_size(to_row), true);
+        }
     }
 
     [[nodiscard]] util::BitMagic& sparse_map();
diff --git a/cpp/arcticdb/column_store/memory_segment_impl.cpp b/cpp/arcticdb/column_store/memory_segment_impl.cpp
index 47e8e29a2a..bd24ca4633 100644
--- a/cpp/arcticdb/column_store/memory_segment_impl.cpp
+++ b/cpp/arcticdb/column_store/memory_segment_impl.cpp
@@ -349,8 +349,6 @@ std::shared_ptr<SegmentInMemoryImpl> SegmentInMemoryImpl::get_output_segment(siz
 
 std::vector<std::shared_ptr<SegmentInMemoryImpl>> SegmentInMemoryImpl::partition(const std::vector<uint8_t>& row_to_segment,
                                                                    const std::vector<uint64_t>& segment_counts) const {
-    schema::check<ErrorCode::E_UNSUPPORTED_COLUMN_TYPE>(!is_sparse(),
-                                                        "SegmentInMemory::partition not supported with sparse columns");
     internal::check<ErrorCode::E_ASSERTION_FAILURE>(row_count() == row_to_segment.size(),
                                                     "row_to_segment size does not match segment row count: {} != {}", row_to_segment.size(), row_count());
     std::vector<std::shared_ptr<SegmentInMemoryImpl>> output(segment_counts.size());
@@ -362,7 +360,6 @@ std::vector<std::shared_ptr<SegmentInMemoryImpl>> SegmentInMemoryImpl::partition
         if (*segment_count > 0) {
             auto& seg = output.at(segment_count.index);
             seg = get_output_segment(*segment_count);
-            seg->set_row_data(*segment_count - 1);
             seg->set_string_pool(string_pool_);
             seg->set_compacted(compacted_);
             if (metadata_) {
@@ -376,24 +373,50 @@ std::vector<std::shared_ptr<SegmentInMemoryImpl>> SegmentInMemoryImpl::partition
     for(const auto& column : folly::enumerate(columns())) {
         details::visit_type((*column)->type().data_type(), [&](auto col_tag) {
             using type_info = ScalarTypeInfo<decltype(col_tag)>;
-
-            std::vector<typename type_info::RawType*> output_ptrs{output.size(), nullptr};
-            for (const auto& segment: folly::enumerate(output)) {
-                if (static_cast<bool>(*segment)) {
-                    output_ptrs.at(segment.index) = reinterpret_cast<typename type_info::RawType*>((*segment)->column(column.index).ptr());
+            if ((*column)->is_sparse()) {
+                std::vector<std::shared_ptr<Column>> output_columns(output.size());
+                for (const auto& segment: folly::enumerate(output)) {
+                    if (static_cast<bool>(*segment)) {
+                        (*segment)->add_column(field(column.index), 0, false);
+                        output_columns.at(segment.index) = (*segment)->column_ptr(column.index);
+                    }
                 }
-            }
-
-            auto row_to_segment_it = row_to_segment.cbegin();
-            auto input_data = (*column)->data();
-            auto cend = input_data.cend<typename type_info::TDT>();
-            for (auto input_it = input_data.cbegin<typename type_info::TDT>(); input_it != cend; ++input_it, ++row_to_segment_it) {
-                if (ARCTICDB_LIKELY(*row_to_segment_it != std::numeric_limits<uint8_t>::max())) {
-                    *(output_ptrs[*row_to_segment_it]++) = *input_it;
+                for (const auto& segment_idx: folly::enumerate(row_to_segment)) {
+                    if (*segment_idx != std::numeric_limits<uint8_t>::max()) {
+                        auto opt_value = (*column)->scalar_at<typename type_info::RawType>(segment_idx.index);
+                        if (opt_value.has_value()) {
+                            output_columns[*segment_idx]->push_back(*opt_value);
+                        } else {
+                            output_columns[*segment_idx]->mark_absent_rows(1);
+                        }
+                    }
                 }
+            } else {
+                std::vector<typename type_info::RawType*> output_ptrs{output.size(), nullptr};
+                for (const auto& segment: folly::enumerate(output)) {
+                    if (static_cast<bool>(*segment)) {
+                        if (is_sparse()) {
+                            (*segment)->add_column(field(column.index), segment_counts[segment.index], true);
+                        }
+                        output_ptrs.at(segment.index) = reinterpret_cast<typename type_info::RawType*>((*segment)->column(column.index).ptr());
+                    }
+                }
+                auto row_to_segment_it = row_to_segment.cbegin();
+                Column::for_each<typename type_info::TDT>(**column, [&row_to_segment_it, &output_ptrs](auto val) {
+                    if (ARCTICDB_LIKELY(*row_to_segment_it != std::numeric_limits<uint8_t>::max())) {
+                        *(output_ptrs[*row_to_segment_it]++) = val;
+                    }
+                    ++row_to_segment_it;
+                });
             }
         });
     }
+    for (const auto& segment_count: folly::enumerate(segment_counts)) {
+        if (*segment_count > 0) {
+            auto& seg = output.at(segment_count.index);
+            seg->set_row_data(*segment_count - 1);
+        }
+    }
     return output;
 }
 
diff --git a/cpp/arcticdb/processing/clause.cpp b/cpp/arcticdb/processing/clause.cpp
index e7c6c9ab3d..a64cac4aa5 100644
--- a/cpp/arcticdb/processing/clause.cpp
+++ b/cpp/arcticdb/processing/clause.cpp
@@ -387,7 +387,6 @@ Composite<EntityIds> AggregationClause::process(Composite<EntityIds>&& entity_id
                         ++next_group_id;
                     }
                     ssize_t previous_value_index = 0;
-                    constexpr size_t missing_value_group_id = 0;
 
                     Column::for_each_enumerated<typename col_type_info::TDT>(
                             *col.column_,
@@ -419,6 +418,7 @@ Composite<EntityIds> AggregationClause::process(Composite<EntityIds>&& entity_id
                                 }
 
                                 if (is_sparse) {
+                                    constexpr size_t missing_value_group_id = 0;
                                     for (auto j = previous_value_index;
                                          j != enumerating_it.idx(); ++j) {
                                         *row_to_group_ptr++ = missing_value_group_id;
diff --git a/cpp/arcticdb/processing/processing_unit.hpp b/cpp/arcticdb/processing/processing_unit.hpp
index 2ec109484c..d97b651127 100644
--- a/cpp/arcticdb/processing/processing_unit.hpp
+++ b/cpp/arcticdb/processing/processing_unit.hpp
@@ -133,28 +133,38 @@ namespace arcticdb {
             const ColumnWithStrings& col,
             const Grouper& grouper,
             const Bucketizer& bucketizer) {
-        schema::check<ErrorCode::E_UNSUPPORTED_COLUMN_TYPE>(!col.column_->is_sparse(),
-                                                            "GroupBy not supported with sparse columns");
         // Mapping from row to bucket
         // 255 reserved for Nones and NaNs in string/float columns
         // Faster to initialise to 255 and use a raw ptr for the output than to call emplace_back repeatedly
-        std::vector<bucket_id> row_to_bucket(col.column_->row_count(), std::numeric_limits<bucket_id>::max());
+        std::vector<bucket_id> row_to_bucket(col.column_->last_row() + 1, std::numeric_limits<bucket_id>::max());
         auto out_ptr = row_to_bucket.data();
         // Tracks how many rows are in each bucket
         // Use to skip empty buckets, and presize columns in the output ProcessingUnit
         std::vector<uint64_t> bucket_counts(bucketizer.num_buckets(), 0);
 
         using TDT = typename Grouper::GrouperDescriptor;
-        Column::for_each<TDT>(*col.column_, [&] (auto val) {
-            auto opt_group = grouper.group(val, col.string_pool_);
-            if (ARCTICDB_LIKELY(opt_group.has_value())) {
-                auto bucket = bucketizer.bucket(*opt_group);
-                *out_ptr++ = bucket;
-                ++bucket_counts[bucket];
-            } else {
-                ++out_ptr;
-            }
-        });
+
+        if (col.column_->is_sparse()) {
+            Column::for_each_enumerated<TDT>(*col.column_, [&](auto enumerating_it) {
+                auto opt_group = grouper.group(enumerating_it.value(), col.string_pool_);
+                if (ARCTICDB_LIKELY(opt_group.has_value())) {
+                    auto bucket = bucketizer.bucket(*opt_group);
+                    row_to_bucket[enumerating_it.idx()] = bucket;
+                    ++bucket_counts[bucket];
+                }
+            });
+        } else {
+            Column::for_each<TDT>(*col.column_, [&](auto val) {
+                auto opt_group = grouper.group(val, col.string_pool_);
+                if (ARCTICDB_LIKELY(opt_group.has_value())) {
+                    auto bucket = bucketizer.bucket(*opt_group);
+                    *out_ptr++ = bucket;
+                    ++bucket_counts[bucket];
+                } else {
+                    ++out_ptr;
+                }
+            });
+        }
         return {std::move(row_to_bucket), std::move(bucket_counts)};
     }
 
diff --git a/cpp/arcticdb/processing/unsorted_aggregation.cpp b/cpp/arcticdb/processing/unsorted_aggregation.cpp
index 91d39563b4..82efa51d37 100644
--- a/cpp/arcticdb/processing/unsorted_aggregation.cpp
+++ b/cpp/arcticdb/processing/unsorted_aggregation.cpp
@@ -314,7 +314,16 @@ namespace
                 } else {
                     auto in_ptr = reinterpret_cast<MaybeValueType*>(aggregated_.data());
                     for (auto it = column_data.begin<typename col_type_info::TDT>(); it != column_data.end<typename col_type_info::TDT>(); ++it, ++in_ptr) {
-                        *it = in_ptr->value_;
+                        if constexpr (is_floating_point_type(col_type_info::data_type)) {
+                            *it = in_ptr->written_ ? in_ptr->value_ : std::numeric_limits<typename col_type_info::RawType>::quiet_NaN();
+                        } else {
+                            if constexpr(T == Extremum::MAX) {
+                                *it = in_ptr->written_ ? in_ptr->value_ : std::numeric_limits<typename col_type_info::RawType>::lowest();
+                            } else {
+                                // T == Extremum::MIN
+                                *it = in_ptr->written_ ? in_ptr->value_ : std::numeric_limits<typename col_type_info::RawType>::max();
+                            }
+                        }
                     }
                 }
             });
diff --git a/cpp/arcticdb/version/version_core.cpp b/cpp/arcticdb/version/version_core.cpp
index 9172903b5a..ad2356e89c 100644
--- a/cpp/arcticdb/version/version_core.cpp
+++ b/cpp/arcticdb/version/version_core.cpp
@@ -883,8 +883,6 @@ void copy_frame_data_to_buffer(const SegmentInMemory& destination, size_t target
         return;
     }
     auto& src_column = source.column(static_cast<position_t>(source_index));
-    schema::check<ErrorCode::E_UNSUPPORTED_COLUMN_TYPE>(!src_column.opt_sparse_map().has_value(),
-                                                   "QueryBuilder not yet supported with sparse data");
     auto& dst_column = destination.column(static_cast<position_t>(target_index));
     auto& buffer = dst_column.data().buffer();
     auto dst_rawtype_size = sizeof_datatype(dst_column.type());
@@ -901,6 +899,19 @@ void copy_frame_data_to_buffer(const SegmentInMemory& destination, size_t target
         dst_column.type().visit_tag([&](auto dst_desc_tag) {
             util::default_initialize<decltype(dst_desc_tag)>(dst_ptr, num_rows * dst_rawtype_size);
         });
+    // Do not use src_column.is_sparse() here, as that misses columns that are dense, but have fewer than num_rows values
+    } else if (src_column.opt_sparse_map().has_value() && has_valid_type_promotion(src_column.type(), dst_column.type())) {
+        details::visit_type(dst_column.type().data_type(), [&](auto dst_tag) {
+            using dst_type_info = ScalarTypeInfo<decltype(dst_tag)>;
+            util::default_initialize<typename dst_type_info::TDT>(dst_ptr, num_rows * dst_rawtype_size);
+            auto typed_dst_ptr = reinterpret_cast<typename dst_type_info::RawType*>(dst_ptr);
+            details::visit_type(src_column.type().data_type(), [&](auto src_tag) {
+                using src_type_info = ScalarTypeInfo<decltype(src_tag)>;
+                Column::for_each_enumerated<typename src_type_info::TDT>(src_column, [typed_dst_ptr](auto enumerating_it) {
+                    typed_dst_ptr[enumerating_it.idx()] = static_cast<typename dst_type_info::RawType>(enumerating_it.value());
+                });
+            });
+        });
     } else if (trivially_compatible_types(src_column.type(), dst_column.type())) {
         details::visit_type(src_column.type().data_type() ,[&src_data, &dst_ptr] (auto src_desc_tag) {
             using SourceTDT = ScalarTagType<decltype(src_desc_tag)>;


Test patch:
diff --git a/python/tests/unit/arcticdb/version_store/test_query_builder.py b/python/tests/unit/arcticdb/version_store/test_query_builder.py
index 70cec8ddd8..7f3060689b 100644
--- a/python/tests/unit/arcticdb/version_store/test_query_builder.py
+++ b/python/tests/unit/arcticdb/version_store/test_query_builder.py
@@ -512,64 +512,3 @@ def test_querybuilder_pickling():
     q = q.agg({"col2": "sum", "new_col": ("col2", "sum")})
 
     assert pickle.loads(pickle.dumps(q)) == q
-
-
-# Remove the following test and replace with more extensive ones once this issue is fixed:
-# https://github.com/man-group/ArcticDB/issues/1404
-def test_query_builder_sparse(lmdb_version_store):
-    lib = lmdb_version_store
-    sym = "test_filter_sparse"
-    df = pd.DataFrame(
-        {
-            "sparse_col": [0.0, np.nan, 0.0],
-            "dense_col": [0.0, 1.0, 2.0]
-         },
-        index=pd.date_range("2024-01-01", periods=3)
-    )
-    lib.write(sym, df, sparsify_floats=True)
-
-    # Filters
-    # These 2 filters exercise different code paths
-    q = QueryBuilder()
-    q = q[q["sparse_col"].isnull()]
-    with pytest.raises(SchemaException):
-        lib.read(sym, query_builder=q)
-    q = QueryBuilder()
-    q = q[q["sparse_col"] == np.float64(0.0)]
-    with pytest.raises(SchemaException):
-        lib.read(sym, query_builder=q)
-
-    # Projections
-    q = QueryBuilder().apply("projected_col", q["sparse_col"] + 1)
-    with pytest.raises(SchemaException):
-        lib.read(sym, query_builder=q)
-
-    # Groupbys
-    q = QueryBuilder().groupby("sparse_col").agg({"dense_col": "sum"})
-    with pytest.raises(SchemaException):
-        lib.read(sym, query_builder=q)
-
-    # Aggregations
-    q = QueryBuilder().groupby("dense_col").agg({"sparse_col": "sum"})
-    with pytest.raises(SchemaException):
-        lib.read(sym, query_builder=q)
-
-    # Date range
-    q = QueryBuilder().date_range((pd.Timestamp("2024-01-01"), pd.Timestamp("2024-01-02")))
-    with pytest.raises(SchemaException):
-        lib.read(sym, query_builder=q)
-
-    # Head
-    q = QueryBuilder()._head(2)
-    with pytest.raises(SchemaException):
-        lib.read(sym, query_builder=q)
-
-    # Tail
-    q = QueryBuilder()._tail(2)
-    with pytest.raises(SchemaException):
-        lib.read(sym, query_builder=q)
-
-    # Row range
-    q = QueryBuilder()._row_range((1, 2))
-    with pytest.raises(SchemaException):
-        lib.read(sym, query_builder=q)
diff --git a/python/tests/unit/arcticdb/version_store/test_query_builder_sparse.py b/python/tests/unit/arcticdb/version_store/test_query_builder_sparse.py
new file mode 100644
index 0000000000..079c691b2c
--- /dev/null
+++ b/python/tests/unit/arcticdb/version_store/test_query_builder_sparse.py
@@ -0,0 +1,209 @@
+from hypothesis import assume, given, settings, strategies
+from hypothesis.extra.pandas import columns, data_frames, range_indexes
+import numpy as np
+import pandas as pd
+import pytest
+
+from arcticdb.version_store.processing import QueryBuilder
+from arcticdb.util.test import assert_frame_equal
+from arcticdb.util.hypothesis import use_of_function_scoped_fixtures_in_hypothesis_checked
+
+
+class TestQueryBuilderSparse:
+    sym = "TestQueryBuilderSparse"
+    df = None
+
+    @pytest.fixture(autouse=True)
+    def write_test_data(self, lmdb_version_store):
+        lib = lmdb_version_store
+        df_0 = pd.DataFrame(
+            {
+                "sparse1": [1.0, np.nan, 2.0, np.nan],
+                "sparse2": [np.nan, 1.0, 2.0, np.nan],
+            },
+            index=pd.date_range("2024-01-01", periods=4, tz="UTC")
+        )
+        df_1 = pd.DataFrame(
+            {
+                "sparse1": [1.0, np.nan, 2.0, np.nan],
+                "sparse2": [np.nan, 1.0, 2.0, np.nan],
+            },
+            index=pd.date_range("2024-01-05", periods=4, tz="UTC")
+        )
+        # Use parallel write to generate 2 segments as append does not have the sparsify_floats kwarg
+        lib.write(self.sym, df_0, parallel=True, sparsify_floats=True)
+        lib.write(self.sym, df_1, parallel=True, sparsify_floats=True)
+        lib.compact_incomplete(self.sym, False, False, sparsify=True)
+        self.df = pd.concat([df_0, df_1])
+
+    def test_filter_isnull(self, lmdb_version_store):
+        expected = self.df[self.df["sparse1"].isnull()]
+        q = QueryBuilder()
+        q = q[q["sparse1"].isnull()]
+        received = lmdb_version_store.read(self.sym, query_builder=q).data
+        assert_frame_equal(expected, received)
+
+    def test_filter_notnull(self, lmdb_version_store):
+        expected = self.df[self.df["sparse1"].notnull()]
+        q = QueryBuilder()
+        q = q[q["sparse1"].notnull()]
+        received = lmdb_version_store.read(self.sym, query_builder=q).data
+        assert_frame_equal(expected, received)
+
+    def test_filter_col_equals_val(self, lmdb_version_store):
+        expected = self.df.query("sparse1 == 1")
+        q = QueryBuilder()
+        q = q[q["sparse1"] == 1]
+        received = lmdb_version_store.read(self.sym, query_builder=q).data
+        assert_frame_equal(expected, received)
+
+    def test_filter_col_not_equals_val(self, lmdb_version_store):
+        expected = self.df.query("sparse1 != 2")
+        q = QueryBuilder()
+        q = q[q["sparse1"] != 2]
+        received = lmdb_version_store.read(self.sym, query_builder=q).data
+        assert_frame_equal(expected, received)
+
+    def test_filter_col_isin_value_set(self, lmdb_version_store):
+        expected = self.df.query("sparse1 in [1]")
+        q = QueryBuilder()
+        q = q[q["sparse1"].isin([1])]
+        received = lmdb_version_store.read(self.sym, query_builder=q).data
+        assert_frame_equal(expected, received)
+
+    def test_filter_col_isnotin_value_set(self, lmdb_version_store):
+        expected = self.df.query("sparse1 not in [1]")
+        q = QueryBuilder()
+        q = q[q["sparse1"].isnotin([1])]
+        received = lmdb_version_store.read(self.sym, query_builder=q).data
+        assert_frame_equal(expected, received)
+
+    def test_filter_col_equals_col(self, lmdb_version_store):
+        expected = self.df.query("sparse1 == sparse2")
+        q = QueryBuilder()
+        q = q[q["sparse1"] == q["sparse2"]]
+        received = lmdb_version_store.read(self.sym, query_builder=q).data
+        assert_frame_equal(expected, received)
+
+    def test_filter_col_not_equals_col(self, lmdb_version_store):
+        expected = self.df.query("sparse1 != sparse2")
+        q = QueryBuilder()
+        q = q[q["sparse1"] != q["sparse2"]]
+        received = lmdb_version_store.read(self.sym, query_builder=q).data
+        assert_frame_equal(expected, received)
+
+    def test_project_minus_col(self, lmdb_version_store):
+        expected = self.df
+        expected["projected"] = -expected["sparse1"]
+        q = QueryBuilder()
+        q = q.apply("projected", -q["sparse1"])
+        received = lmdb_version_store.read(self.sym, query_builder=q).data
+        assert_frame_equal(expected, received)
+
+    def test_project_col_plus_val(self, lmdb_version_store):
+        expected = self.df
+        expected["projected"] = expected["sparse1"] + 1
+        q = QueryBuilder()
+        q = q.apply("projected", q["sparse1"] + 1)
+        received = lmdb_version_store.read(self.sym, query_builder=q).data
+        assert_frame_equal(expected, received)
+
+    def test_project_col_divided_by_col(self, lmdb_version_store):
+        expected = self.df
+        expected["projected"] = expected["sparse1"] / expected["sparse2"]
+        q = QueryBuilder()
+        q = q.apply("projected", q["sparse1"] / q["sparse2"])
+        received = lmdb_version_store.read(self.sym, query_builder=q).data
+        assert_frame_equal(expected, received)
+
+    def test_groupby(self, lmdb_version_store):
+        aggs = {
+            "sum": ("sparse2", "sum"),
+            "min": ("sparse2", "min"),
+            "max": ("sparse2", "max"),
+            "mean": ("sparse2", "mean"),
+            "count": ("sparse2", "count"),
+        }
+        expected = self.df.groupby("sparse1").agg(None, **aggs)
+        expected = expected.reindex(columns=sorted(expected.columns))
+        q = QueryBuilder()
+        q = q.groupby("sparse1").agg(aggs)
+        received = lmdb_version_store.read(self.sym, query_builder=q).data
+        received = received.reindex(columns=sorted(received.columns))
+        assert_frame_equal(expected, received, check_dtype=False)
+
+
+def test_query_builder_sparse_dynamic_schema_type_change(lmdb_version_store_dynamic_schema):
+    lib = lmdb_version_store_dynamic_schema
+    sym = "test_query_builder_sparse_dynamic_schema_type_change"
+    df_0 = pd.DataFrame(
+        {
+            "sparse1": [1.0, np.nan, 2.0, np.nan],
+            "sparse2": [np.nan, 1.0, 2.0, np.nan],
+        },
+        dtype=np.float64,
+        index=pd.date_range("2024-01-01", periods=4, tz="UTC")
+    )
+    df_1 = pd.DataFrame(
+        {
+            "sparse1": [1.0, np.nan, 2.0, np.nan],
+            "sparse2": [np.nan, 1.0, 2.0, np.nan],
+        },
+        dtype=np.float32,
+        index=pd.date_range("2024-01-05", periods=4, tz="UTC")
+    )
+    lib.write(sym, df_0, parallel=True, sparsify_floats=True)
+    lib.write(sym, df_1, parallel=True, sparsify_floats=True)
+    lib.compact_incomplete(sym, False, False, sparsify=True)
+
+    expected = pd.concat([df_0, df_1])
+    expected = expected[expected["sparse1"].isnull()]
+    q = QueryBuilder()
+    q = q[q["sparse1"].isnull()]
+    received = lib.read(sym, query_builder=q).data
+    assert_frame_equal(expected, received)
+
+
+@use_of_function_scoped_fixtures_in_hypothesis_checked
+@settings(deadline=None)
+@given(
+    df=data_frames(
+        columns(
+            ["sparse1", "sparse2"],
+            elements=strategies.floats(min_value=0, max_value=1000, allow_nan=False, allow_subnormal=False),
+            fill=strategies.just(np.nan)
+        ),
+    ),
+)
+def test_query_builder_sparse_hypothesis(lmdb_version_store, df):
+    assume(not df.empty and not df["sparse1"].isnull().all() and not df["sparse2"].isnull().all())
+    lib = lmdb_version_store
+    sym = "test_query_builder_sparse_hypothesis"
+
+    df.index = pd.date_range("2024-01-01", periods=len(df))
+
+    lib.write(sym, df, sparsify_floats=True)
+
+    # Filter
+    expected = df[df["sparse1"].isnull()]
+    q = QueryBuilder()
+    q = q[q["sparse1"].isnull()]
+    received = lib.read(sym, query_builder=q).data
+    assert_frame_equal(expected, received)
+
+    # Projection
+    expected = df
+    expected["projected"] = expected["sparse1"] + expected["sparse2"]
+    q = QueryBuilder()
+    q = q.apply("projected", q["sparse1"] + q["sparse2"])
+    received = lib.read(sym, query_builder=q).data
+    assert_frame_equal(expected, received)
+
+    # Groupby + aggregation
+    expected = df.groupby("sparse1").agg({"sparse2": "sum"})
+    q = QueryBuilder().groupby("sparse1").agg({"sparse2": "sum"})
+    received = lib.read(sym, query_builder=q).data
+    received.sort_index(inplace=True)
+    assert_frame_equal(expected, received)
+
+