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

Issue:
read_batch raises a DataError with as_of a deleted version
### Describe the bug

When reading a version number that has been deleted a low level data error is returned. Also a warning is logged

`[arcticdb] [warning] Failed to find segment for key 'i:asof_slow_read:499:0x1f40d440a9ebd654@1709222092432355772[1704067200000000000,1704240000000000001]' : No response body.`

The error exception_string is
`arcticdb::storage::KeyNotFoundException: Composite: i:asof_slow_read:499:0x1f40d440a9ebd654@1709222092432355772[1704067200000000000,1704240000000000001],`

### Steps/Code to Reproduce

### Steps/Code to Reproduce

import arcticdb as adb
import pandas as pd
import numpy as np
from datetime import datetime, timedelta

arctic = adb.Arctic(<AWS S3 uri>)
lib = arctic.get_library('adb_bugs', create_if_missing=True)

N = 3
df = pd.DataFrame(
    index=pd.date_range("20240101", periods=N),
    data={'col': np.arange(0., N)}
)

#### write 500 versions
sym1 = 'asof_slow_read'
for i in range(500):
    lib.write(sym1, df)
	
#### remove early versions
lib.delete(sym1)

#### add one more version
lib.write(sym1, df)

#### this gives the error
v = lib.read_batch([adb.ReadRequest(sym1, as_of=499)])
v[0].exception_string

### Expected Results

Error should be
* Version not found
* The same as for lib.read()
* Not refer to low level storage issues

### OS, Python Version and ArcticDB Version

Python 3.10
Linux Linux version 5.15.133.1-microsoft-standard-WSL2
arcticdb 4.3.1

### Backend storage used

AWS S3

### Additional Context

This is possibly related to this issue (failure to observe tombstone correctly). It may be easier to solve the two together

https://github.com/man-group/ArcticDB/issues/1384


Discussion / hints:

Fix patch:
diff --git a/cpp/arcticdb/version/version_map_batch_methods.cpp b/cpp/arcticdb/version/version_map_batch_methods.cpp
index 19eb975b4e..6e18165c40 100644
--- a/cpp/arcticdb/version/version_map_batch_methods.cpp
+++ b/cpp/arcticdb/version/version_map_batch_methods.cpp
@@ -81,7 +81,8 @@ void StreamVersionData::do_react(const pipelines::SnapshotVersionQuery &snapshot
 
 std::optional<AtomKey> get_specific_version_from_entry(
     const std::shared_ptr<VersionMapEntry>& version_map_entry,
-    const pipelines::SpecificVersionQuery& specific_version
+    const pipelines::SpecificVersionQuery& specific_version,
+    bool include_deleted = false
     ) {
     auto signed_version_id = specific_version.version_id_;
     VersionId version_id;
@@ -101,7 +102,7 @@ std::optional<AtomKey> get_specific_version_from_entry(
             return std::nullopt;
         }
     }
-    return find_index_key_for_version_id(version_id, version_map_entry);
+    return find_index_key_for_version_id(version_id, version_map_entry, include_deleted);
 }
 
 std::optional<AtomKey> get_version_map_entry_by_timestamp(


Test patch:
diff --git a/python/tests/hypothesis/arcticdb/test_hypothesis_version_store.py b/python/tests/hypothesis/arcticdb/test_hypothesis_version_store.py
index 60be2170e8..bb27d1f428 100644
--- a/python/tests/hypothesis/arcticdb/test_hypothesis_version_store.py
+++ b/python/tests/hypothesis/arcticdb/test_hypothesis_version_store.py
@@ -210,33 +210,10 @@ def _check_batch_read(self, syms: List[str], vers: Optional[List[int]]):
             else:
                 assert_frame_equal(data[sym].data, expected.data)
 
-    def _check_batch_read_metadata_multi(self, syms: List[str], vers: List[int]):
-        meta = self._lib.batch_read_metadata_multi(syms, as_ofs=vers)
-
-        for sym, versions in meta.items():
-            expected_vers = self._versions[sym]
-            for v in range(len(versions)):
-                expected = expected_vers[v]
-                result = versions[v]
-                assert result.symbol == sym
-                assert result.version == v
-                assert result.metadata == expected.metadata
-
     @invariant()
     def test_batch_read_latest(self):
         self._check_batch_read(list(self._visible_symbols), None)
 
-    @invariant()
-    def test_batch_read_metadata_multi(self):
-        syms = []
-        vers = []
-        for sym, versions in self._versions.items():
-            for ver in range(len(versions)):
-                syms.append(sym)
-                vers.append(ver)
-
-        self._check_batch_read_metadata_multi(syms, vers)
-
     @invariant()
     def test_list_versions_latest_only(self):
         expected = {sym: self._get_latest_undeleted_version(sym)[0] for sym in self._visible_symbols}
@@ -306,12 +283,6 @@ def test_list_versions_snapshot(self, latest_only=False):
     def test_list_versions_latest_only_snapshot(self):
         self.test_list_versions_snapshot(latest_only=True)
 
-    @rule(snap=snaps)
-    def test_batch_read_snapshot(self, snap: str):
-        snap = self._visible_snapshots.get(snap)
-        assume(snap)
-        self._check_batch_read(list(snap.sym_vers), list(snap.sym_vers.values()))
-
 
 @pytest.mark.parametrize("lib_type", ["lmdb_version_store_delayed_deletes_v1", "lmdb_version_store_delayed_deletes_v2"])
 @SLOW_TESTS_MARK
diff --git a/python/tests/integration/arcticdb/version_store/test_basic_version_store.py b/python/tests/integration/arcticdb/version_store/test_basic_version_store.py
index 502b01cb0a..4987a5066e 100644
--- a/python/tests/integration/arcticdb/version_store/test_basic_version_store.py
+++ b/python/tests/integration/arcticdb/version_store/test_basic_version_store.py
@@ -28,7 +28,7 @@
     InternalException,
     UserInputException,
 )
