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

Issue:
date_and_time does not work with pre-epoch timestamps
### Describe the bug

[date_and_time](https://github.com/man-group/ArcticDB/blob/52cdc81b7f175ad410b4a3626b366e3ab1aec3a1/cpp/arcticdb/util/timer.hpp#L296) uses gmtime which does not work with pre-epoch dates in that case gmtime returns nullptr.

### Steps/Code to Reproduce

N/A

### Expected Results

Make it so that it works with pre-epoch dates or throw reasonable exception.

### OS, Python Version and ArcticDB Version

Python: 3.10.11 (tags/v3.10.11:7d4cc5a, Apr 5 2023, 00:38:17) [MSC v.1929 64 bit (AMD64)]
OS: Windows-10-10.0.22631-SP0
ArcticDB: 4.5.0

### Backend storage used

_No response_

### Additional Context

_No response_


Discussion / hints:

Fix patch:
diff --git a/cpp/arcticdb/CMakeLists.txt b/cpp/arcticdb/CMakeLists.txt
index ef9ddd12e3..ed01826ec2 100644
--- a/cpp/arcticdb/CMakeLists.txt
+++ b/cpp/arcticdb/CMakeLists.txt
@@ -491,6 +491,7 @@ set(arcticdb_srcs
         util/offset_string.cpp
         util/sparse_utils.cpp
         util/string_utils.cpp
+        util/timer.cpp
         util/trace.cpp
         util/type_handler.cpp
         version/local_versioned_engine.cpp
diff --git a/cpp/arcticdb/column_store/column.hpp b/cpp/arcticdb/column_store/column.hpp
index e97737f1e0..a5e7c50a42 100644
--- a/cpp/arcticdb/column_store/column.hpp
+++ b/cpp/arcticdb/column_store/column.hpp
@@ -277,10 +277,15 @@ class Column {
         return TypedColumnIterator<TagType, const RawType>(*this, false);
     }
 
-    template<class T, std::enable_if_t<std::is_integral_v<T> || std::is_floating_point_v<T>, int> = 0>
+    template<typename T>
+    requires std::integral<T> || std::floating_point<T>
     void set_scalar(ssize_t row_offset, T val) {
-        util::check(sizeof(T) == get_type_size(type_.data_type()), "Type mismatch in set_scalar, expected {}",
-                    get_type_size(type_.data_type()));
+        util::check(
+            sizeof(T) == get_type_size(type_.data_type()),
+            "Type mismatch in set_scalar, expected {} byte scalar got {} byte scalar",
+            get_type_size(type_.data_type()),
+            sizeof(T)
+        );
 
         auto prev_logical_row = last_logical_row_;
         last_logical_row_ = row_offset;
diff --git a/cpp/arcticdb/column_store/memory_segment.hpp b/cpp/arcticdb/column_store/memory_segment.hpp
index 18c8bb297b..07fc6b6f83 100644
--- a/cpp/arcticdb/column_store/memory_segment.hpp
+++ b/cpp/arcticdb/column_store/memory_segment.hpp
@@ -74,13 +74,15 @@ class SegmentInMemory {
         impl_->end_row();
     }
 
-    template<class T, std::enable_if_t<std::is_integral_v<T> || std::is_floating_point_v<T>, int> = 0>
+    template<typename T>
+    requires std::integral<T> || std::floating_point<T>
     void set_scalar(position_t idx, T val) {
         impl_->set_scalar(idx, val);
     }
 
-    template<class T, std::enable_if_t<std::is_same_v<std::decay_t<T>, std::string>, int> = 0>
-    void set_scalar(position_t idx, T val) {
+    template<typename T>
+    requires std::same_as<std::decay_t<T>, std::string>
+    void set_scalar(position_t idx, const T& val) {
         impl_->set_string(idx, val);
     }
 
@@ -117,16 +119,14 @@ class SegmentInMemory {
         impl_->init_column_map();
     }
 
-    template<class T, template<class> class Tensor, std::enable_if_t<
-            std::is_integral_v<T> || std::is_floating_point_v<T>,
-            int> = 0>
+    template<class T, template<class> class Tensor>
+    requires std::integral<T> || std::floating_point<T>
     void set_array(position_t pos, Tensor<T> &val) {
         impl_->set_array(pos, val);
     }
 
-    template<class T, std::enable_if_t<
-        std::is_integral_v<T> || std::is_floating_point_v<T>,
-        int> = 0>
+    template<class T>
+    requires std::integral<T> || std::floating_point<T>
     void set_array(position_t pos, py::array_t<T>& val) {
         impl_->set_array(pos, val);
     }
@@ -251,12 +251,14 @@ class SegmentInMemory {
         impl_->sparsify();
     }
 
-    template<class T, std::enable_if_t<std::is_integral_v<T> || std::is_floating_point_v<T>, int> = 0>
+    template<class T>
+    requires std::integral<T> || std::floating_point<T>
     void set_external_block(position_t idx, T *val, size_t size) {
         impl_->set_external_block(idx, val, size);
     }
 
-    template<class T, std::enable_if_t<std::is_integral_v<T> || std::is_floating_point_v<T>, int> = 0>
+    template<class T>
+    requires std::integral<T> || std::floating_point<T>
     void set_sparse_block(position_t idx, T *val, size_t rows_to_write) {
         impl_->set_sparse_block(idx, val, rows_to_write);
     }
@@ -460,6 +462,10 @@ class SegmentInMemory {
         return output;
     }
 
+    void drop_empty_columns() {
+        impl_->drop_empty_columns();
+    }
+
 private:
     explicit SegmentInMemory(std::shared_ptr<SegmentInMemoryImpl> impl) :
             impl_(std::move(impl)) {}
diff --git a/cpp/arcticdb/column_store/memory_segment_impl.cpp b/cpp/arcticdb/column_store/memory_segment_impl.cpp
index a562ca289e..8c01debd4f 100644
--- a/cpp/arcticdb/column_store/memory_segment_impl.cpp
+++ b/cpp/arcticdb/column_store/memory_segment_impl.cpp
@@ -9,8 +9,8 @@
 #include <arcticdb/column_store/string_pool.hpp>
 #include <arcticdb/entity/type_utils.hpp>
 #include <arcticdb/pipeline/string_pool_utils.hpp>
+#include <arcticdb/util/preconditions.hpp>
 
-#include <google/protobuf/any.h>
 #include <google/protobuf/any.pb.h>
 
 namespace arcticdb {
@@ -684,4 +684,27 @@ const google::protobuf::Any* SegmentInMemoryImpl::metadata() const {
     return metadata_.get();
 }
 
+void SegmentInMemoryImpl::drop_empty_columns() {
+    internal::check<ErrorCode::E_ASSERTION_FAILURE>(
+        row_count() > 0,
+        "Dropping all empty columns from an empty segment would result in removing all columns"
+    );
+    const StreamDescriptor& original = descriptor();
+    StreamDescriptor only_non_empty_cols;
+    only_non_empty_cols.set_id(original.id());
+    only_non_empty_cols.set_index(descriptor().index());
+    size_t field_index = 0;
+    while (field_index < original.index().field_count()) {
+        only_non_empty_cols.add_field(original.field(field_index++));
+    }
+    while (field_index < original.field_count()) {
+        const Column& col = column(field_index);
+        if (col.row_count() > 0) {
+            only_non_empty_cols.add_field(original.field(field_index));
+        }
+        field_index++;
+    }
+    change_schema(std::move(only_non_empty_cols));
+}
+
 }
diff --git a/cpp/arcticdb/column_store/memory_segment_impl.hpp b/cpp/arcticdb/column_store/memory_segment_impl.hpp
index 78e050e196..5b52c752df 100644
--- a/cpp/arcticdb/column_store/memory_segment_impl.hpp
+++ b/cpp/arcticdb/column_store/memory_segment_impl.hpp
@@ -67,31 +67,31 @@ class SegmentInMemoryImpl {
 
         template<class Callable>
         auto visit(Callable &&c) const {
-            return entity::visit_field(parent_->descriptor().field(column_id_), [that=this, c=std::forward<Callable>(c)](auto type_desc_tag) {
+            return entity::visit_field(parent_->descriptor().field(column_id_), [this, c=std::forward<Callable>(c)](auto type_desc_tag) {
                 using RawType = typename std::decay_t<decltype(type_desc_tag)>::DataTypeTag::raw_type;
-                return c(that->parent_->scalar_at<RawType>(that->row_id_, that->column_id_));
+                return c(parent_->scalar_at<RawType>(row_id_, column_id_));
             });
         }
 
         template<class Callable>
         auto visit_string(Callable &&c) const {
-            return entity::visit_field(parent_->descriptor().field(column_id_), [that=this, c = std::forward<Callable>(c)](auto type_desc_tag) {
+            return entity::visit_field(parent_->descriptor().field(column_id_), [this, c = std::forward<Callable>(c)](auto type_desc_tag) {
                 using DTT = typename std::decay_t<decltype(type_desc_tag)>::DataTypeTag;
                 if constexpr(is_sequence_type(DTT::data_type))
-                    return c(that->parent_->string_at(that->row_id_, position_t(that->column_id_)));
+                    return c(parent_->string_at(row_id_, position_t(column_id_)));
             });
         }
 
         template<class Callable>
         auto visit_field(Callable &&c) const {
             const auto& field = parent_->descriptor().field(column_id_);
-            return entity::visit_field(parent_->descriptor().field(column_id_), [&field, that=this, c = std::forward<Callable>(c)](auto type_desc_tag) {
+            return entity::visit_field(field, [&field, this, c = std::forward<Callable>(c)](auto type_desc_tag) {
                 using DataTypeTag = typename std::decay_t<decltype(type_desc_tag)>::DataTypeTag;
                 using RawType = typename DataTypeTag::raw_type;
                 if constexpr (is_sequence_type(DataTypeTag::data_type))
-                    return c(that->parent_->string_at(that->row_id_, position_t(that->column_id_)), std::string_view{field.name()}, field.type());
+                    return c(parent_->string_at(row_id_, position_t(column_id_)), std::string_view{field.name()}, type_desc_tag);
                 else if constexpr (is_numeric_type(DataTypeTag::data_type) || is_bool_type(DataTypeTag::data_type))
-                    return c(that->parent_->scalar_at<RawType>(that->row_id_, that->column_id_), std::string_view{field.name()}, field.type());
+                    return c(parent_->scalar_at<RawType>(row_id_, column_id_), std::string_view{field.name()}, type_desc_tag);
                 else if constexpr(is_empty_type(DataTypeTag::data_type))
                     internal::raise<ErrorCode::E_ASSERTION_FAILURE>("visit_field does not support empty-type columns");
                 else
@@ -101,9 +101,9 @@ class SegmentInMemoryImpl {
 
         template<class Callable>
         auto visit(Callable &&c) {
-            return entity::visit_field(parent_->descriptor().field(column_id_), [that=this, c=std::forward<Callable>(c)](auto type_desc_tag) {
+            return entity::visit_field(parent_->descriptor().field(column_id_), [this, c=std::forward<Callable>(c)](auto type_desc_tag) {
                 using RawType = typename std::decay_t<decltype(type_desc_tag)>::DataTypeTag::raw_type;
-                return c(that->parent_->reference_at<RawType>(that->row_id_, that->column_id_));
+                return c(parent_->reference_at<RawType>(row_id_, column_id_));
             });
         }
 
@@ -454,18 +454,21 @@ class SegmentInMemoryImpl {
         });
     }
 
-    template<class T, std::enable_if_t<std::is_integral_v<T> || std::is_floating_point_v<T>, int> = 0>
-        void set_scalar(position_t idx, T val) {
+    template<class T>
+    requires std::integral<T> || std::floating_point<T>
+    void set_scalar(position_t idx, T val) {
         ARCTICDB_TRACE(log::version(), "Segment setting scalar {} at row {} column {}", val, row_id_ + 1, idx);
         column(idx).set_scalar(row_id_ + 1, val);
     }
 
-    template<class T, std::enable_if_t<std::is_integral_v<T> || std::is_floating_point_v<T>, int> = 0>
-        void set_external_block(position_t idx, T *val, size_t size) {
+    template<class T>
+    requires std::integral<T> || std::floating_point<T>
+    void set_external_block(position_t idx, T *val, size_t size) {
         column_unchecked(idx).set_external_block(row_id_ + 1, val, size);
     }
 
-    template<class T, std::enable_if_t<std::is_integral_v<T> || std::is_floating_point_v<T>, int> = 0>
+    template<class T>
+    requires std::integral<T> || std::floating_point<T>
     void set_sparse_block(position_t idx, T *val,  size_t rows_to_write) {
         column_unchecked(idx).set_sparse_block(row_id_ + 1, val, rows_to_write);
     }
@@ -478,23 +481,22 @@ class SegmentInMemoryImpl {
         column_unchecked(idx).set_sparse_block(std::move(buffer), std::move(shapes), std::move(bitset));
     }
 
-    template<class T, std::enable_if_t<std::is_same_v<std::decay_t<T>, std::string>, int> = 0>
-        void set_scalar(position_t idx, T val) {
+    template<class T>
+    requires std::same_as<std::decay_t<T>, std::string>
+    void set_scalar(position_t idx, const T& val) {
         set_string(idx, val);
     }
 
-    template<class T, template<class> class Tensor, std::enable_if_t<
-        std::is_integral_v<T> || std::is_floating_point_v<T>,
-        int> = 0>
-            void set_array(position_t pos, Tensor<T> &val) {
+    template<class T, template<class> class Tensor>
+    requires std::integral<T> || std::floating_point<T>
+    void set_array(position_t pos, Tensor<T> &val) {
         magic_.check();
         ARCTICDB_SAMPLE(MemorySegmentSetArray, 0)
         column_unchecked(pos).set_array(row_id_ + 1, val);
     }
 
-    template<class T, std::enable_if_t<
-        std::is_integral_v<T> || std::is_floating_point_v<T>,
-        int> = 0>
+    template<class T>
+    requires std::integral<T> || std::floating_point<T>
     void set_array(position_t pos, py::array_t<T>& val) {
         magic_.check();
         ARCTICDB_SAMPLE(MemorySegmentSetArray, 0)
@@ -786,6 +788,7 @@ class SegmentInMemoryImpl {
                                                        const std::vector<uint64_t>& segment_counts) const;
 
     std::vector<std::shared_ptr<SegmentInMemoryImpl>> split(size_t rows) const;
+    void drop_empty_columns();
 
 private:
     ssize_t row_id_ = -1;
diff --git a/cpp/arcticdb/entity/merge_descriptors.cpp b/cpp/arcticdb/entity/merge_descriptors.cpp
index b41892f085..afa7210149 100644
--- a/cpp/arcticdb/entity/merge_descriptors.cpp
+++ b/cpp/arcticdb/entity/merge_descriptors.cpp
@@ -13,7 +13,7 @@
 namespace arcticdb {
 StreamDescriptor merge_descriptors(
     const StreamDescriptor &original,
-    const std::vector<std::shared_ptr<FieldCollection>> &entries,
+    std::span<const std::shared_ptr<FieldCollection>> entries,
     const std::unordered_set<std::string_view> &filtered_set,
     const std::optional<IndexDescriptorImpl>& default_index) {
     using namespace arcticdb::stream;
@@ -34,6 +34,7 @@ StreamDescriptor merge_descriptors(
                 merged_fields.emplace_back(idx.name());
                 merged_fields_map.try_emplace(idx.name(), TypeDescriptor{typename IndexType::TypeDescTag{}});
             });
+            index = default_index_type_from_descriptor(*default_index);
         } else {
             util::raise_rte("Descriptor has uninitialized index and no default supplied");
         }
@@ -71,7 +72,12 @@ StreamDescriptor merge_descriptors(
                         if(new_descriptor) {
                             merged_fields_map[field.name()] = *new_descriptor;
                         } else {
-                            util::raise_rte("No valid common type between {} and {} for column {}", existing_type_desc, type_desc, field.name());
+                            schema::raise<ErrorCode::E_DESCRIPTOR_MISMATCH>(
+                                "No valid common type between {} and {} for column {}",
+                                existing_type_desc,
+                                type_desc,
+                                field.name()
+                            );
                         }
                     }
                 } else {
@@ -99,6 +105,17 @@ StreamDescriptor merge_descriptors(
     return merge_descriptors(original, entries, filtered_set, default_index);
 }
 
+StreamDescriptor merge_descriptors(
+    const StreamDescriptor& original,
+    std::span<const std::shared_ptr<FieldCollection>> entries,
+    const std::optional<std::vector<std::string>>& filtered_columns,
+    const std::optional<IndexDescriptorImpl>& default_index) {
+    std::unordered_set<std::string_view> filtered_set = filtered_columns.has_value()
+        ? std::unordered_set<std::string_view>(filtered_columns->begin(), filtered_columns->end())
+        : std::unordered_set<std::string_view>{};
+    return merge_descriptors(original, entries, filtered_set, default_index);
+}
+
 StreamDescriptor merge_descriptors(
     const StreamDescriptor &original,
     const std::vector<pipelines::SliceAndKey> &entries,
diff --git a/cpp/arcticdb/entity/merge_descriptors.hpp b/cpp/arcticdb/entity/merge_descriptors.hpp
index c968be930a..14644beb70 100644
--- a/cpp/arcticdb/entity/merge_descriptors.hpp
+++ b/cpp/arcticdb/entity/merge_descriptors.hpp
@@ -7,11 +7,12 @@
 
 #include <arcticdb/entity/stream_descriptor.hpp>
 #include <arcticdb/pipeline/frame_slice.hpp>
+#include <span>
 
 namespace arcticdb {
 StreamDescriptor merge_descriptors(
     const StreamDescriptor &original,
-    const std::vector<std::shared_ptr<FieldCollection>> &entries,
+    std::span<const std::shared_ptr<FieldCollection>> entries,
     const std::unordered_set<std::string_view> &filtered_set,
     const std::optional<IndexDescriptorImpl>& default_index);
 
@@ -21,6 +22,12 @@ entity::StreamDescriptor merge_descriptors(
     const std::optional<std::vector<std::string>> &filtered_columns,
     const std::optional<entity::IndexDescriptorImpl>& default_index = std::nullopt);
 
+entity::StreamDescriptor merge_descriptors(
+    const entity::StreamDescriptor& original,
+    std::span<const std::shared_ptr<FieldCollection>> entries,
+    const std::optional<std::vector<std::string>>& filtered_columns,
+    const std::optional<entity::IndexDescriptorImpl>& default_index = std::nullopt);
+
 entity::StreamDescriptor merge_descriptors(
     const entity::StreamDescriptor &original,
     const std::vector<pipelines::SliceAndKey> &entries,
diff --git a/cpp/arcticdb/entity/types.hpp b/cpp/arcticdb/entity/types.hpp
index 5b21c640a7..8dd5e6c099 100644
--- a/cpp/arcticdb/entity/types.hpp
+++ b/cpp/arcticdb/entity/types.hpp
@@ -441,11 +441,11 @@ struct TypeDescriptor {
     template<typename Callable>
     constexpr auto visit_tag(Callable &&callable) const;
 
-    bool operator==(const TypeDescriptor &o) const {
+    [[nodiscard]] constexpr bool operator==(const TypeDescriptor& o) const {
         return data_type_ == o.data_type_ && dimension_ == o.dimension_;
     }
 
-    bool operator!=(const TypeDescriptor &o) const {
+    [[nodiscard]] constexpr bool operator!=(const TypeDescriptor& o) const {
         return !(*this == o);
     }
 
diff --git a/cpp/arcticdb/pipeline/index_utils.cpp b/cpp/arcticdb/pipeline/index_utils.cpp
index e2eb239ef4..943147dcd1 100644
--- a/cpp/arcticdb/pipeline/index_utils.cpp
+++ b/cpp/arcticdb/pipeline/index_utils.cpp
@@ -92,9 +92,10 @@ TimeseriesDescriptor get_merged_tsd(
     }
     else if (dynamic_schema) {
         // In case of dynamic schema
+        const std::array fields_ptr = {new_frame->desc.fields_ptr()};
         merged_descriptor = merge_descriptors(
                 existing_descriptor,
-                std::vector<std::shared_ptr<FieldCollection>>{new_frame->desc.fields_ptr()},
+                fields_ptr,
                 {}
         );
     } else {
diff --git a/cpp/arcticdb/pipeline/pipeline_context.hpp b/cpp/arcticdb/pipeline/pipeline_context.hpp
index 8c7d0fdd1c..6133bc5fc6 100644
--- a/cpp/arcticdb/pipeline/pipeline_context.hpp
+++ b/cpp/arcticdb/pipeline/pipeline_context.hpp
@@ -108,6 +108,9 @@ struct PipelineContext : public std::enable_shared_from_this<PipelineContext> {
     // written in, desc_ will be modified such that the return matches what's requested, and this'll be set to the
     // original value. It's only set in this edge case.
     std::optional<StreamDescriptor> orig_desc_;
+    // When there are staged segments this holds the combined stream descriptor for all staged segments
+    // This can be different than desc_ in case dynamic schema is used. Otherwise they must be the same.
+    std::optional<StreamDescriptor> staged_descriptor_;
     StreamId stream_id_;
     VersionId version_id_ = 0;
     size_t total_rows_ = 0;
@@ -200,6 +203,7 @@ struct PipelineContext : public std::enable_shared_from_this<PipelineContext> {
         swap(left.segment_descriptors_, right.segment_descriptors_);
         swap(left.filter_columns_set_, right.filter_columns_set_);
         swap(left.compacted_, right.compacted_);
+        swap(left.staged_descriptor_, right.staged_descriptor_);
     }
 
     using iterator = PipelineContextIterator<PipelineContextRow>;
diff --git a/cpp/arcticdb/pipeline/read_frame.cpp b/cpp/arcticdb/pipeline/read_frame.cpp
index 546a608e91..471313cc36 100644
--- a/cpp/arcticdb/pipeline/read_frame.cpp
+++ b/cpp/arcticdb/pipeline/read_frame.cpp
@@ -601,7 +601,7 @@ struct ReduceColumnTask : async::BaseTask {
             } else {
                 column.default_initialize_rows(0, frame_.row_count(), false);
             }
-        } else {
+        } else if (column_data != slice_map_->columns_.end()) {
             if(dynamic_schema_) {
                 NullValueReducer null_reducer{column, context_, frame_, shared_data_, handler_data_};
                 for (const auto &row : column_data->second) {
@@ -623,6 +623,8 @@ struct ReduceColumnTask : async::BaseTask {
 
                 column.set_inflated(frame_.row_count());
             }
+        } else if (!dynamic_schema_ && column_data == slice_map_->columns_.end() && is_sequence_type(column.type().data_type())) {
+            internal::raise<ErrorCode::E_ASSERTION_FAILURE>("Column with index {} is not in static schema slice map.", column_index_);
         }
         return folly::Unit{};
     }
diff --git a/cpp/arcticdb/processing/clause.cpp b/cpp/arcticdb/processing/clause.cpp
index 23c97f5e5c..5e947f6914 100644
--- a/cpp/arcticdb/processing/clause.cpp
+++ b/cpp/arcticdb/processing/clause.cpp
@@ -13,6 +13,7 @@
 #include <arcticdb/processing/processing_unit.hpp>
 #include <arcticdb/column_store/string_pool.hpp>
 #include <arcticdb/util/offset_string.hpp>
+#include <arcticdb/stream/merge.hpp>
 
 #include <arcticdb/processing/clause.hpp>
 #include <arcticdb/pipeline/column_stats.hpp>
@@ -888,13 +889,12 @@ std::vector<EntityId> SortClause::process(std::vector<EntityId>&& entity_ids) co
     return push_entities(component_manager_, std::move(proc));
 }
 
-template<typename IndexType, typename DensityPolicy, typename QueueType, typename Comparator, typename StreamId>
+template<typename IndexType, typename DensityPolicy, typename QueueType, bool dynamic_schema>
 void merge_impl(
         std::shared_ptr<ComponentManager> component_manager,
         std::vector<std::vector<EntityId>>& ret,
         QueueType &input_streams,
         bool add_symbol_column,
-        StreamId stream_id,
         const RowRange& row_range,
         const ColRange& col_range,
         IndexType index,
@@ -907,21 +907,48 @@ void merge_impl(
         ret.emplace_back(push_entities(component_manager, ProcessingUnit{std::forward<SegmentInMemory>(segment), row_range, col_range}));
     };
 
-    using AggregatorType = stream::Aggregator<IndexType, stream::DynamicSchema, SegmentationPolicy, DensityPolicy>;
-    const auto& fields = stream_descriptor.fields();
-    FieldCollection new_fields{};
-    (void)new_fields.add(fields[0].ref());
-
-    auto index_desc = index_descriptor_from_range(stream_id, index, new_fields);
-    auto desc = StreamDescriptor{index_desc};
+    using Schema = std::conditional_t<dynamic_schema, stream::DynamicSchema, stream::FixedSchema>;
+    using AggregatorType = stream::Aggregator<IndexType, Schema, SegmentationPolicy, DensityPolicy>;
 
     AggregatorType agg{
-            stream::DynamicSchema{desc, index},
-            std::move(func), std::move(segmentation_policy), desc, std::nullopt
+        Schema{stream_descriptor, index},
+        std::move(func),
+        std::move(segmentation_policy),
+        stream_descriptor,
+        std::nullopt
     };
 
-    stream::do_merge<IndexType, SegmentWrapper, AggregatorType, decltype(input_streams)>(
-        input_streams, agg, add_symbol_column);
+    stream::do_merge<IndexType, AggregatorType, decltype(input_streams)>(input_streams, agg, add_symbol_column);
+}
+
+MergeClause::MergeClause(
+        stream::Index index,
+        const stream::VariantColumnPolicy &density_policy,
+        const StreamId& stream_id,
+        const StreamDescriptor& stream_descriptor,
+        bool dynamic_schema) :
+        index_(std::move(index)),
+        density_policy_(density_policy),
+        stream_id_(stream_id),
+        stream_descriptor_(stream_descriptor),
+        dynamic_schema_(dynamic_schema) {
+    clause_info_.requires_repartition_ = true;
+}
+
+void MergeClause::set_processing_config(const ProcessingConfig&) {}
+
+void MergeClause::set_component_manager(std::shared_ptr<ComponentManager> component_manager) {
+    component_manager_ = std::move(component_manager);
+}
+
+const ClauseInfo& MergeClause::clause_info() const {
+    return clause_info_;
+}
+
+std::vector<std::vector<size_t>> MergeClause::structure_for_processing(
+    std::vector<RangesAndKey>& ranges_and_keys,
+    size_t) const {
+    return structure_all_together(ranges_and_keys);
 }
 
 // MergeClause receives a list of DataFrames as input and merge them into a single one where all 
@@ -975,18 +1002,31 @@ std::optional<std::vector<std::vector<EntityId>>> MergeClause::repartition(std::
     const RowRange row_range{min_start_row, max_end_row};
     const ColRange col_range{min_start_col, max_end_col};
     std::vector<std::vector<EntityId>> ret;
-    std::visit(
-            [this, &ret, &input_streams, &comp=compare, stream_id=stream_id_, &row_range, &col_range](auto idx, auto density) {
-                merge_impl<decltype(idx), decltype(density), decltype(input_streams), decltype(comp), decltype(stream_id)>(component_manager_,
-                                                                                                      ret,
-                                                                                                      input_streams,
-                                                                                                      add_symbol_column_,
-                                                                                                      stream_id,
-                                                                                                      row_range,
-                                                                                                      col_range,
-                                                                                                      idx,
-                                                                                                      stream_descriptor_);
-            }, index_, density_policy_);
+    std::visit([this, &ret, &input_streams, &comp=compare, stream_id=stream_id_, &row_range, &col_range](auto idx, auto density) {
+            if (dynamic_schema_) {
+                merge_impl<decltype(idx), decltype(density), decltype(input_streams), true>(
+                    component_manager_,
+                    ret,
+                    input_streams,
+                    add_symbol_column_,
+                    row_range,
+                    col_range,
+                    idx,
+                    stream_descriptor_
+                );
+            } else {
+                merge_impl<decltype(idx), decltype(density), decltype(input_streams), false>(
+                    component_manager_,
+                    ret,
+                    input_streams,
+                    add_symbol_column_,
+                    row_range,
+                    col_range,
+                    idx,
+                    stream_descriptor_
+                );
+            }
+        }, index_, density_policy_);
     return ret;
 }
 
diff --git a/cpp/arcticdb/processing/clause.hpp b/cpp/arcticdb/processing/clause.hpp
index 107da44e63..5de69e6dd5 100644
--- a/cpp/arcticdb/processing/clause.hpp
+++ b/cpp/arcticdb/processing/clause.hpp
@@ -21,7 +21,6 @@
 #include <arcticdb/processing/grouper.hpp>
 #include <arcticdb/stream/aggregator.hpp>
 #include <arcticdb/util/movable_priority_queue.hpp>
-#include <arcticdb/stream/merge.hpp>
 #include <arcticdb/pipeline/index_utils.hpp>
 
 #include <folly/Poly.h>
@@ -573,42 +572,32 @@ struct MergeClause {
     stream::Index index_;
     stream::VariantColumnPolicy density_policy_;
     StreamId stream_id_;
-    bool add_symbol_column_ = false;
     StreamId target_id_;
     StreamDescriptor stream_descriptor_;
+    bool add_symbol_column_ = false;
+    bool dynamic_schema_;
 
     MergeClause(
-            stream::Index index,
-            const stream::VariantColumnPolicy &density_policy,
-            const StreamId& stream_id,
-            const StreamDescriptor& stream_descriptor) :
-            index_(std::move(index)),
-            density_policy_(density_policy),
-            stream_id_(stream_id),
-            stream_descriptor_(stream_descriptor) {
-        clause_info_.requires_repartition_ = true;
-    }
+        stream::Index index,
+        const stream::VariantColumnPolicy& density_policy,
+        const StreamId& stream_id,
+        const StreamDescriptor& stream_descriptor,
+        bool dynamic_schema
+    );
 
     [[nodiscard]] std::vector<std::vector<size_t>> structure_for_processing(
             std::vector<RangesAndKey>& ranges_and_keys,
-            size_t) {
-        return structure_all_together(ranges_and_keys);
-    }
+            size_t) const;
 
     [[nodiscard]] std::vector<EntityId> process(std::vector<EntityId>&& entity_ids) const;
 
     [[nodiscard]] std::optional<std::vector<std::vector<EntityId>>> repartition(std::vector<std::vector<EntityId>>&& entity_ids_vec) const;
 
-    [[nodiscard]] const ClauseInfo& clause_info() const {
-        return clause_info_;
-    }
+    [[nodiscard]] const ClauseInfo& clause_info() const;
 
-    void set_processing_config(ARCTICDB_UNUSED const ProcessingConfig& processing_config) {
-    }
+    void set_processing_config(const ProcessingConfig& processing_config);
 
-    void set_component_manager(std::shared_ptr<ComponentManager> component_manager) {
-        component_manager_ = component_manager;
-    }
+    void set_component_manager(std::shared_ptr<ComponentManager> component_manager);
 };
 
 struct ColumnStatsGenerationClause {
diff --git a/cpp/arcticdb/storage/library_manager.hpp b/cpp/arcticdb/storage/library_manager.hpp
index ddb32edaa8..75c1f90e97 100644
--- a/cpp/arcticdb/storage/library_manager.hpp
+++ b/cpp/arcticdb/storage/library_manager.hpp
@@ -11,7 +11,6 @@
 #include <arcticdb/storage/library.hpp>
 #include <arcticdb/storage/library_path.hpp>
 #include <arcticdb/storage/storage_override.hpp>
-#include <arcticdb/pipeline/index_segment_reader.hpp>
 
 
 namespace arcticdb::storage {
diff --git a/cpp/arcticdb/stream/aggregator.hpp b/cpp/arcticdb/stream/aggregator.hpp
index 19cbd6a0c0..06d2089c9b 100644
--- a/cpp/arcticdb/stream/aggregator.hpp
+++ b/cpp/arcticdb/stream/aggregator.hpp
@@ -275,12 +275,14 @@ class Aggregator {
     void commit_impl(bool final);
 
 private:
-    template<class T, std::enable_if_t<std::is_integral_v<T> || std::is_floating_point_v<T>, int> = 0>
+    template <typename T>
+    requires std::integral<T> || std::floating_point<T>
     void set_scalar(std::size_t pos, T val) {
         segment_.set_scalar(pos, val);
     }
 
-    template<class T, std::enable_if_t<std::is_integral_v<T> || std::is_floating_point_v<T>, int> = 0>
+    template<typename T>
+    requires std::integral<T> || std::floating_point<T>
     void set_scalar_by_name(std::string_view name, T val, DataType data_type) {
         position_t pos = schema_policy_.get_column_idx_by_name(segment_, name, make_scalar_type(data_type), segmenting_policy_.expected_row_size(), segment_.row_count());
         set_scalar(pos, val);
diff --git a/cpp/arcticdb/stream/index.cpp b/cpp/arcticdb/stream/index.cpp
index 32cb1bb8e7..e24c1e6e3a 100644
--- a/cpp/arcticdb/stream/index.cpp
+++ b/cpp/arcticdb/stream/index.cpp
@@ -69,7 +69,7 @@ void TimeseriesIndex::check(const FieldCollection& fields) const {
 
 IndexValue TimeseriesIndex::start_value_for_segment(const SegmentInMemory& segment) {
     if (segment.row_count() == 0)
-        return {NumericIndex{0}};
+        return {NumericIndex{min_index_value()}};
     auto first_ts = segment.template scalar_at<timestamp>(0, 0).value();
     return {first_ts};
 }
@@ -77,14 +77,14 @@ IndexValue TimeseriesIndex::start_value_for_segment(const SegmentInMemory& segme
 IndexValue TimeseriesIndex::end_value_for_segment(const SegmentInMemory& segment) {
     auto row_count = segment.row_count();
     if (row_count == 0)
-        return {NumericIndex{0}};
+        return {NumericIndex{min_index_value()}};
     auto last_ts = segment.template scalar_at<timestamp>(row_count - 1, 0).value();
     return {last_ts};
 }
 
 IndexValue TimeseriesIndex::start_value_for_keys_segment(const SegmentInMemory& segment) {
     if (segment.row_count() == 0)
-        return {NumericIndex{0}};
+        return {NumericIndex{min_index_value()}};
     auto start_index_id = int(pipelines::index::Fields::start_index);
     auto first_ts = segment.template scalar_at<timestamp>(0, start_index_id).value();
     return {first_ts};
@@ -93,7 +93,7 @@ IndexValue TimeseriesIndex::start_value_for_keys_segment(const SegmentInMemory&
 IndexValue TimeseriesIndex::end_value_for_keys_segment(const SegmentInMemory& segment) {
     auto row_count = segment.row_count();
     if (row_count == 0)
-        return {NumericIndex{0}};
+        return {NumericIndex{min_index_value()}};
     auto end_index_id = int(pipelines::index::Fields::end_index);
     auto last_ts = segment.template scalar_at<timestamp>(row_count - 1, end_index_id).value();
     return {last_ts};
diff --git a/cpp/arcticdb/stream/index.hpp b/cpp/arcticdb/stream/index.hpp
index 2fcdd07708..44c47b55ca 100644
--- a/cpp/arcticdb/stream/index.hpp
+++ b/cpp/arcticdb/stream/index.hpp
@@ -52,6 +52,12 @@ class TimeseriesIndex : public BaseIndex<TimeseriesIndex> {
     static constexpr IndexDescriptorImpl::Type type() {
         return IndexDescriptorImpl::Type::TIMESTAMP;
     }
+
+    static constexpr timestamp min_index_value() {
+        // std::numeric_limits<timestamp>::min() is reserved for NaT
+        return std::numeric_limits<timestamp>::min() + 1;
+    }
+
     TimeseriesIndex(const std::string& name);
     static TimeseriesIndex default_index();
     void check(const FieldCollection& fields) const;
@@ -76,7 +82,7 @@ class TimeseriesIndex : public BaseIndex<TimeseriesIndex> {
 
   private:
     std::string name_;
-    timestamp ts_ = 0;
+    timestamp ts_ = min_index_value();
 };
 
 class TableIndex : public BaseIndex<TableIndex> {
diff --git a/cpp/arcticdb/stream/merge.hpp b/cpp/arcticdb/stream/merge.hpp
index 0be30bfedf..7564331e77 100644
--- a/cpp/arcticdb/stream/merge.hpp
+++ b/cpp/arcticdb/stream/merge.hpp
@@ -7,31 +7,107 @@
 
 #pragma once
 
+#include <type_traits>
+#include <string_view>
 #include <arcticdb/pipeline/index_utils.hpp>
+#include <arcticdb/util/constants.hpp>
+#include <arcticdb/stream/schema.hpp>
+#include <arcticdb/storage/memory_layout.hpp>
+#include <arcticdb/entity/stream_descriptor.hpp>
+#include <arcticdb/entity/types.hpp>
+#include <arcticdb/util/preconditions.hpp>
+#include <arcticdb/util/error_code.hpp>
+#include <ankerl/unordered_dense.h>
+
+template<typename Aggregator>
+inline consteval bool is_static_schema() {
+    if constexpr (std::is_same_v<typename Aggregator::SchemaPolicy, arcticdb::stream::DynamicSchema>) {
+        return false;
+    } else if constexpr (std::is_same_v<typename Aggregator::SchemaPolicy, arcticdb::stream::FixedSchema>) {
+        return true;
+    } else {
+        static_assert(sizeof(Aggregator) == 0, "Unknown schema type");
+    }
+}
+
+template<typename Aggregator>
+inline consteval bool is_dynamic_schema() {
+    return !is_static_schema<Aggregator>();
+}
+
 
 namespace arcticdb::stream {
-template<typename IndexType, typename WrapperType, typename AggregatorType, typename QueueType>
+template<typename IndexType, typename AggregatorType, typename QueueType>
 void do_merge(
     QueueType& input_streams,
     AggregatorType& agg,
     bool add_symbol_column
     ) {
+    while (!input_streams.empty() && input_streams.top()->seg_.row_count() == 0) {
+        input_streams.pop_top();
+    }
+
+    // NaT is definied as std::numeric_limits<int64_t>::min(), if there are any NaT values they will be on the top of the queue
+    if (!input_streams.empty()) {
+        const auto& next = input_streams.top();
+        const auto index_value =
+            std::get<timestamp>(*pipelines::index::index_value_from_row(next->row(), IndexDescriptorImpl::Type::TIMESTAMP, 0));
+        sorting::check<ErrorCode::E_UNSORTED_DATA>(index_value != NaT, "NaT values are not allowed in the index");
+    }
+
+    [[maybe_unused]] const ankerl::unordered_dense::map<std::string_view, size_t> field_name_to_index = [&](){
+        ankerl::unordered_dense::map<std::string_view, size_t> res;
+        if constexpr (is_dynamic_schema<AggregatorType>()) {
+            const StreamDescriptor& desc = agg.descriptor();
+            for (size_t field_idx = 0; field_idx < desc.field_count(); ++field_idx) {
+                const Field& field = desc.field(field_idx);
+                res[field.name()] = field_idx;
+            }
+        }
+        return res;
+    }();
+
+
     while (!input_streams.empty()) {
         auto next = input_streams.pop_top();
-
-        if (next->row().parent_->row_count() == 0) {
+        if (next->seg_.row_count() == 0) {
             continue;
         }
-        agg.start_row(pipelines::index::index_value_from_row(next->row(), IndexDescriptorImpl::Type::TIMESTAMP, 0).value()) ([&next, add_symbol_column](auto &rb) {
+        const auto index_value =
+            *pipelines::index::index_value_from_row(next->row(), IndexDescriptorImpl::Type::TIMESTAMP, 0);
+        agg.start_row(index_value) ([&](auto &rb) {
             if(add_symbol_column)
                 rb.set_scalar_by_name("symbol", std::string_view(std::get<StringId>(next->id())), DataType::UTF_DYNAMIC64);
 
             auto val = next->row().begin();
             std::advance(val, IndexType::field_count());
             for(; val != next->row().end(); ++val) {
-                val->visit_field([&rb] (const auto& opt_v, std::string_view name, const TypeDescriptor& type_desc) {
-                    if(opt_v)
-                        rb.set_scalar_by_name(name, opt_v.value(), type_desc.data_type());
+                val->visit_field([&] (const auto& opt_v, std::string_view name, auto row_field_descriptor_tag) {
+                    if (opt_v) {
+                        if constexpr (is_static_schema<AggregatorType>()) {
+                            rb.set_scalar_by_name(name, opt_v.value(), row_field_descriptor_tag.data_type());
+                        } else {
+                            const TypeDescriptor& final_type = agg.descriptor().field(field_name_to_index.find(name)->second).type();
+                            details::visit_type(final_type.data_type(), [&](auto merged_descriptor_type) {
+                                using merged_type_info = ScalarTypeInfo<decltype(merged_descriptor_type)>;
+                                using row_type_info = ScalarTypeInfo<typename decltype(row_field_descriptor_tag)::DataTypeTag>;
+                                // At this point all staged descriptors were merged using merge_descritpros and it
+                                // ensured that all staged descriptors are either the same or are convertible to the
+                                // stream descriptor in the aggregator.
+                                if constexpr (merged_type_info::data_type == row_type_info::data_type) {
+                                    rb.set_scalar_by_name(name, opt_v.value(), merged_type_info::data_type);
+                                } else if constexpr (std::is_convertible_v<decltype(*opt_v), typename merged_type_info::RawType>) {
+                                    rb.set_scalar_by_name(name, static_cast<merged_type_info::RawType>(*opt_v), merged_type_info::data_type);
+                                } else {
+                                    schema::raise<ErrorCode::E_DESCRIPTOR_MISMATCH>(
+                                        "Cannot convert {} to {}",
+                                        merged_type_info::TDT::type_descriptor(),
+                                        row_type_info::TDT::type_descriptor()
+                                    );
+                                }
+                            });
+                        }
+                    }
                 });
             }
         });
diff --git a/cpp/arcticdb/util/timer.cpp b/cpp/arcticdb/util/timer.cpp
new file mode 100644
index 0000000000..658a26ed2e
--- /dev/null
+++ b/cpp/arcticdb/util/timer.cpp
@@ -0,0 +1,13 @@
+#include <arcticdb/util/timer.hpp>
+#include <boost/date_time/posix_time/time_formatters.hpp>
+#include <boost/date_time/posix_time/posix_time_duration.hpp>
+#include <boost/date_time/posix_time/conversion.hpp>
+
+namespace arcticdb {
+    std::string date_and_time(int64_t ts) {
+        const std::time_t seconds_since_epoch = ts / BILLION;
+        return boost::posix_time::to_simple_string(
+            boost::posix_time::from_time_t(0) + boost::posix_time::seconds(seconds_since_epoch)
+        );
+    }
+}
diff --git a/cpp/arcticdb/util/timer.hpp b/cpp/arcticdb/util/timer.hpp
index 2caa1f57ae..c438aac8c0 100644
--- a/cpp/arcticdb/util/timer.hpp
+++ b/cpp/arcticdb/util/timer.hpp
@@ -19,7 +19,6 @@
 #include <string>
 #include <map>
 #include <ctime>
-#include <sstream>
 #include <iomanip>
 
 namespace arcticdb {
@@ -293,12 +292,7 @@ arcticdb::ScopedTimerTotal timer2{#name, [&data](auto totals) { \
     std::copy(std::begin(totals), std::end(totals), std::back_inserter(data)); \
 }};
 
-inline std::string date_and_time(int64_t ts) {
-    std::time_t seconds_since_epoch = ts / BILLION;
-    std::stringstream ss;
-    ss << std::put_time(std::gmtime(&seconds_since_epoch), "%F %T");
-    return ss.str();
-}
+[[nodiscard]] std::string date_and_time(int64_t ts);
 
 
 } //namespace arcticdb
diff --git a/cpp/arcticdb/version/local_versioned_engine.cpp b/cpp/arcticdb/version/local_versioned_engine.cpp
index e3a12cec41..e5bf4ccdf9 100644
--- a/cpp/arcticdb/version/local_versioned_engine.cpp
+++ b/cpp/arcticdb/version/local_versioned_engine.cpp
@@ -1661,7 +1661,9 @@ VersionedItem LocalVersionedEngine::sort_merge_internal(
     const std::optional<arcticdb::proto::descriptors::UserDefinedMetadata>& user_meta,
     const CompactIncompleteOptions& options) {
     auto update_info = get_latest_undeleted_version_and_next_version_id(store(), version_map(), stream_id);
-    auto versioned_item = sort_merge_impl(store_, stream_id, user_meta, update_info, options);
+    const WriteOptions& write_opts = get_write_options();
+    auto versioned_item =
+        sort_merge_impl(store_, stream_id, user_meta, update_info, options, write_opts);
     write_version_and_prune_previous(options.prune_previous_versions_, versioned_item.key_, update_info.previous_index_key_);
     add_to_symbol_list_on_compaction(stream_id, options, update_info);
     return versioned_item;
diff --git a/cpp/arcticdb/version/schema_checks.hpp b/cpp/arcticdb/version/schema_checks.hpp
index 695f27e383..831341b45f 100644
--- a/cpp/arcticdb/version/schema_checks.hpp
+++ b/cpp/arcticdb/version/schema_checks.hpp
@@ -127,7 +127,7 @@ inline void fix_descriptor_mismatch_or_throw(
     if (!dynamic_schema && !columns_match(old_sd, new_frame.desc)) {
         throw StreamDescriptorMismatch(
             "The columns (names and types) in the argument are not identical to that of the existing version",
-            StreamDescriptor{old_sd},
+            old_sd,
             new_frame.desc,
             operation);
     }
diff --git a/cpp/arcticdb/version/version_core-inl.hpp b/cpp/arcticdb/version/version_core-inl.hpp
index 6f210f6968..c42087b93e 100644
--- a/cpp/arcticdb/version/version_core-inl.hpp
+++ b/cpp/arcticdb/version/version_core-inl.hpp
@@ -79,7 +79,7 @@ void merge_frames_for_keys_impl(
 
     using AggregatorType = Aggregator<IndexType, DynamicSchema, SegmentationPolicy, DensityPolicy>;
     AggregatorType agg{DynamicSchema{index.create_stream_descriptor(target_id, {}), index}, std::move(func), std::move(segmentation_policy)};
-    do_merge<IndexType, StreamMergeWrapper, AggregatorType, decltype(input_streams)>(input_streams, agg, true);
+    do_merge<IndexType, AggregatorType, decltype(input_streams)>(input_streams, agg, true);
 }
 
 template <typename SegmentationPolicy, typename Callable>
@@ -108,7 +108,8 @@ void do_compact(
     std::vector<pipelines::FrameSlice>& slices,
     const std::shared_ptr<Store>& store,
     bool convert_int_to_float,
-    std::optional<size_t> segment_size){
+    std::optional<size_t> segment_size,
+    bool validate_index){
         auto index = stream::index_type_from_descriptor(pipeline_context->descriptor());
         stream::SegmentAggregator<IndexType, SchemaType, SegmentationPolicy, DensityPolicy>
         aggregator{
@@ -136,6 +137,14 @@ void do_compact(
             if (sk.slice().rows().diff() == 0) {
                 continue;
             }
+
+            const auto& segment = sk.segment(store);
+            sorting::check<ErrorCode::E_UNSORTED_DATA>(
+                !validate_index || segment.descriptor().sorted() == SortedValue::ASCENDING ||
+                    segment.descriptor().sorted() == SortedValue::UNKNOWN,
+                "Cannot compact unordered segment."
+            );
+
             aggregator.add_segment(
                 std::move(sk.segment(store)),
                 sk.slice(),
diff --git a/cpp/arcticdb/version/version_core.cpp b/cpp/arcticdb/version/version_core.cpp
index 9452b2b76e..b24fc1f157 100644
--- a/cpp/arcticdb/version/version_core.cpp
+++ b/cpp/arcticdb/version/version_core.cpp
@@ -853,14 +853,16 @@ void read_indexed_keys_to_pipeline(
     pipeline_context->bucketize_dynamic_ = bucketize_dynamic;
 }
 
-void read_incompletes_to_pipeline(
+// Returns true if there are staged segments
+bool read_incompletes_to_pipeline(
     const std::shared_ptr<Store>& store,
     std::shared_ptr<PipelineContext>& pipeline_context,
     const ReadQuery& read_query,
     const ReadOptions& read_options,
     bool convert_int_to_float,
     bool via_iteration,
-    bool sparsify) {
+    bool sparsify,
+    bool dynamic_schema) {
 
     auto incomplete_segments = get_incomplete(
         store,
@@ -871,46 +873,81 @@ void read_incompletes_to_pipeline(
         false);
 
     if(incomplete_segments.empty())
-        return;
+        return false;
 
     // In order to have the right normalization metadata and descriptor we need to find the first non-empty segment.
     // Picking an empty segment when there are non-empty ones will impact the index type and column namings.
     // If all segments are empty we will procede as if were appending/writing and empty dataframe.
     debug::check<ErrorCode::E_ASSERTION_FAILURE>(!incomplete_segments.empty(), "Incomplete segments must be non-empty");
-    const auto* seg = &incomplete_segments.front().segment(store);
-    for (auto& slice : incomplete_segments) {
-        if (slice.segment(store).row_count() > 0) {
-            seg = &slice.segment_.value();
-            break;
-        }
-    }
+    const auto first_non_empty_seg = std::find_if(incomplete_segments.begin(), incomplete_segments.end(), [&](auto& slice){
+        return slice.segment(store).row_count() > 0;
+    });
+    const auto& seg =
+        first_non_empty_seg != incomplete_segments.end() ? first_non_empty_seg->segment(store) : incomplete_segments.begin()->segment(store);
     // Mark the start point of the incompletes, so we know that there is no column slicing after this point
     pipeline_context->incompletes_after_ = pipeline_context->slice_and_keys_.size();
 
     // If there are only incompletes we need to add the index here
     if(pipeline_context->slice_and_keys_.empty()) {
-        add_index_columns_to_query(read_query, seg->index_descriptor());
+        add_index_columns_to_query(read_query, seg.index_descriptor());
     }
-
-
-    if (!pipeline_context->desc_)
-        pipeline_context->desc_ = seg->descriptor();
+    pipeline_context->slice_and_keys_.insert(std::end(pipeline_context->slice_and_keys_), incomplete_segments.begin(), incomplete_segments.end());
 
     if (!pipeline_context->norm_meta_) {
         pipeline_context->norm_meta_ = std::make_unique<arcticdb::proto::descriptors::NormalizationMetadata>();
-        pipeline_context->norm_meta_->CopyFrom(seg->index_descriptor().proto().normalization());
+        pipeline_context->norm_meta_->CopyFrom(seg.index_descriptor().proto().normalization());
         ensure_timeseries_norm_meta(*pipeline_context->norm_meta_, pipeline_context->stream_id_, sparsify);
     }
 
-    pipeline_context->desc_ = merge_descriptors(pipeline_context->descriptor(), incomplete_segments, read_query.columns);
+    if (!dynamic_schema) {
+        const auto first_different_seg = std::adjacent_find(
+            incomplete_segments.begin(),
+            incomplete_segments.end(),
+            [&](const SliceAndKey& left, const SliceAndKey& right) {
+                const StreamDescriptor& left_desc = left.segment(store).descriptor();
+                const StreamDescriptor& right_desc = right.segment(store).descriptor();
+                return !columns_match(left_desc, right_desc);
+            }
+        );
+        if (first_different_seg != incomplete_segments.end()) {
+            schema::raise<ErrorCode::E_DESCRIPTOR_MISMATCH>(
+                "When static schema is used all staged segments must have the same column and column types."
+                "{} is different than {}",
+                first_different_seg->segment(store).descriptor(),
+                (first_different_seg + 1)->segment(store).descriptor()
+            );
+        }
+        const StreamDescriptor& staged_desc = incomplete_segments[0].segment(store).descriptor();
+        if (pipeline_context->desc_) {
+            schema::check<ErrorCode::E_DESCRIPTOR_MISMATCH>(
+                columns_match(staged_desc, *pipeline_context->desc_),
+                "When static schema is used the staged stream descriptor {} must equal the stream descriptor on storage {}",
+                staged_desc,
+                *pipeline_context->desc_
+            );
+        }
+        pipeline_context->staged_descriptor_ = staged_desc;
+        pipeline_context->desc_ = staged_desc;
+    } else {
+        pipeline_context->staged_descriptor_ =
+            merge_descriptors(seg.descriptor(), incomplete_segments, read_query.columns);
+        if (pipeline_context->desc_) {
+            const std::array fields_ptr = {pipeline_context->desc_->fields_ptr()};
+            pipeline_context->desc_ =
+                merge_descriptors(*pipeline_context->staged_descriptor_, fields_ptr, read_query.columns);
+        } else {
+            pipeline_context->desc_ = pipeline_context->staged_descriptor_;
+        }
+    }
+
     modify_descriptor(pipeline_context, read_options);
     if (convert_int_to_float) {
-        stream::convert_descriptor_types(*pipeline_context->desc_);
+        stream::convert_descriptor_types(*pipeline_context->staged_descriptor_);
     }
 
     generate_filtered_field_descriptors(pipeline_context, read_query.columns);
-    pipeline_context->slice_and_keys_.insert(std::end(pipeline_context->slice_and_keys_), std::begin(incomplete_segments),  std::end(incomplete_segments));
     pipeline_context->total_rows_ = pipeline_context->calc_rows();
+    return true;
 }
 
 void check_incompletes_index_ranges_dont_overlap(const std::shared_ptr<PipelineContext>& pipeline_context,
@@ -1357,12 +1394,39 @@ VersionedItem collate_and_write(
     });
 }
 
+class IncompleteKeysRAII {
+public:
+    IncompleteKeysRAII(std::shared_ptr<PipelineContext> pipeline_context, std::shared_ptr<Store> store)
+            : context_(pipeline_context),
+              store_(store) {
+    }
+
+    ~IncompleteKeysRAII(){
+        std::vector<entity::VariantKey> keys_to_delete;
+        for (auto sk = context_->incompletes_begin(); sk != context_->end(); ++sk) {
+            const auto& slice_and_key = sk->slice_and_key();
+            internal::check<ErrorCode::E_ASSERTION_FAILURE>(
+                slice_and_key.key().type() == KeyType::APPEND_DATA,
+                "Deleting incorrect key type {}",
+                slice_and_key.key().type()
+            );
+            keys_to_delete.emplace_back(slice_and_key.key());
+        }
+        store_->remove_keys(keys_to_delete).get();
+    }
+
+private:
+    std::shared_ptr<PipelineContext> context_;
+    std::shared_ptr<Store> store_;
+};
+
 VersionedItem sort_merge_impl(
     const std::shared_ptr<Store>& store,
     const StreamId& stream_id,
     const std::optional<arcticdb::proto::descriptors::UserDefinedMetadata>& norm_meta,
     const UpdateInfo& update_info,
-    const CompactIncompleteOptions& options) {
+    const CompactIncompleteOptions& options,
+    const WriteOptions& write_options) {
     auto pipeline_context = std::make_shared<PipelineContext>();
     pipeline_context->stream_id_ = stream_id;
     pipeline_context->version_id_ = update_info.next_version_id_;
@@ -1373,28 +1437,25 @@ VersionedItem sort_merge_impl(
         read_indexed_keys_to_pipeline(store, pipeline_context, *(update_info.previous_index_key_), read_query, ReadOptions{});
         previous_sorted_value.emplace(pipeline_context->desc_->sorted());
     }
+    pipeline_context->incompletes_after_ = pipeline_context->slice_and_keys_.size();
+    IncompleteKeysRAII incomplete_keys(pipeline_context, store);
+    const auto num_versioned_rows = pipeline_context->total_rows_;
 
-    auto num_versioned_rows = pipeline_context->total_rows_;
-
-    read_incompletes_to_pipeline(store,
-                                 pipeline_context,
-                                 read_query,
-                                 ReadOptions{},
-                                 options.convert_int_to_float_,
-                                 options.via_iteration_,
-                                 options.sparsify_);
+    const bool has_incomplete_segments = read_incompletes_to_pipeline(
+        store,
+        pipeline_context,
+        read_query,
+        ReadOptions{},
+        options.convert_int_to_float_,
+        options.via_iteration_,
+        options.sparsify_,
+        write_options.dynamic_schema
+    );
     user_input::check<ErrorCode::E_NO_STAGED_SEGMENTS>(
-        pipeline_context->slice_and_keys_.size() > 0,
+        has_incomplete_segments,
         "Finalizing staged data is not allowed with empty staging area"
     );
 
-    std::vector<entity::VariantKey> delete_keys;
-    for(auto sk = pipeline_context->incompletes_begin(); sk != pipeline_context->end(); ++sk) {
-        const auto& slice_and_key = sk->slice_and_key();
-        util::check(slice_and_key.key().type() == KeyType::APPEND_DATA, "Deleting incorrect key type {}", slice_and_key.key().type());
-        delete_keys.emplace_back(slice_and_key.key());
-    }
-
     std::vector<FrameSlice> slices;
     std::vector<folly::Future<VariantKey>> fut_vec;
     auto index = stream::index_type_from_descriptor(pipeline_context->descriptor());
@@ -1405,15 +1466,23 @@ VersionedItem sort_merge_impl(
             //const auto split_size = ConfigsMap::instance()->get_int("Split.RowCount", 10000);
             //read_query.clauses_.emplace_back(std::make_shared<Clause>(SplitClause{static_cast<size_t>(split_size)}));
 
-            read_query.clauses_.emplace_back(std::make_shared<Clause>(MergeClause{timeseries_index, SparseColumnPolicy{}, stream_id, pipeline_context->descriptor()}));
+            read_query.clauses_.emplace_back(std::make_shared<Clause>(MergeClause{
+                timeseries_index,
+                SparseColumnPolicy{},
+                stream_id,
+                pipeline_context->descriptor(),
+                write_options.dynamic_schema
+            }));
             auto segments = read_and_process(store, pipeline_context, read_query, ReadOptions{}, pipeline_context->incompletes_after());
-            if (options.append_ && update_info.previous_index_key_.has_value() &&
-                update_info.previous_index_key_->end_time() - 1 > std::get<timestamp>(TimeseriesIndex::start_value_for_segment(segments[0].segment_.value()))) {
-                store->remove_keys(delete_keys).get();
-                sorting::raise<ErrorCode::E_UNSORTED_DATA>(
-                    "Cannot append staged segments to existing data as incomplete segment contains index value < existing data (in UTC): {} <= {}",
-                    date_and_time(std::get<timestamp>(TimeseriesIndex::start_value_for_segment(segments[0].segment_.value()))),
-                    date_and_time(update_info.previous_index_key_->end_time() - 1)
+            if (options.append_ && update_info.previous_index_key_ && !segments.empty()) {
+                const timestamp last_index_on_disc = update_info.previous_index_key_->end_time() - 1;
+                const timestamp incomplete_start =
+                    std::get<timestamp>(TimeseriesIndex::start_value_for_segment(segments[0].segment(store)));
+                sorting::check<ErrorCode::E_UNSORTED_DATA>(
+                    last_index_on_disc <= incomplete_start,
+                    "Cannot append staged segments to existing data as incomplete segment contains index value {} < existing data {}",
+                    date_and_time(incomplete_start),
+                    date_and_time(last_index_on_disc)
                 );
             }
             pipeline_context->total_rows_ = num_versioned_rows + get_slice_rowcounts(segments);
@@ -1424,22 +1493,25 @@ VersionedItem sort_merge_impl(
                 [&slices](FrameSlice &&slice) {
                     slices.emplace_back(std::move(slice));
                 },
-                DynamicSchema{pipeline_context->descriptor(), index},
-                [pipeline_context=pipeline_context, &fut_vec, &store](SegmentInMemory &&segment) {
-                    auto local_index_start = TimeseriesIndex::start_value_for_segment(segment);
-                    auto local_index_end = TimeseriesIndex::end_value_for_segment(segment);
+                DynamicSchema{*pipeline_context->staged_descriptor_, index},
+                [pipeline_context, &fut_vec, &store](SegmentInMemory &&segment) {
+                    StreamDescriptor only_non_empty_cols;
+                    const auto local_index_start = TimeseriesIndex::start_value_for_segment(segment);
+                    const auto local_index_end = TimeseriesIndex::end_value_for_segment(segment);
                     stream::StreamSink::PartialKey
                     pk{KeyType::TABLE_DATA, pipeline_context->version_id_, pipeline_context->stream_id_, local_index_start, local_index_end};
                     fut_vec.emplace_back(store->write(pk, std::move(segment)));
                 }};
 
             for(auto sk = segments.begin(); sk != segments.end(); ++sk) {
-                aggregator.add_segment(
-                    std::move(sk->segment(store)),
-                    sk->slice(),
-                    options.convert_int_to_float_);
-
-                sk->unset_segment();
+                SegmentInMemory segment = sk->release_segment(store);
+                // Empty columns can appear only of one staged segment is empty and adds column which
+                // does not appear in any other segment. There can also be empty columns if all segments
+                // are empty in that case this loop won't be reached as segments.size() will be 0
+                if (write_options.dynamic_schema) {
+                    segment.drop_empty_columns();
+                }
+                aggregator.add_segment(std::move(segment), sk->slice(), options.convert_int_to_float_);
             }
             aggregator.commit();
             pipeline_context->desc_->set_sorted(deduce_sorted(previous_sorted_value.value_or(SortedValue::ASCENDING), SortedValue::ASCENDING));
@@ -1458,7 +1530,6 @@ VersionedItem sort_merge_impl(
         pipeline_context->incompletes_after(),
         norm_meta);
 
-    store->remove_keys(delete_keys).get();
     return vit;
 }
 
@@ -1483,17 +1554,21 @@ VersionedItem compact_incomplete_impl(
         read_indexed_keys_to_pipeline(store, pipeline_context, *(update_info.previous_index_key_), read_query, read_options);
         previous_sorted_value.emplace(pipeline_context->desc_->sorted());
     }
+    IncompleteKeysRAII incomplete_keys(pipeline_context, store);
+
+    const bool has_incomplete_segments = read_incompletes_to_pipeline(
+        store,
+        pipeline_context,
+        read_query,
+        ReadOptions{},
+        options.convert_int_to_float_,
+        options.via_iteration_,
+        options.sparsify_,
+        write_options.dynamic_schema
+    );
 
-    auto prev_size = pipeline_context->slice_and_keys_.size();
-    read_incompletes_to_pipeline(store,
-                                 pipeline_context,
-                                 ReadQuery{},
-                                 ReadOptions{},
-                                 options.convert_int_to_float_,
-                                 options.via_iteration_,
-                                 options.sparsify_);
     user_input::check<ErrorCode::E_NO_STAGED_SEGMENTS>(
-        pipeline_context->slice_and_keys_.size() != prev_size,
+        has_incomplete_segments,
         "Finalizing staged data is not allowed with empty staging area"
     );
     if (options.validate_index_) {
@@ -1501,12 +1576,6 @@ VersionedItem compact_incomplete_impl(
     }
     const auto& first_seg = pipeline_context->slice_and_keys_.begin()->segment(store);
 
-    std::vector<entity::VariantKey> delete_keys;
-    for(auto sk = pipeline_context->incompletes_begin(); sk != pipeline_context->end(); ++sk) {
-        util::check(sk->slice_and_key().key().type() == KeyType::APPEND_DATA, "Deleting incorrect key type {}", sk->slice_and_key().key().type());
-        delete_keys.emplace_back(sk->slice_and_key().key());
-    }
-
     std::vector<folly::Future<VariantKey>> fut_vec;
     std::vector<FrameSlice> slices;
     bool dynamic_schema = write_options.dynamic_schema;
@@ -1522,6 +1591,7 @@ VersionedItem compact_incomplete_impl(
         using IndexType = std::remove_reference_t<decltype(idx)>;
         using SchemaType = std::remove_reference_t<decltype(schema)>;
         using ColumnPolicyType = std::remove_reference_t<decltype(column_policy)>;
+        constexpr bool validate_index_sorted = IndexType::type() == IndexDescriptorImpl::Type::TIMESTAMP;
         do_compact<IndexType, SchemaType, RowCountSegmentPolicy, ColumnPolicyType>(
                 pipeline_context->incompletes_begin(),
                 pipeline_context->end(),
@@ -1530,7 +1600,8 @@ VersionedItem compact_incomplete_impl(
                 slices,
                 store,
                 options.convert_int_to_float_,
-                write_options.segment_row_size);
+                write_options.segment_row_size,
+                validate_index_sorted);
         if constexpr(std::is_same_v<IndexType, TimeseriesIndex>) {
             pipeline_context->desc_->set_sorted(deduce_sorted(previous_sorted_value.value_or(SortedValue::ASCENDING), SortedValue::ASCENDING));
         }
@@ -1545,7 +1616,6 @@ VersionedItem compact_incomplete_impl(
         pipeline_context->incompletes_after(),
         user_meta);
 
-    store->remove_keys(delete_keys).get();
     return vit;
 }
 
@@ -1634,7 +1704,8 @@ VersionedItem defragment_symbol_data_impl(
             slices,
             store,
             false,
-            segment_size);
+            segment_size,
+            false);
     });
 
     auto keys = folly::collect(fut_vec).get();
@@ -1688,7 +1759,7 @@ FrameAndDescriptor read_frame_for_version(
         const auto& query_range = std::get<IndexRange>(read_query.row_filter);
         const auto existing_range = pipeline_context->index_range();
         if(!existing_range.specified_ || query_range.end_ > existing_range.end_)
-            read_incompletes_to_pipeline(store, pipeline_context, read_query, read_options, false, false, false);
+            read_incompletes_to_pipeline(store, pipeline_context, read_query, read_options, false, false, false,  opt_false(read_options.dynamic_schema_));
     }
 
     if(std::holds_alternative<StreamId>(version_info) && !pipeline_context->incompletes_after_) {
diff --git a/cpp/arcticdb/version/version_core.hpp b/cpp/arcticdb/version/version_core.hpp
index e885801af2..27bdd65aa5 100644
--- a/cpp/arcticdb/version/version_core.hpp
+++ b/cpp/arcticdb/version/version_core.hpp
@@ -198,7 +198,8 @@ VersionedItem sort_merge_impl(
     const StreamId& stream_id,
     const std::optional<arcticdb::proto::descriptors::UserDefinedMetadata>& user_meta,
     const UpdateInfo& update_info,
-    const CompactIncompleteOptions& options);
+    const CompactIncompleteOptions& options,
+    const WriteOptions& write_options);
 
 void modify_descriptor(
     const std::shared_ptr<pipelines::PipelineContext>& pipeline_context,
diff --git a/cpp/arcticdb/version/version_map.hpp b/cpp/arcticdb/version/version_map.hpp
index 50afa11b0f..cb9a15786c 100644
--- a/cpp/arcticdb/version/version_map.hpp
+++ b/cpp/arcticdb/version/version_map.hpp
@@ -509,7 +509,7 @@ class VersionMapImpl {
 
         util::check(key.type() != KeyType::TABLE_INDEX || !entry->head_.has_value() || key.version_id() > entry->head_->version_id(),
                     "Trying to write a non-increasing TABLE_INDEX key. New version: {}, Last version: {}",
-                    key.version_id(), entry->head_->version_id());
+                    key.version_id(), entry->head_ ? entry->head_->version_id() : VariantId{""});
         auto journal_key = to_atom(std::move(journal_single_key(store, key, entry->head_)));
         write_to_entry(entry, key, journal_key);
         auto previous_index = entry->get_second_undeleted_index();


Test patch:
diff --git a/cpp/arcticdb/processing/test/benchmark_clause.cpp b/cpp/arcticdb/processing/test/benchmark_clause.cpp
index 35ce8d56e3..ecd78ec98b 100644
--- a/cpp/arcticdb/processing/test/benchmark_clause.cpp
+++ b/cpp/arcticdb/processing/test/benchmark_clause.cpp
@@ -49,7 +49,7 @@ void time_merge_on_segments(const std::vector<SegmentInMemory> &segments, benchm
     auto stream_id = StreamId("Merge");
     StreamDescriptor descriptor{};
     descriptor.add_field(FieldRef{make_scalar_type(DataType::NANOSECONDS_UTC64),"time"});
-    MergeClause merge_clause{TimeseriesIndex{"time"}, DenseColumnPolicy{}, stream_id, descriptor};
+    MergeClause merge_clause{TimeseriesIndex{"time"}, DenseColumnPolicy{}, stream_id, descriptor, false};
     merge_clause.set_component_manager(component_manager);
     state.ResumeTiming();
 
diff --git a/cpp/arcticdb/processing/test/test_clause.cpp b/cpp/arcticdb/processing/test/test_clause.cpp
index 1da137317c..442d5821a9 100644
--- a/cpp/arcticdb/processing/test/test_clause.cpp
+++ b/cpp/arcticdb/processing/test/test_clause.cpp
@@ -371,13 +371,11 @@ TEST(Clause, Merge) {
     const auto num_segs = 4;
     const auto num_rows = 20;
 
-    auto stream_id = StreamId("Merge");
-    StreamDescriptor descriptor{};
-    descriptor.add_field(FieldRef{make_scalar_type(DataType::NANOSECONDS_UTC64),"time"});
-    MergeClause merge_clause{TimeseriesIndex{"time"}, SparseColumnPolicy{}, stream_id, descriptor};
+    const auto stream_id = StreamId("Merge");
+    const auto seg = get_standard_timeseries_segment(std::get<StringId>(stream_id), num_rows);
+    MergeClause merge_clause{TimeseriesIndex{"time"}, SparseColumnPolicy{}, stream_id, seg.descriptor(), false};
     merge_clause.set_component_manager(component_manager);
 
-    auto seg = get_standard_timeseries_segment(std::get<StringId>(stream_id), num_rows);
 
     std::vector<SegmentInMemory> segs;
     for(auto x = 0u; x < num_segs; ++x) {
diff --git a/cpp/arcticdb/version/test/test_sparse.cpp b/cpp/arcticdb/version/test/test_sparse.cpp
index 02611f6fdd..92241eaa72 100644
--- a/cpp/arcticdb/version/test/test_sparse.cpp
+++ b/cpp/arcticdb/version/test/test_sparse.cpp
@@ -592,6 +592,8 @@ TEST_F(SparseTestStore, Compact) {
     };
 
     DynamicAggregator aggregator(std::move(schema), [&](SegmentInMemory &&segment) {
+        ASSERT_TRUE(segment.is_index_sorted());
+        segment.descriptor().set_sorted(SortedValue::ASCENDING);
         test_store_->append_incomplete_segment(stream_id, std::move(segment));
     }, RowCountSegmentPolicy{10});
 
@@ -647,6 +649,8 @@ TEST_F(SparseTestStore, CompactWithStrings) {
     };
 
     DynamicAggregator aggregator(std::move(schema), [&](SegmentInMemory &&segment) {
+        ASSERT_TRUE(segment.is_index_sorted());
+        segment.descriptor().set_sorted(SortedValue::ASCENDING);
         test_store_->append_incomplete_segment(stream_id, std::move(segment));
     }, RowCountSegmentPolicy{10});
 
diff --git a/cpp/arcticdb/version/test/test_version_store.cpp b/cpp/arcticdb/version/test/test_version_store.cpp
index d70371bf1b..e2cd1e9805 100644
--- a/cpp/arcticdb/version/test/test_version_store.cpp
+++ b/cpp/arcticdb/version/test/test_version_store.cpp
@@ -250,6 +250,8 @@ TEST_F(VersionStoreTest, CompactIncompleteDynamicSchema) {
     std::shuffle(data.begin(), data.end(), mt);
 
     for(auto& frame : data) {
+        ASSERT_TRUE(frame.segment_.is_index_sorted());
+        frame.segment_.descriptor().set_sorted(SortedValue::ASCENDING);
         test_store_->write_parallel_frame(symbol, std::move(frame.input_frame_), true);
     }
 
diff --git a/python/tests/conftest.py b/python/tests/conftest.py
index 161ae93b82..51decbba58 100644
--- a/python/tests/conftest.py
+++ b/python/tests/conftest.py
@@ -19,6 +19,7 @@
 from datetime import datetime
 from functools import partial
 
+from arcticdb import LibraryOptions
 from arcticdb.storage_fixtures.api import StorageFixture
 from arcticdb.storage_fixtures.azure import AzuriteStorageFixtureFactory
 from arcticdb.storage_fixtures.lmdb import LmdbStorageFixture
@@ -109,6 +110,19 @@ def lmdb_storage(tmp_path):
 def lmdb_library(lmdb_storage, lib_name):
     return lmdb_storage.create_arctic().create_library(lib_name)
 
+@pytest.fixture
+def lmdb_library_dynamic_schema(lmdb_storage, lib_name):
+    return lmdb_storage.create_arctic().create_library(lib_name, library_options=LibraryOptions(dynamic_schema=True))
+
+@pytest.fixture(
+    scope="function",
+    params=(
+        "lmdb_library",
+        "lmdb_library_dynamic_schema"
+    ),
+)
+def lmdb_library_static_dynamic(request):
+    yield request.getfixturevalue(request.param)
 
 # ssl is enabled by default to maximize test coverage as ssl is enabled most of the times in real world
 @pytest.fixture(scope="session")
diff --git a/python/tests/hypothesis/arcticdb/test_sort_merge.py b/python/tests/hypothesis/arcticdb/test_sort_merge.py
index e39f1ccfa3..490158dac6 100644
--- a/python/tests/hypothesis/arcticdb/test_sort_merge.py
+++ b/python/tests/hypothesis/arcticdb/test_sort_merge.py
@@ -1,71 +1,229 @@
-from re import S
 import pandas as pd
 import pytest
-from hypothesis import assume
+from hypothesis import assume, given, settings
 import hypothesis.extra.pandas as hs_pd
 import hypothesis.strategies as st
 from hypothesis.stateful import RuleBasedStateMachine, rule, initialize, run_state_machine_as_test, precondition
 from collections import namedtuple
 from pandas.testing import assert_frame_equal
 from arcticdb.version_store.library import StagedDataFinalizeMethod
+from arcticdb.exceptions import UserInputException, StreamDescriptorMismatch, UnsortedDataException, NoSuchVersionException, SchemaException
+import numpy as np
+import string
+from arcticdb.util._versions import IS_PANDAS_TWO
+from pandas.api.types import is_numeric_dtype, is_integer_dtype, is_float_dtype
+from arcticdb.util.hypothesis import use_of_function_scoped_fixtures_in_hypothesis_checked
+from arcticdb.toolbox.library_tool import KeyType
 
 ColumnInfo = namedtuple('ColumnInfo', ['name', 'dtype'])
 
-COLUMN_DESCRIPTIONS = [ColumnInfo("a", "float"), ColumnInfo("b", "int64"), ColumnInfo("c", "str"), ColumnInfo("d", "datetime64[ns]")]
+COLUMNS = [f"col_{i}" for i in range(0, 5)]
+DTYPES = ["int16", "int64", "float", "object", "datetime64[ns]"]
+COLUMN_DESCRIPTIONS = [ColumnInfo(name, dtype) for name in COLUMNS for dtype in DTYPES]
+
+def are_dtypes_compatible(left, right):
+    if left == right:
+        return True
+    if is_numeric_dtype(left) and is_numeric_dtype(right):
+        return True
+    return False
+
+def string_column_strategy(name):
+    return hs_pd.column(name=name, elements=st.text(alphabet=string.ascii_letters))
 
 @st.composite
-def df(draw, column_list):
-    column_infos = draw(st.lists(st.sampled_from(column_list), unique=True))
-    columns = [hs_pd.column(name=ci.name, dtype=ci.dtype) for ci in column_infos]
-    return draw(hs_pd.data_frames(columns, index=hs_pd.indexes(dtype="datetime64[ns]")))
-
-class StagedWrite(RuleBasedStateMachine):
-
-    lib = None
- 
-    @initialize()
-    def init(self):
-        self.df = pd.DataFrame([])
-
-    @rule(df=df(COLUMN_DESCRIPTIONS))
-    def stage(self, df):
-        self.df = pd.concat([self.df, df])
-        self.lib.write("sym", df, staged=True)
-
-    @rule()
-    def finalize_write(self):
-        self.df.sort_index(inplace=True)
-        self.lib.sort_and_finalize_staged_data("sym")
-        arctic_df = self.lib.read("sym").data
-        assert arctic_df.index.equals(self.df.index)
-        assert_frame_equal(arctic_df, self.df, check_like=True)
-        self.df = pd.DataFrame([])
-        
-    @rule()
-    def finalize_append(self):
-        self.df.sort_index(inplace=True)
-        pre_append_arctic = self.lib.read("sym").data if "sym" in self.lib.list_symbols() else pd.DataFrame([])
-
-        if len(pre_append_arctic) > 0 and len(self.df) > 0 and pre_append_arctic.index[-1] > self.df.index[0]:
-            # Make sure appends keep the index ordered
-            with pytest.raises(Exception) as exception_info:
-                self.lib.sort_and_finalize_staged_data("sym", mode=StagedDataFinalizeMethod.APPEND)
-            assert "append" in str(exception_info.value)
-        else:
-            self.lib.sort_and_finalize_staged_data("sym", mode=StagedDataFinalizeMethod.APPEND)
-            post_append_arctic = self.lib.read("sym").data
-            appended_arctic = post_append_arctic[len(pre_append_arctic):]
-            assert appended_arctic.index.equals(self.df.index)
-            assert_frame_equal(appended_arctic, self.df, check_like=True)
-        self.df = pd.DataFrame([])
-
-    def teardown(self):
-        self.df = pd.DataFrame([])
-        for sym in self.lib.list_symbols():
-            self.lib.delete(sym)
-
-@pytest.mark.skip(reason="Needs to resolve the issues found in unit tests.")
-def test_sort_and_finalize_staged_data(lmdb_storage, lib_name):
-    ac = lmdb_storage.create_arctic()
-    StagedWrite.lib = ac.create_library(lib_name)
-    run_state_machine_as_test(StagedWrite)
+def generate_single_dataframe(draw, column_list, min_size=0, allow_nat_in_index=True):
+    column_infos = draw(st.lists(st.sampled_from(column_list), unique_by=lambda x: x.name, min_size=1))
+    columns = [hs_pd.column(name=ci.name, dtype=ci.dtype) if ci.dtype != 'object' else string_column_strategy(ci.name) for ci in column_infos]
+    if not IS_PANDAS_TWO:
+        # Due to https://github.com/man-group/ArcticDB/blob/7479c0b0caa8121bc2ca71a73e29769bbc41c66a/python/arcticdb/version_store/_normalization.py#L184
+        # we change the dtype of empty float columns. This makes hypothesis tests extremely hard to write as we must
+        # keep addional state about is there a mix of empty/non-empty float columns in the staging area, did we write
+        # empty float column (if so it's type would be object). These edge cases are covered in the unit tests.
+        index = hs_pd.indexes(dtype="datetime64[ns]", min_size=1 if min_size <= 0 else min_size).filter(lambda x: allow_nat_in_index or not pd.NaT in x)
+    else:
+        index = hs_pd.indexes(dtype="datetime64[ns]", min_size=min_size).filter(lambda x: allow_nat_in_index or not pd.NaT in x)
+    return draw(hs_pd.data_frames(columns, index=index))
+
+@st.composite
+def generate_dataframes(draw, column_list):
+    return draw(st.lists(generate_single_dataframe(COLUMN_DESCRIPTIONS)))
+
+def assert_equal(left, right, dynamic=False):
+    """
+    The sorting Arctic does is not stable. Thus when there are repeated index values the
+    result from our sorting and the result from Pandas' sort might differ where the index
+    values are repeated.
+    """
+    if any(left.index.duplicated()):
+        assert left.index.equals(right.index), f"Indexes are different {left.index} != {right.index}"
+        assert set(left.columns) == set(right.columns), f"Column sets are different {set(left.columns)} != {set(right.columns)}"
+        assert left.shape == right.shape, f"Shapes are different {left.shape} != {right.shape}"
+        left_groups = left.groupby(left.index, sort=False).apply(lambda x: x.sort_values(list(left.columns)))
+        right_groups = right.groupby(right.index, sort=False).apply(lambda x: x.sort_values(list(left.columns)))
+        assert_frame_equal(left_groups, right_groups, check_like=True, check_dtype=False)
+    else:
+        assert_frame_equal(left, right, check_like=True, check_dtype=False)
+
+def assert_cannot_finalize_without_staged_data(lib, symbol, mode):
+    with pytest.raises(UserInputException) as exception_info:
+        lib.sort_and_finalize_staged_data(symbol, mode=mode)
+    assert "E_NO_STAGED_SEGMENTS" in str(exception_info.value)
+    assert len(get_append_keys(lib, symbol)) == 0
+  
+def assert_nat_is_not_supported(lib, symbol, mode):
+    with pytest.raises(UnsortedDataException) as exception_info:
+        lib.sort_and_finalize_staged_data(symbol, mode=mode)
+    assert "E_UNSORTED_DATA" in str(exception_info.value)
+    assert len(get_append_keys(lib, symbol)) == 0
+
+
+def assert_staged_columns_are_incompatible(lib, symbol, mode):
+    with pytest.raises(SchemaException) as exception_info:
+        lib.sort_and_finalize_staged_data(symbol, mode)
+    assert "E_DESCRIPTOR_MISMATCH" in str(exception_info.value)
+    assert len(get_append_keys(lib, symbol)) == 0
+
+def has_nat_in_index(segment_list):
+    return any(pd.NaT in segment.index for segment in segment_list)
+
+def merge_and_sort_segment_list(segment_list, int_columns_in_df=None):
+    merged = pd.concat(segment_list)
+    # pd.concat promotes dtypes. If there are missing values in an int typed column
+    # it will become float column and the missing values will be NaN.
+    int_columns_in_df = int_columns_in_df if int_columns_in_df else []
+    for col in int_columns_in_df:
+        merged[col] = merged[col].replace(np.NaN, 0)
+    merged.sort_index(inplace=True)
+    return merged
+
+def assert_appended_data_does_not_overlap_with_storage(lib, symbol):
+    with pytest.raises(UnsortedDataException) as exception_info:
+        lib.sort_and_finalize_staged_data(symbol, mode=StagedDataFinalizeMethod.APPEND)
+    assert "E_UNSORTED_DATA" in str(exception_info.value)
+    assert "append" in str(exception_info.value)
+    assert len(get_append_keys(lib, symbol)) == 0
+
+def segments_have_compatible_schema(segment_list):
+    """
+    Used to check dynamic schemas. Considers all numeric types for compatible.
+    """
+    dtypes = {}
+    for segment in segment_list:
+        for col in segment:
+            if col not in dtypes:
+                dtypes[col] = segment.dtypes[col]
+            elif not are_dtypes_compatible(dtypes[col], segment.dtypes[col]):
+                return False
+    return True
+
+def get_append_keys(lib, sym):
+    lib_tool = lib._nvs.library_tool()
+    keys = lib_tool.find_keys_for_symbol(KeyType.APPEND_DATA, sym)
+    return keys
+
+@use_of_function_scoped_fixtures_in_hypothesis_checked
+@settings(deadline=None)
+@given(df_list=generate_dataframes(COLUMN_DESCRIPTIONS))
+def test_sort_merge_static_schema_write(lmdb_library, df_list):
+    lib = lmdb_library
+    sym = "test_sort_merge_static_schema_write"
+    for df in df_list:
+        lib.write(sym, df, staged=True, validate_index=False)
+    if len(df_list) == 0:
+        assert_cannot_finalize_without_staged_data(lib, sym, StagedDataFinalizeMethod.WRITE)
+        return
+    if not all(df_list[0].dtypes.equals(segment.dtypes) for segment in df_list):
+        assert_staged_columns_are_incompatible(lib, sym, StagedDataFinalizeMethod.WRITE)
+        return
+    if has_nat_in_index(df_list):
+        assert_nat_is_not_supported(lib, sym, StagedDataFinalizeMethod.WRITE)
+        return
+    lib.sort_and_finalize_staged_data(sym, mode=StagedDataFinalizeMethod.WRITE)
+    assert len(get_append_keys(lib, sym)) == 0
+    expected = merge_and_sort_segment_list(df_list)
+    data = lib.read(sym).data
+    assert_equal(expected, data)
+
+@use_of_function_scoped_fixtures_in_hypothesis_checked
+@settings(deadline=None)
+@given(df_list=generate_dataframes(COLUMN_DESCRIPTIONS), initial_df=generate_single_dataframe(COLUMN_DESCRIPTIONS, min_size=1, allow_nat_in_index=False))
+def test_sort_merge_static_schema_append(lmdb_library, df_list, initial_df):
+    lib = lmdb_library
+    sym = "test_sort_merge_static_schema_append"
+    initial_df.sort_index(inplace=True)
+    lib.write(sym, initial_df)
+    for df in df_list:
+        lib.write(sym, df, staged=True, validate_index=False)
+    if len(df_list) == 0:
+        assert_cannot_finalize_without_staged_data(lib, sym, StagedDataFinalizeMethod.APPEND)
+        return
+    if not all(initial_df.dtypes.equals(segment.dtypes) for segment in df_list):
+        assert_staged_columns_are_incompatible(lib, sym, StagedDataFinalizeMethod.APPEND)
+        return
+    if has_nat_in_index(df_list):
+        assert_nat_is_not_supported(lib, sym, StagedDataFinalizeMethod.APPEND)
+        return
+    merged_staging = merge_and_sort_segment_list(df_list)
+    if len(merged_staging) > 0 and initial_df.index[-1] > merged_staging.index[0]:
+        assert_appended_data_does_not_overlap_with_storage(lib, sym)
+        return
+    lib.sort_and_finalize_staged_data(sym, mode=StagedDataFinalizeMethod.APPEND)
+    assert len(get_append_keys(lib, sym)) == 0
+    expected = pd.concat([initial_df, merged_staging])
+    data = lib.read(sym).data
+    assert_equal(expected, data)
+
+@use_of_function_scoped_fixtures_in_hypothesis_checked
+@settings(deadline=None)
+@given(df_list=generate_dataframes(COLUMN_DESCRIPTIONS))
+def test_sort_merge_dynamic_schema_write(lmdb_library_dynamic_schema, df_list):
+    lib = lmdb_library_dynamic_schema
+    sym = "test_sort_merge_dynamic_schema_write"
+    for df in df_list:
+        lib.write(sym, df, staged=True, validate_index=False)
+    if len(df_list) == 0:
+        assert_cannot_finalize_without_staged_data(lib, sym, StagedDataFinalizeMethod.WRITE)
+        return
+    if not segments_have_compatible_schema(df_list):
+        assert_staged_columns_are_incompatible(lib, sym, StagedDataFinalizeMethod.WRITE)
+        return
+    if has_nat_in_index(df_list):
+        assert_nat_is_not_supported(lib, sym, StagedDataFinalizeMethod.WRITE)
+        return
+    lib.sort_and_finalize_staged_data(sym, mode=StagedDataFinalizeMethod.WRITE)
+    assert len(get_append_keys(lib, sym)) == 0
+    data = lib.read(sym).data
+    int_columns_in_df = [col_name for col_name in data if is_integer_dtype(data.dtypes[col_name])]
+    expected = merge_and_sort_segment_list(df_list, int_columns_in_df=int_columns_in_df)
+    assert_equal(expected, data)
+
+@use_of_function_scoped_fixtures_in_hypothesis_checked
+@settings(deadline=None)
+@given(df_list=generate_dataframes(COLUMN_DESCRIPTIONS), initial_df=generate_single_dataframe(COLUMN_DESCRIPTIONS, min_size=1, allow_nat_in_index=False))
+def test_sort_merge_dynamic_schema_append(lmdb_library_dynamic_schema, df_list, initial_df):    
+    lib = lmdb_library_dynamic_schema
+    sym = "test_sort_merge_dynamic_schema_append"
+    initial_df.sort_index(inplace=True)
+    lib.write(sym, initial_df)
+    for df in df_list:
+        lib.write(sym, df, staged=True, validate_index=False)
+    if len(df_list) == 0:
+        assert_cannot_finalize_without_staged_data(lib, sym, StagedDataFinalizeMethod.APPEND)
+        return
+    if not segments_have_compatible_schema([initial_df, *df_list]):
+        assert_staged_columns_are_incompatible(lib, sym, StagedDataFinalizeMethod.APPEND)
+        return
+    if has_nat_in_index(df_list):
+        assert_nat_is_not_supported(lib, sym, StagedDataFinalizeMethod.APPEND)
+        return
+    merged_staging = merge_and_sort_segment_list(df_list)
+    if len(merged_staging) > 0 and initial_df.index[-1] > merged_staging.index[0]:
+        assert_appended_data_does_not_overlap_with_storage(lib, sym)
+        return
+    lib.sort_and_finalize_staged_data(sym, mode=StagedDataFinalizeMethod.APPEND)
+    assert len(get_append_keys(lib, sym)) == 0
+    data = lib.read(sym).data
+    int_columns_in_df = [col_name for col_name in data if is_integer_dtype(data.dtypes[col_name])]
+    expected = merge_and_sort_segment_list([initial_df, merged_staging], int_columns_in_df=int_columns_in_df)
+    assert_equal(expected, data)
\ No newline at end of file
diff --git a/python/tests/integration/arcticdb/test_arctic.py b/python/tests/integration/arcticdb/test_arctic.py
index e5ccbfc801..86ccef336e 100644
--- a/python/tests/integration/arcticdb/test_arctic.py
+++ b/python/tests/integration/arcticdb/test_arctic.py
@@ -19,7 +19,7 @@
 from arcticdb_ext.exceptions import InternalException, SortingException, UserInputException
 from arcticdb_ext.storage import NoDataFoundException
 from arcticdb_ext.version_store import SortedValue
-from arcticdb.exceptions import ArcticDbNotYetImplemented, LibraryNotFound, MismatchingLibraryOptions
+from arcticdb.exceptions import ArcticDbNotYetImplemented, LibraryNotFound, MismatchingLibraryOptions, StreamDescriptorMismatch, SchemaException
 from arcticdb.adapters.mongo_library_adapter import MongoLibraryAdapter
 from arcticdb.arctic import Arctic
 from arcticdb.options import LibraryOptions, EnterpriseLibraryOptions
@@ -292,42 +292,15 @@ def test_parallel_writes_and_appends_index_validation(arctic_library, finalize_m
 
 
 @pytest.mark.parametrize("finalize_method", (StagedDataFinalizeMethod.APPEND, StagedDataFinalizeMethod.WRITE))
-class TestFinalizeWithEmptySegments:
-    def test_staged_segment_is_only_empty_dfs(self, lmdb_library, finalize_method):
-        lib = lmdb_library
-        lib.write("sym", pd.DataFrame([]), staged=True)
-        lib.write("sym", pd.DataFrame([]), staged=True)
-        lib.finalize_staged_data("sym", mode=finalize_method)
-        assert_frame_equal(lib.read("sym").data, pd.DataFrame([], index=pd.DatetimeIndex([])))
-
-    def test_staged_segment_has_empty_df(self, lmdb_library, finalize_method):
-        lib = lmdb_library
-        index = pd.DatetimeIndex([pd.Timestamp(2024, 1, 1), pd.Timestamp(2024, 1, 3), pd.Timestamp(2024, 1, 4)])
-        df1 = pd.DataFrame({"col": [1, 2, 3]}, index=index)
-        df2 = pd.DataFrame({})
-        df3 = pd.DataFrame({"col": [4]}, index=pd.DatetimeIndex([pd.Timestamp(2024, 1, 5)]))
-        lib.write("sym", df1, staged=True)
-        lib.write("sym", df2, staged=True)
-        lib.write("sym", df3, staged=True)
-        lib.finalize_staged_data("sym", mode=finalize_method)
-        assert_frame_equal(lib.read("sym").data, pd.concat([df1, df2, df3]))
-
-    def test_df_without_rows(self, lmdb_library, finalize_method):
-        lib = lmdb_library
-        df = pd.DataFrame({"col": []}, index=pd.DatetimeIndex([]))
-        lib.write("sym", df, staged=True)
-        lib.finalize_staged_data("sym", mode=finalize_method)
-        assert_frame_equal(lib.read("sym").data, df)
-
-@pytest.mark.parametrize("finalize_method", (StagedDataFinalizeMethod.APPEND, StagedDataFinalizeMethod.WRITE))
-def test_finalize_without_adding_segments(lmdb_library, finalize_method):
-    lib = lmdb_library
+def test_finalize_without_adding_segments(arctic_library, finalize_method):
+    lib = arctic_library
     with pytest.raises(UserInputException) as exception_info:
         lib.finalize_staged_data("sym", mode=finalize_method)
 
+
 class TestAppendStagedData:
-    def test_appended_df_interleaves_with_storage(self, lmdb_library):
-        lib = lmdb_library
+    def test_appended_df_interleaves_with_storage(self, arctic_library):
+        lib = arctic_library
         initial_df = pd.DataFrame({"col": [1, 3]}, index=pd.DatetimeIndex([np.datetime64('2023-01-01'), np.datetime64('2023-01-03')], dtype="datetime64[ns]"))
         lib.write("sym", initial_df)
         df1 = pd.DataFrame({"col": [2]}, index=pd.DatetimeIndex([np.datetime64('2023-01-02')], dtype="datetime64[ns]"))
@@ -336,8 +309,8 @@ def test_appended_df_interleaves_with_storage(self, lmdb_library):
             lib.finalize_staged_data("sym", mode=StagedDataFinalizeMethod.APPEND)
         assert "append" in str(exception_info.value)
 
-    def test_appended_df_start_same_as_df_end(self, lmdb_library):
-        lib = lmdb_library
+    def test_appended_df_start_same_as_df_end(self, arctic_library):
+        lib = arctic_library
         df = pd.DataFrame(
             {"col": [1, 2, 3]},
             index=pd.DatetimeIndex([np.datetime64('2023-01-01'), np.datetime64('2023-01-02'), np.datetime64('2023-01-03')], dtype="datetime64[ns]")
diff --git a/python/tests/nonreg/arcticdb/version_store/test_nonreg_sort_merge.py b/python/tests/nonreg/arcticdb/version_store/test_nonreg_sort_merge.py
new file mode 100644
index 0000000000..ebf3ba317a
--- /dev/null
+++ b/python/tests/nonreg/arcticdb/version_store/test_nonreg_sort_merge.py
@@ -0,0 +1,28 @@
+import pandas as pd
+import numpy as np
+
+# This was a added as a bug repro for GH issue #1795.
+def test_two_columns_with_different_dtypes(lmdb_library_dynamic_schema):
+    lib = lmdb_library_dynamic_schema
+
+    idx1 = pd.DatetimeIndex([
+        pd.Timestamp("2024-01-02")
+    ])
+    df1 = pd.DataFrame({
+         "a": np.array([1], dtype="float"),
+         "b": np.array([2], dtype="int64")
+    }, index=idx1)
+    
+    b = np.array([3, 4], dtype="int64")
+
+    idx = pd.DatetimeIndex([
+        pd.Timestamp("2024-01-03"),
+        pd.Timestamp("2024-01-01")
+    ])
+
+    df2 = pd.DataFrame({"b": b}, index=idx)
+    
+    lib.write("sym", df1, staged=True, validate_index=False)
+    lib.write("sym", df2, staged=True, validate_index=False)
+    lib.sort_and_finalize_staged_data("sym")
+    lib.read("sym")
\ No newline at end of file
diff --git a/python/tests/unit/arcticdb/version_store/test_parallel.py b/python/tests/unit/arcticdb/version_store/test_parallel.py
index c41b068a84..d7d82da0e8 100644
--- a/python/tests/unit/arcticdb/version_store/test_parallel.py
+++ b/python/tests/unit/arcticdb/version_store/test_parallel.py
@@ -134,16 +134,15 @@ def test_write_parallel_sort_merge(basic_arctic_library, prune_previous_versions
 
     num_rows_per_day = 10
     num_days = 10
-    num_columns = 8
+    num_columns = 4
     column_length = 8
     dt = datetime.datetime(2019, 4, 8, 0, 0, 0)
-    columns = random_strings_of_length(num_columns, column_length, True)
+    cols = random_strings_of_length(num_columns, column_length, True)
     symbol = "test_write_parallel_sort_merge"
     dataframes = []
     df = pd.DataFrame()
 
     for _ in range(num_days):
-        cols = random.sample(columns, 4)
         index = pd.Index([dt + datetime.timedelta(seconds=s) for s in range(num_rows_per_day)])
         vals = {c: random_floats(num_rows_per_day) for c in cols}
         new_df = pd.DataFrame(data=vals, index=index)
@@ -632,3 +631,116 @@ def test_parallel_append_dynamic_schema_missing_column(lmdb_version_store_tiny_s
     assert_frame_equal(expected, received)
 
 
+class TestFinalizeStagedDataStaticSchemaMismatch:
+    def test_append_throws_with_missmatched_column_set(self, lmdb_version_store_v1):
+        lib = lmdb_version_store_v1
+
+        initial_df = pd.DataFrame({"col_0": [1]}, index=pd.DatetimeIndex([pd.Timestamp(2024, 1, 1)]))
+        lib.write("sym", initial_df)
+
+        appended_df = pd.DataFrame({"col_1": [1]}, index=pd.DatetimeIndex([pd.Timestamp(2024, 1, 2)]))
+        lib.write("sym", appended_df, parallel=True)
+        with pytest.raises(SchemaException) as exception_info:
+            lib.compact_incomplete("sym", True, False)
+        assert "col_1" in str(exception_info.value)
+
+    def test_append_throws_column_subset(self, lmdb_version_store_v1):
+        lib = lmdb_version_store_v1
+
+        df1 = pd.DataFrame(
+            {"a": np.array([1.1], dtype="float"), "b": np.array([2], dtype="int64")},
+            index=pd.DatetimeIndex([pd.Timestamp("2024-01-01")])
+        )
+        lib.write("sym", df1)
+        df2 = pd.DataFrame({"b": [1]}, index=pd.DatetimeIndex([pd.Timestamp("2024-01-02")]))
+        lib.write("sym", df2, parallel=True)
+        with pytest.raises(SchemaException) as exception_info: 
+            lib.compact_incomplete("sym", True, False)
+        assert "a" in str(exception_info.value)
+        assert "b" in str(exception_info.value)
+
+    def test_append_throws_on_incompatible_dtype(self, lmdb_version_store_v1):
+        lib = lmdb_version_store_v1
+
+        initial_df = pd.DataFrame({"col_0": np.array([1], dtype="int64")}, index=pd.DatetimeIndex([pd.Timestamp(2024, 1, 1)]))
+        lib.write("sym", initial_df)
+
+        appended_df = pd.DataFrame({"col_0": ["asd"]}, index=pd.DatetimeIndex([pd.Timestamp(2024, 1, 2)]))
+        lib.write("sym", appended_df, parallel=True)
+        with pytest.raises(SchemaException) as exception_info:
+            lib.compact_incomplete("sym", True, False)
+        assert "col_0" in str(exception_info.value)
+        assert "INT64" in str(exception_info.value)
+
+    def test_type_mismatch_in_staged_segments_throws(self, lmdb_version_store_v1):
+        lib = lmdb_version_store_v1
+        lib.write("sym", pd.DataFrame({"col": [1]}, index=pd.DatetimeIndex([np.datetime64('2023-01-01')])), parallel=True)
+        lib.write("sym", pd.DataFrame({"col": ["a"]}, index=pd.DatetimeIndex([np.datetime64('2023-01-02')])), parallel=True)
+        with pytest.raises(Exception) as exception_info:
+            lib.compact_incomplete("sym", False, False)
+        assert all(x in str(exception_info.value) for x in ["INT64", "type"])
+
+    def test_types_cant_be_promoted(self, lmdb_version_store_v1):
+        lib = lmdb_version_store_v1
+        lib.write("sym", pd.DataFrame({"col": np.array([1], dtype="int64")}, index=pd.DatetimeIndex([pd.Timestamp(2024, 1, 1)])))
+
+        lib.write("sym", pd.DataFrame({"col": np.array([1], dtype="int32")}, index=pd.DatetimeIndex([pd.Timestamp(2024, 1, 2)])), parallel=True)
+        with pytest.raises(SchemaException) as exception_info:
+            lib.compact_incomplete("sym", True, False)
+        assert "INT32" in str(exception_info.value)
+        assert "INT64" in str(exception_info.value)
+
+    def test_appending_reordered_column_set_throws(self, lmdb_version_store_v1):
+        lib = lmdb_version_store_v1
+
+        lib.write("sym", pd.DataFrame({"col_0": [1], "col_1": ["test"], "col_2": [1.2]}, index=pd.DatetimeIndex([pd.Timestamp(2024, 1, 1)])))
+
+        lib.write("sym", pd.DataFrame({"col_1": ["asd"], "col_2": [2.5], "col_0": [2]}, index=pd.DatetimeIndex([pd.Timestamp(2024, 1, 2)])), parallel=True)
+        with pytest.raises(SchemaException) as exception_info:
+            lib.compact_incomplete("sym", True, False)
+        assert "col_0" in str(exception_info.value)
+        assert "col_1" in str(exception_info.value)
+        assert "col_2" in str(exception_info.value)
+
+    @pytest.mark.parametrize("mode", [True, False])
+    def test_staged_segments_can_be_reordered(self, lmdb_version_store_v1, mode):
+        lib = lmdb_version_store_v1
+        df1 = pd.DataFrame({"col_0": [1], "col_1": ["test"], "col_2": [1.2]}, index=pd.DatetimeIndex([pd.Timestamp(2024, 1, 1)]))
+        df2 = pd.DataFrame({"col_1": ["asd"], "col_2": [2.5], "col_0": [2]}, index=pd.DatetimeIndex([pd.Timestamp(2024, 1, 2)]))
+        lib.write("sym", df1, parallel=True)
+        lib.write("sym", df2, parallel=True)
+        expected = pd.concat([df1, df2]).sort_index()
+        with pytest.raises(SchemaException) as exception_info:
+            lib.compact_incomplete("sym", mode, False)
+        assert "col_0" in str(exception_info.value)
+        assert "col_1" in str(exception_info.value)
+        assert "col_2" in str(exception_info.value)
+
+# finalize_method True -> append, finalize_method False -> write
+@pytest.mark.parametrize("finalize_method", (True, False))
+class TestFinalizeWithEmptySegments:
+    def test_staged_segment_is_only_empty_dfs(self, lmdb_version_store_v1, finalize_method):
+        lib = lmdb_version_store_v1
+        lib.write("sym", pd.DataFrame([]), parallel=True)
+        lib.write("sym", pd.DataFrame([]), parallel=True)
+        lib.compact_incomplete("sym", finalize_method, False)
+        assert_frame_equal(lib.read("sym").data, pd.DataFrame([], index=pd.DatetimeIndex([])))
+
+    def test_staged_segment_has_empty_df(self, lmdb_version_store_v1, finalize_method):
+        lib = lmdb_version_store_v1
+        index = pd.DatetimeIndex([pd.Timestamp(2024, 1, 1), pd.Timestamp(2024, 1, 3), pd.Timestamp(2024, 1, 4)])
+        df1 = pd.DataFrame({"col": [1, 2, 3]}, index=index)
+        df2 = pd.DataFrame({})
+        df3 = pd.DataFrame({"col": [4]}, index=pd.DatetimeIndex([pd.Timestamp(2024, 1, 5)]))
+        lib.write("sym", df1, parallel=True)
+        lib.write("sym", df2, parallel=True)
+        lib.write("sym", df3, parallel=True)
+        with pytest.raises(SchemaException):
+            lib.compact_incomplete("sym", finalize_method, False)
+
+    def test_df_without_rows(self, lmdb_version_store_v1, finalize_method):
+        lib = lmdb_version_store_v1
+        df = pd.DataFrame({"col": []}, index=pd.DatetimeIndex([]))
+        lib.write("sym", df, parallel=True)
+        lib.compact_incomplete("sym", finalize_method, False)
+        assert_frame_equal(lib.read("sym").data, df)
\ No newline at end of file
diff --git a/python/tests/unit/arcticdb/version_store/test_sort_merge.py b/python/tests/unit/arcticdb/version_store/test_sort_merge.py
index 081a1e9803..3a9b5c0761 100644
--- a/python/tests/unit/arcticdb/version_store/test_sort_merge.py
+++ b/python/tests/unit/arcticdb/version_store/test_sort_merge.py
@@ -1,13 +1,14 @@
 import pandas as pd
 import numpy as np
-from pandas.testing import assert_frame_equal
+from arcticdb.util.test import assert_frame_equal
 import pytest
 from arcticdb_ext.storage import KeyType
 from arcticdb.version_store.library import StagedDataFinalizeMethod
-from arcticdb.exceptions import UserInputException, SortingException
+from arcticdb.exceptions import UserInputException, SortingException, StreamDescriptorMismatch, InternalException, SchemaException
+from arcticdb.util._versions import IS_PANDAS_TWO
 
-def test_merge_single_column(lmdb_library):
-    lib = lmdb_library
+def test_merge_single_column(lmdb_library_static_dynamic):
+    lib = lmdb_library_static_dynamic
 
     dates1 = [np.datetime64('2023-01-01'), np.datetime64('2023-01-03'), np.datetime64('2023-01-05')]
     dates2 = [np.datetime64('2023-01-02'), np.datetime64('2023-01-04'), np.datetime64('2023-01-06')]
@@ -36,8 +37,8 @@ def test_merge_single_column(lmdb_library):
     assert sort_and_finalize_res.version == 0
     assert lib.read(sym1).metadata == metadata
 
-def test_merge_two_column(lmdb_library):
-    lib = lmdb_library
+def test_merge_two_column(lmdb_library_static_dynamic):
+    lib = lmdb_library_static_dynamic
 
     dates1 = [np.datetime64('2023-01-01'), np.datetime64('2023-01-03'), np.datetime64('2023-01-05')]
     dates2 = [np.datetime64('2023-01-02'), np.datetime64('2023-01-04'), np.datetime64('2023-01-06')]
@@ -61,8 +62,8 @@ def test_merge_two_column(lmdb_library):
     assert_frame_equal(lib.read(sym1).data, expected_df)
 
 
-def test_merge_dynamic(lmdb_library):
-    lib = lmdb_library
+def test_merge_dynamic(lmdb_library_dynamic_schema):
+    lib = lmdb_library_dynamic_schema
 
     dates1 = [np.datetime64('2023-01-01'), np.datetime64('2023-01-03'), np.datetime64('2023-01-05')]
     dates2 = [np.datetime64('2023-01-02'), np.datetime64('2023-01-04'), np.datetime64('2023-01-06')]
@@ -87,8 +88,8 @@ def test_merge_dynamic(lmdb_library):
 
 
 
-def test_merge_strings(lmdb_library):
-    lib = lmdb_library
+def test_merge_strings(lmdb_library_static_dynamic):
+    lib = lmdb_library_static_dynamic
 
     dates1 = [np.datetime64('2023-01-01'), np.datetime64('2023-01-03'), np.datetime64('2023-01-05')]
     dates2 = [np.datetime64('2023-01-02'), np.datetime64('2023-01-04'), np.datetime64('2023-01-06')]
@@ -112,8 +113,8 @@ def test_merge_strings(lmdb_library):
     assert_frame_equal(lib.read(sym1).data, expected_df)
 
 
-def test_merge_strings_dynamic(lmdb_library):
-    lib = lmdb_library
+def test_merge_strings_dynamic(lmdb_library_dynamic_schema):
+    lib = lmdb_library_dynamic_schema
 
     dates1 = [np.datetime64('2023-01-01'), np.datetime64('2023-01-03'), np.datetime64('2023-01-05')]
     dates2 = [np.datetime64('2023-01-02'), np.datetime64('2023-01-04'), np.datetime64('2023-01-06')]
@@ -136,35 +137,34 @@ def test_merge_strings_dynamic(lmdb_library):
     expected_df = pd.DataFrame(expected_values, index=expected_dates)
     assert_frame_equal(lib.read(sym1).data, expected_df)
 
-
-@pytest.mark.xfail(reason="Unsorted segments are not implemented")
-def test_unordered_segment(lmdb_library):
-    lib = lmdb_library
+def test_unordered_segment(lmdb_library_static_dynamic):
+    lib = lmdb_library_static_dynamic
     dates = [np.datetime64('2023-01-03'), np.datetime64('2023-01-01'), np.datetime64('2023-01-05')]
     df = pd.DataFrame({"col": [2, 1, 3]}, index=dates)
-    lib.write("sym", df, staged=True)
+    lib.write("sym", df, staged=True, validate_index=False)
     lib.sort_and_finalize_staged_data("sym")
-    assert_frame_equal(lib.read('sym'), pd.DataFrame({"col": [1, 2, 3]}, index=[np.datetime64('2023-01-01'), np.datetime64('2023-01-03'), np.datetime64('2023-01-05')]))
+    assert_frame_equal(lib.read('sym').data, pd.DataFrame({"col": [1, 2, 3]}, index=[np.datetime64('2023-01-01'), np.datetime64('2023-01-03'), np.datetime64('2023-01-05')]))
 
-def test_repeating_index_values(lmdb_library):
-    lib = lmdb_library
+def test_repeating_index_values(lmdb_library_static_dynamic):
+    lib = lmdb_library_static_dynamic
     dates = [np.datetime64('2023-01-01'), np.datetime64('2023-01-03'), np.datetime64('2023-01-05')]
     df1 = pd.DataFrame({"col": [1,2,3]}, index=dates)
     df2 = pd.DataFrame({"col": [4,5,6]}, index=dates)
     lib.write("sym", df1, staged=True)
     lib.write("sym", df2, staged=True)
     lib.sort_and_finalize_staged_data("sym")
-    expected_index = pd.DatetimeIndex([np.datetime64('2023-01-01'), np.datetime64('2023-01-01'), np.datetime64('2023-01-03'),
-                                       np.datetime64('2023-01-03'), np.datetime64('2023-01-05'), np.datetime64('2023-01-05')],
-                                      dtype="datetime64[ns]")
     data = lib.read("sym").data
-    assert data.index.equals(expected_index)
-    for i in range(0, len(data["col"])):
-        assert data["col"][i] == df1["col"][i // 2] or data["col"][i] == df2["col"][i // 2]
+    expected = pd.concat([df1, df2]).sort_index()
+    assert data.index.equals(expected.index)
+    for i in range(0, len(df1)):
+        row = 2 * i
+        assert data["col"][row] == df1["col"][i] or data["col"][row] == df2["col"][i]
+        assert (data["col"][row + 1] == df1["col"][i] or data["col"][row + 1] == df2["col"][i]) and data["col"][row] != data["col"][row + 1]
+
 
 class TestMergeSortAppend:
-    def test_appended_values_are_after(self, lmdb_library):
-        lib = lmdb_library
+    def test_appended_values_are_after(self, lmdb_library_static_dynamic):
+        lib = lmdb_library_static_dynamic
         initial_df = pd.DataFrame(
             {"col": [1, 2, 3]},
             index=pd.DatetimeIndex([np.datetime64('2023-01-01'), np.datetime64('2023-01-02'), np.datetime64('2023-01-03')], dtype="datetime64[ns]")
@@ -184,8 +184,8 @@ def test_appended_values_are_after(self, lmdb_library):
         expected_df = pd.DataFrame({"col": range(1, 9)}, index=expected_index)
         assert_frame_equal(lib.read("sym").data, expected_df)
 
-    def test_appended_df_interleaves_with_storage(self, lmdb_library):
-        lib = lmdb_library
+    def test_appended_df_interleaves_with_storage(self, lmdb_library_static_dynamic):
+        lib = lmdb_library_static_dynamic
         initial_df = pd.DataFrame({"col": [1, 3]}, index=pd.DatetimeIndex([np.datetime64('2023-01-01'), np.datetime64('2023-01-03')], dtype="datetime64[ns]"))
         lib.write("sym", initial_df)
         df1 = pd.DataFrame({"col": [2]}, index=pd.DatetimeIndex([np.datetime64('2023-01-02')], dtype="datetime64[ns]"))
@@ -194,8 +194,8 @@ def test_appended_df_interleaves_with_storage(self, lmdb_library):
             lib.sort_and_finalize_staged_data("sym", mode=StagedDataFinalizeMethod.APPEND)
         assert "append" in str(exception_info.value)
 
-    def test_appended_df_start_same_as_df_end(self, lmdb_library):
-        lib = lmdb_library
+    def test_appended_df_start_same_as_df_end(self, lmdb_library_static_dynamic):
+        lib = lmdb_library_static_dynamic
         df = pd.DataFrame(
             {"col": [1, 2, 3]},
             index=pd.DatetimeIndex([np.datetime64('2023-01-01'), np.datetime64('2023-01-02'), np.datetime64('2023-01-03')], dtype="datetime64[ns]")
@@ -212,8 +212,8 @@ def test_appended_df_start_same_as_df_end(self, lmdb_library):
         assert_frame_equal(lib.read("sym").data, expected_df)
 
 
-def test_prune_previous(lmdb_library):
-    lib = lmdb_library
+def test_prune_previous(lmdb_library_static_dynamic):
+    lib = lmdb_library_static_dynamic
     idx = pd.DatetimeIndex([np.datetime64('2023-01-01'), np.datetime64('2023-01-03')], dtype="datetime64[ns]")
     df = pd.DataFrame({"col": [1, 3]}, index=idx)
     lib.write("sym", df)
@@ -223,16 +223,17 @@ def test_prune_previous(lmdb_library):
     assert_frame_equal(df, lib.read("sym").data)
     assert len(lib.list_versions("sym")) == 1
 
+@pytest.mark.parametrize("mode", [StagedDataFinalizeMethod.APPEND, StagedDataFinalizeMethod.WRITE])
 class TestEmptySegments:
-    def test_staged_segment_is_only_empty_dfs(self, lmdb_library):
-        lib = lmdb_library
+    def test_staged_segment_is_only_empty_dfs(self, lmdb_library_static_dynamic, mode):
+        lib = lmdb_library_static_dynamic
         lib.write("sym", pd.DataFrame([]), staged=True)
         lib.write("sym", pd.DataFrame([]), staged=True)
-        lib.sort_and_finalize_staged_data("sym")
+        lib.sort_and_finalize_staged_data("sym", mode=mode)
         assert_frame_equal(lib.read("sym").data, pd.DataFrame([], index=pd.DatetimeIndex([])))
 
-    def test_staged_segment_has_empty_df(self, lmdb_library):
-        lib = lmdb_library
+    def test_staged_segment_has_empty_df(self, lmdb_library_dynamic_schema, mode):
+        lib = lmdb_library_dynamic_schema
         index = pd.DatetimeIndex([pd.Timestamp(2024, 1, 1), pd.Timestamp(2024, 1, 3), pd.Timestamp(2024, 1, 4)])
         df1 = pd.DataFrame({"col": [1, 2, 3]}, index=index)
         df2 = pd.DataFrame({})
@@ -240,29 +241,35 @@ def test_staged_segment_has_empty_df(self, lmdb_library):
         lib.write("sym", df1, staged=True)
         lib.write("sym", df2, staged=True)
         lib.write("sym", df3, staged=True)
-        lib.sort_and_finalize_staged_data("sym")
+        lib.sort_and_finalize_staged_data("sym", mode=mode)
         assert_frame_equal(lib.read("sym").data, pd.concat([df1, df2, df3]).sort_index())
 
-    def test_df_without_rows(self, lmdb_library):
-        lib = lmdb_library
+    def test_df_without_rows(self, lmdb_library_static_dynamic, mode):
+        lib = lmdb_library_static_dynamic
         df = pd.DataFrame({"col": []}, index=pd.DatetimeIndex([]))
         lib.write("sym", df, staged=True)
-        lib.sort_and_finalize_staged_data("sym")
+        lib.sort_and_finalize_staged_data("sym", mode=mode)
         assert_frame_equal(lib.read("sym").data, df)
 
+    def test_finalize_without_adding_segments(self, lmdb_library_static_dynamic, mode):
+        lib = lmdb_library_static_dynamic
+        with pytest.raises(UserInputException) as exception_info:
+            lib.sort_and_finalize_staged_data("sym", mode=mode)
+        assert "E_NO_STAGED_SEGMENTS" in str(exception_info.value)
 
-def test_finalize_without_adding_segments(lmdb_library):
-    lib = lmdb_library
-    with pytest.raises(UserInputException) as exception_info:
-        lib.sort_and_finalize_staged_data("sym")
+    def test_mixing_empty_and_non_empty_columns(self, lmdb_library_dynamic_schema, mode):
+        lib = lmdb_library_dynamic_schema
 
-def test_type_mismatch_throws(lmdb_library):
-    lib = lmdb_library
-    lib.write("sym", pd.DataFrame({"col": [1]}, index=pd.DatetimeIndex([np.datetime64('2023-01-01')])), staged=True)
-    lib.write("sym", pd.DataFrame({"col": ["a"]}, index=pd.DatetimeIndex([np.datetime64('2023-01-02')])), staged=True)
-    with pytest.raises(Exception) as exception_info:
-        lib.sort_and_finalize_staged_data("sym")
-    assert all(x in str(exception_info.value) for x in ["INT64", "type"])
+        df = pd.DataFrame({"a": [1]},index=pd.DatetimeIndex([pd.Timestamp("1970-01-01")]))
+        df2 = pd.DataFrame({"b": np.array([], dtype="float"), "c": np.array([], dtype="int64"), "d": np.array([], dtype="object")},index=pd.DatetimeIndex([]))
+        lib.write("sym", df, staged=True)
+        lib.write("sym", df2, staged=True)
+        lib.sort_and_finalize_staged_data("sym", mode=mode)
+        if IS_PANDAS_TWO:
+            expected = pd.DataFrame({"a": [1], "b": np.array([np.nan], dtype="float"), "c": np.array([0], dtype="int64"), "d": np.array([None], dtype="object")}, index=[pd.Timestamp("1970-01-01")])
+        else:
+            expected = pd.DataFrame({"a": [1], "b": np.array([np.nan], dtype="object"), "c": np.array([0], dtype="int64"), "d": np.array([None], dtype="object")}, index=[pd.Timestamp("1970-01-01")])
+        assert_frame_equal(lib.read("sym").data, expected)
 
 def test_append_to_missing_symbol(lmdb_library):
     lib = lmdb_library
@@ -271,6 +278,310 @@ def test_append_to_missing_symbol(lmdb_library):
     lib.sort_and_finalize_staged_data("sym", mode=StagedDataFinalizeMethod.APPEND)
     assert_frame_equal(lib.read("sym").data, df)
 
+def test_pre_epoch(lmdb_library):
+    lib = lmdb_library
+
+    df = pd.DataFrame({"col": [1]}, pd.DatetimeIndex([pd.Timestamp(1969, 12, 31)]))
+    lib.write("sym", df, staged=True)
+    lib.sort_and_finalize_staged_data("sym")
+
+    assert_frame_equal(lib.read("sym").data, df)
+
+
+class TestDescriptorMismatchBetweenStagedSegments:
+    def test_append_throws_with_missmatched_column_set(self, lmdb_library):
+        lib = lmdb_library
+
+        initial_df = pd.DataFrame({"col_0": [1]}, index=pd.DatetimeIndex([pd.Timestamp(2024, 1, 1)]))
+        lib.write("sym", initial_df, staged=True)
+
+        appended_df = pd.DataFrame({"col_1": [1]}, index=pd.DatetimeIndex([pd.Timestamp(2024, 1, 2)]))
+        lib.write("sym", appended_df, staged=True)
+
+        with pytest.raises(SchemaException) as exception_info:
+            lib.sort_and_finalize_staged_data("sym", mode=StagedDataFinalizeMethod.APPEND)
+        assert "col_1" in str(exception_info.value)
+
+    def test_append_throws_column_subset(self, lmdb_library):
+        lib = lmdb_library
+
+        df1 = pd.DataFrame(
+            {
+                "col_0": np.array([1.1], dtype="float"),
+                "col_1": np.array([2], dtype="int64")
+            },
+            index=pd.DatetimeIndex([pd.Timestamp("2024-01-01")])
+        )
+        lib.write("sym", df1, staged=True)
+
+        df2 = pd.DataFrame({"col_1": [1]}, index=pd.DatetimeIndex([pd.Timestamp("2024-01-02")]))
+        lib.write("sym", df2, staged=True)
+
+        with pytest.raises(SchemaException) as exception_info: 
+            lib.sort_and_finalize_staged_data("sym", StagedDataFinalizeMethod.APPEND)
+        assert "col_0" in str(exception_info.value)
+        assert "col_1" in str(exception_info.value)
+
+    def test_appending_reordered_column_set_throws(self, lmdb_library):
+        lib = lmdb_library
+
+        df1 = pd.DataFrame({"col_0": [1], "col_1": ["test"], "col_2": [1.2]}, index=pd.DatetimeIndex([pd.Timestamp(2024, 1, 1)]))
+        lib.write("sym", df1, staged=True)
+
+        df2 = pd.DataFrame({"col_1": ["asd"], "col_2": [2.5], "col_0": [2]}, index=pd.DatetimeIndex([pd.Timestamp(2024, 1, 2)]))
+        lib.write("sym", df2, staged=True)
+
+        with pytest.raises(SchemaException) as exception_info:
+            lib.sort_and_finalize_staged_data("sym", mode=StagedDataFinalizeMethod.APPEND)
+        assert "col_0" in str(exception_info.value)
+        assert "col_1" in str(exception_info.value)
+        assert "col_2" in str(exception_info.value)
+
+    def test_append_throws_on_incompatible_dtype(self, lmdb_library):
+        lib = lmdb_library
+
+        df1 = pd.DataFrame({"col_0": np.array([1], dtype="int64")}, index=pd.DatetimeIndex([pd.Timestamp(2024, 1, 1)]))
+        lib.write("sym", df1, staged=True)
+
+        df2 = pd.DataFrame({"col_0": ["asd"]}, index=pd.DatetimeIndex([pd.Timestamp(2024, 1, 2)]))
+        lib.write("sym", df2, staged=True)
+
+        with pytest.raises(SchemaException) as exception_info:
+            lib.sort_and_finalize_staged_data("sym", mode=StagedDataFinalizeMethod.APPEND)
+        assert "col_0" in str(exception_info.value)
+        assert "INT64" in str(exception_info.value)
+
+    @pytest.mark.parametrize("mode", [StagedDataFinalizeMethod.APPEND, StagedDataFinalizeMethod.WRITE])
+    def test_types_cant_be_promoted(self, lmdb_library, mode):
+        lib = lmdb_library
+
+        df1 = pd.DataFrame({"col_0": np.array([1], dtype="float")}, index=pd.DatetimeIndex([np.datetime64('2023-01-01')]))
+        lib.write("sym", df1, staged=True)
+        
+        df2 = pd.DataFrame({"col_0": np.array([1], dtype="int")}, index=pd.DatetimeIndex([np.datetime64('2023-01-02')]))
+        lib.write("sym", df2, staged=True)
+
+        with pytest.raises(SchemaException) as exception_info:
+            lib.sort_and_finalize_staged_data("sym", mode=mode)
+        assert "col_0" in str(exception_info.value)
+        assert "FLOAT" in str(exception_info.value)
+        assert "INT" in str(exception_info.value)
+
+    @pytest.mark.parametrize("mode", [StagedDataFinalizeMethod.APPEND, StagedDataFinalizeMethod.WRITE])
+    def test_type_mismatch_in_staged_segments_throws_with_non_promotoable_types(self, lmdb_library, mode):
+        lib = lmdb_library
+
+        df1 = pd.DataFrame({"col": np.array([1], dtype="int64")}, index=pd.DatetimeIndex([np.datetime64('2023-01-01')]))
+        lib.write("sym", df1, staged=True)
+
+        df2 = pd.DataFrame({"col": ["test"]}, index=pd.DatetimeIndex([np.datetime64('2023-01-02')]))
+        lib.write("sym", df2, staged=True)
+
+        with pytest.raises(SchemaException) as exception_info:
+            lib.sort_and_finalize_staged_data("sym", mode=mode)
+        assert "INT" in str(exception_info.value)
+        assert "UTF_DYNAMIC" in str(exception_info.value)
+
+    @pytest.mark.parametrize("mode", [StagedDataFinalizeMethod.APPEND, StagedDataFinalizeMethod.WRITE])
+    def test_staged_segments_cant_be_reordered(self, lmdb_library, mode):
+        lib = lmdb_library
+
+        df1 = pd.DataFrame({"col_0": [1], "col_1": ["test"], "col_2": [1.2]}, index=pd.DatetimeIndex([pd.Timestamp(2024, 1, 1)]))
+        lib.write("sym", df1, staged=True)
+        
+        df2 = pd.DataFrame({"col_1": ["asd"], "col_2": [2.5], "col_0": [2]}, index=pd.DatetimeIndex([pd.Timestamp(2024, 1, 2)]))
+        lib.write("sym", df2, staged=True)
+
+        with pytest.raises(SchemaException) as exception_info:
+            lib.sort_and_finalize_staged_data("sym", mode=mode)
+        assert "col_0" in str(exception_info.value)
+        assert "col_1" in str(exception_info.value)
+        assert "col_2" in str(exception_info.value)
+
+
+class TestStreamDescriptorMismatchOnFinalizeAppend:
+    def init_symbol(self, lib, sym):
+        df = pd.DataFrame(
+            {"col_0": np.array([1], dtype="int32"), "col_1": np.array([0.5], dtype="float64"), "col_2": np.array(["val"], dtype="object")},
+            index=pd.DatetimeIndex([pd.Timestamp(2024, 1, 1)])
+        )
+        lib.write(sym, df)
+
+    def test_cannot_append_column_subset(self, lmdb_library):
+        lib = lmdb_library
+        self.init_symbol(lib, "sym")
+        
+        df = pd.DataFrame({"col_0": np.array([1], dtype="int32")}, index=pd.DatetimeIndex([pd.Timestamp(2024, 1, 1)]))
+        lib.write("sym", df, staged=True)
+        with pytest.raises(SchemaException) as exception_info:
+            lib.sort_and_finalize_staged_data("sym", mode=StagedDataFinalizeMethod.APPEND)
+        assert "col_1" in str(exception_info.value)
+        assert "col_2" in str(exception_info.value)
+
+    def test_cannot_append_reordered_columns(self, lmdb_library):
+        lib = lmdb_library
+        self.init_symbol(lib, "sym")
+        
+        df = pd.DataFrame(
+            {"col_1": np.array([1.4], dtype="float64"), "col_0": np.array([5], dtype="int32"), "col_2": np.array(["val"], dtype="object")},
+            index=pd.DatetimeIndex([pd.Timestamp(2024, 1, 1)])
+        )
+        lib.write("sym", df, staged=True)
+        with pytest.raises(SchemaException) as exception_info:
+            lib.sort_and_finalize_staged_data("sym", mode=StagedDataFinalizeMethod.APPEND)
+        assert "col_0" in str(exception_info.value)
+        assert "col_1" in str(exception_info.value)
+        assert "col_2" in str(exception_info.value)
+
+    def test_cannot_promote_stored_type(self, lmdb_library):
+        lib = lmdb_library
+        self.init_symbol(lib, "sym")
+        
+        df = pd.DataFrame(
+            {"col_0": np.array([1], dtype="int64"), "col_1": np.array([5], dtype="float64"), "col_2": np.array(["val"], dtype="object")},
+            index=pd.DatetimeIndex([pd.Timestamp(2024, 1, 1)])
+        )
+        lib.write("sym", df, staged=True)
+        with pytest.raises(SchemaException) as exception_info:
+            lib.sort_and_finalize_staged_data("sym", mode=StagedDataFinalizeMethod.APPEND)
+        assert "col_0" in str(exception_info.value)
+        assert "INT32" in str(exception_info.value)
+        assert "INT64" in str(exception_info.value)
+ 
+    def test_cannot_promote_input_type(self, lmdb_library):
+        lib = lmdb_library
+        self.init_symbol(lib, "sym")
+        
+        df = pd.DataFrame(
+            {"col_0": np.array([1], dtype="int16"), "col_1": np.array([5], dtype="float64"), "col_2": np.array(["val"], dtype="object")},
+            index=pd.DatetimeIndex([pd.Timestamp(2024, 1, 1)])
+        )
+        lib.write("sym", df, staged=True)
+        with pytest.raises(SchemaException) as exception_info:
+            lib.sort_and_finalize_staged_data("sym", mode=StagedDataFinalizeMethod.APPEND)
+        assert "col_0" in str(exception_info.value)
+        assert "INT32" in str(exception_info.value)
+        assert "INT16" in str(exception_info.value)
+
+    def test_cannot_add_new_columns(self, lmdb_library):
+        lib = lmdb_library
+        self.init_symbol(lib, "sym")
+
+        df = pd.DataFrame(
+            {
+                "col_0": np.array([1], dtype="int32"),
+                "col_1": np.array([5], dtype="float64"),
+                "col_2": np.array(["val"], dtype="object"),
+                "col_3": np.array([1], dtype="int32")
+            },
+            index=pd.DatetimeIndex([pd.Timestamp(2024, 1, 1)])
+        )
+        lib.write("sym", df, staged=True)
+        with pytest.raises(SchemaException) as exception_info:
+            lib.sort_and_finalize_staged_data("sym", mode=StagedDataFinalizeMethod.APPEND)
+        assert "col_3" in str(exception_info.value)
+
+@pytest.mark.parametrize("mode", [StagedDataFinalizeMethod.APPEND, StagedDataFinalizeMethod.WRITE])
+class TestNatInIndexNotAllowed:
+
+    @classmethod
+    def assert_nat_not_allowed(cls, lib, symbol, mode):
+        with pytest.raises(SortingException) as exception_info:
+            lib.sort_and_finalize_staged_data(symbol, mode=mode)
+        assert "NaT" in str(exception_info.value)
+
+    def test_index_only_nat(self, lmdb_library_static_dynamic, mode):
+        lib = lmdb_library_static_dynamic
+
+        df1 = pd.DataFrame({"a": [1, 2]}, index=pd.DatetimeIndex([pd.NaT, pd.NaT]))
+        lib.write("sym", df1, staged=True, validate_index=False)
+        self.assert_nat_not_allowed(lib, "sym", mode)
+
+    def test_nat_and_valid_date(self, lmdb_library_static_dynamic, mode):
+        lib = lmdb_library_static_dynamic
+
+        df1 = pd.DataFrame({"a": [1, 2]}, index=pd.DatetimeIndex([pd.NaT, pd.Timestamp(2024, 1, 1)]))
+        lib.write("sym", df1, staged=True, validate_index=False)
+        self.assert_nat_not_allowed(lib, "sym", mode)
+        lib.delete_staged_data("sym")
+
+        df1 = pd.DataFrame({"a": [1, 2]}, index=pd.DatetimeIndex([pd.Timestamp(2024, 1, 1), pd.NaT]))
+        lib.write("sym", df1, staged=True, validate_index=False)
+        self.assert_nat_not_allowed(lib, "sym", mode)
+
+class TestSortMergeDynamicSchema:
+
+    def test_appended_columns_are_subset(self, lmdb_library_dynamic_schema):
+        lib = lmdb_library_dynamic_schema
+
+        lib.write("sym", pd.DataFrame({"a": [1], "b": [1.2]}, index=pd.DatetimeIndex([pd.Timestamp(2024, 1, 1)])))
+
+        lib.write("sym", pd.DataFrame({"a": [2]}, index=pd.DatetimeIndex([pd.Timestamp(2024, 1, 2)])), staged=True)
+        lib.write("sym", pd.DataFrame({"b": [5.3]}, index=pd.DatetimeIndex([pd.Timestamp(2024, 1, 3)])), staged=True)
+        lib.sort_and_finalize_staged_data("sym", mode=StagedDataFinalizeMethod.APPEND)
+
+        expected = pd.DataFrame({"a": [1, 2, 0], "b": [1.2, np.nan, 5.3]}, index=pd.date_range("2024-01-01", "2024-01-03"))
+        stored = lib.read("sym").data
+
+        assert_frame_equal(expected, stored)
+
+    def test_can_append_new_columns(self, lmdb_library_dynamic_schema):
+        lib = lmdb_library_dynamic_schema
+
+        lib.write("sym", pd.DataFrame({"a": [1]}, index=pd.DatetimeIndex([pd.Timestamp(2024, 1, 1)])))
+
+        lib.write("sym", pd.DataFrame({"b": [1.5]}, index=pd.DatetimeIndex([pd.Timestamp(2024, 1, 2)])), staged=True)
+        lib.write("sym", pd.DataFrame({"c": ["c"]}, index=pd.DatetimeIndex([pd.Timestamp(2024, 1, 3)])), staged=True)
+        lib.sort_and_finalize_staged_data("sym", mode=StagedDataFinalizeMethod.APPEND)
+
+        expected = pd.DataFrame({"a": [1, 0, 0], "b": [np.nan, 1.5, np.nan], "c": [None, None, "c"]}, index=pd.date_range("2024-01-01", "2024-01-03"))
+        stored = lib.read("sym").data
+
+        assert_frame_equal(expected, stored, check_like=True)
+
+    def test_staged_segments_are_promoted(self, lmdb_library_dynamic_schema):
+        lib = lmdb_library_dynamic_schema
+        df1 =pd.DataFrame(
+            {"col_0": np.array([1], dtype="int16"), "col_1": np.array([2], dtype="int64"), "col_3": np.array([3], dtype="int32")},
+            index=pd.DatetimeIndex([pd.Timestamp(2024, 1, 1)])
+        )
+        lib.write("sym", df1, staged=True)
+
+        df2 = pd.DataFrame(
+            {"col_0": np.array([10], dtype="int32"), "col_1": np.array([20], dtype="int16"), "col_3": np.array([30], dtype="float32")},
+            index=pd.DatetimeIndex([pd.Timestamp(2024, 1, 2)])
+        )
+        lib.write("sym", df2, staged=True)
+
+        lib.sort_and_finalize_staged_data("sym")
+
+        expected = pd.DataFrame(
+            {"col_0": np.array([1, 10], dtype="int32"), "col_1": np.array([2, 20], dtype="int64"), "col_3": np.array([3, 30], dtype="float32")},
+            index=pd.DatetimeIndex([pd.Timestamp(2024, 1, 1), pd.Timestamp(2024, 1, 2)])
+        )
+        assert_frame_equal(lib.read("sym").data, expected, check_dtype=True)
+
+    def test_finalize_append_promotes_types(self, lmdb_library_dynamic_schema):
+        lib = lmdb_library_dynamic_schema
+        df1 =pd.DataFrame(
+            {"col_0": np.array([1], dtype="int16"), "col_1": np.array([2], dtype="int64"), "col_3": np.array([3], dtype="int32")},
+            index=pd.DatetimeIndex([pd.Timestamp(2024, 1, 1)])
+        )
+        lib.write("sym", df1)
+
+        df2 = pd.DataFrame(
+            {"col_0": np.array([10], dtype="int32"), "col_1": np.array([20], dtype="int16"), "col_3": np.array([30], dtype="float32")},
+            index=pd.DatetimeIndex([pd.Timestamp(2024, 1, 2)])
+        )
+        lib.write("sym", df2, staged=True)
+
+        lib.sort_and_finalize_staged_data("sym", mode=StagedDataFinalizeMethod.APPEND)
+
+        expected = pd.DataFrame(
+            {"col_0": np.array([1, 10], dtype="int32"), "col_1": np.array([2, 20], dtype="int64"), "col_3": np.array([3, 30], dtype="float32")},
+            index=pd.DatetimeIndex([pd.Timestamp(2024, 1, 1), pd.Timestamp(2024, 1, 2)])
+        )
+        assert_frame_equal(lib.read("sym").data, expected, check_dtype=True)
 
 def test_update_symbol_list(lmdb_library):
     lib = lmdb_library
@@ -301,5 +612,4 @@ def test_update_symbol_list(lmdb_library):
     lib.write(sym_2, df, staged=True)
     lib.sort_and_finalize_staged_data(sym_2, mode=StagedDataFinalizeMethod.APPEND)
     assert lib_tool.count_keys(KeyType.SYMBOL_LIST) == 3
-    assert set(lib.list_symbols()) == set([sym, sym_2])
-
+    assert set(lib.list_symbols()) == set([sym, sym_2])
\ No newline at end of file