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

Issue:
`compact_incomplete` should work or raise useful error messages in all scenarios with dynamic schema
- Be restrictive with static schema to all columns being present and types not changeing
- Types changing, both between the existing symbol and the incompletes, and between the incompletes themselves
- Columns missing from all incomplete segments
- Columns missing from some incomplete segments
- Respect the lib config segment row size
Note there is a test called `CompactIncompleteDynamicSchema` in `test_version_store.cpp` that passes, and an `xfail` test in `test_parallel.py` that does not.


Discussion / hints:

Fix patch:
diff --git a/cpp/arcticdb/column_store/column.cpp b/cpp/arcticdb/column_store/column.cpp
index 3fe55c4c1f..40e35609ee 100644
--- a/cpp/arcticdb/column_store/column.cpp
+++ b/cpp/arcticdb/column_store/column.cpp
@@ -366,23 +366,25 @@ void Column::mark_absent_rows(size_t num_rows) {
 }
 
 void Column::default_initialize_rows(size_t start_pos, size_t num_rows, bool ensure_alloc) {
-    type_.visit_tag([that=this, start_pos, num_rows, ensure_alloc](auto tag) {
-        using T= std::decay_t<decltype(tag)>;
-        using RawType = typename T::DataTypeTag::raw_type;
-        const auto bytes = (num_rows * sizeof(RawType));
+    if (num_rows > 0) {
+        type_.visit_tag([this, start_pos, num_rows, ensure_alloc](auto tag) {
+            using T = std::decay_t<decltype(tag)>;
+            using RawType = typename T::DataTypeTag::raw_type;
+            const auto bytes = (num_rows * sizeof(RawType));
 
-        if(ensure_alloc)
-            that->data_.ensure<uint8_t>(bytes);
+            if (ensure_alloc)
+                data_.ensure<uint8_t>(bytes);
 
-        auto type_ptr = that->data_.ptr_cast<RawType>(start_pos, bytes);
-        util::default_initialize<T>(reinterpret_cast<uint8_t*>(type_ptr), bytes);
+            auto type_ptr = data_.ptr_cast<RawType>(start_pos, bytes);
+            util::default_initialize<T>(reinterpret_cast<uint8_t *>(type_ptr), bytes);
 
-        if(ensure_alloc)
-            that->data_.commit();
+            if (ensure_alloc)
+                data_.commit();
 
-        that->last_logical_row_ += static_cast<ssize_t>(num_rows);
-        that->last_physical_row_ += static_cast<ssize_t>(num_rows);
-    });
+            last_logical_row_ += static_cast<ssize_t>(num_rows);
+            last_physical_row_ += static_cast<ssize_t>(num_rows);
+        });
+    }
 }
 
 void Column::set_row_data(size_t row_id) {
@@ -538,6 +540,7 @@ void Column::change_type(DataType target_type) {
             });
         });
     }
+    buf.commit();
     type_ = TypeDescriptor{target_type, type_.dimension()};
     std::swap(data_, buf);
 }
