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

Issue:
`compact_incomplete` with `append=True` should throw if the earliest timestamp in the index of an incomplete segment is earlier than the latest timestamp in the index of the existing symbol



Discussion / hints:

Fix patch:
diff --git a/cpp/arcticdb/version/version_core.cpp b/cpp/arcticdb/version/version_core.cpp
index e38245814d..626ca8cd14 100644
--- a/cpp/arcticdb/version/version_core.cpp
+++ b/cpp/arcticdb/version/version_core.cpp
@@ -819,6 +819,43 @@ 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
+    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()) {
+            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;
+        }
+
+        // Use ordered set so we only need to compare adjacent elements
+        std::set<TimestampRange> unique_timestamp_ranges;
+        for (auto it = pipeline_context->incompletes_begin(); it!= pipeline_context->end(); it++) {
+            sorting::check<ErrorCode::E_UNSORTED_DATA>(
+                    !last_existing_index_value.has_value() || it->slice_and_key().key().start_time() > *last_existing_index_value,
+                    "Cannot append staged segments to existing data as incomplete segment contains index value <= existing data: {} <= {}",
+                    it->slice_and_key().key().start_time(),
+                    *last_existing_index_value);
+            unique_timestamp_ranges.insert({it->slice_and_key().key().start_time(), it->slice_and_key().key().end_time()});
+        }
+        for (auto it = unique_timestamp_ranges.begin(); it != unique_timestamp_ranges.end(); it++) {
+            auto next_it = std::next(it);
+            if (next_it != unique_timestamp_ranges.end()) {
+                sorting::check<ErrorCode::E_UNSORTED_DATA>(
+                        next_it->first >= it->second,
+                        "Cannot append staged segments to existing data as incomplete segment index values overlap one another: ({}, {}) intersects ({}, {})",
+                        it->first, it->second - 1, next_it->first, next_it->second - 1);
+            }
+        }
+    }
+}
+
 void copy_frame_data_to_buffer(const SegmentInMemory& destination, size_t target_index, SegmentInMemory& source, size_t source_index, const RowRange& row_range) {
     auto num_rows = row_range.diff();
     if (num_rows == 0) {
@@ -1268,6 +1305,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);
     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 8c45a0af10..27e85e9b46 100644
--- a/python/tests/unit/arcticdb/version_store/test_parallel.py
+++ b/python/tests/unit/arcticdb/version_store/test_parallel.py
@@ -319,7 +319,6 @@ def test_parallel_non_timestamp_index(lmdb_version_store, append):
     assert_frame_equal(expected_df, read_df)
 
 
-@pytest.mark.xfail(reason="See https://github.com/man-group/ArcticDB/issues/1248")
 @pytest.mark.parametrize("append", (True, False))
 def test_parallel_overlapping_incomplete_segments(lmdb_version_store, append):
     lib = lmdb_version_store
@@ -335,11 +334,10 @@ def test_parallel_overlapping_incomplete_segments(lmdb_version_store, append):
     else:
         lib.write(sym, df_2, parallel=True)
         lib.write(sym, df_1, parallel=True)
-    with pytest.raises(Exception):
+    with pytest.raises(SortingException):
         lib.compact_incomplete(sym, append, False)
 
 
-@pytest.mark.xfail(reason="See https://github.com/man-group/ArcticDB/issues/1249")
 def test_parallel_append_overlapping_with_existing(lmdb_version_store):
     lib = lmdb_version_store
     sym = "test_parallel_append_overlapping_with_existing"
@@ -347,7 +345,7 @@ def test_parallel_append_overlapping_with_existing(lmdb_version_store):
     lib.write(sym, df_0)
     df_1 = pd.DataFrame({"col": [3, 4]}, index=[pd.Timestamp("2024-01-01T12"), pd.Timestamp("2024-01-03")])
     lib.append(sym, df_1, incomplete=True)
-    with pytest.raises(Exception):
+    with pytest.raises(SortingException):
         lib.compact_incomplete(sym, True, False)