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

Issue:
`compact_incomplete` with `append=True` should fail if the existing data is `UNSORTED` or `DESCENDING`
i.e. the following test should pass:
```
@pytest.mark.parametrize("sortedness", ("DESCENDING", "UNSORTED"))
def test_parallel_append_to_known_unsorted_data(lmdb_version_store, sortedness):
    lib = lmdb_version_store
    sym = "test_parallel_append_to_known_unsorted_data"
    last_index_date = "2024-01-01" if sortedness == "DESCENDING" else "2024-01-03"
    df_0 = pd.DataFrame(
        {"col": [1, 2, 3]},
        index=[pd.Timestamp("2024-01-04"), pd.Timestamp("2024-01-02"), pd.Timestamp(last_index_date)]
    )
    lib.write(sym, df_0)
    assert lib.get_info(sym)["sorted"] == sortedness
    df_1 = pd.DataFrame({"col": [3, 4]}, index=[pd.Timestamp("2024-01-05"), pd.Timestamp("2024-01-06")])
    lib.append(sym, df_1, incomplete=True)
    with pytest.raises(SortingException):
        lib.compact_incomplete(sym, True, False)
```
Currently no exception is thrown


Discussion / hints:

Fix patch:
diff --git a/cpp/arcticdb/version/version_core.cpp b/cpp/arcticdb/version/version_core.cpp
index b31c957826..07bf4aa340 100644
--- a/cpp/arcticdb/version/version_core.cpp
+++ b/cpp/arcticdb/version/version_core.cpp
@@ -821,16 +821,26 @@ void read_incompletes_to_pipeline(
     pipeline_context->total_rows_ = pipeline_context->calc_rows();
 }
 
-void check_incompletes_index_ranges_dont_overlap(const std::shared_ptr<PipelineContext>& pipeline_context) {
-    // Does nothing if the symbol is not timestamp-indexed
-    // Checks both that the index ranges of incomplete segments do not overlap with one another, and that the earliest
-    // timestamp in an incomplete segment is greater than the latest timestamp existing in the symbol in the case of a
-    // parallel append
+void check_incompletes_index_ranges_dont_overlap(const std::shared_ptr<PipelineContext>& pipeline_context,
+                                                 const std::optional<SortedValue>& previous_sorted_value) {
+    /*
+     Does nothing if the symbol is not timestamp-indexed
+     Checks:
+      - that the existing data is not known to be unsorted in the case of a parallel append
+      - that the index ranges of incomplete segments do not overlap with one another
+      - that the earliest timestamp in an incomplete segment is greater than the latest timestamp existing in the
+        symbol in the case of a parallel append
+     */
     if (pipeline_context->descriptor().index().type() == IndexDescriptor::TIMESTAMP) {
         std::optional<timestamp> last_existing_index_value;
         // Beginning of incomplete segments == beginning of all segments implies all segments are incompletes, so we are
         // writing, not appending
         if (pipeline_context->incompletes_begin() != pipeline_context->begin()) {
+            sorting::check<ErrorCode::E_UNSORTED_DATA>(
+                    !previous_sorted_value.has_value() ||
+                    *previous_sorted_value == SortedValue::ASCENDING ||
+                    *previous_sorted_value == SortedValue::UNKNOWN,
+                    "Cannot append staged segments to existing data as existing data is not sorted in ascending order");
             auto last_indexed_slice_and_key = std::prev(pipeline_context->incompletes_begin())->slice_and_key();
             // -1 as end_time is stored as 1 greater than the last index value in the segment
             last_existing_index_value = last_indexed_slice_and_key.key().end_time() - 1;
@@ -1322,7 +1332,7 @@ VersionedItem compact_incomplete_impl(
     if (pipeline_context->slice_and_keys_.size() == prev_size) {
         util::raise_rte("No incomplete segments found for {}", stream_id);
     }
-    check_incompletes_index_ranges_dont_overlap(pipeline_context);
+    check_incompletes_index_ranges_dont_overlap(pipeline_context, previous_sorted_value);
     const auto& first_seg = pipeline_context->slice_and_keys_.begin()->segment(store);
 
     std::vector<entity::VariantKey> delete_keys;


Test patch:
diff --git a/python/tests/unit/arcticdb/version_store/test_parallel.py b/python/tests/unit/arcticdb/version_store/test_parallel.py
index 27e85e9b46..676160baf8 100644
--- a/python/tests/unit/arcticdb/version_store/test_parallel.py
+++ b/python/tests/unit/arcticdb/version_store/test_parallel.py
@@ -349,6 +349,23 @@ def test_parallel_append_overlapping_with_existing(lmdb_version_store):
         lib.compact_incomplete(sym, True, False)
 
 
+@pytest.mark.parametrize("sortedness", ("DESCENDING", "UNSORTED"))
+def test_parallel_append_existing_data_unsorted(lmdb_version_store, sortedness):
+    lib = lmdb_version_store
+    sym = "test_parallel_append_existing_data_unsorted"
+    last_index_date = "2024-01-01" if sortedness == "DESCENDING" else "2024-01-03"
+    df_0 = pd.DataFrame(
+        {"col": [1, 2, 3]},
+        index=[pd.Timestamp("2024-01-04"), pd.Timestamp("2024-01-02"), pd.Timestamp(last_index_date)]
+    )
+    lib.write(sym, df_0)
+    assert lib.get_info(sym)["sorted"] == sortedness
+    df_1 = pd.DataFrame({"col": [3, 4]}, index=[pd.Timestamp("2024-01-05"), pd.Timestamp("2024-01-06")])
+    lib.append(sym, df_1, incomplete=True)
+    with pytest.raises(SortingException):
+        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):