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

Issue:
Appending empty type to non-empty column and vice versa is not working
### Describe the bug

When trying to append an empty segment to a column that already has data (or trying to append data to an empty segment) it fails due to type mismatch. Two things need to be done:

1. Make it possible to append empty segments to non-empty columns (and vice versa)
2. Perform filling of data. If appending empty column to column of int values fill the "empty" ones with zero, if appending to floats fill with NaN if appending to strings fill with None.

### Steps/Code to Reproduce

```python
from arcticdb import Arctic
from pandas.testing import assert_frame_equal

ac = Arctic("lmdb://test")
lib = ac.get_library("test", create_if_missing=True)
df_empty = pd.DataFrame({"col1": [None, None]})
lib.write("test_can_append_to_empty_column", df_empty)
df_non_empty = pd.DataFrame({"col1": [1, 2, 3]})
lib .append("test_can_append_to_empty_column", df_non_empty)
expected_result = pd.concat([df_empty, df_non_empty], ignore_index=True)
assert_frame_equal(lib .read("test_can_append_to_empty_column").data, expected_result)
```

### Expected Results

No error is thrown.

{"col1": [0,0,1,2,3]}

### OS, Python Version and ArcticDB Version

All

### Backend storage used

_No response_

### Additional Context

_No response_


Discussion / hints:
This needs to include truly empty dataframes (with zero records). Especially the case where the initial write is empty. This is a requirement for a client.
What needs to be done is to add a check just before the call to trivially_compatible_types where the error is thrown (which is protecting the call to decode_and_expand, so that if the column has a null type we just skip that column.

Then in the operator or ReduceColumnTask where it currently calls default_initialize_rows when the column is entirely missing, it also needs to do that when the column has a none type. 

It needs to be this way because this is the point where we know what the overall column type is, so that we can use the correct backfill value (i.e. floats should be backfilled with NaN etc)
I have now made a long list of issues related to empty and missing data. All of these need to be tackled as part of this issue. The list and an explanation is here 
https://github.com/man-group/arcticdb-bloomberg/blob/main/notes/testing/missing_empty_data_tests.ipynb


Fix patch:
diff --git a/cpp/arcticdb/CMakeLists.txt b/cpp/arcticdb/CMakeLists.txt
index fd415734a2..abb6d9eadf 100644
--- a/cpp/arcticdb/CMakeLists.txt
+++ b/cpp/arcticdb/CMakeLists.txt
@@ -157,7 +157,7 @@ ENDIF()
 
 set(CMAKE_INSTALL_RPATH "\\$ORIGIN/lib")
 
-if (NOT WIN32)
+if (NOT MSVC)
     if(NOT ${ARCTICDB_USING_CONDA})
         # Compilers on conda-forge are relatively strict.
         # For now, we do not want to treat warnings as errors for those builds.
@@ -166,6 +166,13 @@ if (NOT WIN32)
     add_compile_options("-ggdb")
     add_compile_options("-fvisibility=hidden")
     add_compile_options("-fno-omit-frame-pointer")
+else()
+    add_compile_options(
+        /WX
+        /wd4250 # Suppress inherits via dominance warning
+        /we4189 # Unused local variable is an error
+        /we4100 # Unused function parameter is an error
+    )
 endif ()
 
 ## Core library without python bindings ##
@@ -200,6 +207,7 @@ set(arcticdb_srcs
         entity/key.hpp
         entity/metrics.hpp
         entity/data_error.hpp
+        entity/descriptors.hpp
         entity/native_tensor.hpp
         entity/performance_tracing.hpp
         entity/protobuf_mappings.hpp
@@ -208,6 +216,7 @@ set(arcticdb_srcs
         entity/serialized_key.hpp
         entity/type_conversion.hpp
         entity/types.hpp
+        entity/type_utils.hpp
         entity/types_proto.hpp
         entity/types-inl.hpp
         entity/variant_key.hpp
@@ -236,7 +245,6 @@ set(arcticdb_srcs
         pipeline/read_pipeline.hpp
         pipeline/slicing.hpp
         pipeline/string_pool_utils.hpp
-        pipeline/string_pool_utils.hpp
         pipeline/value.hpp
         pipeline/value_set.hpp
         pipeline/write_frame.hpp
@@ -380,8 +388,10 @@ set(arcticdb_srcs
         entity/metrics.cpp
         entity/performance_tracing.cpp
         entity/types.cpp
+        entity/type_utils.cpp
         entity/types_proto.cpp
         log/log.cpp
+        pipeline/column_mapping.cpp
         pipeline/column_stats.cpp
         pipeline/frame_slice.cpp
         pipeline/frame_utils.cpp
diff --git a/cpp/arcticdb/codec/codec-inl.hpp b/cpp/arcticdb/codec/codec-inl.hpp
index a799def672..cb170cf6a8 100644
--- a/cpp/arcticdb/codec/codec-inl.hpp
+++ b/cpp/arcticdb/codec/codec-inl.hpp
@@ -18,10 +18,6 @@
 
 #include <arcticdb/util/bitset.hpp>
 
-#include <arcticdb/util/buffer.hpp>
-
-#include <google/protobuf/text_format.h>
-
 #include <type_traits>
 
 namespace arcticdb {
@@ -135,6 +131,7 @@ std::size_t decode_ndarray(
         }
 
         if(field.sparse_map_bytes()) {
+            util::check(!is_empty_type(type_desc_tag.data_type()), "Empty typed columns should not have sparse map");
             util::check_magic<util::BitMagicStart>(data_in);
             const auto bitmap_size = field.sparse_map_bytes() - util::combined_bit_magic_delimiters_size();
             bv = util::deserialize_bytes_to_bitmap(data_in, bitmap_size);
diff --git a/cpp/arcticdb/codec/codec.cpp b/cpp/arcticdb/codec/codec.cpp
index 4e3f9e835e..1aeebc689a 100644
--- a/cpp/arcticdb/codec/codec.cpp
+++ b/cpp/arcticdb/codec/codec.cpp
@@ -310,11 +310,12 @@ void decode_string_pool( const arcticdb::proto::encoding::SegmentHeader& hdr,
     }
 }
 
-void decode_v2(const Segment& segment,
-           arcticdb::proto::encoding::SegmentHeader& hdr,
-           SegmentInMemory& res,
-           const StreamDescriptor& desc)
-           {
+void decode_v2(
+    const Segment& segment,
+    arcticdb::proto::encoding::SegmentHeader& hdr,
+    SegmentInMemory& res,
+    const StreamDescriptor& desc
+) {
     ARCTICDB_SAMPLE(DecodeSegment, 0)
     const auto [begin, end] = get_segment_begin_end(segment, hdr);
     auto encoded_fields_ptr = end;
@@ -364,11 +365,12 @@ void decode_v2(const Segment& segment,
         res.set_compacted(segment.header().compacted());
     }}
 
-void decode_v1(const Segment& segment,
-            const arcticdb::proto::encoding::SegmentHeader& hdr,
-            SegmentInMemory& res,
-            StreamDescriptor::Proto& desc)
-{
+void decode_v1(
+    const Segment& segment,
+    const arcticdb::proto::encoding::SegmentHeader& hdr,
+    SegmentInMemory& res,
+    StreamDescriptor::Proto& desc
+) {
     ARCTICDB_SAMPLE(DecodeSegment, 0)
     const uint8_t* data = segment.buffer().data();
     util::check(data != nullptr, "Got null data ptr from segment");
@@ -384,17 +386,29 @@ void decode_v1(const Segment& segment,
         const auto seg_row_count = fields_size ? ssize_t(hdr.fields(0).ndarray().items_count()) : 0LL;
         res.init_column_map();
 
-        for (std::size_t i = 0; i < static_cast<size_t>(fields_size); ++i) {
-            const auto& field = hdr.fields(static_cast<int>(i));
-            const auto& field_name = desc.fields(static_cast<int>(i)).name();
-            util::check(data!=end, "Reached end of input block with {} fields to decode", fields_size-i);
-            if(auto col_index = res.column_index(field_name)) {
+        for (int i = 0; i < fields_size; ++i) {
+            const auto& field = hdr.fields(i);
+            const auto& field_name = desc.fields(i).name();
+            if (auto col_index = res.column_index(field_name)) {
                 auto& col = res.column(static_cast<position_t>(*col_index));
-                data += decode_field(res.field(*col_index).type(), field, data, col, col.opt_sparse_map(), to_encoding_version(hdr.encoding_version()));
-            } else
+                util::check(
+                    data != end || is_empty_type(col.type().data_type()),
+                    "Reached end of input block with {} fields to decode",
+                    fields_size - i
+                );
+                data += decode_field(
+                    res.field(*col_index).type(),
+                    field,
+                    data,
+                    col,
+                    col.opt_sparse_map(),
+                    to_encoding_version(hdr.encoding_version())
+                );
+            } else {
+                util::check(data != end, "Reached end of input block with {} fields to decode", fields_size - i);
                 data += encoding_sizes::field_compressed_size(field);
-
-            ARCTICDB_TRACE(log::codec(), "Decoded column {} to position {}", i, data-begin);
+            }
+            ARCTICDB_TRACE(log::codec(), "Decoded column {} to position {}", i, data - begin);
         }
 
         decode_string_pool(hdr, data, begin, end, res);
@@ -500,7 +514,8 @@ void encode_sparse_map(
     Buffer& out,
     std::ptrdiff_t& pos
 ) {
-    if (column_data.bit_vector() != nullptr && column_data.bit_vector()->count() > 0)   {
+    if (column_data.bit_vector() != nullptr && column_data.bit_vector()->count() > 0) {
+        util::check(!is_empty_type(column_data.type().data_type()), "Empty typed columns should not have sparse maps");
         ARCTICDB_DEBUG(log::codec(), "Sparse map count = {} pos = {}", column_data.bit_vector()->count(), pos);
         const size_t sparse_bm_bytes = encode_bitmap(*column_data.bit_vector(), out, pos);
         util::variant_match(variant_field, [&](auto field) {
diff --git a/cpp/arcticdb/codec/zstd.hpp b/cpp/arcticdb/codec/zstd.hpp
index 712c26543d..4f69ff07d2 100644
--- a/cpp/arcticdb/codec/zstd.hpp
+++ b/cpp/arcticdb/codec/zstd.hpp
@@ -51,28 +51,31 @@ struct ZstdBlockEncoder {
 
 struct ZstdDecoder {
 
-/*
- * encoder_version is here to support multiple versions but won't be used before we have them
- */
-#pragma GCC diagnostic push
-#pragma GCC diagnostic ignored "-Wunused-parameter"
-
-    template<typename T>
+    /// @param[in] encoder_version Used to support multiple versions but won't be used before we have them
+    template <typename T>
     static void decode_block(
-        std::uint32_t encoder_version,
-        const std::uint8_t *in,
+        [[maybe_unused]] std::uint32_t encoder_version,
+        const std::uint8_t* in,
         std::size_t in_bytes,
-        T *t_out,
-        std::size_t out_bytes) {
+        T* t_out,
+        std::size_t out_bytes
+    ) {
 
         const std::size_t decomp_size = ZSTD_getFrameContentSize(in, in_bytes);
-        codec::check<ErrorCode::E_DECODE_ERROR>(decomp_size == out_bytes, "expected out_bytes == zstd deduced bytes, actual {} != {}",
-                        out_bytes, decomp_size);
+        codec::check<ErrorCode::E_DECODE_ERROR>(
+            decomp_size == out_bytes,
+            "expected out_bytes == zstd deduced bytes, actual {} != {}",
+            out_bytes,
+            decomp_size
+        );
         std::size_t real_decomp = ZSTD_decompress(t_out, out_bytes, in, in_bytes);
-        codec::check<ErrorCode::E_DECODE_ERROR>(real_decomp == out_bytes, "expected out_bytes == zstd decompressed bytes, actual {} != {}",
-                        out_bytes, real_decomp);
+        codec::check<ErrorCode::E_DECODE_ERROR>(
+            real_decomp == out_bytes,
+            "expected out_bytes == zstd decompressed bytes, actual {} != {}",
+            out_bytes,
+            real_decomp
+        );
     }
-#pragma GCC diagnostic pop
 };
 
 } // namespace arcticdb::detail
diff --git a/cpp/arcticdb/column_store/column.cpp b/cpp/arcticdb/column_store/column.cpp
index 23ba7cf152..c8d991efc8 100644
--- a/cpp/arcticdb/column_store/column.cpp
+++ b/cpp/arcticdb/column_store/column.cpp
@@ -90,14 +90,14 @@ void Column::unsparsify(size_t num_rows) {
     if(!sparse_map_)
         return;
 
-    type_.visit_tag([that=this, num_rows] (const auto tdt) {
+    type_.visit_tag([this, num_rows] (const auto tdt) {
         using TagType = decltype(tdt);
         using RawType = typename TagType::DataTypeTag::raw_type;
         const auto dest_bytes = num_rows * sizeof(RawType);
         auto dest = ChunkedBuffer::presized(dest_bytes);
         util::default_initialize<TagType>(dest.data(), dest_bytes);
-        util::expand_dense_buffer_using_bitmap<RawType>(that->sparse_map_.value(), that->data_.buffer().data(), dest.data());
-        std::swap(dest, that->data_.buffer());
+        util::expand_dense_buffer_using_bitmap<RawType>(sparse_map_.value(), data_.buffer().data(), dest.data());
+        std::swap(dest, data_.buffer());
     });
     sparse_map_ = std::nullopt;
     last_logical_row_ = last_physical_row_ = static_cast<ssize_t>(num_rows) - 1;
@@ -106,14 +106,13 @@ void Column::unsparsify(size_t num_rows) {
 }
 
 void Column::sparsify() {
-    type().visit_tag([that=this](auto type_desc_tag) {
-        using TDT = decltype(type_desc_tag);
-        using RawType = typename TDT::DataTypeTag::raw_type;
-        if constexpr(is_floating_point_type(TDT::DataTypeTag::data_type)) {
-            auto raw_ptr =reinterpret_cast<const RawType*>(that->ptr());
-            auto buffer = util::scan_floating_point_to_sparse(raw_ptr, that->row_count(), that->sparse_map());
-            std::swap(that->data().buffer(), buffer);
-            that->last_physical_row_ = that->sparse_map().count() - 1;
+    type().visit_tag([this](auto type_desc_tag) {
+        using RawType = typename decltype(type_desc_tag)::DataTypeTag::raw_type;
+        if constexpr (is_floating_point_type(type_desc_tag.data_type())) {
+            auto raw_ptr = reinterpret_cast<const RawType*>(ptr());
+            auto buffer = util::scan_floating_point_to_sparse(raw_ptr, row_count(), sparse_map());
+            std::swap(data().buffer(), buffer);
+            last_physical_row_ = sparse_map().count() - 1;
         }
     });
 }
@@ -324,6 +323,9 @@ void Column::default_initialize_rows(size_t start_pos, size_t num_rows, bool ens
 }
 
 void Column::set_row_data(size_t row_id) {
+    if (is_empty_type(type_.data_type())) {
+        return;
+    }
     last_logical_row_ = row_id;
     const auto last_stored_row = row_count() - 1;
     if(sparse_map_) {
@@ -513,66 +515,87 @@ std::vector<std::shared_ptr<Column>> Column::split(const std::shared_ptr<Column>
     return output;
 }
 
-// Produces a new column
-// Inclusive of start_row, exclusive of end_row
-std::shared_ptr<Column> Column::truncate(const std::shared_ptr<Column>& column, size_t start_row, size_t end_row) {
-    auto type_size = get_type_size(column->type().data_type());
-    // Bytes from the underlying chunked buffer to include. Inclusive of start_byte, exclusive of end_byte
-    size_t start_byte;
-    size_t end_byte;
-    util::BitMagic input_sparse_map;
-    if (column->is_sparse()) {
-        input_sparse_map = column->sparse_map();
-        internal::check<ErrorCode::E_ASSERTION_FAILURE>(input_sparse_map.size() > 0, "Unexpected empty sparse map in Column::truncate");
-        // Sparse columns do not include trailing 0s in the bitset, so the relevant end_row is capped at the size of the biset
+/// Bytes from the underlying chunked buffer to include when truncating. Inclusive of start_byte, exclusive of end_byte
+[[nodiscard]] static std::pair<size_t, size_t> column_start_end_bytes(
+    const Column& column,
+    size_t start_row,
+    size_t end_row
+) {
+    const size_t type_size = get_type_size(column.type().data_type());
+    size_t start_byte = start_row * type_size;
+    size_t end_byte = end_row * type_size;
+    if (column.is_sparse()) {
+        const util::BitMagic& input_sparse_map = column.sparse_map();
+        internal::check<ErrorCode::E_ASSERTION_FAILURE>(
+            input_sparse_map.size() > 0,
+            "Unexpected empty sparse map in Column::truncate"
+        );
+        // Sparse columns do not include trailing 0s in the bitset, so the relevant end_row is capped at the size of the
+        // biset
         end_row = std::min(end_row, static_cast<size_t>(input_sparse_map.size()));
         // count_range is inclusive at both ends
-        auto set_bits_before_range = start_row > 0 ? input_sparse_map.count_range(0, start_row - 1) : 0;
-        auto set_bits_in_range = input_sparse_map.count_range(start_row, end_row - 1);
+        const size_t set_bits_before_range = start_row > 0 ? input_sparse_map.count_range(0, start_row - 1) : 0;
+        const size_t set_bits_in_range = input_sparse_map.count_range(start_row, end_row - 1);
         start_byte = set_bits_before_range * type_size;
         end_byte = start_byte + (set_bits_in_range * type_size);
-    } else {
-        start_byte = start_row * type_size;
-        end_byte = end_row * type_size;
     }
     internal::check<ErrorCode::E_ASSERTION_FAILURE>(
-            start_byte % type_size == 0,
-            "start_byte {} is not a multiple of type size {}", start_byte, column->type());
+        start_byte % type_size == 0,
+        "start_byte {} is not a multiple of type size {}",
+        start_byte,
+        column.type()
+    );
     internal::check<ErrorCode::E_ASSERTION_FAILURE>(
-            end_byte % type_size == 0,
-            "start_byte {} is not a multiple of type size {}", end_byte, column->type());
+        end_byte % type_size == 0,
+        "start_byte {} is not a multiple of type size {}",
+        end_byte,
+        column.type()
+    );
+    return {start_byte, end_byte};
+}
+
+[[nodiscard]] static util::BitMagic truncate_sparse_map(
+    const util::BitMagic& input_sparse_map,
+    size_t start_row,
+    size_t end_row
+) {
+    // The output sparse map is the slice [start_row, end_row) of the input sparse map
+    // BitMagic doesn't have a method for this, so hand-roll it here
+    // Ctor parameter is the size
+    util::BitMagic output_sparse_map(end_row - start_row);
+    util::BitSet::bulk_insert_iterator inserter(output_sparse_map);
+    util::BitSetSizeType set_input_bit;
+    if (start_row == 0) {
+        // get_first can return 0 if no bits are set, but we checked earlier that input_sparse_map.size() > 0,
+        // and we do not have sparse maps with no bits set (except for the empty type)
+        set_input_bit = input_sparse_map.get_first();
+        if (set_input_bit < end_row) {
+            inserter = set_input_bit;
+        }
+    } else {
+        set_input_bit = input_sparse_map.get_next(start_row - 1);
+        // get_next returns 0 if no more bits are set
+        if (set_input_bit != 0 && set_input_bit < end_row) {
+            // Shift start_row elements to the left
+            inserter = set_input_bit - start_row;
+        }
+    }
+    do {
+        set_input_bit = input_sparse_map.get_next(set_input_bit);
+        if (set_input_bit != 0 && set_input_bit < end_row) {
+            inserter = set_input_bit - start_row;
+        }
+    } while (set_input_bit != 0);
+    inserter.flush();
+    return output_sparse_map;
+}
+
+std::shared_ptr<Column> Column::truncate(const std::shared_ptr<Column>& column, size_t start_row, size_t end_row) {
+    const auto [start_byte, end_byte] = column_start_end_bytes(*column, start_row, end_row);
     auto buffer = ::arcticdb::truncate(column->data_.buffer(), start_byte, end_byte);
     auto res = std::make_shared<Column>(column->type(), column->allow_sparse_, std::move(buffer));
     if (column->is_sparse()) {
-        // The output sparse map is the slice [start_row, end_row) of the input sparse map
-        // BitMagic doesn't have a method for this, so hand-roll it here
-        // Ctor parameter is the size
-        util::BitMagic output_sparse_map(end_row - start_row);
-        util::BitSet::bulk_insert_iterator inserter(output_sparse_map);
-        util::BitSetSizeType set_input_bit;
-        if (start_row == 0) {
-            // get_first can return 0 if no bits are set, but we checked earlier that input_sparse_map.size() > 0,
-            // and we do not have sparse maps with no bits set
-            set_input_bit = input_sparse_map.get_first();
-            if (set_input_bit < end_row) {
-                inserter = set_input_bit;
-            }
-        } else {
-            set_input_bit = input_sparse_map.get_next(start_row - 1);
-            // get_next returns 0 if no more bits are set
-            if (set_input_bit != 0 && set_input_bit < end_row) {
-                // Shift start_row elements to the left
-                inserter = set_input_bit - start_row;
-            }
-        }
-        do {
-            set_input_bit = input_sparse_map.get_next(set_input_bit);
-            if (set_input_bit != 0 && set_input_bit < end_row) {
-                inserter = set_input_bit - start_row;
-            }
-        } while (set_input_bit != 0);
-        inserter.flush();
-        res->set_sparse_map(std::move(output_sparse_map));
+        res->set_sparse_map(truncate_sparse_map(column->sparse_map(), start_row, end_row));
     }
     res->set_row_data(end_row - (start_row + 1));
     return res;
@@ -675,7 +698,7 @@ void Column::regenerate_offsets() const {
     }
 }
 
-[[nodiscard]] util::BitMagic& Column::sparse_map() {
+util::BitMagic& Column::sparse_map() {
     if(!sparse_map_)
         sparse_map_ = std::make_optional<util::BitMagic>(0);
 
@@ -687,4 +710,8 @@ const util::BitMagic& Column::sparse_map() const {
     return sparse_map_.value();
 }
 
+std::optional<util::BitMagic>& Column::opt_sparse_map() {
+    return sparse_map_;
+}
+
 } //namespace arcticdb
diff --git a/cpp/arcticdb/column_store/column.hpp b/cpp/arcticdb/column_store/column.hpp
index 2d94fac56e..b5698318d8 100644
--- a/cpp/arcticdb/column_store/column.hpp
+++ b/cpp/arcticdb/column_store/column.hpp
@@ -222,7 +222,7 @@ class Column {
             // it pointers to an array type (at numpy arrays at the moment). When we allocate a column for an array we
             // need to allocate space for one pointer per row. This also affects how we handle arrays to python as well.
             // Check cpp/arcticdb/column_store/column_utils.hpp::array_at and cpp/arcticdb/column_store/column.hpp::Column
-            data_(expected_rows * (type.dimension() == Dimension::Dim0 ? get_type_size(type.data_type()) : sizeof(void*)), presize),
+            data_(expected_rows * entity::sizeof_datatype(type), presize),
             type_(type),
             allow_sparse_(allow_sparse){
         ARCTICDB_TRACE(log::inmem(), "Creating column with descriptor {}", type);
@@ -244,9 +244,9 @@ class Column {
         sparse_map().set_range(0, bv_size(to_row), true);
     }
 
-    std::optional<util::BitMagic>& opt_sparse_map() {
-        return sparse_map_;
-    }
+    [[nodiscard]] util::BitMagic& sparse_map();
+    [[nodiscard]] const util::BitMagic& sparse_map() const;
+    [[nodiscard]] std::optional<util::BitMagic>& opt_sparse_map();
 
     template<typename TagType>
     auto begin() const {
@@ -651,11 +651,27 @@ class Column {
         return *res;
     }
 
+    [[nodiscard]] static std::vector<std::shared_ptr<Column>> split(
+        const std::shared_ptr<Column>& column,
+        size_t num_rows
+    );
+    /// @brief Produces a new column containing only the data in range [start_row, end_row)
+    /// @param[in] start_row Inclusive start of the row range
+    /// @param[in] end_row Exclusive end of the row range
+    [[nodiscard]] static std::shared_ptr<Column> truncate(
+        const std::shared_ptr<Column>& column,
+        size_t start_row,
+        size_t end_row
+    );
+
     /*
      * !!!!!!!!!!----------IMPORTANT----------!!!!!!!!!!
      * When adding new uses of these static methods, add an internal::check<...>(f.heapAllocatedMemory() == 0,...)
      * and run all of the tests in CI on all supported platforms. We do not want these checks in the released code,
      * but also do not want any passed in lambdas to be heap allocated.
+     *
+     * TODO: Use type-deduced functor (with enable_if to verify it's in/out types) in all functions similar to transform.
+     *   In case C++20 is available use concepts.
      */
 
     template<typename input_tdt>
@@ -667,13 +683,21 @@ class Column {
         });
     }
 
-    template<typename input_tdt, typename output_tdt>
+    template <
+        typename input_tdt,
+        typename output_tdt,
+        typename FunctorT,
+        typename = std::enable_if_t<std::is_invocable_r_v<
+            typename output_tdt::DataTypeTag::raw_type,
+            FunctorT,
+            typename input_tdt::DataTypeTag::raw_type>>>
     static void transform(const Column& input_column,
                           Column& output_column,
-                          folly::Function<typename output_tdt::DataTypeTag::raw_type(typename input_tdt::DataTypeTag::raw_type)>&& f) {
+                          FunctorT&& f
+    ) {
         auto input_data = input_column.data();
         auto output_data = output_column.data();
-        std::transform(input_data.cbegin<input_tdt>(), input_data.cend<input_tdt>(), output_data.begin<output_tdt>(), std::move(f));
+        std::transform(input_data.cbegin<input_tdt>(), input_data.cend<input_tdt>(), output_data.begin<output_tdt>(), std::forward<FunctorT>(f));
     }
 
     template<typename left_input_tdt, typename right_input_tdt, typename output_tdt>
@@ -726,10 +750,6 @@ class Column {
 
     // end IMPORTANT
 
-    static std::vector<std::shared_ptr<Column>> split(const std::shared_ptr<Column>& column, size_t num_rows);
-
-    static std::shared_ptr<Column> truncate(const std::shared_ptr<Column>& column, size_t start_row, size_t end_row);
-
 private:
 
     position_t last_offset() const;
@@ -748,9 +768,6 @@ class Column {
     // Permutes the physical column storage based on the given sorted_pos.
     void physical_sort_external(std::vector<uint32_t> &&sorted_pos);
 
-    [[nodiscard]] util::BitMagic& sparse_map();
-    const util::BitMagic& sparse_map() const;
-
     // Members
     CursoredBuffer<ChunkedBuffer> data_;
     CursoredBuffer<Buffer> shapes_;
diff --git a/cpp/arcticdb/column_store/column_utils.hpp b/cpp/arcticdb/column_store/column_utils.hpp
index 71fc1192a9..90a35a1048 100644
--- a/cpp/arcticdb/column_store/column_utils.hpp
+++ b/cpp/arcticdb/column_store/column_utils.hpp
@@ -7,7 +7,6 @@
 
 #pragma once
 
-#include <arcticdb/column_store/column.hpp>
 #include <arcticdb/column_store/memory_segment.hpp>
 #include <arcticdb/pipeline/frame_data_wrapper.hpp>
 
@@ -44,7 +43,7 @@ inline py::array array_at(const SegmentInMemory& frame, std::size_t col_pos, py:
                 } else {
                     dtype = fmt::format("{}{:d}", get_dtype_specifier(data_type), esize);
                 }
-            } else if constexpr (is_empty_type(data_type) || is_py_bool_type(data_type) || is_numpy_array(TypeDescriptor(tag))) {
+            } else if constexpr (is_empty_type(data_type) || is_bool_object_type(data_type) || is_numpy_array(TypeDescriptor(tag))) {
                 dtype= "O";
                 // The python representation of multidimensional columns differs from the in-memory/on-storage. In memory,
                 // we hold all scalars in a contiguous buffer with the shapes buffer telling us how many elements are there
@@ -53,9 +52,7 @@ inline py::array array_at(const SegmentInMemory& frame, std::size_t col_pos, py:
                 // (numpy) array, thus the size of the element is not the size of the raw type, but the size of a pointer.
                 // This also affects how we allocate columns. Check cpp/arcticdb/column_store/column.hpp::Column and
                 // cpp/arcticdb/pipeline/column_mapping.hpp::sizeof_datatype
-                if constexpr(tag.dimension() > Dimension::Dim0) {
-                    esize = sizeof(PyObject*);
-                }
+                esize = sizeof_datatype(TypeDescriptor{tag});
             } else if constexpr(tag.dimension() == Dimension::Dim2) {
                 util::raise_rte("Read resulted in two dimensional type. This is not supported.");
             } else {
@@ -96,7 +93,7 @@ inline py::array array_at(const SegmentInMemory& frame, std::size_t col_pos, py:
             } else {
                 dtype = fmt::format("{}{:d}", get_dtype_specifier(data_type), esize);
             }
-        } else if constexpr (is_empty_type(data_type) || is_py_bool_type(data_type) || is_numpy_array(TypeDescriptor(tag))){
+        } else if constexpr (is_empty_type(data_type) || is_bool_object_type(data_type) || is_numpy_array(TypeDescriptor(tag))){
             dtype = "O";
             // The python representation of multidimensional columns differs from the in-memory/on-storage. In memory,
             // we hold all scalars in a contiguous buffer with the shapes buffer telling us how many elements are there
@@ -105,9 +102,7 @@ inline py::array array_at(const SegmentInMemory& frame, std::size_t col_pos, py:
             // (numpy) array, thus the size of the element is not the size of the raw type, but the size of a pointer.
             // This also affects how we allocate columns. Check cpp/arcticdb/column_store/column.hpp::Column and
             // cpp/arcticdb/pipeline/column_mapping.hpp::sizeof_datatype
-            if constexpr(tag.dimension() > Dimension::Dim0) {
-                esize = sizeof(PyObject*);
-            }
+            esize = sizeof_datatype(TypeDescriptor{tag});
         } else if constexpr(tag.dimension() == Dimension::Dim2) {
             util::raise_rte("Read resulted in two dimensional type. This is not supported.");
         } else {
diff --git a/cpp/arcticdb/column_store/memory_segment.hpp b/cpp/arcticdb/column_store/memory_segment.hpp
index ab34b6f5d6..561feaee0f 100644
--- a/cpp/arcticdb/column_store/memory_segment.hpp
+++ b/cpp/arcticdb/column_store/memory_segment.hpp
@@ -391,8 +391,9 @@ class SegmentInMemory {
         return SegmentInMemory(impl_->filter(filter_bitset, filter_down_stringpool, validate));
     }
 
-    SegmentInMemory truncate(size_t start_row, size_t end_row) const{
-        return SegmentInMemory(impl_->truncate(start_row, end_row));
+    /// @see SegmentInMemoryImpl::truncate
+    SegmentInMemory truncate(size_t start_row, size_t end_row, bool reconstruct_string_pool) const{
+        return SegmentInMemory(impl_->truncate(start_row, end_row, reconstruct_string_pool));
     }
 
     std::vector<SegmentInMemory> partition(const std::vector<std::optional<uint8_t>>& row_to_segment,
diff --git a/cpp/arcticdb/column_store/memory_segment_impl.cpp b/cpp/arcticdb/column_store/memory_segment_impl.cpp
index 13a4542628..4f03653664 100644
--- a/cpp/arcticdb/column_store/memory_segment_impl.cpp
+++ b/cpp/arcticdb/column_store/memory_segment_impl.cpp
@@ -8,6 +8,7 @@
 #include <arcticdb/column_store/memory_segment_impl.hpp>
 #include <arcticdb/column_store/string_pool.hpp>
 #include <arcticdb/entity/type_utils.hpp>
+#include <arcticdb/pipeline/string_pool_utils.hpp>
 
 #include <google/protobuf/any.h>
 #include <google/protobuf/any.pb.h>
@@ -436,8 +437,11 @@ bool operator==(const SegmentInMemoryImpl& left, const SegmentInMemoryImpl& righ
     return true;
 }
 
-// Inclusive of start_row, exclusive of end_row
-std::shared_ptr<SegmentInMemoryImpl> SegmentInMemoryImpl::truncate(size_t start_row, size_t end_row) const {
+std::shared_ptr<SegmentInMemoryImpl> SegmentInMemoryImpl::truncate(
+    size_t start_row,
+    size_t end_row,
+    bool reconstruct_string_pool
+) const {
     auto num_values = end_row - start_row;
     internal::check<ErrorCode::E_ASSERTION_FAILURE>(
             is_sparse() || (start_row < row_count() && end_row <= row_count()),
@@ -446,24 +450,43 @@ std::shared_ptr<SegmentInMemoryImpl> SegmentInMemoryImpl::truncate(size_t start_
     auto output = std::make_shared<SegmentInMemoryImpl>();
 
     output->set_row_data(num_values - 1);
-    output->set_string_pool(string_pool_);
     output->set_compacted(compacted_);
+    if (!reconstruct_string_pool) {
+        output->set_string_pool(string_pool_);
+    }
     if (metadata_) {
         google::protobuf::Any metadata;
         metadata.CopyFrom(*metadata_);
         output->set_metadata(std::move(metadata));
     }
 
-    for(const auto&& [idx, column] : folly::enumerate(columns_)) {
-        auto truncated_column = Column::truncate(column, start_row, end_row);
-        output->add_column(descriptor_->field(idx), truncated_column);
+    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)>(
+                    *truncated_column,
+                    *truncated_column,
+                    [this, &output](auto string_pool_offset) -> arcticdb::OffsetString::offset_t {
+                        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();
+                        }
+                        return string_pool_offset;
+                    }
+                );
+            }
+        });
+        output->add_column(field, std::move(truncated_column));
     }
     output->attach_descriptor(descriptor_);
     return output;
 }
 
-// Combine 2 segments that hold different columns associated with the same rows
-// If unique_column_names is true, any columns from other with names matching those in this are ignored
+/// @brief Combine 2 segments that hold different columns associated with the same rows
+/// @param[in] unique_column_names If true, any columns from other with names matching those in this are ignored
 void SegmentInMemoryImpl::concatenate(SegmentInMemoryImpl&& other, bool unique_column_names) {
         internal::check<ErrorCode::E_INVALID_ARGUMENT>(
                 row_count() == other.row_count(),
diff --git a/cpp/arcticdb/column_store/memory_segment_impl.hpp b/cpp/arcticdb/column_store/memory_segment_impl.hpp
index 0c30e8e8ef..b09b726d56 100644
--- a/cpp/arcticdb/column_store/memory_segment_impl.hpp
+++ b/cpp/arcticdb/column_store/memory_segment_impl.hpp
@@ -752,8 +752,8 @@ class SegmentInMemoryImpl {
 
     SegmentInMemoryImpl clone() const;
 
-    void set_string_pool(const std::shared_ptr<StringPool>& string_pool) {
-        string_pool_ = string_pool;
+    void set_string_pool(std::shared_ptr<StringPool> string_pool) {
+        string_pool_ = std::move(string_pool);
     }
 
     std::shared_ptr<SegmentInMemoryImpl> get_output_segment(size_t num_values, bool pre_allocate=true) const;
@@ -778,8 +778,18 @@ class SegmentInMemoryImpl {
 
     void set_timeseries_descriptor(TimeseriesDescriptor&& tsd);
 
-    // Inclusive of start_row, exclusive of end_row
-    std::shared_ptr<SegmentInMemoryImpl> truncate(size_t start_row, size_t end_row) const;
+    /// @brief Construct a copy of the segment containing only rows in [start_row; end_row)
+    /// @param start_row Start of the row range (inclusive)
+    /// @param end_Row End of the row range (exclusive)
+    /// @param reconstruct_string_pool When truncating some of the strings values of the original
+    ///  segment might not be referenced in the resulting segment. In this case, reconstructing the
+    ///  string pool will save some memory. Note that reconstructing the string pool is an expensive
+    ///  operation and should be avoided if possible.
+    std::shared_ptr<SegmentInMemoryImpl> truncate(
+        size_t start_row,
+        size_t end_row,
+        bool reconstruct_string_pool
+    ) const;
 
     // Partitions the segment into n new segments. Each row in the starting segment is mapped to one of the output segments
     // by the row_to_segment vector (std::nullopt means the row is not included in any output segment).
diff --git a/cpp/arcticdb/entity/descriptors.hpp b/cpp/arcticdb/entity/descriptors.hpp
new file mode 100644
index 0000000000..87ffaa041f
--- /dev/null
+++ b/cpp/arcticdb/entity/descriptors.hpp
@@ -0,0 +1,14 @@
+/* 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 <descriptors.pb.h>
+namespace arcticdb::proto {
+    namespace descriptors = arcticc::pb2::descriptors_pb2;
+}
\ No newline at end of file
diff --git a/cpp/arcticdb/entity/frame_and_descriptor.hpp b/cpp/arcticdb/entity/frame_and_descriptor.hpp
index 15f520fcb2..4a6a19a3f3 100644
--- a/cpp/arcticdb/entity/frame_and_descriptor.hpp
+++ b/cpp/arcticdb/entity/frame_and_descriptor.hpp
@@ -8,7 +8,6 @@
 #pragma once
 
 #include <arcticdb/column_store/memory_segment.hpp>
-#include <arcticdb/entity/protobufs.hpp>
 #include <arcticdb/util/buffer_holder.hpp>
 
 namespace arcticdb {
diff --git a/cpp/arcticdb/entity/merge_descriptors.cpp b/cpp/arcticdb/entity/merge_descriptors.cpp
index 0be36d5c74..ed8ea636cf 100644
--- a/cpp/arcticdb/entity/merge_descriptors.cpp
+++ b/cpp/arcticdb/entity/merge_descriptors.cpp
@@ -32,11 +32,11 @@ StreamDescriptor merge_descriptors(
             util::variant_match(temp_idx, [&merged_fields, &merged_fields_map] (const auto& idx) {
                 using IndexType = std::decay_t<decltype(idx)>;
                 merged_fields.emplace_back(idx.name());
-                merged_fields_map.try_emplace(idx.name(), static_cast<TypeDescriptor>(typename IndexType::TypeDescTag{}));
+                merged_fields_map.try_emplace(idx.name(), TypeDescriptor{typename IndexType::TypeDescTag{}});
             });
-        }
-        else
+        } else {
             util::raise_rte("Descriptor has uninitialized index and no default supplied");
+        }
     } else {
         index = index_type_from_descriptor(original);
     }
diff --git a/cpp/arcticdb/entity/timeseries_descriptor.hpp b/cpp/arcticdb/entity/timeseries_descriptor.hpp
index 806051f129..3802fa831d 100644
--- a/cpp/arcticdb/entity/timeseries_descriptor.hpp
+++ b/cpp/arcticdb/entity/timeseries_descriptor.hpp
@@ -7,7 +7,6 @@
 
 #pragma once
 
-#include <proto/arcticc/pb2/proto/descriptors.pb.h>
 #include <arcticdb/entity/field_collection.hpp>
 #include <arcticdb/entity/stream_descriptor.hpp>
 
diff --git a/cpp/arcticdb/entity/type_utils.cpp b/cpp/arcticdb/entity/type_utils.cpp
new file mode 100644
index 0000000000..33b0d0a6b9
--- /dev/null
+++ b/cpp/arcticdb/entity/type_utils.cpp
@@ -0,0 +1,161 @@
+/* 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 <arcticdb/entity/type_utils.hpp>
+#include <arcticdb/entity/types.hpp>
+#include <arcticdb/entity/types_proto.hpp>
+
+namespace arcticdb {
+    bool trivially_compatible_types(const entity::TypeDescriptor& left, const entity::TypeDescriptor& right) {
+        if (left == right)
+            return true;
+
+        // Multidimensional types are pointers
+        if (left.dimension() >= entity::Dimension::Dim1 && right.dimension() >= entity::Dimension::Dim1)
+            return true;
+
+        // Multidimensional types are pointers the empty type is pointer as well
+        if (left.dimension() >= entity::Dimension::Dim1 && is_empty_type(right.data_type()))
+            return true;
+
+        // Multidimensional types are pointers the empty type is pointer as well
+        if (right.dimension() >= entity::Dimension::Dim1 && is_empty_type(left.data_type()))
+            return true;
+
+        if (is_sequence_type(left.data_type()) && is_sequence_type(right.data_type())) {
+            // TODO coercion of utf strings is not always safe, should allow safe conversion and reinstate the
+            // stronger requirement for trivial conversion below.
+            //        if(!is_utf_type(slice_value_type(left.data_type)) &&
+            //        !is_utf_type(slice_value_type(right.data_type)))
+            //            return true;
+
+            return is_utf_type(slice_value_type(left.data_type())) == is_utf_type(slice_value_type(right.data_type()));
+        }
+
+        return false;
+    }
+
+    std::optional<entity::TypeDescriptor> has_valid_type_promotion(
+        const entity::TypeDescriptor& source,
+        const entity::TypeDescriptor& target
+    ) {
+
+        if (source.dimension() != target.dimension()) {
+            // Empty of dimension 0 means lack of any given type and can be promoted to anything (even if the dimensions
+            // don't match), e.g. empty type can become int or array of ints. Empty type of higher dimension is used to
+            // specify an empty array or an empty matrix, thus it cannot become any other type unless the dimensionality
+            // matches
+            if (is_empty_type(source.data_type()) && source.dimension() == entity::Dimension::Dim0)
+                return target;
+            return std::nullopt;
+        }
+
+        if (source == target)
+            return target;
+
+        // Empty type is coercible to any type
+        if (is_empty_type(source.data_type())) {
+            return target;
+        }
+
+        // Nothing is coercible to the empty type.
+        if (is_empty_type(target.data_type())) {
+            return std::nullopt;
+        }
+
+        auto source_type = source.data_type();
+        auto target_type = target.data_type();
+        auto source_size = slice_bit_size(source_type);
+        auto target_size = slice_bit_size(target_type);
+
+        if (is_time_type(source_type)) {
+            if (!is_time_type(target_type))
+                return std::nullopt;
+        } else if (is_unsigned_type(source_type)) {
+            if (is_unsigned_type(target_type)) {
+                // UINT->UINT, target_size must be >= source_size
+                if (source_size > target_size)
+                    return std::nullopt;
+            } else if (is_signed_type(target_type)) {
+                // UINT->INT, target_size must be > source_size
+                if (source_size >= target_size)
+                    return std::nullopt;
+            } else if (is_floating_point_type(target_type)) {
+                // UINT->FLOAT, no restrictions on relative sizes
+            } else {
+                // Non-numeric target type
+                return std::nullopt;
+            }
+        } else if (is_signed_type(source_type)) {
+            if (is_unsigned_type(target_type)) {
+                // INT->UINT never promotable
+                return std::nullopt;
+            } else if (is_signed_type(target_type)) {
+                // INT->INT, target_size must be >= source_size
+                if (source_size > target_size)
+                    return std::nullopt;
+            } else if (is_floating_point_type(target_type)) {
+                // INT->FLOAT, no restrictions on relative sizes
+            } else {
+                // Non-numeric target type
+                return std::nullopt;
+            }
+        } else if (is_floating_point_type(source_type)) {
+            if (is_unsigned_type(target_type) || is_signed_type(target_type)) {
+                // FLOAT->U/INT never promotable
+                return std::nullopt;
+            } else if (is_floating_point_type(target_type)) {
+                // FLOAT->FLOAT, target_size must be >= source_size
+                if (source_size > target_size)
+                    return std::nullopt;
+            } else {
+                // Non-numeric target type
+                return std::nullopt;
+            }
+        } else if (is_sequence_type(source_type) && is_sequence_type(target_type)) {
+            // Only allow promotion with UTF strings, and only to dynamic (never to fixed width)
+            if (!is_utf_type(source_type) || !is_utf_type(target_type) || !is_dynamic_string_type(target_type))
+                return std::nullopt;
+        } else if (is_bool_object_type(source_type)) {
+            return std::nullopt;
+        } else {
+            // Non-numeric source type
+            return std::nullopt;
+        }
+
+        return entity::TypeDescriptor{
+            combine_data_type(slice_value_type(target_type), target_size),
+            target.dimension()};
+    }
+
+    std::optional<entity::TypeDescriptor> has_valid_type_promotion(
+        const proto::descriptors::TypeDescriptor& source,
+        const proto::descriptors::TypeDescriptor& target
+    ) {
+        return has_valid_type_promotion(entity::type_desc_from_proto(source), entity::type_desc_from_proto(target));
+    }
+
+    std::optional<entity::TypeDescriptor> has_valid_common_type(
+        const entity::TypeDescriptor& left,
+        const entity::TypeDescriptor& right
+    ) {
+        auto maybe_common_type = has_valid_type_promotion(left, right);
+        if (!maybe_common_type) {
+            maybe_common_type = has_valid_type_promotion(right, left);
+        }
+        return maybe_common_type;
+    }
+
+    std::optional<entity::TypeDescriptor> has_valid_common_type(
+        const proto::descriptors::TypeDescriptor& left,
+        const proto::descriptors::TypeDescriptor& right
+    ) {
+        return has_valid_common_type(entity::type_desc_from_proto(left), entity::type_desc_from_proto(right));
+    }
+
+}
\ No newline at end of file
diff --git a/cpp/arcticdb/entity/type_utils.hpp b/cpp/arcticdb/entity/type_utils.hpp
index 13804582f7..4ecc0f2606 100644
--- a/cpp/arcticdb/entity/type_utils.hpp
+++ b/cpp/arcticdb/entity/type_utils.hpp
@@ -6,146 +6,36 @@
  */
 
 #pragma once
-
-#include <arcticdb/entity/types.hpp>
+#include<optional>
+#include<arcticdb/entity/descriptors.hpp>
 
 namespace arcticdb {
 
-inline bool trivially_compatible_types(entity::TypeDescriptor left, entity::TypeDescriptor right) {
-    if(left == right)
-        return true;
-
-    // Multidimensional types are pointers
-    if(left.dimension() >= entity::Dimension::Dim1 && right.dimension() >= entity::Dimension::Dim1)
-        return true;
-
-    // Multidimensional types are pointer the empty type is pointer as well
-    if(left.dimension() >= entity::Dimension::Dim1 && is_empty_type(right.data_type()))
-        return true;
-    
-    // Multidimensional types are pointer the empty type is pointer as well
-    if(right.dimension() >= entity::Dimension::Dim1 && is_empty_type(left.data_type()))
-        return true;
-
-    if(is_sequence_type(left.data_type()) && is_sequence_type(right.data_type())) {
-        //TODO coercion of utf strings is not always safe, should allow safe conversion and reinstate the
-        //stronger requirement for trivial conversion below.
-//        if(!is_utf_type(slice_value_type(left.data_type)) && !is_utf_type(slice_value_type(right.data_type)))
-//            return true;
-
-        return is_utf_type(slice_value_type(left.data_type()))  == is_utf_type(slice_value_type(right.data_type()));
-    }
-
-    if(is_py_bool_type(right.data_type()) && is_py_bool_type(right.data_type())) {
-        return true;
-    }
-
-    return false;
+namespace entity {
+    struct TypeDescriptor;
 }
 
-inline std::optional<entity::TypeDescriptor> has_valid_type_promotion(entity::TypeDescriptor source, entity::TypeDescriptor target) {
-
-    if(source.dimension() != target.dimension()) {
-        // Empty of dimension 0 means lack of any given type and can be promoted to anything (even if the dimensions don't
-        // match), e.g. empty type can become int or array of ints. Empty type of higher dimension is used to specify an
-        // empty array or an empty matrix, thus it cannot become any other type unless the dimensionality matches
-        if(is_empty_type(source.data_type()) && source.dimension() == entity::Dimension::Dim0)
-            return target;
-        return std::nullopt;
-    }
-
-    if (source == target)
-        return target;
+/// Two types are trivially compatible if their byte representation is exactly the same i.e. you can memcpy
+/// n elements of left type from one buffer to n elements of type right in another buffer and get the same result
+[[nodiscard]] bool trivially_compatible_types(const entity::TypeDescriptor& left, const entity::TypeDescriptor& right);
 
-    // Empty type is coercible to any type
-    if(is_empty_type(source.data_type())) {
-        return target;
-    }
+[[nodiscard]] std::optional<entity::TypeDescriptor> has_valid_type_promotion(
+    const entity::TypeDescriptor& source,
+    const entity::TypeDescriptor& target
+);
 
-    // Nothing is coercible to the empty type.
-    if(is_empty_type(target.data_type())) {
-        return std::nullopt;
-    }
-
-    auto source_type = source.data_type();
-    auto target_type = target.data_type();
-    auto source_size = slice_bit_size(source_type);
-    auto target_size = slice_bit_size(target_type);
-
-    if (is_time_type(source_type)) {
-        if (!is_time_type(target_type))
-            return std::nullopt;
-    } else if (is_unsigned_type(source_type)) {
-        if (is_unsigned_type(target_type)) {
-            // UINT->UINT, target_size must be >= source_size
-            if (source_size > target_size)
-                return std::nullopt;
-        } else if (is_signed_type(target_type)) {
-            // UINT->INT, target_size must be > source_size
-            if (source_size >= target_size)
-                return std::nullopt;
-        } else if (is_floating_point_type(target_type)) {
-            // UINT->FLOAT, no restrictions on relative sizes
-        } else {
-            // Non-numeric target type
-            return std::nullopt;
-        }
-    } else if (is_signed_type(source_type)) {
-        if (is_unsigned_type(target_type)) {
-            // INT->UINT never promotable
-            return std::nullopt;
-        } else if (is_signed_type(target_type)) {
-            // INT->INT, target_size must be >= source_size
-            if (source_size > target_size)
-                return std::nullopt;
-        } else if (is_floating_point_type(target_type)) {
-            // INT->FLOAT, no restrictions on relative sizes
-        } else {
-            // Non-numeric target type
-            return std::nullopt;
-        }
-    } else if (is_floating_point_type(source_type)) {
-        if (is_unsigned_type(target_type) || is_signed_type(target_type)) {
-            // FLOAT->U/INT never promotable
-            return std::nullopt;
-        } else if (is_floating_point_type(target_type)) {
-            // FLOAT->FLOAT, target_size must be >= source_size
-            if (source_size > target_size)
-                return std::nullopt;
-        } else {
-            // Non-numeric target type
-            return std::nullopt;
-        }
-    } else if (is_sequence_type(source_type) && is_sequence_type(target_type)) {
-        // Only allow promotion with UTF strings, and only to dynamic (never to fixed width)
-        if (!is_utf_type(source_type) || !is_utf_type(target_type) || !is_dynamic_string_type(target_type))
-            return std::nullopt;
-    } else {
-        // Non-numeric source type
-        return std::nullopt;
-    }
-
-    return entity::TypeDescriptor{combine_data_type(slice_value_type(target_type), target_size), target.dimension()};
-}
-
-inline std::optional<entity::TypeDescriptor> has_valid_type_promotion(
+[[nodiscard]] std::optional<entity::TypeDescriptor> has_valid_type_promotion(
     const proto::descriptors::TypeDescriptor& source,
-    const proto::descriptors::TypeDescriptor& target) {
-    return has_valid_type_promotion(entity::type_desc_from_proto(source), entity::type_desc_from_proto(target));
-}
+    const proto::descriptors::TypeDescriptor& target
+);
 
-inline std::optional<entity::TypeDescriptor> has_valid_common_type(entity::TypeDescriptor left, entity::TypeDescriptor right) {
-    auto maybe_common_type = has_valid_type_promotion(left, right);
-    if (!maybe_common_type) {
-        maybe_common_type = has_valid_type_promotion(right, left);
-    }
-    return maybe_common_type;
-}
+[[nodiscard]] std::optional<entity::TypeDescriptor> has_valid_common_type(
+    const entity::TypeDescriptor& left,
+    const entity::TypeDescriptor& right
+);
 
-inline std::optional<entity::TypeDescriptor> has_valid_common_type(
+[[nodiscard]] std::optional<entity::TypeDescriptor> has_valid_common_type(
     const proto::descriptors::TypeDescriptor& left,
-    const proto::descriptors::TypeDescriptor& right) {
-    return has_valid_common_type(entity::type_desc_from_proto(left), entity::type_desc_from_proto(right));
-}
-
+    const proto::descriptors::TypeDescriptor& right
+);
 }
diff --git a/cpp/arcticdb/entity/types-inl.hpp b/cpp/arcticdb/entity/types-inl.hpp
index 291140bd0b..ba5e001417 100644
--- a/cpp/arcticdb/entity/types-inl.hpp
+++ b/cpp/arcticdb/entity/types-inl.hpp
@@ -18,7 +18,7 @@ namespace arcticdb::entity {
 namespace details {
 
 template<class DimType, class Callable>
-auto visit_dim(DataType dt, Callable &&c) {
+constexpr auto visit_dim(DataType dt, Callable &&c) {
     switch (dt) {
 #define DT_CASE(__T__) case DataType::__T__: \
         return c(TypeDescriptorTag<DataTypeTag<DataType::__T__>, DimType>());
@@ -39,8 +39,7 @@ auto visit_dim(DataType dt, Callable &&c) {
         DT_CASE(UTF_FIXED64)
         DT_CASE(UTF_DYNAMIC64)
         DT_CASE(EMPTYVAL)
-        DT_CASE(PYBOOL8)
-        DT_CASE(PYBOOL64)
+        DT_CASE(BOOL_OBJECT8)
 #undef DT_CASE
         default: util::raise_rte("Invalid dtype '{}' in visit dim", datatype_to_str(dt));
     }
@@ -68,7 +67,7 @@ auto visit_type(DataType dt, Callable &&c) {
         DT_CASE(UTF_FIXED64)
         DT_CASE(UTF_DYNAMIC64)
         DT_CASE(EMPTYVAL)
-        DT_CASE(PYBOOL8)
+        DT_CASE(BOOL_OBJECT8)
 #undef DT_CASE
     default: util::raise_rte("Invalid dtype '{}' in visit type", datatype_to_str(dt));
     }
@@ -77,7 +76,7 @@ auto visit_type(DataType dt, Callable &&c) {
 } // namespace details
 
 template<class Callable>
-auto TypeDescriptor::visit_tag(Callable &&callable) const {
+constexpr auto TypeDescriptor::visit_tag(Callable &&callable) const {
     switch (dimension_) {
         case Dimension::Dim0: return details::visit_dim<DimensionTag<Dimension::Dim0>>(data_type_, callable);
         case Dimension::Dim1: return details::visit_dim<DimensionTag<Dimension::Dim1>>(data_type_, callable);
diff --git a/cpp/arcticdb/entity/types.cpp b/cpp/arcticdb/entity/types.cpp
index 5da45658bd..a03e45cdfd 100644
--- a/cpp/arcticdb/entity/types.cpp
+++ b/cpp/arcticdb/entity/types.cpp
@@ -6,6 +6,7 @@
  */
 
 #include <arcticdb/entity/types.hpp>
+#include <arcticdb/util/type_handler.hpp>
 
 namespace arcticdb::entity {
 
@@ -36,11 +37,18 @@ std::string_view datatype_to_str(const DataType dt) {
         TO_STR(UTF_FIXED64)
         TO_STR(UTF_DYNAMIC64)
         TO_STR(EMPTYVAL)
-        TO_STR(PYBOOL8)
-        TO_STR(PYBOOL64)
+        TO_STR(BOOL_OBJECT8)
 #undef TO_STR
         default:return std::string_view("UNKNOWN");
     }
 }
 
+std::size_t sizeof_datatype(const TypeDescriptor& td) {
+    return td.visit_tag([](auto tdt) {
+        using RawType = typename std::decay_t<decltype(tdt)>::DataTypeTag::raw_type;
+        auto handler = TypeHandlerRegistry::instance()->get_handler(TypeDescriptor{tdt});
+        return handler ? handler->type_size() : sizeof(RawType);
+    });
+}
+
 } // namespace arcticdb
diff --git a/cpp/arcticdb/entity/types.hpp b/cpp/arcticdb/entity/types.hpp
index eddb4a914d..54c6e8930c 100644
--- a/cpp/arcticdb/entity/types.hpp
+++ b/cpp/arcticdb/entity/types.hpp
@@ -92,7 +92,7 @@ enum class ValueType : uint8_t {
     // When data is appended, the column type is inferred from the data and the column is promoted to the inferred type.
     EMPTY = 13,
     /// Nullable booleans
-    PYBOOL = 14,
+    BOOL_OBJECT = 14,
     COUNT // Not a real value type, should not be added to proto descriptor. Used to count the number of items in the enum
 };
 
@@ -189,8 +189,7 @@ enum class DataType : uint8_t {
     UTF_FIXED64 = detail::combine_val_bits(ValueType::UTF8_FIXED, SizeBits::S64),
     UTF_DYNAMIC64 = detail::combine_val_bits(ValueType::UTF_DYNAMIC, SizeBits::S64),
     EMPTYVAL = detail::combine_val_bits(ValueType::EMPTY, SizeBits::S64),
-    PYBOOL8 = detail::combine_val_bits(ValueType::PYBOOL, SizeBits::S8),
-    PYBOOL64 = detail::combine_val_bits(ValueType::PYBOOL, SizeBits::S64),
+    BOOL_OBJECT8 = detail::combine_val_bits(ValueType::BOOL_OBJECT, SizeBits::S8),
     UNKNOWN = 0,
 };
 
@@ -244,8 +243,8 @@ constexpr bool is_bool_type(DataType dt) {
     return slice_value_type(dt) == ValueType::BOOL;
 }
 
-constexpr bool is_py_bool_type(DataType dt) {
-    return slice_value_type(dt) == ValueType::PYBOOL;
+constexpr bool is_bool_object_type(DataType dt) {
+    return slice_value_type(dt) == ValueType::BOOL_OBJECT;
 }
 
 constexpr bool is_unsigned_type(DataType dt) {
@@ -373,8 +372,7 @@ DATA_TYPE_TAG(ASCII_DYNAMIC64, std::uint64_t)
 DATA_TYPE_TAG(UTF_FIXED64, std::uint64_t)
 DATA_TYPE_TAG(UTF_DYNAMIC64, std::uint64_t)
 DATA_TYPE_TAG(EMPTYVAL, std::uint64_t)
-DATA_TYPE_TAG(PYBOOL8, uint8_t)
-DATA_TYPE_TAG(PYBOOL64, std::uint64_t)
+DATA_TYPE_TAG(BOOL_OBJECT8, uint8_t)
 #undef DATA_TYPE_TAG
 
 enum class Dimension : uint8_t {
@@ -419,7 +417,7 @@ struct TypeDescriptor {
     ARCTICDB_MOVE_COPY_DEFAULT(TypeDescriptor)
 
     template<typename Callable>
-    auto visit_tag(Callable &&callable) const;
+    constexpr auto visit_tag(Callable &&callable) const;
 
     bool operator==(const TypeDescriptor &o) const {
         return data_type_ == o.data_type_ && dimension_ == o.dimension_;
@@ -441,6 +439,10 @@ struct TypeDescriptor {
         data_type_ = combine_data_type(slice_value_type(data_type_), new_size_bits);
     }
 
+    [[nodiscard]] SizeBits get_size_bits() const {
+        return slice_bit_size(data_type_);
+    }
+
     [[nodiscard]] constexpr int get_type_byte_size() const {
         return get_byte_count(slice_bit_size(data_type_));
     }
@@ -463,12 +465,10 @@ constexpr bool is_numpy_array(TypeDescriptor td) {
 }
 
 constexpr bool is_pyobject_type(TypeDescriptor td) {
-    return is_dynamic_string_type(slice_value_type(td.data_type())) ||
-           slice_value_type(td.data_type()) == ValueType::PYBOOL ||
-           is_numpy_array(td);
+	return is_dynamic_string_type(slice_value_type(td.data_type())) || is_bool_object_type(td.data_type()) ||
+		is_numpy_array(td);
 }
 
-
 inline void set_data_type(DataType data_type, TypeDescriptor& type_desc) {
     type_desc.data_type_ = data_type;
 }
@@ -484,16 +484,20 @@ struct TypeDescriptorTag {
     using DataTypeTag = DT;
     using DimensionTag = D;
     explicit constexpr operator TypeDescriptor() const {
-        return TypeDescriptor{DataTypeTag::data_type, DimensionTag::value};
+        return type_descriptor();
     }
 
-    [[nodiscard]] constexpr Dimension dimension() const {
+    [[nodiscard]] static constexpr Dimension dimension() {
         return DimensionTag::value;
     }
 
-    [[nodiscard]] constexpr DataType data_type() const {
+    [[nodiscard]] static constexpr DataType data_type() {
         return DataTypeTag::data_type;
     }
+
+    [[nodiscard]] static constexpr TypeDescriptor type_descriptor() {
+        return TypeDescriptor{DataTypeTag::data_type, DimensionTag::value};
+    }
 };
 
 template <typename DTT>
@@ -624,6 +628,7 @@ inline bool operator!=(const Field& l, const Field& r) {
     return !(l == r);
 }
 
+std::size_t sizeof_datatype(const TypeDescriptor& td);
 } // namespace arcticdb::entity
 
 // StreamId ordering - numbers before strings
diff --git a/cpp/arcticdb/pipeline/column_mapping.cpp b/cpp/arcticdb/pipeline/column_mapping.cpp
new file mode 100644
index 0000000000..f348cb9f49
--- /dev/null
+++ b/cpp/arcticdb/pipeline/column_mapping.cpp
@@ -0,0 +1,137 @@
+/* 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 <arcticdb/pipeline/column_mapping.hpp>
+#include <arcticdb/column_store/memory_segment.hpp>
+#include <arcticdb/pipeline/pipeline_context.hpp>
+
+namespace arcticdb {
+    ColumnMapping::ColumnMapping(
+        SegmentInMemory& frame,
+        size_t dst_col,
+        size_t field_col,
+        pipelines::PipelineContextRow& context
+    ) :
+            source_type_desc_(context.descriptor().fields(field_col).type()),
+            dest_type_desc_(frame.field(dst_col).type()),
+            frame_field_descriptor_(frame.field(dst_col)),
+            dest_size_(sizeof_datatype(dest_type_desc_)),
+            num_rows_(context.slice_and_key().slice_.row_range.diff()),
+            first_row_(context.slice_and_key().slice_.row_range.first - frame.offset()),
+            offset_bytes_(dest_size_ * first_row_),
+            dest_bytes_(dest_size_ * num_rows_) {
+    }
+
+    StaticColumnMappingIterator::StaticColumnMappingIterator(
+        pipelines::PipelineContextRow& context,
+        size_t index_fieldcount
+    ) :
+            index_fieldcount_(index_fieldcount),
+            field_count_(context.slice_and_key().slice_.col_range.diff() + index_fieldcount),
+            first_slice_col_offset_(context.slice_and_key().slice_.col_range.first),
+            last_slice_col_offset_(context.slice_and_key().slice_.col_range.second),
+            bit_set_(context.get_selected_columns()) {
+        prev_col_offset_ = first_slice_col_offset_ - 1;
+        if (bit_set_) {
+            source_col_ = (*bit_set_)[bv_size(first_slice_col_offset_)]
+                ? first_slice_col_offset_
+                : bit_set_->get_next(bv_size(first_slice_col_offset_));
+            if ((*bit_set_)[bv_size(first_slice_col_offset_)]) {
+                source_col_ = first_slice_col_offset_;
+            } else {
+                auto next_pos = bit_set_->get_next(bv_size(first_slice_col_offset_));
+                // We have to do this extra check in bitmagic, get_next returns 0 in case no next present
+                if (next_pos == 0 && bit_set_->size() > 0 && !bit_set_->test(0))
+                    invalid_ = true;
+                else
+                    source_col_ = next_pos;
+            }
+            if (source_col_ < first_slice_col_offset_)
+                invalid_ = true;
+
+        } else {
+            source_col_ = first_slice_col_offset_;
+        }
+
+        dst_col_ = bit_set_ ? bit_set_->count_range(0, bv_size(source_col_)) - 1 : source_col_;
+        source_field_pos_ = (source_col_ - first_slice_col_offset_) + index_fieldcount_;
+    }
+
+    std::optional<size_t> StaticColumnMappingIterator::get_next_source_col() const {
+        if (!bit_set_) {
+            return source_col_ + 1;
+        } else {
+            auto next_pos = bit_set_->get_next(bv_size(source_col_));
+            if (next_pos == 0)
+                return std::nullopt;
+            else
+                return next_pos;
+        }
+    }
+
+    void StaticColumnMappingIterator::advance() {
+        ++dst_col_;
+        prev_col_offset_ = source_col_;
+        auto new_source_col = get_next_source_col();
+        if (new_source_col) {
+            source_col_ = *new_source_col;
+            source_field_pos_ = (source_col_ - first_slice_col_offset_) + index_fieldcount_;
+        } else {
+            source_field_pos_ = field_count_;
+            source_col_ = last_slice_col_offset_;
+        }
+    }
+
+    bool StaticColumnMappingIterator::invalid() const {
+        return invalid_;
+    }
+
+    bool StaticColumnMappingIterator::has_next() const {
+        return source_field_pos_ < field_count_;
+    }
+
+    bool StaticColumnMappingIterator::at_end_of_selected() const {
+        return !source_col_ || source_col_ >= last_slice_col_offset_;
+    }
+
+    size_t StaticColumnMappingIterator::remaining_fields() const {
+        return field_count_ - source_field_pos_;
+    }
+
+    size_t StaticColumnMappingIterator::prev_col_offset() const {
+        return prev_col_offset_;
+    }
+
+    size_t StaticColumnMappingIterator::source_field_pos() const {
+        return source_field_pos_;
+    }
+
+    size_t StaticColumnMappingIterator::source_col() const {
+        return source_col_;
+    }
+
+    size_t StaticColumnMappingIterator::first_slice_col_offset() const {
+        return first_slice_col_offset_;
+    }
+
+    size_t StaticColumnMappingIterator::last_slice_col_offset() const {
+        return last_slice_col_offset_;
+    }
+
+    size_t StaticColumnMappingIterator::dest_col() const {
+        return dst_col_;
+    }
+
+    size_t StaticColumnMappingIterator::field_count() const {
+        return field_count_;
+    }
+
+    size_t StaticColumnMappingIterator::index_fieldcount() const {
+        return index_fieldcount_;
+    }
+} // namespace arcticdb
\ No newline at end of file
diff --git a/cpp/arcticdb/pipeline/column_mapping.hpp b/cpp/arcticdb/pipeline/column_mapping.hpp
index 3279ea7493..8dcdddb186 100644
--- a/cpp/arcticdb/pipeline/column_mapping.hpp
+++ b/cpp/arcticdb/pipeline/column_mapping.hpp
@@ -5,41 +5,30 @@
  * 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 <arcticdb/entity/types.hpp>
+#include <arcticdb/util/bitset.hpp>
+#include <optional>
+
+namespace arcticdb::pipelines {
+    struct PipelineContextRow;
+}
 
 namespace arcticdb {
 
-inline std::size_t sizeof_datatype(const TypeDescriptor &td) {
-    return td.visit_tag([](auto dt) {
-        using RawType = typename std::decay_t<decltype(dt)>::DataTypeTag::raw_type;
-        // Array types are stored on disk as flat sequences. The python layer cannot work with this. We need to pass
-        // it pointers to an array type (at numpy arrays at the moment). When we allocate a column for an array we
-        // need to allocate space for one pointer per row. This also affects how we handle arrays to python as well.
-        // Check cpp/arcticdb/column_store/column_utils.hpp::array_at and cpp/arcticdb/column_store/column.hpp
-        return dt.dimension() == Dimension::Dim0 ? sizeof(RawType) : sizeof(void*);
-    });
-}
+class SegmentInMemory;
 
 struct ColumnMapping {
-    const TypeDescriptor source_type_desc_;
-    const TypeDescriptor dest_type_desc_;
-    const Field& frame_field_descriptor_;
+    const entity::TypeDescriptor source_type_desc_;
+    const entity::TypeDescriptor dest_type_desc_;
+    const entity::Field& frame_field_descriptor_;
     const size_t dest_size_;
     const size_t num_rows_;
     const size_t first_row_;
     const size_t offset_bytes_;
     const size_t dest_bytes_;
 
-    ColumnMapping(SegmentInMemory &frame, size_t dst_col, size_t field_col, pipelines::PipelineContextRow &context) :
-        source_type_desc_(context.descriptor().fields(field_col).type()),
-        dest_type_desc_(frame.field(dst_col).type()),
-        frame_field_descriptor_(frame.field(dst_col)),
-        dest_size_(sizeof_datatype(dest_type_desc_)),
-        num_rows_(context.slice_and_key().slice_.row_range.diff()),
-        first_row_(context.slice_and_key().slice_.row_range.first - frame.offset()),
-        offset_bytes_(dest_size_ * first_row_),
-        dest_bytes_(dest_size_ * num_rows_) {
-    }
+    ColumnMapping(SegmentInMemory& frame, size_t dst_col, size_t field_col, pipelines::PipelineContextRow& context);
 };
 
 struct StaticColumnMappingIterator {
@@ -53,109 +42,23 @@ struct StaticColumnMappingIterator {
     size_t dst_col_ = 0;
     bool invalid_ = false;
     const std::optional<util::BitSet>& bit_set_;
-    
-    StaticColumnMappingIterator(pipelines::PipelineContextRow& context, size_t index_fieldcount) :
-        index_fieldcount_(index_fieldcount),
-        field_count_(context.slice_and_key().slice_.col_range.diff() + index_fieldcount),
-        first_slice_col_offset_(context.slice_and_key().slice_.col_range.first),
-        last_slice_col_offset_(context.slice_and_key().slice_.col_range.second),
-        bit_set_(context.get_selected_columns()) {
-            prev_col_offset_ = first_slice_col_offset_ - 1;
-            if(bit_set_) {
-                source_col_ = (*bit_set_)[bv_size(first_slice_col_offset_)] ? first_slice_col_offset_ : bit_set_->get_next(bv_size(first_slice_col_offset_));
-                if ((*bit_set_)[bv_size(first_slice_col_offset_)]) {
-                    source_col_ = first_slice_col_offset_;
-                } else {
-                    auto next_pos = bit_set_->get_next(bv_size(first_slice_col_offset_));
-                    // We have to do this extra check in bitmagic, get_next returns 0 in case no next present
-                    if (next_pos == 0 && bit_set_->size() > 0 && !bit_set_->test(0))
-                        invalid_ = true;
-                    else
-                        source_col_ = next_pos;
-                }
-                if(source_col_ < first_slice_col_offset_)
-                    invalid_ = true;
-
-            } else {
-                source_col_ = first_slice_col_offset_;
-            }
-
-            dst_col_ = bit_set_ ? bit_set_->count_range(0, bv_size(source_col_)) - 1 : source_col_;
-            source_field_pos_ = (source_col_ - first_slice_col_offset_) + index_fieldcount_;
-    }
-
-    std::optional<size_t> get_next_source_col() const {
-        if(!bit_set_) {
-            return source_col_ + 1;
-        } else {
-            auto next_pos = bit_set_->get_next(bv_size(source_col_));
-            if(next_pos == 0)
-                return std::nullopt;
-            else
-                return next_pos;
-        }
-    }
-
-    void advance() {
-        ++dst_col_;
-        prev_col_offset_ = source_col_;
-        auto new_source_col = get_next_source_col();
-        if(new_source_col) {
-            source_col_ = *new_source_col;
-            source_field_pos_ = (source_col_ - first_slice_col_offset_) + index_fieldcount_;
-        } else {
-            source_field_pos_ = field_count_;
-            source_col_ = last_slice_col_offset_;
-        }
-    }
-
-    bool invalid() const {
-        return invalid_;
-    }
-
-    bool has_next() const {
-        return source_field_pos_ < field_count_;
-    }
-
-    bool at_end_of_selected() const {
-        return !source_col_ || source_col_ >= last_slice_col_offset_;
-    }
-
-    size_t remaining_fields() const {
-        return field_count_ - source_field_pos_;
-    }
-
-    size_t prev_col_offset() const {
-        return prev_col_offset_;
-    }
-
-    size_t source_field_pos() const {
-        return source_field_pos_;
-    }
-
-    size_t source_col() const {
-        return source_col_;
-    }
-
-    size_t first_slice_col_offset() const {
-        return first_slice_col_offset_;
-    }
-
-    size_t last_slice_col_offset() const {
-        return last_slice_col_offset_;
-    }
-
-    size_t dest_col() const {
-        return dst_col_;
-    }
-
-    size_t field_count() const {
-        return field_count_;
-    }
 
-    size_t index_fieldcount() const {
-        return index_fieldcount_;
-    }
+    StaticColumnMappingIterator(pipelines::PipelineContextRow& context, size_t index_fieldcount);
+
+    void advance();
+    [[nodiscard]] std::optional<size_t> get_next_source_col() const;
+    [[nodiscard]] bool invalid() const;
+    [[nodiscard]] bool has_next() const;
+    [[nodiscard]] bool at_end_of_selected() const;
+    [[nodiscard]] size_t remaining_fields() const;
+    [[nodiscard]] size_t prev_col_offset() const;
+    [[nodiscard]] size_t source_field_pos() const;
+    [[nodiscard]] size_t source_col() const;
+    [[nodiscard]] size_t first_slice_col_offset() const;
+    [[nodiscard]] size_t last_slice_col_offset() const;
+    [[nodiscard]] size_t dest_col() const;
+    [[nodiscard]] size_t field_count() const;
+    [[nodiscard]] size_t index_fieldcount() const;
 };
 
 
diff --git a/cpp/arcticdb/pipeline/filter_segment.hpp b/cpp/arcticdb/pipeline/filter_segment.hpp
index 1e6699a3b7..142d132bd0 100644
--- a/cpp/arcticdb/pipeline/filter_segment.hpp
+++ b/cpp/arcticdb/pipeline/filter_segment.hpp
@@ -9,8 +9,6 @@
 
 #include <arcticdb/column_store/memory_segment.hpp>
 #include <arcticdb/util/bitset.hpp>
-#include <folly/container/Enumerate.h>
-#include <arcticdb/entity/types.hpp>
 
 namespace arcticdb {
 
@@ -21,12 +19,6 @@ inline SegmentInMemory filter_segment(const SegmentInMemory& input,
     return input.filter(filter_bitset, filter_down_stringpool, validate);
 }
 
-inline SegmentInMemory truncate_segment(const SegmentInMemory& input,
-                                        size_t start,
-                                        size_t end) {
-    return input.truncate(start, end);
-}
-
 inline std::vector<SegmentInMemory> partition_segment(const SegmentInMemory& input,
                                       const std::vector<std::optional<uint8_t>>& row_to_segment,
                                       const std::vector<uint64_t>& segment_counts) {
diff --git a/cpp/arcticdb/pipeline/frame_slice_map.hpp b/cpp/arcticdb/pipeline/frame_slice_map.hpp
index 7b23eb4ee2..9579694215 100644
--- a/cpp/arcticdb/pipeline/frame_slice_map.hpp
+++ b/cpp/arcticdb/pipeline/frame_slice_map.hpp
@@ -8,8 +8,6 @@
 #pragma once
 
 #include <arcticdb/pipeline/pipeline_context.hpp>
-#include <unordered_map>
-#include <map>
 
 namespace arcticdb::pipelines {
 
@@ -35,13 +33,30 @@ struct FrameSliceMap {
                     continue;
                 }
 
-                if(!dynamic_schema && !is_sequence_type(field->type().data_type())) {
-                    ARCTICDB_DEBUG(log::version(), "{} not a string type in dynamic schema, skipping", field->name());
-                    continue;
+                const entity::DataType row_range_type = field->type().data_type();
+                if(!dynamic_schema && !is_sequence_type(row_range_type)) {
+                    // In case we end up with static schema and empty we must check the type of the whole column
+                    // Because we could be reading an empty segment of a string column. Example: start with [None],
+                    // then append ["string"]. If we read with row range [0;1) we would end up reading the empty segment
+                    // On read the empty type handler will fill the segment with not_a_string() and the reducer must
+                    // run over them.
+                    // TODO: This logic won't be needed when we move string handling into separate type handler
+                    if(is_empty_type(row_range_type)) {
+                        const entity::StreamDescriptor& descriptor = context_->descriptor();
+                        const size_t global_field_idx = descriptor.find_field(field->name()).value();
+                        const Field& global_field = descriptor.field(global_field_idx);
+                        const entity::DataType global_field_type = global_field.type().data_type();
+                        if(!is_sequence_type(global_field_type)) {
+                            ARCTICDB_DEBUG(log::version(), "{} not a string type in dynamic schema, skipping", field->name());
+                            continue;
+                        }
+                    } else {
+                        ARCTICDB_DEBUG(log::version(), "{} not a string type in dynamic schema, skipping", field->name());
+                        continue;
+                    }
                 }
 
                 auto& column = columns_[field->name()];
-
                 ContextData data{context_row.index_, field.index};
                 column.insert(std::make_pair(row_range, data));
             }
diff --git a/cpp/arcticdb/pipeline/frame_utils.hpp b/cpp/arcticdb/pipeline/frame_utils.hpp
index 193528c187..f9ecf52311 100644
--- a/cpp/arcticdb/pipeline/frame_utils.hpp
+++ b/cpp/arcticdb/pipeline/frame_utils.hpp
@@ -195,7 +195,7 @@ std::optional<convert::StringEncodingError> aggregator_set_data(
                     agg.set_array(col, t);
                 }
             }
-        } else if constexpr(is_py_bool_type(dt)) {
+        } else if constexpr(is_bool_object_type(dt)) {
             normalization::check<ErrorCode::E_UNIMPLEMENTED_INPUT_TYPE>(tag.dimension() == Dimension::Dim0,
                 "Multidimensional nullable booleans are not supported");
             auto data = const_cast<void *>(tensor.data());
@@ -209,12 +209,9 @@ std::optional<convert::StringEncodingError> aggregator_set_data(
 
             const auto num_values = bitset.count();
             auto bool_buffer = ChunkedBuffer::presized(num_values * sizeof(uint8_t));
-            auto en = bitset.first();
-            auto en_end = bitset.end();
             auto bool_ptr = bool_buffer.ptr_cast<uint8_t>(0u, num_values);
-            while (en < en_end) {
-                *bool_ptr = static_cast<uint8_t>(PyObject_IsTrue(ptr_data[*en]));
-                ++en;
+            for (auto it = bitset.first(); it < bitset.end(); ++it) {
+                *bool_ptr = static_cast<uint8_t>(PyObject_IsTrue(ptr_data[*it]));
                 ++bool_ptr;
             }
             if(bitset.count() > 0)
diff --git a/cpp/arcticdb/pipeline/read_frame.cpp b/cpp/arcticdb/pipeline/read_frame.cpp
index 9a39383ad8..2c85564f6f 100644
--- a/cpp/arcticdb/pipeline/read_frame.cpp
+++ b/cpp/arcticdb/pipeline/read_frame.cpp
@@ -56,27 +56,14 @@ void mark_index_slices(
         column_groups).value();
 }
 
-StreamDescriptor modify_output_sizes(StreamDescriptor&& desc) {
-    for(auto& field : desc.mutable_fields()) {
-        if(is_py_bool_type(field.type().data_type())) {
-            field.mutable_type_desc()->set_size_bits(SizeBits::S64);
-        }
-    }
-    return std::move(desc);
-}
-
 StreamDescriptor get_filtered_descriptor(StreamDescriptor&& descriptor, const std::shared_ptr<FieldCollection>& filter_columns) {
     // We assume here that filter_columns_ will always contain the index.
 
     auto desc = std::move(descriptor);
     auto index = stream::index_type_from_descriptor(desc);
     return util::variant_match(index, [&desc, &filter_columns] (const auto& idx) {
-        if(filter_columns) {
-            return modify_output_sizes(StreamDescriptor{index_descriptor(desc.id(), idx, *filter_columns)});
-        }
-        else {
-            return modify_output_sizes(StreamDescriptor{index_descriptor(desc.id(), idx, *desc.fields_ptr())});
-        }
+        const std::shared_ptr<FieldCollection>& fields = filter_columns ? filter_columns : desc.fields_ptr();
+        return StreamDescriptor{index_descriptor(desc.id(), idx, *fields)};
     });
 }
 
@@ -225,12 +212,21 @@ void decode_or_expand_impl(
     const uint8_t*& data,
     uint8_t* dest,
     const EncodedFieldType& encoded_field_info,
-    const TypeDescriptor& type_descriptor,
     size_t dest_bytes,
     std::shared_ptr<BufferHolder> buffers,
-	EncodingVersion encding_version) {
-    if(auto handler = TypeHandlerRegistry::instance()->get_handler(type_descriptor); handler) {
-        handler->handle_type(data, dest, VariantField{&encoded_field_info}, type_descriptor, dest_bytes, std::move(buffers), encding_version);
+    EncodingVersion encding_version,
+    const ColumnMapping& m
+) {
+    if (auto handler = TypeHandlerRegistry::instance()->get_handler(m.source_type_desc_); handler) {
+        handler->handle_type(
+            data,
+            dest,
+            VariantField{&encoded_field_info},
+            dest_bytes,
+            std::move(buffers),
+            encding_version,
+            m
+        );
     } else {
         std::optional<util::BitMagic> bv;
         if (encoded_field_info.has_ndarray() && encoded_field_info.ndarray().sparse_map_bytes() > 0) {
@@ -238,8 +234,8 @@ void decode_or_expand_impl(
             const auto bytes = encoding_sizes::data_uncompressed_size(ndarray);
             ChunkedBuffer sparse{bytes};
             SliceDataSink sparse_sink{sparse.data(), bytes};
-            data += decode_field(type_descriptor, encoded_field_info, data, sparse_sink, bv, encding_version);
-            type_descriptor.visit_tag([dest, dest_bytes, &bv, &sparse](const auto tdt) {
+            data += decode_field(m.source_type_desc_, encoded_field_info, data, sparse_sink, bv, encding_version);
+            m.source_type_desc_.visit_tag([dest, dest_bytes, &bv, &sparse](const auto tdt) {
                 using TagType = decltype(tdt);
                 using RawType = typename TagType::DataTypeTag::raw_type;
                 util::default_initialize<TagType>(dest, dest_bytes);
@@ -249,12 +245,12 @@ void decode_or_expand_impl(
             SliceDataSink sink(dest, dest_bytes);
             const auto &ndarray = encoded_field_info.ndarray();
             if (const auto bytes = encoding_sizes::data_uncompressed_size(ndarray); bytes < dest_bytes) {
-                type_descriptor.visit_tag([dest, bytes, dest_bytes](const auto tdt) {
+                m.source_type_desc_.visit_tag([dest, bytes, dest_bytes](const auto tdt) {
                     using TagType = decltype(tdt);
                     util::default_initialize<TagType>(dest + bytes, dest_bytes - bytes);
                 });
             }
-            data += decode_field(type_descriptor, encoded_field_info, data, sink, bv, encding_version);
+            data += decode_field(m.source_type_desc_, encoded_field_info, data, sink, bv, encding_version);
         }
     }
 }
@@ -279,13 +275,21 @@ void decode_or_expand(
     const uint8_t*& data,
     uint8_t* dest,
     const VariantField& variant_field,
-    const TypeDescriptor& type_descriptor,
     size_t dest_bytes,
     std::shared_ptr<BufferHolder> buffers,
-    EncodingVersion encoding_version
+    EncodingVersion encoding_version,
+    const ColumnMapping& m
 ) {
     util::variant_match(variant_field, [&](auto field) {
-        decode_or_expand_impl(data, dest, *field, type_descriptor, dest_bytes, buffers, encoding_version);
+        decode_or_expand_impl(
+            data,
+            dest,
+            *field,
+            dest_bytes,
+            buffers,
+            encoding_version,
+            m
+        );
     });
 }
 
@@ -369,15 +373,32 @@ void decode_into_frame_static(
             auto field_name = context.descriptor().fields(it.source_field_pos()).name();
             auto& buffer = frame.column(static_cast<ssize_t>(it.dest_col())).data().buffer();
             ColumnMapping m{frame, it.dest_col(), it.source_field_pos(), context};
-            util::check(trivially_compatible_types(m.source_type_desc_, m.dest_type_desc_), "Column type conversion from {} to {} not implemented in column {}:{} -> {}:{}",
-                        m.source_type_desc_,
-                        m.dest_type_desc_,
-                        it.source_col(),
-                        field_name,
-                        it.dest_col(),
-                        m.frame_field_descriptor_.name());
-            util::check(data != end || remaining_fields_empty(it, context), "Reached end of input block with {} fields to decode", it.remaining_fields());
-            decode_or_expand(data, buffer.data() + m.offset_bytes_, encoded_field, m.source_type_desc_,  m.dest_bytes_, buffers, encoding_version);
+            const bool types_trivially_compatible = trivially_compatible_types(m.source_type_desc_, m.dest_type_desc_);
+            const bool any_type_is_empty = is_empty_type(m.source_type_desc_.data_type()) || is_empty_type(m.dest_type_desc_.data_type());
+            util::check(
+                types_trivially_compatible || any_type_is_empty,
+                "Column type conversion from {} to {} not implemented in column {}:{} -> {}:{}",
+                m.source_type_desc_,
+                m.dest_type_desc_,
+                it.source_col(),
+                field_name,
+                it.dest_col(),
+                m.frame_field_descriptor_.name()
+            );
+            util::check(
+                data != end || remaining_fields_empty(it, context),
+                "Reached end of input block with {} fields to decode",
+                it.remaining_fields()
+            );
+            decode_or_expand(
+                data,
+                buffer.data() + m.offset_bytes_,
+                encoded_field,
+                m.dest_bytes_,
+                buffers,
+                encoding_version,
+                m
+            );
             ARCTICDB_TRACE(log::codec(), "Decoded column {} to position {}", field_name, data - begin);
 
             it.advance();
@@ -403,10 +424,11 @@ void decode_into_frame_static(
 }
 
 void decode_into_frame_dynamic(
-        SegmentInMemory &frame,
-        PipelineContextRow &context,
-        Segment &&s,
-        const std::shared_ptr<BufferHolder>& buffers) {
+    SegmentInMemory& frame,
+    PipelineContextRow& context,
+    Segment&& s,
+    const std::shared_ptr<BufferHolder>& buffers
+) {
     ARCTICDB_SAMPLE_DEFAULT(DecodeIntoFrame)
     auto seg = std::move(s);
     const uint8_t *data = seg.buffer().data();
@@ -419,9 +441,11 @@ void decode_into_frame_dynamic(
     context.set_compacted(hdr.compacted());
     const EncodingVersion encdoing_version = EncodingVersion(hdr.encoding_version());
     const bool has_magic_numbers = encdoing_version == EncodingVersion::V2;
+    VariantEncodedFieldCollection fields(seg);
 
-    if (data != end) {
-        VariantEncodedFieldCollection fields(seg);
+    // data == end in case we have empty data types (e.g. {EMPTYVAL, Dim0}, {EMPTYVAL, Dim1}) for which we store nothing
+    // in storage as they can be reconstructed in the type handler on the read path.
+    if (data != end || fields.size() > 0) {
         auto index_field = fields.at(0u);
         decode_index_field(frame, index_field, data, begin, end, context, encdoing_version);
 
@@ -439,15 +463,38 @@ void decode_into_frame_dynamic(
             auto dst_col = *frame_loc_opt;
             auto& buffer = frame.column(static_cast<position_t>(dst_col)).data().buffer();
             ColumnMapping m{frame, dst_col, field_col, context};
-            util::check(static_cast<bool>(has_valid_type_promotion(m.source_type_desc_, m.dest_type_desc_)), "Can't promote type {} to type {} in field {}",
-                        m.source_type_desc_, m.dest_type_desc_, m.frame_field_descriptor_.name());
-            ARCTICDB_TRACE(log::storage(), "Creating data slice at {} with total size {} ({} rows)", m.offset_bytes_, m.dest_bytes_,
-                           context.slice_and_key().slice_.row_range.diff());
-            util::check(data != end,
-                        "Reached end of input block with {} fields to decode",
-                        field_count - field_col);
-            decode_or_expand(data, buffer.data() + m.offset_bytes_, encoded_field, m.source_type_desc_, m.dest_bytes_, buffers, encdoing_version);
-            if (!trivially_compatible_types(m.source_type_desc_, m.dest_type_desc_)) {
+            util::check(
+                static_cast<bool>(has_valid_type_promotion(m.source_type_desc_, m.dest_type_desc_)),
+                "Can't promote type {} to type {} in field {}",
+                m.source_type_desc_,
+                m.dest_type_desc_,
+                m.frame_field_descriptor_.name()
+            );
+            ARCTICDB_TRACE(
+                log::storage(),
+                "Creating data slice at {} with total size {} ({} rows)",
+                m.offset_bytes_,
+                m.dest_bytes_,
+                context.slice_and_key().slice_.row_range.diff()
+            );
+            const bool source_is_empty = is_empty_type(m.source_type_desc_.data_type());
+            util::check(
+                data != end || source_is_empty,
+                "Reached end of input block with {} fields to decode",
+                field_count - field_col
+            );
+            decode_or_expand(
+                data,
+                buffer.data() + m.offset_bytes_,
+                encoded_field,
+                m.dest_bytes_,
+                buffers,
+                encdoing_version,
+                m
+            );
+            // decode_or_expand will invoke the empty type handler which will do backfilling with the default value depending on the
+            // destination type.
+            if (!trivially_compatible_types(m.source_type_desc_, m.dest_type_desc_) && !source_is_empty) {
                 m.dest_type_desc_.visit_tag([&buffer, &m, &data, encoded_field, buffers, encdoing_version] (auto dest_desc_tag) {
                     using DestinationType =  typename decltype(dest_desc_tag)::DataTypeTag::raw_type;
                     m.source_type_desc_.visit_tag([&buffer, &m, &data, &encoded_field, &buffers, encdoing_version] (auto src_desc_tag ) {
@@ -956,18 +1003,24 @@ class DynamicStringReducer : public StringReducer {
         const auto& segment_descriptor = context_row.descriptor();
         const auto& segment_field = segment_descriptor[column_index];
 
-        auto has_type_conversion = frame_field_.type() != segment_field.type();
-        util::check(!has_type_conversion || trivially_compatible_types(frame_field_.type(), segment_field.type()),
-                    "Cannot convert from type {} to {} in frame field", frame_field_.type(), segment_field.type());
+        const bool trivially_compatible = trivially_compatible_types(frame_field_.type(), segment_field.type());
+        // In case the segment type is EMPTYVAL the empty type handler should have ran and set all missing values
+        // to not_a_string()
+        util::check(trivially_compatible || is_empty_type(segment_field.type().data_type()),
+            "String types are not trivially compatible. Cannot convert from type {} to {} in frame field.",
+            frame_field_.type(),
+            segment_field.type()
+        );
 
         auto is_utf = is_utf_type(slice_value_type(frame_field_.type().data_type()));
         size_t end =  context_row.slice_and_key().slice_.row_range.second - frame_.offset();
 
+        const auto is_type_different = frame_field_.type() != segment_field.type();
         auto ptr_src = get_offset_ptr_at(row_, src_buffer_);
         if(do_lock_)
-            process_string_views<LockActive>(has_type_conversion, is_utf, end, ptr_src, context_row.string_pool());
+            process_string_views<LockActive>(is_type_different, is_utf, end, ptr_src, context_row.string_pool());
         else
-            process_string_views<LockDisabled>(has_type_conversion, is_utf, end, ptr_src, context_row.string_pool());
+            process_string_views<LockDisabled>(is_type_different, is_utf, end, ptr_src, context_row.string_pool());
     }
 
     void finalize() override {
diff --git a/cpp/arcticdb/pipeline/read_frame.hpp b/cpp/arcticdb/pipeline/read_frame.hpp
index 0249c91c4d..d0c28348b8 100644
--- a/cpp/arcticdb/pipeline/read_frame.hpp
+++ b/cpp/arcticdb/pipeline/read_frame.hpp
@@ -8,16 +8,18 @@
 #pragma once
 
 #include <arcticdb/stream/stream_source.hpp>
-#include <arcticdb/async/task_scheduler.hpp>
 #include <arcticdb/pipeline/pipeline_context.hpp>
 #include <arcticdb/pipeline/read_options.hpp>
 #include <arcticdb/util/bitset.hpp>
-#include <arcticdb/util/buffer_holder.hpp>
 
 #include <folly/futures/Future.h>
 
 #include <memory>
 
+namespace arcticdb {
+    struct BufferHolder;
+}
+
 namespace arcticdb::pipelines {
 
 SegmentInMemory allocate_frame(const std::shared_ptr<PipelineContext>& context);
diff --git a/cpp/arcticdb/pipeline/string_pool_utils.hpp b/cpp/arcticdb/pipeline/string_pool_utils.hpp
index 3a1212bc18..fac34f7894 100644
--- a/cpp/arcticdb/pipeline/string_pool_utils.hpp
+++ b/cpp/arcticdb/pipeline/string_pool_utils.hpp
@@ -36,8 +36,16 @@ inline auto get_string_from_pool(entity::position_t offset_val, const StringPool
     return string_pool.get_const_view(offset_val);
 }
 
-inline std::variant<std::string_view, entity::position_t> get_string_from_buffer(size_t offset, const ChunkedBuffer& src, const StringPool& string_pool) {
-    auto offset_val = get_offset_string_at(offset, src);
+/// @brief Get the i-th string from a buffer
+/// @param string_pos The index of the string to be returned
+/// @return If the string at @p string_pos is actual string inside @p string_pool return a
+///  string view, otherwise return an (integer) placeholder representing not a string.
+inline std::variant<std::string_view, entity::position_t> get_string_from_buffer(
+    size_t string_pos,
+    const ChunkedBuffer& src,
+    const StringPool& string_pool
+) {
+    auto offset_val = get_offset_string_at(string_pos, src);
     if (offset_val == nan_placeholder() || offset_val == not_a_string())
         return offset_val;
     else
diff --git a/cpp/arcticdb/pipeline/write_frame.cpp b/cpp/arcticdb/pipeline/write_frame.cpp
index 81002c2ef1..af689e9788 100644
--- a/cpp/arcticdb/pipeline/write_frame.cpp
+++ b/cpp/arcticdb/pipeline/write_frame.cpp
@@ -28,8 +28,9 @@
 #include <arcticdb/entity/merge_descriptors.hpp>
 #include <arcticdb/async/task_scheduler.hpp>
 #include <arcticdb/util/format_date.hpp>
+#include <vector>
+#include <array>
 
-#include <pybind11/pybind11.h>
 
 namespace arcticdb::pipelines {
 
@@ -247,114 +248,141 @@ folly::Future<entity::AtomKey> append_frame(
         slices_to_write.insert(std::end(slices_to_write), std::make_move_iterator(std::begin(slice_and_keys_to_append)), std::make_move_iterator(std::end(slice_and_keys_to_append)));
         std::sort(std::begin(slices_to_write), std::end(slices_to_write));
         if(dynamic_schema) {
-            auto merged_descriptor =
-                merge_descriptors(frame->desc, std::vector< std::shared_ptr<FieldCollection>>{ index_segment_reader.tsd().fields_ptr()}, {});
+            auto merged_descriptor = merge_descriptors(
+                frame->desc,
+                std::vector<std::shared_ptr<FieldCollection>>{index_segment_reader.tsd().fields_ptr()},
+                {}
+            );
             merged_descriptor.set_sorted(deduce_sorted(index_segment_reader.get_sorted(), frame->desc.get_sorted()));
-            auto tsd =
-                make_timeseries_descriptor(frame->num_rows + frame->offset, std::move(merged_descriptor), std::move(frame->norm_meta), std::move(frame->user_meta), std::nullopt, std::nullopt, frame->bucketize_dynamic);
+            auto tsd = make_timeseries_descriptor(
+                frame->num_rows + frame->offset,
+                std::move(merged_descriptor),
+                std::move(frame->norm_meta),
+                std::move(frame->user_meta),
+                std::nullopt,
+                std::nullopt,
+                frame->bucketize_dynamic
+            );
             return index::write_index(stream::index_type_from_descriptor(frame->desc), std::move(tsd), std::move(slices_to_write), key, store);
         } else {
+            const FieldCollection& new_fields{index_segment_reader.tsd().fields()};
+            for (size_t i = 0; i < new_fields.size(); ++i) {
+                const Field& new_field = new_fields.at(i);
+                TypeDescriptor& original_type = frame->desc.mutable_field(i).mutable_type();
+                if (is_empty_type(original_type.data_type()) && !is_empty_type(new_field.type().data_type())) {
+                    original_type = new_field.type();
+                }
+            }
             frame->desc.set_sorted(deduce_sorted(index_segment_reader.get_sorted(), frame->desc.get_sorted()));
             return index::write_index(frame, std::move(slices_to_write), key, store);
         }
     });
 }
 
-void update_string_columns(const SegmentInMemory& original, SegmentInMemory output) {
-    util::check(original.descriptor() == output.descriptor(), "Update string column handling expects identical descriptors");
-    for (size_t column = 0; column < static_cast<size_t>(original.descriptor().fields().size()); ++column) {
-        auto &frame_field = original.field(column);
-        auto field_type = frame_field.type().data_type();
-
-        if (is_sequence_type(field_type)) {
-            auto &target = output.column(static_cast<position_t>(column)).data().buffer();
-            size_t end = output.row_count();
-            for(auto row = 0u; row < end; ++row) {
-                auto val = get_string_from_buffer(row, target, original.const_string_pool());
-                util::variant_match(val,
-                                    [&] (std::string_view sv) {
-                                        auto off_str = output.string_pool().get(sv);
-                                        set_offset_string_at(row, target, off_str.offset());
-                                    },
-                                    [&] (entity::position_t offset) {
-                                        set_offset_string_at(row, target, offset);
-                                    });
+/// @brief Find the row range of affected rows during a partial rewrite (on update)
+/// During partial rewrite the segment is either affected from the beginning to a
+/// certain row or from a certain row to the end. Thus the row range will always be
+/// either [0, row) or [row, end).
+static RowRange partial_rewrite_row_range(
+    const SegmentInMemory& segment,
+    const IndexRange& range,
+    AffectedSegmentPart affected_end
+) {
+    if (affected_end == AffectedSegmentPart::START) {
+        const timestamp start = std::get<timestamp>(range.start_);
+        auto bound = std::lower_bound(std::begin(segment), std::end(segment), start, [](const auto& row, timestamp t) {
+            return row.template index<TimeseriesIndex>() < t;
+        });
+        return {0, bound->row_pos()};
+    } else {
+        const timestamp end = std::get<timestamp>(range.end_);
+        auto bound = std::upper_bound(std::begin(segment), std::end(segment), end, [](timestamp t, const auto& row) {
+            return t < row.template index<TimeseriesIndex>();
+        });
+        return {bound->row_pos(), segment.row_count()};
+    }
+}
 
-            }
-        }
+/// @brief Find the index range of affected rows during a partial rewrite (on update)
+/// Similar to partial_rewrite_row_range the segment is affected either at the beginning
+/// or at the end.
+static IndexRange partial_rewrite_index_range(
+    const IndexRange& segment_range,
+    const IndexRange& update_range,
+    AffectedSegmentPart affected_part
+) {
+    if (affected_part == AffectedSegmentPart::START) {
+        util::check(
+            segment_range.start_ < update_range.start_,
+            "Unexpected index range in after: {} !< {}",
+            segment_range.start_,
+            update_range.start_
+        );
+        return {segment_range.start_, update_range.start_};
+    } else {
+        util::check(
+            segment_range.end_ > update_range.end_,
+            "Unexpected non-intersection of update indices: {} !> {}",
+            segment_range.end_,
+            update_range.end_
+        );
+        return {segment_range.end_, update_range.end_};
     }
 }
 
 std::optional<SliceAndKey> rewrite_partial_segment(
-        const SliceAndKey& existing,
-        IndexRange index_range,
-         VersionId version_id,
-         bool before,
-         const std::shared_ptr<Store>& store) {
-    const auto& key =  existing.key();
-    const auto& existing_range =key.index_range();
+    const SliceAndKey& existing,
+    IndexRange index_range,
+    VersionId version_id,
+    AffectedSegmentPart affected_part,
+    const std::shared_ptr<Store>& store
+) {
+    const auto& key = existing.key();
+    const IndexRange& existing_range = key.index_range();
     auto kv = store->read(key).get();
-    auto &segment = kv.second;
-
-    auto start = std::get<timestamp>(index_range.start_);
-    auto end = std::get<timestamp>(index_range.end_);
-
-    if(!before) {
-        util::check(existing_range.start_ < index_range.start_, "Unexpected index range in after: {} !< {}", existing_range.start_, index_range.start_);
-        auto bound = std::lower_bound(std::begin(segment), std::end(segment), start, [] ( auto& row, timestamp t) {return row.template index<TimeseriesIndex>() < t; });
-        size_t num_rows = std::distance(std::begin(segment), bound);
-        if(num_rows == 0)
-            return std::nullopt;
-
-        auto output = SegmentInMemory{segment.descriptor(), num_rows};
-        std::copy(std::begin(segment), bound, std::back_inserter(output));
-        update_string_columns(segment, output);
-        FrameSlice new_slice{
-            std::make_shared<StreamDescriptor>(output.descriptor()),
-            existing.slice_.col_range,
-            RowRange{0, num_rows},
-            existing.slice_.hash_bucket(),
-            existing.slice_.num_buckets()};
-        auto fut_key = store->write(key.type(), version_id, key.id(), existing_range.start_, index_range.start_, std::move(output));
-        return SliceAndKey{std::move(new_slice), std::get<AtomKey>(std::move(fut_key).get())};
-    }
-    else {
-        util::check(existing_range.end_ > index_range.end_, "Unexpected non-intersection of update indices: {} !> {}", existing_range.end_ , index_range.end_);
-
-        auto bound = std::upper_bound(std::begin(segment), std::end(segment), end, [] ( timestamp t, auto& row) {
-            return t < row.template index<TimeseriesIndex>();
-        });
-        size_t num_rows = std::distance(bound, std::end(segment));
-        if(num_rows == 0)
-            return std::nullopt;
-
-        auto output = SegmentInMemory{segment.descriptor(), num_rows};
-        std::copy(bound, std::end(segment), std::back_inserter(output));
-        update_string_columns(segment, output);
-        FrameSlice new_slice{
-            std::make_shared<StreamDescriptor>(output.descriptor()),
-                    existing.slice_.col_range,
-                    RowRange{0, num_rows},
-                    existing.slice_.hash_bucket(),
-                    existing.slice_.num_buckets()};
-        auto fut_key = store->write(key.type(), version_id, key.id(), index_range.end_, existing_range.end_, std::move(output));
-        return SliceAndKey{std::move(new_slice), std::get<AtomKey>(std::move(fut_key).get())};
+    const SegmentInMemory& segment = kv.second;
+    const IndexRange affected_index_range = partial_rewrite_index_range(existing_range, index_range, affected_part);
+    const RowRange affected_row_range = partial_rewrite_row_range(segment, index_range, affected_part);
+    const int64_t num_rows = affected_row_range.end() - affected_row_range.start();
+    if (num_rows <= 0) {
+        return std::nullopt;
     }
+    SegmentInMemory output = segment.truncate(affected_row_range.start(), affected_row_range.end(), true);
+    FrameSlice new_slice{
+        std::make_shared<StreamDescriptor>(output.descriptor()),
+        existing.slice_.col_range,
+        RowRange{0, num_rows},
+        existing.slice_.hash_bucket(),
+        existing.slice_.num_buckets()};
+    auto fut_key = store->write(
+        key.type(),
+        version_id,
+        key.id(),
+        affected_index_range.start_,
+        affected_index_range.end_,
+        std::move(output)
+    );
+    return SliceAndKey{std::move(new_slice), std::get<AtomKey>(std::move(fut_key).get())};
 }
 
-std::vector<SliceAndKey> flatten_and_fix_rows(const std::vector<std::vector<SliceAndKey>>& groups, size_t& global_count) {
+std::vector<SliceAndKey> flatten_and_fix_rows(const std::array<std::vector<SliceAndKey>, 5>& groups, size_t& global_count) {
     std::vector<SliceAndKey> output;
+    output.reserve(groups.size());
     global_count = 0;
-    for(auto group : groups) {
-        if(group.empty()) continue;
+    for (const std::vector<SliceAndKey>& group : groups) {
+        if (group.empty())
+            continue;
         auto group_start = group.begin()->slice_.row_range.first;
-        auto group_end = std::accumulate(std::begin(group), std::end(group), 0ULL, [](size_t a, const SliceAndKey& sk) { return std::max(a, sk.slice_.row_range.second); });
-        std::transform(std::begin(group), std::end(group), std::back_inserter(output), [&] (auto& sk) {
+        auto group_end = std::accumulate(std::begin(group), std::end(group), 0ULL, [](size_t a, const SliceAndKey& sk) {
+            return std::max(a, sk.slice_.row_range.second);
+        });
+        std::transform(std::begin(group), std::end(group), std::back_inserter(output), [&](SliceAndKey sk) {
             auto range_start = global_count + (sk.slice_.row_range.first - group_start);
             auto new_range = RowRange{range_start, range_start + (sk.slice_.row_range.diff())};
             sk.slice_.row_range = new_range;
-            return sk; });
-        global_count += (group_end - group_start) ;
+            return sk;
+        });
+        global_count += (group_end - group_start);
     }
     return output;
 }
diff --git a/cpp/arcticdb/pipeline/write_frame.hpp b/cpp/arcticdb/pipeline/write_frame.hpp
index 3af445e270..1fee3e7654 100644
--- a/cpp/arcticdb/pipeline/write_frame.hpp
+++ b/cpp/arcticdb/pipeline/write_frame.hpp
@@ -7,12 +7,9 @@
 
 #pragma once
 
-#include <arcticdb/stream/index.hpp>
-#include <arcticdb/entity/protobufs.hpp>
 #include <arcticdb/entity/index_range.hpp>
 
-#include <vector>
-#include <variant>
+
 #include <arcticdb/pipeline/input_tensor_frame.hpp>
 #include <arcticdb/stream/index.hpp>
 #include <folly/futures/Future.h>
@@ -63,15 +60,33 @@ folly::Future<entity::AtomKey> append_frame(
         bool ignore_sort_order
 );
 
+enum class AffectedSegmentPart {
+    START,
+    END
+};
+
 std::optional<SliceAndKey> rewrite_partial_segment(
         const SliceAndKey& existing,
         IndexRange index_range,
         VersionId version_id,
-        bool before,
+        AffectedSegmentPart affected_part,
         const std::shared_ptr<Store>& store);
 
+
+/// Used, when updating a segment, to convert all 5 affected groups into a single list of slices
+/// The 5 groups are:
+/// * Segments before the update range which do not intersect with it and are not affected by
+///   the update
+/// * Segments before the update range which are intersecting with it and are partially affected
+///   by the update.
+/// * Segments which are fully contained inside the update range
+/// * Segments after the update range which are intersecting with it and are partially affected
+///   by the update
+/// * Segments after the update range which do not intersect with it and are not affected by the
+///   update
 std::vector<SliceAndKey> flatten_and_fix_rows(
-        const std::vector<std::vector<SliceAndKey>>& groups,
-        size_t& global_count);
+    const std::array<std::vector<SliceAndKey>, 5>& groups,
+    size_t& global_count
+);
 
 } //namespace arcticdb::pipelines
\ No newline at end of file
diff --git a/cpp/arcticdb/processing/aggregation.cpp b/cpp/arcticdb/processing/aggregation.cpp
index 96b872554b..45422844b0 100644
--- a/cpp/arcticdb/processing/aggregation.cpp
+++ b/cpp/arcticdb/processing/aggregation.cpp
@@ -133,13 +133,8 @@ namespace
     };
 
     template<>
-    struct OutputType<DataTypeTag<DataType::PYBOOL8>, void> {
-        using type = ScalarTagType<DataTypeTag<DataType::PYBOOL8>>;
-    };
-
-    template<>
-    struct OutputType<DataTypeTag<DataType::PYBOOL64>, void> {
-        using type = ScalarTagType<DataTypeTag<DataType::PYBOOL64>>;
+    struct OutputType<DataTypeTag<DataType::BOOL_OBJECT8>, void> {
+        using type = ScalarTagType<DataTypeTag<DataType::BOOL_OBJECT8>>;
     };
 }
 
@@ -172,13 +167,13 @@ 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, that=this] (auto global_type_desc_tag) {
+    entity::details::visit_type(*data_type_, [&input_column, unique_values, &groups, this] (auto global_type_desc_tag) {
         using GlobalInputType = decltype(global_type_desc_tag);
         if constexpr(!is_sequence_type(GlobalInputType::DataTypeTag::data_type)) {
             using GlobalTypeDescriptorTag =  typename OutputType<GlobalInputType>::type;
             using GlobalRawType = typename GlobalTypeDescriptorTag::DataTypeTag::raw_type;
-            that->aggregated_.resize(sizeof(GlobalRawType)* unique_values);
-            auto out_ptr = reinterpret_cast<GlobalRawType*>(that->aggregated_.data());
+            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)>;
@@ -189,7 +184,11 @@ void SumAggregatorData::aggregate(const std::optional<ColumnWithStrings>& input_
                             while (auto block = col_data.next<TypeDescriptorTag<ColumnTagType, DimensionTag<entity::Dimension::Dim0>>>()) {
                                 auto ptr = reinterpret_cast<const ColumnType *>(block.value().data());
                                 for (auto i = 0u; i < block.value().row_count(); ++i, ++ptr, ++iter) {
-                                    out_ptr[groups[deref(iter)]] += GlobalRawType(*ptr);
+                                    if constexpr (std::is_same_v<GlobalRawType, bool>) {
+                                        out_ptr[groups[deref(iter)]] |= GlobalRawType(*ptr);
+                                    } else {
+                                        out_ptr[groups[deref(iter)]] += GlobalRawType(*ptr);
+                                    }
                                 }
                             }
                         };
diff --git a/cpp/arcticdb/processing/clause.cpp b/cpp/arcticdb/processing/clause.cpp
index 43199a9975..d7fddd97a5 100644
--- a/cpp/arcticdb/processing/clause.cpp
+++ b/cpp/arcticdb/processing/clause.cpp
@@ -795,7 +795,7 @@ Composite<EntityIds> RowRangeClause::process(Composite<EntityIds> &&entity_ids)
                 if (end_ > row_range->start() && end_ < row_range->end()) {
                     end_row = end_ - (row_range->start());
                 }
-                auto truncated_segment = truncate_segment(*proc.segments_->at(idx), start_row, end_row);
+                auto truncated_segment = proc.segments_->at(idx)->truncate(start_row, end_row, false);
                 auto num_rows = truncated_segment.is_null() ? 0 : truncated_segment.row_count();
                 proc.row_ranges_->at(idx) = std::make_shared<pipelines::RowRange>(proc.row_ranges_->at(idx)->first, proc.row_ranges_->at(idx)->first + num_rows);
                 auto num_cols = truncated_segment.is_null() ? 0 : truncated_segment.descriptor().field_count() - truncated_segment.descriptor().index().field_count();
diff --git a/cpp/arcticdb/processing/processing_unit.cpp b/cpp/arcticdb/processing/processing_unit.cpp
index 53346dfcaf..f0f98c4eed 100644
--- a/cpp/arcticdb/processing/processing_unit.cpp
+++ b/cpp/arcticdb/processing/processing_unit.cpp
@@ -34,7 +34,7 @@ void ProcessingUnit::truncate(size_t start_row, size_t end_row) {
                                                     "ProcessingUnit::truncate requires all of segments, row_ranges, and col_ranges to be present");
 
     for (auto&& [idx, segment]: folly::enumerate(*segments_)) {
-        auto seg = truncate_segment(*segment, start_row, end_row);
+        auto seg = segment->truncate(start_row, end_row, false);
         auto num_rows = seg.is_null() ? 0 : seg.row_count();
         row_ranges_->at(idx) = std::make_shared<pipelines::RowRange>(row_ranges_->at(idx)->first, row_ranges_->at(idx)->first + num_rows);
         auto num_cols = seg.is_null() ? 0 : seg.descriptor().field_count() - seg.descriptor().index().field_count();
diff --git a/cpp/arcticdb/python/python_handlers.cpp b/cpp/arcticdb/python/python_handlers.cpp
index cadf3e58ef..4e9aca7f33 100644
--- a/cpp/arcticdb/python/python_handlers.cpp
+++ b/cpp/arcticdb/python/python_handlers.cpp
@@ -5,10 +5,10 @@
  * 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 <python/python_handlers.hpp>
-#include <python/gil_lock.hpp>
 #include <arcticdb/codec/slice_data_sink.hpp>
 #include <arcticdb/codec/encoding_sizes.hpp>
 #include <arcticdb/codec/codec.hpp>
+#include <arcticdb/util/buffer_holder.hpp>
 
 namespace arcticdb {
 
@@ -26,8 +26,8 @@ namespace arcticdb {
     }
 
     /// @important This calls pybind's initialize array function which is NOT thread safe. Moreover, numpy arrays can
-    /// be created only by the thread holindg the GIL. In practice we can get away with allocating arrays only from
-    /// a sigle thread (enve if it's not the one holding the GIL). This, however, is not guranteed to work.
+    /// be created only by the thread holding the GIL. In practice we can get away with allocating arrays only from
+    /// a single thread (even if it's not the one holding the GIL). This, however, is not guaranteed to work.
     /// @todo Allocate numpy arrays only from the thread holding the GIL
     [[nodiscard]] static inline PyObject* initialize_array(
         const pybind11::dtype& descr,
@@ -49,7 +49,7 @@ namespace arcticdb {
 
     static inline const PyObject** fill_with_none(const PyObject** dest, size_t count) {
         auto none = py::none();
-        std::generate(dest, dest + count, [&none]() { return none.inc_ref().ptr(); });
+        std::generate_n(dest, count, [&none]() { return none.inc_ref().ptr(); });
         return dest + count;
     }
 
@@ -57,29 +57,37 @@ namespace arcticdb {
         const uint8_t*& input,
         uint8_t* dest,
         const VariantField& variant_field,
-        const entity::TypeDescriptor&,
         size_t dest_bytes,
         std::shared_ptr<BufferHolder>,
-        EncodingVersion
+        EncodingVersion,
+        const ColumnMapping& m
     ) {
         ARCTICDB_SAMPLE(HandleEmpty, 0)
         util::check(dest != nullptr, "Got null destination pointer");
-        const size_t num_rows = dest_bytes / get_type_size(DataType::EMPTYVAL);
-        static_assert(get_type_size(DataType::EMPTYVAL) == sizeof(PyObject *));
-        auto ptr_dest = reinterpret_cast<const PyObject **>(dest);
-        fill_with_none(ptr_dest, num_rows);
+        ARCTICDB_TRACE(
+            log::version(),
+            "Empty type handler invoked for source type: {}, destination type: {}, num rows: {}",
+            m.source_type_desc_,
+            m.dest_type_desc_,
+            m.num_rows_
+        );
+        static_assert(get_type_size(DataType::EMPTYVAL) == sizeof(PyObject*));
+
+        m.dest_type_desc_.visit_tag([&](auto tag) {
+            util::default_initialize<decltype(tag)>(dest, dest_bytes);
+        });
 
-        util::variant_match(variant_field, [&input](const auto &field) {
+        util::variant_match(variant_field, [&input](const auto& field) {
             using EncodedFieldType = std::decay_t<decltype(*field)>;
             if constexpr (std::is_same_v<EncodedFieldType, arcticdb::EncodedField>)
                 util::check_magic<ColumnMagic>(input);
 
             if (field->encoding_case() == EncodedFieldType::kNdarray) {
-                const auto &ndarray_field = field->ndarray();
+                const auto& ndarray_field = field->ndarray();
                 const auto num_blocks = ndarray_field.values_size();
                 util::check(num_blocks <= 1, "Unexpected number of empty type blocks: {}", num_blocks);
                 for (auto block_num = 0; block_num < num_blocks; ++block_num) {
-                    const auto &block_info = ndarray_field.values(block_num);
+                    const auto& block_info = ndarray_field.values(block_num);
                     input += block_info.out_bytes();
                 }
             } else {
@@ -88,87 +96,96 @@ namespace arcticdb {
         });
     }
 
+    int EmptyHandler::type_size() const {
+        return sizeof(PyObject*);
+    }
+
+    void EmptyHandler::default_initialize(void* dest, size_t byte_size) const {
+        fill_with_none(reinterpret_cast<const PyObject**>(dest), byte_size / type_size());
+    }
+
     void BoolHandler::handle_type(
-            const uint8_t*& data,
-            uint8_t* dest,
-            const VariantField& encoded_field_info,
-            const entity::TypeDescriptor& type_descriptor,
-            size_t dest_bytes,
-            std::shared_ptr<BufferHolder>,
-            EncodingVersion encding_version
+        const uint8_t*& data,
+        uint8_t* dest,
+        const VariantField& encoded_field_info,
+        size_t,
+        std::shared_ptr<BufferHolder>,
+        EncodingVersion encding_version,
+        const ColumnMapping& m
     ) {
         std::visit([&](const auto& field){
             ARCTICDB_SAMPLE(HandleBool, 0)
             util::check(dest != nullptr, "Got null destination pointer");
             util::check(field->has_ndarray(), "Bool handler expected array");
             ARCTICDB_DEBUG(log::version(), "Bool handler got encoded field: {}", field->DebugString());
-            size_t row_count = dest_bytes / get_type_size(DataType::PYBOOL64);
             auto ptr_dest = reinterpret_cast<const PyObject**>(dest);
-
-            if(!field->ndarray().sparse_map_bytes()) {
-                ARCTICDB_DEBUG(log::version(), "Bool handler has no values");
-                fill_with_none(ptr_dest, row_count);
-                return;
-            }
-
-            std::optional<util::BitSet> bv;
             const auto& ndarray = field->ndarray();
             const auto bytes = encoding_sizes::data_uncompressed_size(ndarray);
-            auto sparse = ChunkedBuffer::presized(bytes);
-            SliceDataSink sparse_sink{sparse.data(), bytes};
-            data += decode_field(type_descriptor, *field, data, sparse_sink, bv, encding_version);
-            const auto num_bools = bv->count();
-            auto ptr_src = sparse.template ptr_cast<uint8_t>(0, num_bools * sizeof(uint8_t));
-            auto last_row = 0u;
-            const auto ptr_begin = ptr_dest;
-
-            for(auto en = bv->first(); en < bv->end(); ++en) {
-                const auto current_pos = *en;
-                ptr_dest = fill_with_none(ptr_dest, current_pos - last_row);
-                last_row = current_pos;
-                auto py_bool = py::bool_(static_cast<bool>(*ptr_src++));
-                *ptr_dest++ = py_bool.release().ptr();
-                ++last_row;
-                util::check(ptr_begin + last_row == ptr_dest, "Boolean pointer out of alignment {} != {}", last_row, uintptr_t(ptr_begin + last_row));
+            ChunkedBuffer decoded_data = ChunkedBuffer::presized(bytes);
+            SliceDataSink decoded_data_sink{decoded_data.data(), bytes};
+            std::optional<util::BitSet> sparse_map;
+            data += decode_field(m.source_type_desc_, *field, data, decoded_data_sink, sparse_map, encding_version);
+            const auto num_bools = sparse_map.has_value() ? sparse_map->count() : m.num_rows_;
+            auto ptr_src = decoded_data.template ptr_cast<uint8_t>(0, num_bools * sizeof(uint8_t));
+            if (sparse_map.has_value()) {
+                ARCTICDB_TRACE(log::codec(), "Bool handler using a sparse map");
+                unsigned last_row = 0u;
+                for (auto en = sparse_map->first(); en < sparse_map->end(); ++en, last_row++) {
+                    const auto current_pos = *en;
+                    ptr_dest = fill_with_none(ptr_dest, current_pos - last_row);
+                    last_row = current_pos;
+                    *ptr_dest++ = py::bool_(static_cast<bool>(*ptr_src++)).release().ptr();
+                }
+                fill_with_none(ptr_dest, m.num_rows_ - last_row);
+            } else {
+                ARCTICDB_TRACE(log::codec(), "Bool handler didn't find a sparse map. Assuming dense array.");
+                std::transform(ptr_src, ptr_src + num_bools, ptr_dest, [](uint8_t value) {
+                    return py::bool_(static_cast<bool>(value)).release().ptr();
+                });
             }
-            fill_with_none(ptr_dest, row_count - last_row);
-            util::check(ptr_begin + last_row == ptr_dest, "Boolean pointer out of alignment at end: {} != {}", last_row, uintptr_t(ptr_begin + last_row));
         }, encoded_field_info);
     }
 
+    int BoolHandler::type_size() const {
+        return sizeof(PyObject*);
+    }
+
+    void BoolHandler::default_initialize(void* dest, size_t byte_size) const {
+        fill_with_none(reinterpret_cast<const PyObject**>(dest), byte_size / type_size());
+    }
+
     std::mutex ArrayHandler::initialize_array_mutex;
 
     void ArrayHandler::handle_type(
         const uint8_t*& data,
         uint8_t* dest,
         const VariantField& encoded_field_info,
-        const entity::TypeDescriptor& type_descriptor,
-        size_t dest_bytes,
+        size_t,
         std::shared_ptr<BufferHolder> buffers,
-		EncodingVersion encoding_version
+        EncodingVersion encoding_version,
+        const ColumnMapping& m
     ) {
         util::variant_match(encoded_field_info, [&](auto field){
             ARCTICDB_SAMPLE(HandleArray, 0)
             util::check(field->has_ndarray(), "Expected ndarray in array object handler");
-            const auto row_count = dest_bytes / sizeof(PyObject*);
 
             auto ptr_dest = reinterpret_cast<const PyObject**>(dest);
             if(!field->ndarray().sparse_map_bytes()) {
                 log::version().info("Array handler has no values");
-                fill_with_none(ptr_dest, row_count);
+                fill_with_none(ptr_dest, m.num_rows_);
                 return;
             }
-            std::shared_ptr<Column> column = buffers->get_buffer(type_descriptor, true);
+            std::shared_ptr<Column> column = buffers->get_buffer(m.source_type_desc_, true);
             column->check_magic();
             log::version().info("Column got buffer at {}", uintptr_t(column.get()));
             auto bv = std::make_optional(util::BitSet{});
-            data += decode_field(type_descriptor, *field, data, *column, bv, encoding_version);
+            data += decode_field(m.source_type_desc_, *field, data, *column, bv, encoding_version);
 
             auto last_row = 0u;
             ARCTICDB_SUBSAMPLE(InitArrayAcquireGIL, 0)
-            const auto strides = static_cast<stride_t>(get_type_size(type_descriptor.data_type()));
-            const py::dtype py_dtype = generate_python_dtype(type_descriptor, strides);
-            type_descriptor.visit_tag([&] (auto tdt) {
+            const auto strides = static_cast<stride_t>(get_type_size(m.source_type_desc_.data_type()));
+            const py::dtype py_dtype = generate_python_dtype(m.source_type_desc_, strides);
+            m.source_type_desc_.visit_tag([&] (auto tdt) {
                 const auto& blocks = column->blocks();
                 if(blocks.empty())
                     return;
@@ -203,7 +220,15 @@ namespace arcticdb {
             });
 
             ARCTICDB_SUBSAMPLE(ArrayIncNones, 0)
-            fill_with_none(ptr_dest, row_count - last_row);
+            fill_with_none(ptr_dest, m.num_rows_ - last_row);
         });
     }
+
+    int ArrayHandler::type_size() const {
+        return sizeof(PyObject*);
+    }
+
+    void ArrayHandler::default_initialize(void* dest, size_t byte_size) const {
+        fill_with_none(reinterpret_cast<const PyObject**>(dest), byte_size / type_size());
+    }
 }
diff --git a/cpp/arcticdb/python/python_handlers.hpp b/cpp/arcticdb/python/python_handlers.hpp
index f6f750791b..186c87cbd9 100644
--- a/cpp/arcticdb/python/python_handlers.hpp
+++ b/cpp/arcticdb/python/python_handlers.hpp
@@ -10,17 +10,22 @@
 #include <util/type_handler.hpp>
 
 namespace arcticdb {
-    namespace py = pybind11;
+    struct ColumnMapping;
+
     struct EmptyHandler {
         /// @see arcticdb::ITypeHandler
         void handle_type(
             const uint8_t*& data,
             uint8_t* dest,
             const VariantField& encoded_field,
-            const entity::TypeDescriptor& type_descriptor,
             size_t dest_bytes,
             std::shared_ptr<BufferHolder> buffers,
-            EncodingVersion encding_version);
+            EncodingVersion encding_version,
+            const ColumnMapping& columnMapping
+        );
+
+        int type_size() const;
+        void default_initialize(void* dest, size_t byte_size) const;
     };
 
     struct BoolHandler {
@@ -29,24 +34,26 @@ namespace arcticdb {
             const uint8_t *&data,
             uint8_t *dest,
             const VariantField &encoded_field,
-            const entity::TypeDescriptor &type_descriptor,
             size_t dest_bytes,
             std::shared_ptr<BufferHolder> buffers,
-            EncodingVersion encding_version);
+            EncodingVersion encding_version,
+            const ColumnMapping& columnMapping
+        );
+        int type_size() const;
+        void default_initialize(void* dest, size_t byte_size) const;
     };
 
     struct DecimalHandler {
         void handle_type(
-                const uint8_t*& data,
-                uint8_t* dest,
-                const VariantField& encoded_field,
-                const entity::TypeDescriptor& type_descriptor,
-                size_t dest_bytes,
-                std::shared_ptr<BufferHolder> buffers
+            const uint8_t*& data,
+            uint8_t* dest,
+            const VariantField& encoded_field,
+            size_t dest_bytes,
+            std::shared_ptr<BufferHolder> buffers,
+            EncodingVersion encding_version,
+            const ColumnMapping& m
         );
-
-        std::shared_ptr<std::mutex> mutex_ = std::make_shared<std::mutex>();
-        std::shared_ptr<py::object> decimal_ = std::make_shared<py::object>(py::module_::import("decimal").attr("Decimal"));
+        int type_size() const;
     };
 
     struct ArrayHandler {
@@ -55,10 +62,13 @@ namespace arcticdb {
             const uint8_t*& data,
             uint8_t* dest,
             const VariantField& encoded_field,
-            const entity::TypeDescriptor& type_descriptor,
             size_t dest_bytes,
             std::shared_ptr<BufferHolder> buffers,
-            EncodingVersion encding_version);
+            EncodingVersion encding_version,
+            const ColumnMapping& columnMapping
+        );
+        int type_size() const;
+        void default_initialize(void* dest, size_t byte_size) const;
         static std::mutex initialize_array_mutex;
     };
 } //namespace arcticdb
diff --git a/cpp/arcticdb/python/python_module.cpp b/cpp/arcticdb/python/python_module.cpp
index cd6c554a1c..762a74cb1b 100644
--- a/cpp/arcticdb/python/python_module.cpp
+++ b/cpp/arcticdb/python/python_module.cpp
@@ -291,8 +291,7 @@ void register_type_handlers() {
     for(const DataType& data_type : allowed_array_types) {
         TypeHandlerRegistry::instance()->register_handler(TypeDescriptor{data_type, Dimension::Dim1}, arcticdb::ArrayHandler());
     }
-    TypeHandlerRegistry::instance()->register_handler(TypeDescriptor{DataType::PYBOOL64, Dimension::Dim0}, arcticdb::BoolHandler());
-    TypeHandlerRegistry::instance()->register_handler(TypeDescriptor{DataType::PYBOOL8, Dimension::Dim0}, arcticdb::BoolHandler());
+    TypeHandlerRegistry::instance()->register_handler(TypeDescriptor{DataType::BOOL_OBJECT8, Dimension::Dim0}, arcticdb::BoolHandler());
 }
 
 PYBIND11_MODULE(arcticdb_ext, m) {
diff --git a/cpp/arcticdb/python/python_to_tensor_frame.cpp b/cpp/arcticdb/python/python_to_tensor_frame.cpp
index 5da1a8e422..ce0e3f79b3 100644
--- a/cpp/arcticdb/python/python_to_tensor_frame.cpp
+++ b/cpp/arcticdb/python/python_to_tensor_frame.cpp
@@ -50,8 +50,9 @@ std::variant<StringEncodingError, PyStringWrapper> pystring_to_buffer(PyObject *
 std::tuple<ValueType, uint8_t, ssize_t> determine_python_object_type(PyObject* obj) {
     if (is_py_boolean(obj)) {
         normalization::raise<ErrorCode::E_UNIMPLEMENTED_INPUT_TYPE>("Nullable booleans are not supported at the moment");
-        return {ValueType::PYBOOL, 1, 1};
+        return {ValueType::BOOL_OBJECT, 1, 1};
     }
+
     return {ValueType::BYTES, 8, 1};
 }
 
diff --git a/cpp/arcticdb/storage/azure/azure_mock_client.cpp b/cpp/arcticdb/storage/azure/azure_mock_client.cpp
index 5b0bf3b5ce..ef848f3d48 100644
--- a/cpp/arcticdb/storage/azure/azure_mock_client.cpp
+++ b/cpp/arcticdb/storage/azure/azure_mock_client.cpp
@@ -52,7 +52,7 @@ std::optional<Azure::Core::RequestFailedException> has_failure_trigger(const std
         auto error_message = fmt::format("Simulated Error, message: #{}_{}", failure_string_for_operation, (int) status_code);
 
         return get_exception(error_message, status_code);
-    } catch (std::exception& e) {
+    } catch (std::exception&) {
         return std::nullopt;
     }
 }
diff --git a/cpp/arcticdb/storage/library.hpp b/cpp/arcticdb/storage/library.hpp
index 1979afefbd..7fa6af53dd 100644
--- a/cpp/arcticdb/storage/library.hpp
+++ b/cpp/arcticdb/storage/library.hpp
@@ -73,8 +73,11 @@ class Library {
         if (open_mode() < OpenMode::WRITE)
             throw PermissionException(library_path_, open_mode(), "write");
 
-        size_t ARCTICDB_UNUSED total_size = kvs.fold([] (size_t s, const KeySegmentPair& seg) { return s + seg.segment().total_segment_size(); }, size_t(0));
-        auto kv_count ARCTICDB_UNUSED = kvs.size();
+        [[maybe_unused]] const size_t total_size = kvs.fold(
+            [](size_t s, const KeySegmentPair& seg) { return s + seg.segment().total_segment_size(); },
+            size_t(0)
+        );
+        [[maybe_unused]] const auto kv_count = kvs.size();
         storages_->write(std::move(kvs));
         ARCTICDB_TRACE(log::storage(), "{} kv written, {} bytes", kv_count, total_size);
     }
@@ -84,8 +87,11 @@ class Library {
         if (open_mode() < OpenMode::WRITE)
             throw PermissionException(library_path_, open_mode(), "update");
 
-        size_t total_size ARCTICDB_UNUSED = kvs.fold([] (size_t s, const KeySegmentPair& seg) { return s + seg.segment().total_segment_size(); }, size_t(0));
-        auto kv_count ARCTICDB_UNUSED = kvs.size();
+        [[maybe_unused]] const size_t total_size = kvs.fold(
+            [](size_t s, const KeySegmentPair& seg) { return s + seg.segment().total_segment_size(); },
+            size_t(0)
+        );
+        [[maybe_unused]] const auto kv_count = kvs.size();
         storages_->update(std::move(kvs), opts);
         ARCTICDB_TRACE(log::storage(), "{} kv updated, {} bytes", kv_count, total_size);
     }
diff --git a/cpp/arcticdb/storage/lmdb/lmdb_storage.cpp b/cpp/arcticdb/storage/lmdb/lmdb_storage.cpp
index 015726a605..1f66d1c620 100644
--- a/cpp/arcticdb/storage/lmdb/lmdb_storage.cpp
+++ b/cpp/arcticdb/storage/lmdb/lmdb_storage.cpp
@@ -154,7 +154,7 @@ bool LmdbStorage::do_key_exists(const VariantKey&key) {
         MDB_val mdb_key{stored_key.size(), stored_key.data()};
         MDB_val mdb_val;
         return ::lmdb::dbi_get(txn, dbi.handle(), &mdb_key, &mdb_val);
-    } catch (const ::lmdb::not_found_error &ex) {
+    } catch ([[maybe_unused]] const ::lmdb::not_found_error &ex) {
         ARCTICDB_DEBUG(log::storage(), "Caught lmdb not found error: {}", ex.what());
         return false;
     }
diff --git a/cpp/arcticdb/storage/s3/s3_mock_client.cpp b/cpp/arcticdb/storage/s3/s3_mock_client.cpp
index 7ce4176944..a0fb57a3b1 100644
--- a/cpp/arcticdb/storage/s3/s3_mock_client.cpp
+++ b/cpp/arcticdb/storage/s3/s3_mock_client.cpp
@@ -52,7 +52,7 @@ std::optional<Aws::S3::S3Error> has_failure_trigger(const std::string& s3_object
         auto failure_code = Aws::S3::S3Errors(std::stoi(failure_code_string));
         bool retryable = std::stoi(s3_object_name.substr(s3_object_name.find_last_of('_') + 1));
         return Aws::S3::S3Error(Aws::Client::AWSError<Aws::S3::S3Errors>(failure_code, "Simulated error", "Simulated error message", retryable));
-    } catch (std::exception& e) {
+    } catch (std::exception&) {
         return std::nullopt;
     }
 }
diff --git a/cpp/arcticdb/stream/index.hpp b/cpp/arcticdb/stream/index.hpp
index 370e83d67e..006b695f67 100644
--- a/cpp/arcticdb/stream/index.hpp
+++ b/cpp/arcticdb/stream/index.hpp
@@ -82,7 +82,7 @@ class TimeseriesIndex : public BaseIndex<TimeseriesIndex> {
 
     void check(const FieldCollection &fields) const {
         const size_t fields_size = fields.size();
-        const int current_fields_size = int(field_count());
+        constexpr int current_fields_size = int(field_count());
 
         const TypeDescriptor &first_field_type = fields[0].type();
         const TypeDescriptor &current_first_field_type = this->field(0).type();
diff --git a/cpp/arcticdb/util/sparse_utils.hpp b/cpp/arcticdb/util/sparse_utils.hpp
index 86b87676c2..13bb5a78b0 100644
--- a/cpp/arcticdb/util/sparse_utils.hpp
+++ b/cpp/arcticdb/util/sparse_utils.hpp
@@ -9,6 +9,7 @@
 
 #include <arcticdb/util/offset_string.hpp>
 #include <arcticdb/util/preprocess.hpp>
+#include <arcticdb/util/type_handler.hpp>
 
 #include <arcticdb/util/bitset.hpp>
 #include <arcticdb/util/constants.hpp>
@@ -17,7 +18,6 @@
 
 #include <bitmagic/bmserial.h>
 
-#include <algorithm>
 #include <cstdint>
 
 namespace arcticdb::util {
@@ -68,22 +68,31 @@ inline void expand_dense_buffer_using_bitmap(const util::BitMagic &bv, const uin
 }
 
 template <typename TagType>
-void default_initialize(uint8_t* data, size_t bytes) {
+inline void default_initialize(uint8_t* data, size_t bytes) {
     using RawType = typename TagType::DataTypeTag::raw_type;
     const auto num_rows ARCTICDB_UNUSED = bytes / sizeof(RawType);
-
-    constexpr auto data_type =TagType::DataTypeTag::data_type;
+    constexpr auto data_type = TagType::DataTypeTag::data_type;
     auto type_ptr ARCTICDB_UNUSED = reinterpret_cast<RawType*>(data);
     if constexpr (is_sequence_type(data_type)) {
-        std::fill(type_ptr, type_ptr + num_rows, not_a_string());
-    }
-    else if constexpr (is_floating_point_type(data_type)) {
-        std::fill(type_ptr, type_ptr + num_rows, std::numeric_limits<double>::quiet_NaN());
-    }
-    else if constexpr (is_time_type(data_type)) {
-        std::fill(type_ptr, type_ptr + num_rows, NaT);
+        std::fill_n(type_ptr, num_rows, not_a_string());
+    } else if constexpr (is_floating_point_type(data_type)) {
+        std::fill_n(type_ptr, num_rows, std::numeric_limits<RawType>::quiet_NaN());
+    } else if constexpr (is_time_type(data_type)) {
+        std::fill_n(type_ptr, num_rows, NaT);
+    } else if constexpr (is_integer_type(data_type) || is_bool_type(data_type)) {
+        std::memset(data, 0, bytes);
     } else {
-        std::fill(data, data + bytes, 0);
+        constexpr TypeDescriptor type_descriptor = TagType::type_descriptor();
+        const std::shared_ptr<TypeHandler>& handler =
+            arcticdb::TypeHandlerRegistry::instance()->get_handler(type_descriptor);
+        if (handler) {
+            handler->default_initialize(data, bytes);
+        } else {
+            internal::raise<ErrorCode::E_INVALID_ARGUMENT>(
+                "Default initialization for {} is not implemented.",
+                type_descriptor
+            );
+        }
     }
 }
 
@@ -98,7 +107,7 @@ ChunkedBuffer scan_floating_point_to_sparse(
     util::BitMagic& block_bitset) {
     auto scan_ptr = ptr;
     for (size_t idx = 0; idx < rows_to_write; ++idx, ++scan_ptr) {
-        block_bitset[bv_size(idx)] = isnan(*scan_ptr) <= 0;
+        block_bitset[bv_size(idx)] = !isnan(*scan_ptr);
     }
 
     const auto bytes = block_bitset.count() * sizeof(RawType);
diff --git a/cpp/arcticdb/util/trace.cpp b/cpp/arcticdb/util/trace.cpp
index 41f14e050d..5b1c6991a0 100644
--- a/cpp/arcticdb/util/trace.cpp
+++ b/cpp/arcticdb/util/trace.cpp
@@ -11,20 +11,16 @@
 #include <cxxabi.h>
 #endif
 
-#include <iostream>
-
 namespace arcticdb {
 
-    std::string get_type_name(const std::type_info & ti){
+    std::string get_type_name(const std::type_info& ti){
 #ifndef _WIN32
         char* demangled = abi::__cxa_demangle(ti.name(), nullptr, nullptr, nullptr);
         std::string ret = demangled;
         free(demangled);
         return ret;
 #else
-        //TODO: Implement get_name_type for windows
-        return std::string("");
+        return ti.name();
 #endif
     }
-
 }
diff --git a/cpp/arcticdb/util/trace.hpp b/cpp/arcticdb/util/trace.hpp
index 17bfbf7ed0..146c9c9c91 100644
--- a/cpp/arcticdb/util/trace.hpp
+++ b/cpp/arcticdb/util/trace.hpp
@@ -8,14 +8,10 @@
 #pragma once
 
 #include <string>
-#include <ostream>
-#include <fstream>
-#include <fmt/format.h>
-#include <stdexcept>
 #include <typeinfo>
 
 namespace arcticdb {
 
-std::string get_type_name(const std::type_info & );
+std::string get_type_name(const std::type_info& type_info);
 
 }
diff --git a/cpp/arcticdb/util/type_handler.cpp b/cpp/arcticdb/util/type_handler.cpp
index 1bf4734e3f..8a3a1d2e41 100644
--- a/cpp/arcticdb/util/type_handler.cpp
+++ b/cpp/arcticdb/util/type_handler.cpp
@@ -21,16 +21,16 @@ void TypeHandlerRegistry::init() {
 std::shared_ptr<TypeHandlerRegistry> TypeHandlerRegistry::instance_;
 std::once_flag TypeHandlerRegistry::init_flag_;
 
-std::shared_ptr<TypeHandler> TypeHandlerRegistry::get_handler(const util::TypeDescriptor& type_descriptor) const {
+std::shared_ptr<TypeHandler> TypeHandlerRegistry::get_handler(const entity::TypeDescriptor& type_descriptor) const {
     auto it = handlers_.find(type_descriptor);
     return it == std::end(handlers_) ? std::shared_ptr<TypeHandler>{} : it->second;
 }
 
-void TypeHandlerRegistry::register_handler(const util::TypeDescriptor& type_descriptor, TypeHandler&& handler) {
+void TypeHandlerRegistry::register_handler(const entity::TypeDescriptor& type_descriptor, TypeHandler&& handler) {
      handlers_.try_emplace(type_descriptor, std::make_shared<TypeHandler>(std::move(handler)));
 }
 
-size_t TypeHandlerRegistry::Hasher::operator()(const util::TypeDescriptor descriptor) const {
+size_t TypeHandlerRegistry::Hasher::operator()(const entity::TypeDescriptor descriptor) const {
     static_assert(sizeof(descriptor) == sizeof(uint16_t), "Cannot compute util::TypeDescriptor's hash. The size is wrong.");
     static_assert(sizeof(decltype(descriptor.data_type())) == 1);
     static_assert(sizeof(decltype(descriptor.dimension())) == 1);
diff --git a/cpp/arcticdb/util/type_handler.hpp b/cpp/arcticdb/util/type_handler.hpp
index a696b0c9ec..87fcb6296f 100644
--- a/cpp/arcticdb/util/type_handler.hpp
+++ b/cpp/arcticdb/util/type_handler.hpp
@@ -8,9 +8,8 @@
 #pragma once
 
 #include <arcticdb/entity/types.hpp>
-#include <arcticdb/column_store/chunked_buffer.hpp>
-#include <arcticdb/util/buffer_holder.hpp>
 #include <arcticdb/codec/variant_encoded_field_collection.hpp>
+#include <arcticdb/pipeline/column_mapping.hpp>
 
 #include <folly/Poly.h>
 
@@ -19,6 +18,7 @@
 
 namespace arcticdb {
 
+struct BufferHolder;
 
 struct ITypeHandler {
     template<class Base>
@@ -32,16 +32,34 @@ struct ITypeHandler {
             const uint8_t*& source,
             uint8_t* dest,
             const VariantField& encoded_field_info,
-            const entity::TypeDescriptor& type_descriptor,
             size_t dest_bytes,
             std::shared_ptr<BufferHolder> buffers,
-            EncodingVersion encoding_version) {
-            folly::poly_call<0>(*this, source, dest, encoded_field_info, type_descriptor, dest_bytes, buffers, encoding_version);
+            EncodingVersion encoding_version,
+            const ColumnMapping& m
+        ) {
+            folly::poly_call<0>(
+                *this,
+                source,
+                dest,
+                encoded_field_info,
+                dest_bytes,
+                buffers,
+                encoding_version,
+                m
+            );
+        }
+
+        int type_size() const {
+            return folly::poly_call<1>(*this);
+        }
+
+        void default_initialize(void* dest, size_t byte_size) const {
+            folly::poly_call<2>(*this, dest, byte_size);
         }
     };
 
     template<class T>
-    using Members = folly::PolyMembers<&T::handle_type>;
+    using Members = folly::PolyMembers<&T::handle_type, &T::type_size, &T::default_initialize>;
 };
 
 using TypeHandler = folly::Poly<ITypeHandler>;
@@ -56,13 +74,13 @@ class TypeHandlerRegistry {
     static void init();
     static std::shared_ptr<TypeHandlerRegistry> instance();
 
-    std::shared_ptr<TypeHandler> get_handler(const util::TypeDescriptor& type_descriptor) const;
-    void register_handler(const util::TypeDescriptor& type_descriptor, TypeHandler&& handler);
+    std::shared_ptr<TypeHandler> get_handler(const entity::TypeDescriptor& type_descriptor) const;
+    void register_handler(const entity::TypeDescriptor& type_descriptor, TypeHandler&& handler);
 private:
     struct Hasher {
-        size_t operator()(const util::TypeDescriptor val) const;
+        size_t operator()(const entity::TypeDescriptor val) const;
     };
-    std::unordered_map<util::TypeDescriptor, std::shared_ptr<TypeHandler>, Hasher> handlers_;
+    std::unordered_map<entity::TypeDescriptor, std::shared_ptr<TypeHandler>, Hasher> handlers_;
 };
 
 }
diff --git a/cpp/arcticdb/version/version_core.cpp b/cpp/arcticdb/version/version_core.cpp
index 626ca8cd14..a575ed9ca5 100644
--- a/cpp/arcticdb/version/version_core.cpp
+++ b/cpp/arcticdb/version/version_core.cpp
@@ -205,22 +205,35 @@ inline std::pair<std::vector<SliceAndKey>, std::vector<SliceAndKey>> intersectin
     const IndexRange& front_range,
     const IndexRange& back_range,
     VersionId version_id,
-    const std::shared_ptr<Store>& store) {
+    const std::shared_ptr<Store>& store
+) {
     std::vector<SliceAndKey> intersect_before;
     std::vector<SliceAndKey> intersect_after;
 
     for (const auto& affected_slice_and_key : affected_keys) {
         const auto& affected_range = affected_slice_and_key.key().index_range();
-        if (intersects(affected_range, front_range) && !overlaps(affected_range, front_range)
-        && is_before(affected_range, front_range)) {
-            auto front_overlap_key = rewrite_partial_segment(affected_slice_and_key, front_range, version_id, false, store);
+        if (intersects(affected_range, front_range) && !overlaps(affected_range, front_range) &&
+            is_before(affected_range, front_range)) {
+            auto front_overlap_key = rewrite_partial_segment(
+                affected_slice_and_key,
+                front_range,
+                version_id,
+                AffectedSegmentPart::START,
+                store
+            );
             if (front_overlap_key)
                 intersect_before.push_back(*front_overlap_key);
         }
 
-        if (intersects(affected_range, back_range) && !overlaps(affected_range, back_range)
-        && is_after(affected_range, back_range)) {
-            auto back_overlap_key = rewrite_partial_segment(affected_slice_and_key, back_range, version_id, true, store);
+        if (intersects(affected_range, back_range) && !overlaps(affected_range, back_range) &&
+            is_after(affected_range, back_range)) {
+            auto back_overlap_key = rewrite_partial_segment(
+                affected_slice_and_key,
+                back_range,
+                version_id,
+                AffectedSegmentPart::END,
+                store
+            );
             if (back_overlap_key)
                 intersect_after.push_back(*back_overlap_key);
         }
@@ -265,13 +278,12 @@ VersionedItem delete_range_impl(
     auto orig_filter_range = std::holds_alternative<std::monostate>(query.row_filter) ? get_query_index_range(index, index_range) : query.row_filter;
 
     size_t row_count = 0;
-    auto flattened_slice_and_keys = flatten_and_fix_rows(std::vector<std::vector<SliceAndKey>>{
+    const std::array<std::vector<SliceAndKey>, 5> groups{
         strictly_before(orig_filter_range, unaffected_keys),
         std::move(intersect_before),
         std::move(intersect_after),
-        strictly_after(orig_filter_range, unaffected_keys)},
-                                                         row_count
-                                                         );
+        strictly_after(orig_filter_range, unaffected_keys)};
+    auto flattened_slice_and_keys = flatten_and_fix_rows(groups, row_count);
 
     std::sort(std::begin(flattened_slice_and_keys), std::end(flattened_slice_and_keys));
     bool bucketize_dynamic = index_segment_reader.bucketize_dynamic();
@@ -369,14 +381,13 @@ VersionedItem update_impl(
 
     size_t row_count = 0;
     const size_t new_keys_size = new_slice_and_keys.size();
-    auto flattened_slice_and_keys = flatten_and_fix_rows(std::vector<std::vector<SliceAndKey>>{
+    const std::array<std::vector<SliceAndKey>, 5> groups{
         strictly_before(orig_filter_range, unaffected_keys),
         std::move(intersect_before),
         std::move(new_slice_and_keys),
         std::move(intersect_after),
-        strictly_after(orig_filter_range, unaffected_keys)},
-                                                         row_count
-                                                         );
+        strictly_after(orig_filter_range, unaffected_keys)};
+    auto flattened_slice_and_keys = flatten_and_fix_rows(groups, row_count);
 
     util::check(unaffected_keys.size() + new_keys_size + (affected_keys.size() * 2) >= flattened_slice_and_keys.size(),
                 "Output size mismatch: {} + {} + (2 * {}) < {}",
diff --git a/cpp/proto/arcticc/pb2/descriptors.proto b/cpp/proto/arcticc/pb2/descriptors.proto
index 17e0bd0b2a..0dcf769a35 100644
--- a/cpp/proto/arcticc/pb2/descriptors.proto
+++ b/cpp/proto/arcticc/pb2/descriptors.proto
@@ -31,7 +31,7 @@ message TypeDescriptor {
         DYNAMIC_STRING = 11; // offset reference to deduplicated string pool
         DYNAMIC_UTF8 = 12;
         EMPTY = 13;
-        PYBOOL = 14; // Nullable boolean
+        BOOL_OBJECT = 14; // Nullable boolean
     }
 
     enum SizeBits {


Test patch:
diff --git a/cpp/arcticdb/column_store/test/test_memory_segment.cpp b/cpp/arcticdb/column_store/test/test_memory_segment.cpp
index 6782993c5f..3416242e75 100644
--- a/cpp/arcticdb/column_store/test/test_memory_segment.cpp
+++ b/cpp/arcticdb/column_store/test/test_memory_segment.cpp
@@ -17,6 +17,7 @@
 
 #include <cstdint>
 #include <algorithm>
+#include <type_traits>
 
 TEST(MemSegment, Empty) {
     using namespace arcticdb;
@@ -138,9 +139,9 @@ TEST(MemSegment, IterateWithEmptyTypeColumn) {
     for (auto&& [idx, row]: folly::enumerate(seg)) {
         ASSERT_EQ(static_cast<int64_t>(idx), row.scalar_at<int64_t>(0));
         // Exception should be thrown regardless of the type requested for empty type columns
-        EXPECT_THROW(row.scalar_at<int64_t>(1).has_value(), InternalException);
-        EXPECT_THROW(row.scalar_at<float>(1).has_value(), InternalException);
-        EXPECT_THROW(row.scalar_at<uint8_t>(1).has_value(), InternalException);
+        EXPECT_THROW([[maybe_unused]] auto v = row.scalar_at<int64_t>(1).has_value(), InternalException);
+        EXPECT_THROW([[maybe_unused]] auto v = row.scalar_at<float>(1).has_value(), InternalException);
+        EXPECT_THROW([[maybe_unused]] auto v = row.scalar_at<uint8_t>(1).has_value(), InternalException);
     }
 }
 
@@ -176,7 +177,13 @@ TEST(MemSegment, ModifyViaIterator) {
     auto &segment = frame_wrapper.segment_;
     for (auto &row : segment) {
         for (auto &value : row) {
-            value.visit([](auto &v) { v += 1; });
+            value.visit([](auto& v) {
+                if constexpr (std::is_same_v<std::remove_cv_t<std::remove_reference_t<decltype(v)>>, bool>) {
+                    v |= 1;
+                } else {
+                    v += 1;
+                }
+            });
         }
     }
 
@@ -479,9 +486,9 @@ TEST(MemSegment, Filter) {
     for (auto&& [idx, row]: folly::enumerate(filtered_seg)) {
         ASSERT_EQ(static_cast<int64_t>(retained_rows[idx]), row.scalar_at<int64_t>(0));
         // Exception should be thrown regardless of the type requested for empty type columns
-        EXPECT_THROW(row.scalar_at<int64_t>(1).has_value(), InternalException);
-        EXPECT_THROW(row.scalar_at<float>(1).has_value(), InternalException);
-        EXPECT_THROW(row.scalar_at<uint8_t>(1).has_value(), InternalException);
+        EXPECT_THROW([[maybe_unused]] auto v = row.scalar_at<int64_t>(1).has_value(), InternalException);
+        EXPECT_THROW([[maybe_unused]] auto v = row.scalar_at<float>(1).has_value(), InternalException);
+        EXPECT_THROW([[maybe_unused]] auto v = row.scalar_at<uint8_t>(1).has_value(), InternalException);
     }
 }
 
diff --git a/cpp/arcticdb/pipeline/test/test_pipeline.cpp b/cpp/arcticdb/pipeline/test/test_pipeline.cpp
index ac3799d929..0c263c068d 100644
--- a/cpp/arcticdb/pipeline/test/test_pipeline.cpp
+++ b/cpp/arcticdb/pipeline/test/test_pipeline.cpp
@@ -182,7 +182,9 @@ TEST(Pipeline, Basic) {
     Pipeline pipeline(ex);
     TestFilter even_filter{[](const SegmentInMemory::Row& row) {
         return row[0].visit([](auto val) {
-            if constexpr(std::is_integral_v<decltype(val)>)
+            if constexpr (std::is_same_v<bool, decltype(val)>)
+                return val == false;
+            else if constexpr(std::is_integral_v<decltype(val)>)
                 return val % 2 == 0;
             else
                 return false;
diff --git a/python/tests/conftest.py b/python/tests/conftest.py
index 6c76b10066..1b6cb6f7ec 100644
--- a/python/tests/conftest.py
+++ b/python/tests/conftest.py
@@ -440,14 +440,15 @@ def lmdb_version_store_v2(version_store_factory, lib_name):
     return version_store_factory(dynamic_strings=True, encoding_version=int(EncodingVersion.V2), name=library_name)
 
 
-@pytest.fixture
-def lmdb_version_store(lmdb_version_store_v1, lmdb_version_store_v2, encoding_version):
-    if encoding_version == EncodingVersion.V1:
-        return lmdb_version_store_v1
-    elif encoding_version == EncodingVersion.V2:
-        return lmdb_version_store_v2
-    else:
-        raise ValueError(f"Unexpected encoding version: {encoding_version}")
+@pytest.fixture(
+    scope="function",
+    params=(
+        "lmdb_version_store_v1",
+        "lmdb_version_store_v2",
+    )
+)
+def lmdb_version_store(request):
+    yield request.getfixturevalue(request.param)
 
 
 @pytest.fixture
@@ -694,6 +695,22 @@ def get_df(ts, width, max_col_width):
     return get_df
 
 
+@pytest.fixture(
+    scope="function",
+    params=(
+        "lmdb_version_store_v1",
+        "lmdb_version_store_v2",
+        "lmdb_version_store_dynamic_schema_v1",
+        "lmdb_version_store_dynamic_schema_v2",
+    ),
+)
+def lmdb_version_store_static_and_dynamic(request):
+    """
+    Designed to test all combinations between schema and encoding version for LMDB
+    """
+    yield request.getfixturevalue(request.param)
+
+
 @pytest.fixture(
     scope="function",
     params=(
diff --git a/python/tests/unit/arcticdb/version_store/test_empty_column_type.py b/python/tests/unit/arcticdb/version_store/test_empty_column_type.py
index 1dc64f9881..e033889d49 100644
--- a/python/tests/unit/arcticdb/version_store/test_empty_column_type.py
+++ b/python/tests/unit/arcticdb/version_store/test_empty_column_type.py
@@ -5,40 +5,515 @@
 
 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.
 """
+from math import nan
 import pandas as pd
 from pandas.testing import assert_frame_equal
 import numpy as np
 import pytest
 
-@pytest.fixture(params=(int, float, str))
-def element_type_constructor(request):
+@pytest.fixture(params=[t + s for s in ["8", "16", "32", "64"] for t in ["int", "uint"]])
+def int_dtype(request):
     yield request.param
 
 
-@pytest.mark.xfail(reason="Not implemented")
-class TestEmptyColumnTypeAppend:
-    def test_can_append_non_empty_to_empty(self, lmdb_version_store, element_type_constructor):
-        df_empty = pd.DataFrame({"col1": [None, None]})
-        lmdb_version_store.write("test_can_append_to_empty_column", df_empty)
-        df_non_empty = pd.DataFrame({"col1": map(element_type_constructor, [1, 2, 3])})
-        lmdb_version_store.append("test_can_append_to_empty_column", df_non_empty)
-        expected_result = pd.concat([df_empty, df_non_empty], ignore_index=True)
-        assert_frame_equal(lmdb_version_store.read("test_can_append_to_empty_column").data, expected_result)
-
-    def test_can_append_empty_to_non_empty(self, lmdb_version_store, element_type_constructor):
-        df_empty = pd.DataFrame({"col1": [None, None]})
-        df_non_empty = pd.DataFrame({"col1": map(element_type_constructor, [1, 2, 3])})
-        lmdb_version_store.write("test_can_append_to_empty_column", df_non_empty)
-        lmdb_version_store.append("test_can_append_to_empty_column", df_empty)
-        expected_result = pd.concat([df_non_empty, df_empty], ignore_index=True)
-        assert_frame_equal(lmdb_version_store.read("test_can_append_to_empty_column").data, expected_result)
-
-    def test_can_append_empty_to_empty(self, lmdb_version_store, element_type_constructor):
-        df_empty_1 = pd.DataFrame({"col1": [None, None]})
-        df_empty_2 = pd.DataFrame({"col1": []})
-        df_empty_3 = pd.DataFrame({"col1": [None]})
-        lmdb_version_store.write("test_can_append_to_empty_column", df_empty_1)
-        lmdb_version_store.append("test_can_append_to_empty_column", df_empty_2)
-        lmdb_version_store.append("test_can_append_to_empty_column", df_empty_3)
-        expected_result = pd.concat([df_empty_1, df_empty_2, df_empty_3], ignore_index=True)
-        assert_frame_equal(lmdb_version_store.read("test_can_append_to_empty_column").data, expected_result)
+@pytest.fixture(params=["float" + s for s in ["32", "64"]])
+def float_dtype(request):
+    yield request.param
+
+
+# object is for nullable boolean
+@pytest.fixture(params=["object", "bool"])
+def boolean_dtype(request):
+    if request.param == "object":
+        pytest.skip("Nullable booleans are temporary disabled")
+    yield request.param
+
+
+@pytest.fixture(params=["datetime64[ns]"])
+def date_dtype(request):
+    yield request.param
+
+
+class TestCanAppendToEmptyColumn:
+    @pytest.fixture(autouse=True)
+    def create_empty_column(self, lmdb_version_store_static_and_dynamic):
+        lmdb_version_store_static_and_dynamic.write("sym", pd.DataFrame({"col1": 2 * [None]}))
+        yield
+
+    def test_integer(self, lmdb_version_store_static_and_dynamic, int_dtype):
+        df_non_empty = pd.DataFrame({"col1": np.array([1,2,3], dtype=int_dtype)})
+        lmdb_version_store_static_and_dynamic.append("sym", df_non_empty)
+        expected_result = pd.DataFrame({"col1": np.array([0,0,1,2,3], dtype=int_dtype)})
+        assert_frame_equal(lmdb_version_store_static_and_dynamic.read("sym").data, expected_result)
+        assert_frame_equal(
+            lmdb_version_store_static_and_dynamic.read("sym", row_range=[0,2]).data,
+            pd.DataFrame({"col1": np.array([0,0], dtype=int_dtype)})
+        )
+        assert_frame_equal(
+            lmdb_version_store_static_and_dynamic.read("sym", row_range=[2,5]).data,
+            df_non_empty
+        )
+
+    def test_float(self, lmdb_version_store_static_and_dynamic, float_dtype):
+        df_non_empty = pd.DataFrame({"col1": np.array([1,2,3], dtype=float_dtype)})
+        lmdb_version_store_static_and_dynamic.append("sym", df_non_empty)
+        expected_result = pd.DataFrame({"col1": np.array([float("NaN"),float("NaN"),1,2,3], dtype=float_dtype)})
+        assert_frame_equal(lmdb_version_store_static_and_dynamic.read("sym").data, expected_result)
+        assert_frame_equal(
+            lmdb_version_store_static_and_dynamic.read("sym", row_range=[0,2]).data,
+            pd.DataFrame({"col1": np.array([float("NaN"),float("NaN")], dtype=float_dtype)})
+        )
+        assert_frame_equal(
+            lmdb_version_store_static_and_dynamic.read("sym", row_range=[2,5]).data,
+            df_non_empty
+        )
+
+    def test_bool(self, lmdb_version_store_static_and_dynamic, boolean_dtype):
+        # Note: if dtype is bool pandas will convert None to False
+        df_non_empty = pd.DataFrame({"col1": np.array([True, False, True], dtype=boolean_dtype)})
+        lmdb_version_store_static_and_dynamic.append("sym", df_non_empty)
+        expected_result = pd.DataFrame({"col1": np.array([None, None, True, False, True], dtype=boolean_dtype)})
+        assert_frame_equal(lmdb_version_store_static_and_dynamic.read("sym").data, expected_result)
+        assert_frame_equal(
+            lmdb_version_store_static_and_dynamic.read("sym", row_range=[0,2]).data,
+            pd.DataFrame({"col1": np.array([None, None], dtype=boolean_dtype)})
+        )
+        assert_frame_equal(
+            lmdb_version_store_static_and_dynamic.read("sym", row_range=[2,5]).data,
+            df_non_empty
+        )
+
+    def test_empty(self, lmdb_version_store_static_and_dynamic):
+        df_non_empty = pd.DataFrame({"col1": np.array([None, None, None])})
+        lmdb_version_store_static_and_dynamic.append("sym", df_non_empty)
+        expected_result = pd.DataFrame({"col1": np.array([None, None, None, None, None])})
+        assert_frame_equal(lmdb_version_store_static_and_dynamic.read("sym").data, expected_result)
+        assert_frame_equal(
+            lmdb_version_store_static_and_dynamic.read("sym", row_range=[0,2]).data,
+            pd.DataFrame({"col1": np.array([None, None])})
+        )
+        assert_frame_equal(
+            lmdb_version_store_static_and_dynamic.read("sym", row_range=[2,5]).data,
+            df_non_empty
+        )
+
+    def test_string(self, lmdb_version_store_static_and_dynamic):
+        df_non_empty = pd.DataFrame({"col1": np.array(["some_string", "long_string"*100])})
+        lmdb_version_store_static_and_dynamic.append("sym", df_non_empty)
+        expected_result = pd.DataFrame({"col1": np.array([None, None, "some_string", "long_string"*100])})
+        assert_frame_equal(lmdb_version_store_static_and_dynamic.read("sym").data, expected_result)
+        assert_frame_equal(
+            lmdb_version_store_static_and_dynamic.read("sym", row_range=[0,2]).data,
+            pd.DataFrame({"col1": np.array([None, None])})
+        )
+        assert_frame_equal(
+            lmdb_version_store_static_and_dynamic.read("sym", row_range=[2,5]).data,
+            df_non_empty
+        )
+
+    def test_date(self, lmdb_version_store_static_and_dynamic, date_dtype):
+        df_non_empty = pd.DataFrame(
+            {
+                "col1": np.array(
+                    [
+                        np.datetime64('2005-02'),
+                        np.datetime64('2005-03'),
+                        np.datetime64('2005-03')
+                    ]
+                )
+            },
+            dtype=date_dtype
+        )
+        lmdb_version_store_static_and_dynamic.append("sym", df_non_empty)
+        expected_result = pd.DataFrame(
+            {
+                "col1": np.array(
+                    [
+                        np.datetime64('NaT'),
+                        np.datetime64('NaT'),
+                        np.datetime64('2005-02'),
+                        np.datetime64('2005-03'),
+                        np.datetime64('2005-03')
+                    ],
+                    dtype=date_dtype
+                )
+            }
+        )
+        assert_frame_equal(lmdb_version_store_static_and_dynamic.read("sym").data, expected_result)
+        assert_frame_equal(
+            lmdb_version_store_static_and_dynamic.read("sym", row_range=[0,2]).data,
+            pd.DataFrame({"col1": np.array([np.datetime64('NaT'), np.datetime64('NaT')], dtype=date_dtype)})
+        )
+        assert_frame_equal(
+            lmdb_version_store_static_and_dynamic.read("sym", row_range=[2,5]).data,
+            df_non_empty
+        )
+
+
+class TestCanAppendEmptyToColumn:
+    def test_integer(self, lmdb_version_store_static_and_dynamic, int_dtype):
+        df_initial = pd.DataFrame({"col1": np.array([1,2,3], dtype=int_dtype)})
+        lmdb_version_store_static_and_dynamic.write("sym", df_initial)
+        lmdb_version_store_static_and_dynamic.append("sym", pd.DataFrame({"col1": [None, None]}))
+        assert_frame_equal(
+            lmdb_version_store_static_and_dynamic.read("sym").data,
+            pd.DataFrame({"col1": np.array([1,2,3,0,0], dtype=int_dtype)})
+        )
+        assert_frame_equal(
+            lmdb_version_store_static_and_dynamic.read("sym", row_range=[0,3]).data,
+            df_initial
+        )
+        assert_frame_equal(
+            lmdb_version_store_static_and_dynamic.read("sym", row_range=[3,5]).data,
+            pd.DataFrame({"col1": np.array([0,0], dtype=int_dtype)})
+        )
+
+    def test_float(self, lmdb_version_store_static_and_dynamic, float_dtype):
+        df_initial = pd.DataFrame({"col1": np.array([1,2,3], dtype=float_dtype)})
+        lmdb_version_store_static_and_dynamic.write("sym", df_initial)
+        lmdb_version_store_static_and_dynamic.append("sym", pd.DataFrame({"col1": [None, None]}))
+        assert_frame_equal(
+            lmdb_version_store_static_and_dynamic.read("sym").data,
+            pd.DataFrame({"col1": np.array([1,2,3,float("NaN"),float("NaN")], dtype=float_dtype)})
+        )
+        assert_frame_equal(
+            lmdb_version_store_static_and_dynamic.read("sym", row_range=[0,3]).data,
+            df_initial
+        )
+        assert_frame_equal(
+            lmdb_version_store_static_and_dynamic.read("sym", row_range=[3,5]).data,
+            pd.DataFrame({"col1": np.array([float("NaN"), float("NaN")], dtype=float_dtype)})
+        )
+
+    def test_bool(self, lmdb_version_store_static_and_dynamic, boolean_dtype):
+        # Note: if dtype is bool pandas will convert None to False
+        df_initial = pd.DataFrame({"col1": np.array([True, False, True], dtype=boolean_dtype)})
+        df_with_none = pd.DataFrame({"col1": np.array([None, None])})
+        lmdb_version_store_static_and_dynamic.write("sym", df_initial)
+        lmdb_version_store_static_and_dynamic.append("sym", df_with_none)
+        assert_frame_equal(
+            lmdb_version_store_static_and_dynamic.read("sym").data,
+            pd.DataFrame({"col1": np.array([True, False, True, None, None], dtype=boolean_dtype)})
+        )
+        assert_frame_equal(
+            lmdb_version_store_static_and_dynamic.read("sym", row_range=[0,3]).data,
+            df_initial
+        )
+        assert_frame_equal(
+            lmdb_version_store_static_and_dynamic.read("sym", row_range=[3,5]).data,
+            pd.DataFrame(df_with_none, dtype=boolean_dtype)
+        )
+
+    def test_string(self, lmdb_version_store_static_and_dynamic):
+        df_initial = pd.DataFrame({"col1": np.array(["some_string", "long_string"*100, ""])})
+        df_with_none = pd.DataFrame({"col1": np.array([None, None])})
+        lmdb_version_store_static_and_dynamic.write("sym", df_initial)
+        lmdb_version_store_static_and_dynamic.append("sym", df_with_none)
+        assert_frame_equal(
+            lmdb_version_store_static_and_dynamic.read("sym").data,
+            pd.DataFrame({"col1": np.array(["some_string", "long_string"*100, "", None, None])})
+        )
+        assert_frame_equal(
+            lmdb_version_store_static_and_dynamic.read("sym", row_range=[0,3]).data,
+            df_initial
+        )
+        assert_frame_equal(
+            lmdb_version_store_static_and_dynamic.read("sym", row_range=[3,5]).data,
+            df_with_none
+        )
+
+    def test_date(self, lmdb_version_store_static_and_dynamic, date_dtype):
+        df_initial = pd.DataFrame(
+            {
+                "col1": np.array(
+                    [
+                        np.datetime64('2005-02'),
+                        np.datetime64('2005-03'),
+                        np.datetime64('2005-03')
+                    ]
+                )
+            },
+            dtype=date_dtype
+        )
+        lmdb_version_store_static_and_dynamic.write("sym", df_initial)
+        lmdb_version_store_static_and_dynamic.append("sym", pd.DataFrame({"col1": np.array([None, None])}))
+        expected_result = pd.DataFrame(
+            {
+                "col1": np.array(
+                    [
+                        np.datetime64('2005-02'),
+                        np.datetime64('2005-03'),
+                        np.datetime64('2005-03'),
+                        np.datetime64('NaT'),
+                        np.datetime64('NaT')
+                    ],
+                    dtype=date_dtype)
+            }
+        )
+        assert_frame_equal(lmdb_version_store_static_and_dynamic.read("sym").data, expected_result)
+        assert_frame_equal(
+            lmdb_version_store_static_and_dynamic.read("sym", row_range=[0,3]).data,
+            df_initial
+        )
+        assert_frame_equal(
+            lmdb_version_store_static_and_dynamic.read("sym", row_range=[3,5]).data,
+            pd.DataFrame({"col1": np.array([np.datetime64('NaT'), np.datetime64('NaT')], dtype=date_dtype)})
+        )
+
+
+class TestCanUpdateWithEmpty:
+    def index(self):
+        return list(pd.date_range(start="1/1/2024", end="1/4/2024"))
+
+    def update_index(self):
+        return list(pd.date_range(start="1/2/2024", end="1/3/2024"))
+
+    def test_int(self, lmdb_version_store_static_and_dynamic, int_dtype):
+        lmdb_version_store_static_and_dynamic.write(
+            "sym",
+            pd.DataFrame({"col": [1, 2, 3, 4]}, dtype=int_dtype, index=self.index())
+        )
+        lmdb_version_store_static_and_dynamic.update(
+            "sym",
+            pd.DataFrame({"col": [None, None]}, index=self.update_index())
+        )
+        assert_frame_equal(
+            lmdb_version_store_static_and_dynamic.read("sym").data,
+            pd.DataFrame({"col": [1, 0, 0, 4]}, index=self.index(), dtype=int_dtype)
+        )
+
+    def test_float(self, lmdb_version_store_static_and_dynamic, float_dtype):
+        lmdb_version_store_static_and_dynamic.write(
+            "sym",
+            pd.DataFrame({"col": [1, 2, 3, 4]}, dtype=float_dtype, index=self.index())
+        )
+        lmdb_version_store_static_and_dynamic.update(
+            "sym",
+            pd.DataFrame({"col": [None, None]}, index=self.update_index())
+        )
+        assert_frame_equal(
+            lmdb_version_store_static_and_dynamic.read("sym").data,
+            pd.DataFrame({"col": [1, float("NaN"), float("NaN"), 4]}, index=self.index(), dtype=float_dtype)
+        )
+
+    def test_bool(self, lmdb_version_store_static_and_dynamic, boolean_dtype):
+        # Note: if dtype is bool pandas will convert None to False
+        lmdb_version_store_static_and_dynamic.write(
+            "sym",
+            pd.DataFrame({"col": [True, True, True, True]}, dtype=boolean_dtype, index=self.index())
+        )
+        lmdb_version_store_static_and_dynamic.update(
+            "sym",
+            pd.DataFrame({"col": [None, None]}, dtype=boolean_dtype, index=self.update_index())
+        )
+        assert_frame_equal(
+            lmdb_version_store_static_and_dynamic.read("sym").data,
+            pd.DataFrame({"col": [True, None, None, True]}, index=self.index(), dtype=boolean_dtype)
+        )
+
+    def test_string(self, lmdb_version_store_static_and_dynamic):
+        lmdb_version_store_static_and_dynamic.write(
+            "sym",
+            pd.DataFrame({"col": ["a", "longstr"*20, "b", "longstr"*20]}, index=self.index())
+        )
+        lmdb_version_store_static_and_dynamic.update(
+            "sym",
+            pd.DataFrame({"col": [None, None]}, index=self.update_index())
+        )
+        assert_frame_equal(
+            lmdb_version_store_static_and_dynamic.read("sym").data,
+            pd.DataFrame({"col": ["a", None, None, "longstr"*20]}, index=self.index())
+        )
+
+    def test_date(self, lmdb_version_store_static_and_dynamic, date_dtype):
+        lmdb_version_store_static_and_dynamic.write(
+            "sym",
+            pd.DataFrame(
+                {
+                    "col": [
+                        np.datetime64('2005-02'),
+                        np.datetime64('2005-03'),
+                        np.datetime64('2005-04'),
+                        np.datetime64('2005-05')
+                    ]
+                },
+                dtype=date_dtype,
+                index=self.index()
+            )
+        )
+        lmdb_version_store_static_and_dynamic.update(
+            "sym",
+            pd.DataFrame({"col": [None, None]}, index=self.update_index())
+        )
+        assert_frame_equal(
+            lmdb_version_store_static_and_dynamic.read("sym").data,
+            pd.DataFrame(
+                {
+                    "col": [
+                        np.datetime64('2005-02'),
+                        np.datetime64('NaT'),
+                        np.datetime64('NaT'),
+                        np.datetime64('2005-05')
+                    ]
+                },
+                index=self.index()
+            )
+        )
+
+
+class TestCanUpdateEmpty:
+    """
+    Check that a column containing only empty type and date index can be updated and that
+    the type of the column will be changed to the type of the new data. The update range
+    is right in the middle of the initial date range, so there will be values unaffected
+    by the update. Check that on read the type of the unaffected entries will be changed
+    from empty to the new type of the column.
+    """
+
+
+    def index(self):
+        return list(pd.date_range(start="1/1/2024", end="1/4/2024"))
+
+    def update_index(self):
+        return list(pd.date_range(start="1/2/2024", end="1/3/2024"))
+
+    @pytest.fixture(autouse=True)
+    def create_empty_column(self, lmdb_version_store_static_and_dynamic):
+        lmdb_version_store_static_and_dynamic.write("sym", pd.DataFrame({"col": 4 * [None]}, index=self.index()))
+        yield
+
+    def test_integer(self, lmdb_version_store_static_and_dynamic, int_dtype):
+        lmdb_version_store_static_and_dynamic.update(
+            "sym",
+            pd.DataFrame({"col": [1, 2]}, dtype=int_dtype, index=self.update_index())
+        )
+        assert_frame_equal(
+            lmdb_version_store_static_and_dynamic.read("sym").data,
+            pd.DataFrame({"col": [0,1,2,0]}, dtype=int_dtype, index=self.index())
+        )
+        assert_frame_equal(
+            lmdb_version_store_static_and_dynamic.read(
+                "sym",
+                date_range=(pd.to_datetime("1/1/2024"), pd.to_datetime("1/1/2024"))
+            ).data,
+            pd.DataFrame({"col": [0]}, dtype=int_dtype, index=[pd.to_datetime("1/1/2024")])
+        )
+
+    def test_float(self, lmdb_version_store_static_and_dynamic, float_dtype):
+        lmdb_version_store_static_and_dynamic.update(
+            "sym",
+            pd.DataFrame({"col": [1, 2]}, dtype=float_dtype, index=self.update_index())
+        )
+        assert_frame_equal(
+            lmdb_version_store_static_and_dynamic.read("sym").data,
+            pd.DataFrame(
+                {"col": [float("NaN"), 1, 2, float("NaN")]},
+                dtype=float_dtype,
+                index=self.index()
+            )
+        )
+        assert_frame_equal(
+            lmdb_version_store_static_and_dynamic.read(
+                "sym",
+                date_range=(pd.to_datetime("1/1/2024"), pd.to_datetime("1/1/2024"))
+            ).data,
+            pd.DataFrame({"col": [float("NaN")]}, dtype=float_dtype, index=[pd.to_datetime("1/1/2024")])
+        )
+
+    def test_bool(self, lmdb_version_store_static_and_dynamic, boolean_dtype):
+        # Note: if dtype is bool pandas will convert None to False
+        lmdb_version_store_static_and_dynamic.update(
+            "sym",
+            pd.DataFrame({"col": [True, False]}, dtype=boolean_dtype, index=self.update_index())
+        )
+        assert_frame_equal(
+            lmdb_version_store_static_and_dynamic.read("sym").data,
+            pd.DataFrame(
+                {"col": [None, True, False, None]},
+                dtype=boolean_dtype,
+                index=self.index()
+            )
+        )
+        assert_frame_equal(
+            lmdb_version_store_static_and_dynamic.read(
+                "sym",
+                date_range=(pd.to_datetime("1/1/2024"), pd.to_datetime("1/1/2024"))
+            ).data,
+            pd.DataFrame({"col": [None]}, dtype=boolean_dtype, index=[pd.to_datetime("1/1/2024")])
+        )
+
+    def test_string(self, lmdb_version_store_static_and_dynamic):
+        lmdb_version_store_static_and_dynamic.update(
+            "sym",
+            pd.DataFrame({"col": ["a", 20*"long_string"]}, index=self.update_index())
+        )
+        assert_frame_equal(
+            lmdb_version_store_static_and_dynamic.read("sym").data,
+            pd.DataFrame(
+                {"col": [None, "a", 20*"long_string", None]},
+                index=self.index()
+            )
+        )
+        assert_frame_equal(
+            lmdb_version_store_static_and_dynamic.read(
+                "sym",
+                date_range=(pd.to_datetime("1/1/2024"), pd.to_datetime("1/1/2024"))
+            ).data,
+            pd.DataFrame({"col": [None]}, index=[pd.to_datetime("1/1/2024")])
+        )
+
+    def test_empty(self, lmdb_version_store_static_and_dynamic):
+        lmdb_version_store_static_and_dynamic.update(
+            "sym",
+            pd.DataFrame({"col": 2*[None]}, index=self.update_index())
+        )
+        assert_frame_equal(
+            lmdb_version_store_static_and_dynamic.read("sym").data,
+            pd.DataFrame({"col": 4*[None]}, index=self.index())
+        )
+
+    def test_date(self, lmdb_version_store_static_and_dynamic, date_dtype):
+        lmdb_version_store_static_and_dynamic.update(
+            "sym",
+            pd.DataFrame(
+                {"col": [np.datetime64('2005-02'), np.datetime64('2005-03')]},
+                dtype=date_dtype,
+                index=self.update_index()
+            )
+        )
+        assert_frame_equal(
+            lmdb_version_store_static_and_dynamic.read("sym").data,
+            pd.DataFrame(
+                {
+                    "col": [
+                        np.datetime64('NaT'),
+                        np.datetime64('2005-02'),
+                        np.datetime64('2005-03'),
+                        np.datetime64('NaT')
+                    ]
+                },
+                index=self.index(),
+                dtype=date_dtype
+            )
+        )
+        assert_frame_equal(
+            lmdb_version_store_static_and_dynamic.read(
+                "sym",
+                date_range=(pd.to_datetime("1/1/2024"), pd.to_datetime("1/1/2024"))
+            ).data,
+            pd.DataFrame(
+                {
+                    "col": [np.datetime64('NaT', 'ns')]
+                },
+                index=[pd.to_datetime("1/1/2024")],
+                dtype=date_dtype
+            )
+        )
+
+class TestEmptyTypesIsOverriden:
+    def test_cannot_append_different_type_after_first_not_none(self, lmdb_version_store_static_and_dynamic):
+        lmdb_version_store_static_and_dynamic.write("sym", pd.DataFrame({"col": [None, None]}))
+        lmdb_version_store_static_and_dynamic.append("sym", pd.DataFrame({"col": [1, 2, 3]}))
+        lmdb_version_store_static_and_dynamic.append("sym", pd.DataFrame({"col": [None, None]}))
+        with pytest.raises(Exception):
+            lmdb_version_store_static_and_dynamic.append("sym", pd.DataFrame({"col": ["some", "string"]}))
\ No newline at end of file