diff --git a/cpp/arcticdb/column_store/memory_segment_impl.cpp b/cpp/arcticdb/column_store/memory_segment_impl.cpp
index 8f8acf55df..2e87bb92d0 100644
--- a/cpp/arcticdb/column_store/memory_segment_impl.cpp
+++ b/cpp/arcticdb/column_store/memory_segment_impl.cpp
@@ -59,6 +59,7 @@ void SegmentInMemoryImpl::append(const SegmentInMemoryImpl& other) {
 
             if (this_type != *opt_common_type) {
                 column_unchecked(col).change_type(opt_common_type->data_type_);
+                descriptor_->mutable_field(col).mutable_type().data_type_ = opt_common_type->data_type_;
             }
             if (other_type != *opt_common_type) {
                 auto type_promoted_col = other.column_unchecked(*other_col_index).clone();
@@ -543,8 +544,10 @@ void SegmentInMemoryImpl::change_schema(StreamDescriptor descriptor) {
         const auto& other_type = descriptor.field(col).type();
         if(col_index) {
             auto this_type = column_unchecked(*col_index).type();
-            util::check(this_type == other_type, "Could not convert type {} to type {} for column {}, this index {}, other index {}",
-                        other_type, this_type, col_name, *col_index, col);
+            schema::check<ErrorCode::E_DESCRIPTOR_MISMATCH>(
+                    this_type == other_type,
+                    "Could not convert type {} to type {} for column {}, this index {}, other index {}",
+                    other_type, this_type, col_name, *col_index, col);
             new_columns[col] = std::move(columns_[*col_index]);
         } else {
             auto new_column = std::make_shared<Column>(other_type, row_count(), false, allow_sparse_);
diff --git a/cpp/arcticdb/stream/merge_utils.hpp b/cpp/arcticdb/stream/merge_utils.hpp
index d56ed43bde..0e5c3ce338 100644
--- a/cpp/arcticdb/stream/merge_utils.hpp
+++ b/cpp/arcticdb/stream/merge_utils.hpp
@@ -69,10 +69,8 @@ inline void merge_segments(
     timestamp min_idx = std::numeric_limits<timestamp>::max();
     timestamp max_idx = std::numeric_limits<timestamp>::min();
     for (auto &segment : segments) {
-        std::vector<SegmentInMemory> history{{segment}};
-        const auto& latest = *history.rbegin();
-        ARCTICDB_DEBUG(log::version(), "Appending segment with {} rows", latest.row_count());
-        for(const auto& field : latest.descriptor().fields()) {
+        ARCTICDB_DEBUG(log::version(), "Appending segment with {} rows", segment.row_count());
+        for(const auto& field : segment.descriptor().fields()) {
             if(!merged.column_index(field.name())){//TODO: Bottleneck for wide segments
                 auto pos = merged.add_column(field, 0, false);
                 if (!is_sparse){
@@ -80,12 +78,12 @@ inline void merge_segments(
                 }
             }
         }
-        if (latest.row_count() && latest.descriptor().index().type() == IndexDescriptor::TIMESTAMP) {
-            min_idx = std::min(min_idx, latest.begin()->begin()->value<timestamp>());
-            max_idx = std::max(max_idx, (latest.end() - 1)->begin()->value<timestamp>());
+        if (segment.row_count() && segment.descriptor().index().type() == IndexDescriptor::TIMESTAMP) {
+            min_idx = std::min(min_idx, segment.begin()->begin()->value<timestamp>());
+            max_idx = std::max(max_idx, (segment.end() - 1)->begin()->value<timestamp>());
         }
-        merge_string_columns(latest, merged.string_pool_ptr(), false);
-        merged.append(*history.rbegin());
+        merge_string_columns(segment, merged.string_pool_ptr(), false);
+        merged.append(segment);
         merged.set_compacted(true);
         util::print_total_mem_usage(__FILE__, __LINE__, __FUNCTION__);
     }
diff --git a/cpp/arcticdb/stream/segment_aggregator.hpp b/cpp/arcticdb/stream/segment_aggregator.hpp
index 2c848e3f79..6cdb307171 100644
--- a/cpp/arcticdb/stream/segment_aggregator.hpp
+++ b/cpp/arcticdb/stream/segment_aggregator.hpp
@@ -50,6 +50,15 @@ template<class Index, class Schema, class SegmentingPolicy = RowCountSegmentPoli
 
     void add_segment(SegmentInMemory&& seg, const pipelines::FrameSlice& slice, bool convert_int_to_float) {
         auto segment = std::move(seg);
+        if constexpr (std::is_same_v<Schema, FixedSchema>) {
+            if (stream_descriptor_.has_value()) {
+                schema::check<ErrorCode::E_DESCRIPTOR_MISMATCH>(
+                        segment.descriptor().fields() == stream_descriptor_->fields(),
+                        "Stream descriptor mismatch when compacting segments with static schema");
+            } else {
+                stream_descriptor_ = segment.descriptor();
+            }
+        }
         AggregatorType::stats().update_many(segment.row_count(), segment.num_bytes());
         //TODO very specific use-case, you probably don't want this
         if(convert_int_to_float)
@@ -100,6 +109,7 @@ template<class Index, class Schema, class SegmentingPolicy = RowCountSegmentPoli
     std::vector<SegmentInMemory> segments_;
     std::vector<pipelines::FrameSlice> slices_;
     SliceCallBack slice_callback_;
+    std::optional<StreamDescriptor> stream_descriptor_;
 };
 
 } // namespace arcticdb
\ No newline at end of file
diff --git a/cpp/arcticdb/version/local_versioned_engine.hpp b/cpp/arcticdb/version/local_versioned_engine.hpp
index 04ac6fdd32..0c9a1e6bc3 100644
--- a/cpp/arcticdb/version/local_versioned_engine.hpp
+++ b/cpp/arcticdb/version/local_versioned_engine.hpp
@@ -401,6 +401,8 @@ class LocalVersionedEngine : public VersionedEngine {
         return store()->current_timestamp();
     }
 
+    const arcticdb::proto::storage::VersionStoreConfig& cfg() const override { return cfg_; }
+
 protected:
     VersionedItem compact_incomplete_dynamic(
             const StreamId& stream_id,
@@ -423,7 +425,6 @@ class LocalVersionedEngine : public VersionedEngine {
     );
 
     std::shared_ptr<Store>& store() override { return store_; }
-    const arcticdb::proto::storage::VersionStoreConfig& cfg() const override { return cfg_; }
     std::shared_ptr<VersionMap>& version_map() override { return version_map_; }
     SymbolList& symbol_list() override { return *symbol_list_; }
     std::shared_ptr<SymbolList> symbol_list_ptr() { return symbol_list_; }
diff --git a/cpp/arcticdb/version/version_core.cpp b/cpp/arcticdb/version/version_core.cpp
index 35e954fbe3..6c3beec712 100644
--- a/cpp/arcticdb/version/version_core.cpp
+++ b/cpp/arcticdb/version/version_core.cpp
@@ -1352,7 +1352,7 @@ VersionedItem compact_incomplete_impl(
                                     sparsify ? VariantColumnPolicy{SparseColumnPolicy{}} : VariantColumnPolicy{DenseColumnPolicy{}}
                                     );
     util::variant_match(std::move(policies), [
-        &fut_vec, &slices, pipeline_context=pipeline_context, &store, convert_int_to_float, &previous_sorted_value] (auto &&idx, auto &&schema, auto &&column_policy) {
+        &fut_vec, &slices, pipeline_context=pipeline_context, &store, convert_int_to_float, &previous_sorted_value, &write_options] (auto &&idx, auto &&schema, auto &&column_policy) {
         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)>;
@@ -1364,7 +1364,7 @@ VersionedItem compact_incomplete_impl(
                 slices,
                 store,
                 convert_int_to_float,
-                std::nullopt);
+                write_options.segment_row_size);
         if constexpr(std::is_same_v<IndexType, TimeseriesIndex>) {
             pipeline_context->desc_->set_sorted(deduce_sorted(previous_sorted_value.value_or(SortedValue::ASCENDING), SortedValue::ASCENDING));
         }


Test patch:
diff --git a/cpp/arcticdb/version/test/test_version_store.cpp b/cpp/arcticdb/version/test/test_version_store.cpp
index 8ba24ab280..6910ce2154 100644
--- a/cpp/arcticdb/version/test/test_version_store.cpp
+++ b/cpp/arcticdb/version/test/test_version_store.cpp
@@ -203,6 +203,11 @@ TEST_F(VersionStoreTest, CompactIncompleteDynamicSchema) {
     using namespace arcticdb::stream;
     using namespace arcticdb::pipelines;
 
+    arcticdb::proto::storage::VersionStoreConfig cfg;
+    cfg.CopyFrom(test_store_->cfg());
+    cfg.mutable_write_options()->set_dynamic_schema(true);
+    test_store_->configure(std::move(cfg));
+
     size_t count = 0;
 
     std::vector<SegmentToInputFrameAdapter> data;
diff --git a/python/tests/stress/arcticdb/version_store/test_stress_append.py b/python/tests/stress/arcticdb/version_store/test_stress_append.py
index 13dee5abf2..2f0cca9f49 100644
--- a/python/tests/stress/arcticdb/version_store/test_stress_append.py
+++ b/python/tests/stress/arcticdb/version_store/test_stress_append.py
@@ -64,7 +64,7 @@ def test_write_parallel_stress(object_version_store):
     assert_frame_equal(vit.data, df)
 
 
-def test_write_parallel_stress_schema_change(object_version_store):
+def test_write_parallel_stress_schema_change(lmdb_version_store_dynamic_schema):
     num_rows_per_day = 1000
     num_days = 100
     num_columns = 8
@@ -87,17 +87,17 @@ def test_write_parallel_stress_schema_change(object_version_store):
 
     random.shuffle(dataframes)
     for d in dataframes:
-        object_version_store.write(symbol, d, parallel=True)
+        lmdb_version_store_dynamic_schema.write(symbol, d, parallel=True)
 
-    object_version_store.version_store.compact_incomplete(symbol, False, False)
-    vit = object_version_store.read(symbol)
+    lmdb_version_store_dynamic_schema.version_store.compact_incomplete(symbol, False, False)
+    vit = lmdb_version_store_dynamic_schema.read(symbol)
     df.sort_index(axis=1, inplace=True)
     result = vit.data
     result.sort_index(axis=1, inplace=True)
     assert_frame_equal(vit.data, df)
 
 
-def test_write_parallel_stress_schema_change_strings(basic_store_large_data):
+def test_write_parallel_stress_schema_change_strings(lmdb_version_store_dynamic_schema):
     num_rows_per_day = 1000
     num_days = 100
     num_columns = 8
@@ -121,17 +121,17 @@ def test_write_parallel_stress_schema_change_strings(basic_store_large_data):
 
     random.shuffle(dataframes)
     for d in dataframes:
-        basic_store_large_data.write(symbol, d, parallel=True)
+        lmdb_version_store_dynamic_schema.write(symbol, d, parallel=True)
 
-    basic_store_large_data.version_store.compact_incomplete(symbol, False, False)
-    vit = basic_store_large_data.read(symbol)
+    lmdb_version_store_dynamic_schema.version_store.compact_incomplete(symbol, False, False)
+    vit = lmdb_version_store_dynamic_schema.read(symbol)
     df.sort_index(axis=1, inplace=True)
     result = vit.data
     result.sort_index(axis=1, inplace=True)
     assert_frame_equal(vit.data, df)
 
 
-def test_write_parallel_stress_schema_change_strings_with_nan(basic_store_large_data):
+def test_write_parallel_stress_schema_change_strings_with_nan(lmdb_version_store_dynamic_schema):
     num_rows_per_day = 1000
     num_days = 100
     num_columns = 8
@@ -155,10 +155,10 @@ def test_write_parallel_stress_schema_change_strings_with_nan(basic_store_large_
 
     random.shuffle(dataframes)
     for d in dataframes:
-        basic_store_large_data.write(symbol, d, parallel=True)
+        lmdb_version_store_dynamic_schema.write(symbol, d, parallel=True)
 
-    basic_store_large_data.version_store.compact_incomplete(symbol, False, False)
-    vit = basic_store_large_data.read(symbol)
+    lmdb_version_store_dynamic_schema.version_store.compact_incomplete(symbol, False, False)
+    vit = lmdb_version_store_dynamic_schema.read(symbol)
     df.sort_index(axis=1, inplace=True)
     result = vit.data
     result.sort_index(axis=1, inplace=True)
@@ -191,7 +191,7 @@ def test_change_to_dynamic_strings(object_version_store):
     assert_frame_equal(vit.data, df)
 
 
-def test_write_parallel_stress_many_chunks(object_version_store):
+def test_write_parallel_stress_many_chunks(lmdb_version_store_dynamic_schema):
     num_rows_per_day = 10
     num_days = 10
     num_columns = 8
@@ -214,10 +214,10 @@ def test_write_parallel_stress_many_chunks(object_version_store):
 
     random.shuffle(dataframes)
     for d in dataframes:
-        object_version_store.write(symbol, d, parallel=True)
+        lmdb_version_store_dynamic_schema.write(symbol, d, parallel=True)
 
-    object_version_store.version_store.compact_incomplete(symbol, False, False)
-    vit = object_version_store.read(symbol)
+    lmdb_version_store_dynamic_schema.version_store.compact_incomplete(symbol, False, False)
+    vit = lmdb_version_store_dynamic_schema.read(symbol)
     df.sort_index(axis=1, inplace=True)
     result = vit.data
     result.sort_index(axis=1, inplace=True)
diff --git a/python/tests/unit/arcticdb/version_store/test_parallel.py b/python/tests/unit/arcticdb/version_store/test_parallel.py
index 676160baf8..471fc3984d 100644
--- a/python/tests/unit/arcticdb/version_store/test_parallel.py
+++ b/python/tests/unit/arcticdb/version_store/test_parallel.py
@@ -11,7 +11,7 @@
 import datetime
 import pytest
 
-from arcticdb.exceptions import SortingException
+from arcticdb.exceptions import SortingException, SchemaException
 from arcticdb.util.test import (
     assert_frame_equal,
     random_strings_of_length,
@@ -95,7 +95,7 @@ def test_roundtrip_nat(lmdb_version_store):
     vit = lmdb_version_store.read("all_nats")
 
 
-def test_floats_to_nans(lmdb_version_store):
+def test_floats_to_nans(lmdb_version_store_dynamic_schema):
     num_rows_per_day = 10
     num_days = 10
     num_columns = 8
@@ -118,10 +118,10 @@ def test_floats_to_nans(lmdb_version_store):
 
     random.shuffle(dataframes)
     for d in dataframes:
-        lmdb_version_store.write(symbol, d, parallel=True)
+        lmdb_version_store_dynamic_schema.write(symbol, d, parallel=True)
 
-    lmdb_version_store.version_store.compact_incomplete(symbol, False, False)
-    vit = lmdb_version_store.read(symbol)
+    lmdb_version_store_dynamic_schema.version_store.compact_incomplete(symbol, False, False)
+    vit = lmdb_version_store_dynamic_schema.read(symbol)
     df.sort_index(axis=1, inplace=True)
     result = vit.data
     result.sort_index(axis=1, inplace=True)
@@ -204,7 +204,7 @@ def test_sort_merge_append(basic_store_dynamic_schema):
     assert_frame_equal(vit.data, df)
 
 
-def test_datetimes_to_nats(lmdb_version_store):
+def test_datetimes_to_nats(lmdb_version_store_dynamic_schema):
     num_rows_per_day = 10
     num_days = 10
     num_columns = 8
@@ -226,10 +226,10 @@ def test_datetimes_to_nats(lmdb_version_store):
 
     random.shuffle(dataframes)
     for d in dataframes:
-        lmdb_version_store.write(symbol, d, parallel=True)
+        lmdb_version_store_dynamic_schema.write(symbol, d, parallel=True)
 
-    lmdb_version_store.version_store.compact_incomplete(symbol, False, True)
-    vit = lmdb_version_store.read(symbol)
+    lmdb_version_store_dynamic_schema.version_store.compact_incomplete(symbol, False, True)
+    vit = lmdb_version_store_dynamic_schema.read(symbol)
     df.sort_index(axis=1, inplace=True)
     result = vit.data
     result.sort_index(axis=1, inplace=True)
@@ -366,48 +366,104 @@ def test_parallel_append_existing_data_unsorted(lmdb_version_store, sortedness):
         lib.compact_incomplete(sym, True, False)
 
 
-@pytest.mark.xfail(reason="See https://github.com/man-group/ArcticDB/issues/1250")
-@pytest.mark.parametrize("append", (True, False))
-def test_parallel_dynamic_schema_compatible_types(lmdb_version_store_dynamic_schema, append):
-    lib = lmdb_version_store_dynamic_schema
-    sym = "test_parallel_dynamic_schema_compatible_types"
-    if append:
-        df_0 = pd.DataFrame(
-            {
-                "same type": np.arange(1, dtype=np.uint8),
-                "compatible type": np.arange(1, dtype=np.uint8),
-                "column in some incompletes": np.arange(1, dtype=np.uint8),
-                "column in no incompletes": np.arange(1, dtype=np.uint8),
-            }
-            , index=[pd.Timestamp("2024-01-01")]
-        )
-        lib.write(sym, df_0)
-    df_1 = pd.DataFrame(
-        {
-            "same type": np.arange(1, 2, dtype=np.uint8),
-            "compatible type": np.arange(1, 2, dtype=np.uint16),
-            "column in some incompletes": np.arange(1, 2, dtype=np.uint8),
-        }
-        , index=[pd.Timestamp("2024-01-02")]
-    )
-    df_2 = pd.DataFrame(
-        {
-            "same type": np.arange(2, 3, dtype=np.uint8),
-            "compatible type": np.arange(2, 3, dtype=np.uint32),
-        }
-        , index=[pd.Timestamp("2024-01-03")]
-    )
-    if append:
-        lib.append(sym, df_2, incomplete=True)
-        lib.append(sym, df_1, incomplete=True)
-    else:
-        lib.write(sym, df_2, parallel=True)
-        lib.write(sym, df_1, parallel=True)
-    lib.compact_incomplete(sym, append, False)
+def test_parallel_no_column_slicing(lmdb_version_store_tiny_segment):
+    lib = lmdb_version_store_tiny_segment
+    sym = "test_parallel_column_slicing"
+    lib_tool = lib.library_tool()
+    # lmdb_version_store_tiny_segment has 2 columns per slice
+    df = pd.DataFrame({"col_0": [0], "col_1": [1], "col_2": [2]})
+    lib.write(sym, df, parallel=True)
+    assert len(lib_tool.find_keys(KeyType.APPEND_DATA)) == 1
+    lib.remove_incomplete(sym)
+    lib.append(sym, df, incomplete=True)
+    assert len(lib_tool.find_keys(KeyType.APPEND_DATA)) == 1
+
+
+@pytest.mark.parametrize("rows_per_incomplete", (1, 2))
+def test_parallel_write_static_schema_type_changing(lmdb_version_store_tiny_segment, rows_per_incomplete):
+    lib = lmdb_version_store_tiny_segment
+    sym = "test_parallel_write_static_schema_type_changing"
+    lib_tool = lib.library_tool()
+    df_0 = pd.DataFrame({"col": np.arange(rows_per_incomplete, dtype=np.uint8)}, index=pd.date_range("2024-01-01", periods=rows_per_incomplete))
+    df_1 = pd.DataFrame({"col": np.arange(rows_per_incomplete, 2 * rows_per_incomplete, dtype=np.uint16)}, index=pd.date_range("2024-01-03", periods=rows_per_incomplete))
+    lib.write(sym, df_0, parallel=True)
+    lib.write(sym, df_1, parallel=True)
+    with pytest.raises(SchemaException):
+        lib.compact_incomplete(sym, False, False)
+
+
+@pytest.mark.parametrize("rows_per_incomplete", (1, 2))
+def test_parallel_write_dynamic_schema_type_changing(lmdb_version_store_tiny_segment_dynamic, rows_per_incomplete):
+    lib = lmdb_version_store_tiny_segment_dynamic
+    sym = "test_parallel_write_dynamic_schema_type_changing"
+    df_0 = pd.DataFrame({"col": np.arange(rows_per_incomplete, dtype=np.uint8)}, index=pd.date_range("2024-01-01", periods=rows_per_incomplete))
+    df_1 = pd.DataFrame({"col": np.arange(rows_per_incomplete, 2 * rows_per_incomplete, dtype=np.uint16)}, index=pd.date_range("2024-01-02", periods=rows_per_incomplete))
+    lib.write(sym, df_0, parallel=True)
+    lib.write(sym, df_1, parallel=True)
+    lib.compact_incomplete(sym, False, False)
+    expected = pd.concat([df_0, df_1])
+    received = lib.read(sym).data
+    assert_frame_equal(expected, received)
+
+
+def test_parallel_write_static_schema_missing_column(lmdb_version_store_tiny_segment):
+    lib = lmdb_version_store_tiny_segment
+    sym = "test_parallel_write_static_schema_missing_column"
+    df_0 = pd.DataFrame({"col_0": [0], "col_1": [0.5]}, index=pd.date_range("2024-01-01", periods=1))
+    df_1 = pd.DataFrame({"col_0": [1]}, index=pd.date_range("2024-01-02", periods=1))
+    lib.write(sym, df_0, parallel=True)
+    lib.write(sym, df_1, parallel=True)
+    with pytest.raises(SchemaException):
+        lib.compact_incomplete(sym, False, False)
+
+
+def test_parallel_write_dynamic_schema_missing_column(lmdb_version_store_tiny_segment_dynamic):
+    lib = lmdb_version_store_tiny_segment_dynamic
+    sym = "test_parallel_write_dynamic_schema_missing_column"
+    df_0 = pd.DataFrame({"col_0": [0], "col_1": [0.5]}, index=pd.date_range("2024-01-01", periods=1))
+    df_1 = pd.DataFrame({"col_0": [1]}, index=pd.date_range("2024-01-02", periods=1))
+    lib.write(sym, df_0, parallel=True)
+    lib.write(sym, df_1, parallel=True)
+    lib.compact_incomplete(sym, False, False)
+    expected = pd.concat([df_0, df_1])
+    received = lib.read(sym).data
+    assert_frame_equal(expected, received)
+
+
+@pytest.mark.xfail(reason="See https://github.com/man-group/ArcticDB/issues/1466")
+def test_parallel_append_static_schema_type_changing(lmdb_version_store_tiny_segment):
+    lib = lmdb_version_store_tiny_segment
+    sym = "test_parallel_append_static_schema_type_changing"
+    df_0 = pd.DataFrame({"col": np.arange(1, dtype=np.uint8)}, index=pd.date_range("2024-01-01", periods=1))
+    df_1 = pd.DataFrame({"col": np.arange(1, 2, dtype=np.uint16)}, index=pd.date_range("2024-01-02", periods=1))
+    lib.write(sym, df_0)
+    lib.append(sym, df_1, incomplete=True)
+    with pytest.raises(SchemaException):
+        lib.compact_incomplete(sym, True, False)
+
+
+@pytest.mark.xfail(reason="See https://github.com/man-group/ArcticDB/issues/1466")
+def test_parallel_append_static_schema_missing_column(lmdb_version_store_tiny_segment):
+    lib = lmdb_version_store_tiny_segment
+    sym = "test_parallel_append_static_schema_missing_column"
+    df_0 = pd.DataFrame({"col_0": [0], "col_1": [0.5]}, index=pd.date_range("2024-01-01", periods=1))
+    df_1 = pd.DataFrame({"col_0": [1]}, index=pd.date_range("2024-01-02", periods=1))
+    lib.write(sym, df_0)
+    lib.append(sym, df_1, incomplete=True)
+    with pytest.raises(SchemaException):
+        lib.compact_incomplete(sym, True, False)
+
+
+def test_parallel_append_dynamic_schema_missing_column(lmdb_version_store_tiny_segment_dynamic):
+    lib = lmdb_version_store_tiny_segment_dynamic
+    sym = "test_parallel_append_dynamic_schema_missing_column"
+    df_0 = pd.DataFrame({"col_0": [0], "col_1": [0.5]}, index=pd.date_range("2024-01-01", periods=1))
+    df_1 = pd.DataFrame({"col_0": [1]}, index=pd.date_range("2024-01-02", periods=1))
+    lib.write(sym, df_0)
+    lib.append(sym, df_1, incomplete=True)
+    lib.compact_incomplete(sym, True, False)
+    expected = pd.concat([df_0, df_1])
+    received = lib.read(sym).data
+    assert_frame_equal(expected, received)
+
 
-    read_df = lib.read(sym).data
-    if append:
-        expected_df = pd.concat([df_0, df_1, df_2])
-    else:
-        expected_df = pd.concat([df_1, df_2])
-    assert_frame_equal(expected_df, read_df)