-from arcticdb import QueryBuilder
+from arcticdb import QueryBuilder, ReadRequest
 from arcticdb.flattener import Flattener
 from arcticdb.version_store import NativeVersionStore
 from arcticdb.version_store._custom_normalizers import CustomNormalizer, register_normalizer
@@ -1331,18 +1331,6 @@ def test_batch_operations(object_version_store_prune_previous):
         equals(result["sym3"].data, np.arange(10))
 
 
-def test_batch_read_tombstoned_version_via_snapshot(basic_store):  # AN-285
-    lib = basic_store
-    lib.write("a", 0)
-    lib.snapshot("s")
-    lib.write("a", 1, prune_previous_version=True)
-
-    actual = lib.batch_read(["a"], as_ofs=[0])  # Other version query types are not implemented
-    assert actual["a"].data == 0
-    meta = lib.batch_read_metadata(["a"], as_ofs=[0])
-    assert meta["a"].version == 0
-
-
 def test_batch_write(basic_store_tombstone_and_sync_passive):
     lmdb_version_store = basic_store_tombstone_and_sync_passive
     multi_data = {"sym1": np.arange(8), "sym2": np.arange(9), "sym3": np.arange(10)}
@@ -2236,6 +2224,17 @@ def test_batch_read_version_doesnt_exist(basic_store):
     with pytest.raises(NoDataFoundException):
         _ = basic_store.batch_read([sym1, sym2], as_ofs=[0, 1])
 
+def test_read_batch_deleted_version_doesnt_exist(basic_store):
+    sym1 = 'mysymbol'
+    basic_store.write(sym1, 0)
+
+    basic_store.delete(sym1)
+    basic_store.write(sym1, 1)
+    with pytest.raises(NoSuchVersionException):
+        basic_store.read(sym1, as_of=0)
+
+    with pytest.raises(NoSuchVersionException):
+        basic_store.batch_read([sym1],  as_ofs=[0])
 
 def test_index_keys_start_end_index(basic_store, sym):
     idx = pd.date_range("2022-01-01", periods=100, freq="D")