Fix batch append and batch update bust version chain issue

Author: phoebusm

cpp/arcticdb/version/local_versioned_engine.cpp

@@ -1476,7 +1476,11 @@ folly::Future<VersionedItem> LocalVersionedEngine::write_index_key_to_version_ma
     if (add_new_symbol) {
         write_version_fut =
                 std::move(write_version_fut)
-                        .then([this, index_key_id = index_key.id(), reference_id = index_key.version_id()](auto&&) {
+                        .thenValue([this,
+                                    index_key_id = index_key.id(),
+                                    reference_id = index_key.version_id(
+                                    )](auto&&) { // Using .then and not using validating result from previous future
+                                                 // will swallow exception from previous future
                             return async::submit_io_task(
                                     WriteSymbolTask(store(), symbol_list_ptr(), index_key_id, reference_id)
                             );
@@ -1722,10 +1726,9 @@ std::vector<std::variant<VersionedItem, DataError>> LocalVersionedEngine::batch_
                                         missing_data::check<ErrorCode::E_NO_SUCH_VERSION>(
                                                 upsert, "Cannot append to non-existent symbol {}", stream_id
                                         );
-                                        constexpr static auto version_id = 0;
                                         index_key_fut = async_write_dataframe_impl(
                                                 store(),
-                                                version_id,
+                                                update_info.next_version_id_,
                                                 frame,
                                                 write_options,
                                                 std::make_shared<DeDupMap>(),
@@ -1815,7 +1818,7 @@ std::vector<std::variant<VersionedItem, DataError>> LocalVersionedEngine::batch_
                                         );
                                         index_key_fut = async_write_dataframe_impl(
                                                 store(),
-                                                0,
+                                                update_info.next_version_id_,
                                                 std::move(frame),
                                                 std::move(write_options),
                                                 std::make_shared<DeDupMap>(),

python/tests/integration/arcticdb/test_update.py

@@ -25,7 +25,7 @@
 from arcticdb.util.logger import get_logger
 from arcticdb.version_store._store import VersionedItem
 from arcticdb.version_store.library import Library, UpdatePayload, WritePayload
-from arcticdb.util.test import assert_frame_equal
+from arcticdb.util.test import assert_frame_equal, sample_dataframe
 from arcticdb_ext.version_store import DataError, NoSuchVersionException
 from tests.util.mark import LINUX
 
@@ -555,3 +555,18 @@ def test_update_batch_different_updates_dynamic_schema(custom_library):
         ArcticSymbolSimulator.assert_frame_equal_rebuild_index_first(
             expected_results[result.symbol], read_data[result.symbol].data
         )
+
+
+def test_batch_update_after_delete_upsert(arctic_library_lmdb):
+    lib = arctic_library_lmdb
+    lib.write("sym", sample_dataframe())
+    lib.write("sym1", sample_dataframe())
+    lib.write("sym1", sample_dataframe())
+    lib.delete_batch(["sym", "sym1"])
+    df = sample_dataframe()
+    df1 = sample_dataframe()
+    results = lib.update_batch([UpdatePayload("sym", df), UpdatePayload("sym1", df1)], upsert=True)
+    assert results[0].version == 1
+    assert results[1].version == 2
+    assert_frame_equal(lib.read("sym").data, df)
+    assert_frame_equal(lib.read("sym1").data, df1)

python/tests/integration/arcticdb/version_store/test_basic_version_store.py

@@ -726,6 +726,22 @@ def test_prune_previous_versions_append_batch(basic_store):
     assert len(lib_tool.find_keys(KeyType.SYMBOL_LIST)) == 6
 
 
+def test_batch_append_after_delete_upsert(lmdb_version_store_v1):
+    lib = lmdb_version_store_v1
+    lib.write("sym", 1)
+    lib.write("sym1", 1)
+    lib.write("sym1", 1)
+    lib.batch_delete_symbols(["sym", "sym1"])
+
+    df = sample_dataframe()
+    df1 = sample_dataframe()
+    results = lib.batch_append(["sym", "sym1"], [df, df1])
+    assert results[0].version == 1
+    assert results[1].version == 2
+    assert_frame_equal(lib.read("sym").data, df)
+    assert_frame_equal(lib.read("sym1").data, df1)
+
+
 @pytest.mark.storage
 def test_deleting_unknown_symbol(basic_store, symbol):
     df = sample_dataframe()