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

Issue:
Using sort_and_finalize_staged_data with append can create unordered index
**Describe the bug**
When `sort_and_finalize_staged_data` is used with `mode=StagedDataFinalizeMethod.APPEND` it allows to append index such that the overall index becomes out of order. For example it's possible for the appended index values to be less than the last index value in storage.

**Steps to reproduce**
```python
import pandas as pd
import numpy as np
import arcticdb as adb
from arcticdb.version_store.library import StagedDataFinalizeMethod

ac = adb.Arctic("lmdb://test")
lib = ac.get_library("test", create_if_missing=True)
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]"))
lib.write("sym", df1, staged=True)
lib.sort_and_finalize_staged_data("sym", mode=StagedDataFinalizeMethod.APPEND)
print(lib.read("sym"))
```

Output:
```
            col
2023-01-01    1
2023-01-03    3
2023-01-02    2
```

**Expected behavior**
It should behave just like `Library.append` and throw an exception.

**OS, Python, Arctic versions**
```
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.0rc1
```


Discussion / hints:

Fix patch:
diff --git a/cpp/arcticdb/version/version_core.cpp b/cpp/arcticdb/version/version_core.cpp
index 9eff9d77c3..5fb7e0eef1 100644
--- a/cpp/arcticdb/version/version_core.cpp
+++ b/cpp/arcticdb/version/version_core.cpp
@@ -1288,7 +1288,6 @@ VersionedItem sort_merge_impl(
 
     std::vector<FrameSlice> slices;
     std::vector<folly::Future<VariantKey>> fut_vec;
-    std::optional<SegmentInMemory> last_indexed;
     auto index = stream::index_type_from_descriptor(pipeline_context->descriptor());
     util::variant_match(index,
         [&](const stream::TimeseriesIndex &timeseries_index) {
@@ -1299,6 +1298,15 @@ VersionedItem sort_merge_impl(
 
             read_query.clauses_.emplace_back(std::make_shared<Clause>(MergeClause{timeseries_index, SparseColumnPolicy{}, stream_id, pipeline_context->descriptor()}));
             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();
+                user_input::raise<ErrorCode::E_INVALID_USER_ARGUMENT>(
+                    "Cannot append index starting at {} to data frame with index ending at {}",
+                    std::get<timestamp>(TimeseriesIndex::start_value_for_segment(segments[0].segment_.value())),
+                    update_info.previous_index_key_->end_time() - 1
+                );
+            }
             pipeline_context->total_rows_ = num_versioned_rows + get_slice_rowcounts(segments);
 
             auto index = index_type_from_descriptor(pipeline_context->descriptor());


Test patch:
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 8a1dafe01a..237f3faafd 100644
--- a/python/tests/unit/arcticdb/version_store/test_sort_merge.py
+++ b/python/tests/unit/arcticdb/version_store/test_sort_merge.py
@@ -3,6 +3,7 @@
 from pandas.testing import assert_frame_equal
 import pytest
 from arcticdb.version_store.library import StagedDataFinalizeMethod
+from arcticdb.exceptions import UserInputException
 
 def test_merge_single_column(lmdb_library):
     lib = lmdb_library
@@ -176,17 +177,33 @@ 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)
 
-    @pytest.mark.xfail(reason="Does not throw and creates dataframe with unordered index.")
     def test_appended_df_interleaves_with_storage(self, lmdb_library):
         lib = lmdb_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]"))
         lib.write("sym", df1, staged=True)
-        with pytest.raises(Exception) as exception_info:
+        with pytest.raises(UserInputException) as exception_info:
             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
+        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]")
+        )
+        lib.write("sym", df)
+        df_to_append = pd.DataFrame(
+            {"col": [4, 5, 6]},
+            index=pd.DatetimeIndex([np.datetime64('2023-01-03'), np.datetime64('2023-01-04'), np.datetime64('2023-01-05')], dtype="datetime64[ns]")
+        )
+        lib.write("sym", df_to_append, staged=True)
+        lib.sort_and_finalize_staged_data("sym", mode=StagedDataFinalizeMethod.APPEND)
+        res = lib.read("sym").data
+        expected_df = pd.concat([df, df_to_append])
+        assert_frame_equal(lib.read("sym").data, expected_df)
+
 
 def test_prune_previous(lmdb_library):
     lib = lmdb_library