Fix batch_append and batch_update bust version chain if upsert (#2661)

#### Reference Issues/PRs
<!--Example: Fixes #1234. See also #3456.-->
https://man312219.monday.com/boards/7852509418/pulses/9903341116
#### What does this implement or fix?
Existing code incorrectly assumes when the symbol being upsert is not
live, it has never existed. So the version being upserted has always
been set to 0.
But the symbol could exist but deleted at that moment. So the version
being upserted has always been set to 0. If so, this will bust the
version chain.

This PR also fixes exception thrown in writing version keys being
swallowed
#### Any other comments?

#### Checklist

<details>
  <summary>
   Checklist for code changes...
  </summary>
 
- [ ] Have you updated the relevant docstrings, documentation and
copyright notice?
- [ ] Is this contribution tested against [all ArcticDB's
features](../docs/mkdocs/docs/technical/contributing.md)?
- [ ] Do all exceptions introduced raise appropriate [error
messages](https://docs.arcticdb.io/error_messages/)?
 - [ ] Are API changes highlighted in the PR description?
- [ ] Is the PR labelled as enhancement or bug so it appears in
autogenerated release notes?
</details>

<!--
Thanks for contributing a Pull Request to ArcticDB! Please ensure you
have taken a look at:
- ArcticDB's Code of Conduct:
https://github.com/man-group/ArcticDB/blob/master/CODE_OF_CONDUCT.md
- ArcticDB's Contribution Licensing:
https://github.com/man-group/ArcticDB/blob/master/docs/mkdocs/docs/technical/contributing.md#contribution-licensing
-->

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

@@ -27,7 +27,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
 
@@ -575,3 +575,18 @@ def test_update_bool_named_col(lmdb_version_store_dynamic_schema, idx):
         lmdb_version_store_dynamic_schema.update(symbol, bad_df)
 
     assert_frame_equal(lmdb_version_store_dynamic_schema.read(symbol).data, initial)
+
+
+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

@@ -727,6 +727,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()