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

Issue:
read(symbol, as_of) and read_batch([ReadRequest(symbol, as_of)]) behave differently for the same values of symbol and as_of
### Describe the bug

I chose "as_of" to be a datetime *just* after the timestamp of the version 0.

It seems read(symbol, as_of) and read_batch([ReadRequest(symbol, as_of)]) return different results (latter throws when former does not).

I would have expected both of them to return data (i.e. no exception).

### Steps/Code to Reproduce

```python

# lib = existing library with some data for symbol "X"

from arcticdb.version_store.library import ReadRequest
from datetime import datetime, timezone
vi = lib.list_versions("X")

# Let us assume this returns:
# X_v2: (date=2023-06-22 07:38:15.849084990+00:00),
# X_v1: (date=2023-06-22 05:41:31.169344280+00:00),
# X_v0: (date=2023-06-21 22:59:31.163315013+00:00)

as_of = datetime(2023, 6, 21, 22, 59, 32, tzinfo=timezone.utc) # Notice that this is just after the timestamp of v1
vi = lib.read("X", as_of=as_of) # this works fine.

vi = lib.read_batch([ReadRequest("X", as_of=as_of)]) 
# The above call throws an exception:
---------------------------------------------------------------------------
InternalException                         Traceback (most recent call last)
Input In [24], in <cell line: 1>()
----> 1 vi = lib.read_batch([ReadRequest("X", as_of=as_of)])

File /opt/bqplatform/lib/python3.9/site-packages/arcticdb/version_store/library.py:939, in Library.read_batch(self, symbols, query_builder)
    933     else:
    934         raise ArcticInvalidApiUsageException(
    935             f"Unsupported item in the symbols argument s=[{s}] type(s)=[{type(s)}]. Only [str] and"
    936             " [ReadRequest] are supported."
    937         )
--> 939 return self._nvs._batch_read_to_versioned_items(
    940     symbol_strings, as_ofs, date_ranges, columns, query_builder or query_builders
    941 )

File /opt/bqplatform/lib/python3.9/site-packages/arcticdb/version_store/_store.py:976, in NativeVersionStore._batch_read_to_versioned_items(self, symbols, as_ofs, date_ranges, columns, query_builder, kwargs)
    974 read_queries = self._get_read_queries(len(symbols), date_ranges, columns, query_builder)
    975 read_options = self._get_read_options(**kwargs)
--> 976 read_results = self.version_store.batch_read(symbols, version_queries, read_queries, read_options)
    977 versioned_items = []
    978 for i in range(len(read_results)):

InternalException: E_ASSERTION_FAILURE Version not found for stream


### Expected Results

Neither of the two methods should have thrown any exceptions.

### OS, Python Version and ArcticDB Version

Python: 3.9.12 | packaged by conda-forge | (main, Mar 24 2022, 23:22:55) 
[GCC 10.3.0]
OS: Linux-5.10.184-175.731.amzn2.x86_64-x86_64-with-glibc2.17
ArcticDB: 1.5.0

### Backend storage used

AWS S3

### Additional Context

_No response_


Discussion / hints:

Fix patch:
diff --git a/cpp/arcticdb/version/version_map_batch_methods.hpp b/cpp/arcticdb/version/version_map_batch_methods.hpp
index 2a127bc0bd..7421e58512 100644
--- a/cpp/arcticdb/version/version_map_batch_methods.hpp
+++ b/cpp/arcticdb/version/version_map_batch_methods.hpp
@@ -12,7 +12,7 @@
 #include <arcticdb/version/version_tasks.hpp>
 #include <arcticdb/version/version_store_objects.hpp>
 #include <arcticdb/pipeline/query.hpp>
-
+#include <arcticdb/version/version_functions.hpp>
 #include <folly/futures/FutureSplitter.h>
 
 namespace arcticdb {
@@ -192,8 +192,14 @@ inline std::optional<AtomKey> get_key_for_version_query(
         [&version_map_entry] (const pipelines::SpecificVersionQuery& specific_version) {
             return find_index_key_for_version_id(specific_version.version_id_, version_map_entry);
         },
-        [&version_map_entry] (const pipelines::TimestampVersionQuery& timestamp_version) {
-        return find_index_key_for_version_timestamp(timestamp_version.timestamp_, version_map_entry);
+        [&version_map_entry] (const pipelines::TimestampVersionQuery& timestamp_version) -> std::optional<AtomKey> {
+            auto version_key = get_version_key_from_time_for_versions(timestamp_version.timestamp_, version_map_entry->get_indexes(false));
+            if(version_key.has_value()){
+                auto version_id = version_key.value().version_id();
+                return find_index_key_for_version_id(version_id, version_map_entry, false);
+            }else{
+                return std::nullopt;
+            }
         },
         [&version_map_entry] (const std::monostate&) {
         return version_map_entry->get_first_index(false);


Test patch:
diff --git a/python/tests/integration/arcticdb/test_arctic.py b/python/tests/integration/arcticdb/test_arctic.py
index 3aebc263ad..2fd0a93bdc 100644
--- a/python/tests/integration/arcticdb/test_arctic.py
+++ b/python/tests/integration/arcticdb/test_arctic.py
@@ -822,6 +822,52 @@ def test_write_batch_dedup(library_factory):
             assert data_key_version[s] == 0
 
 
+def test_read_batch_time_stamp(arctic_library):
+    """Should be able to read data in batch mode using a timestamp."""
+    lib = arctic_library
+    sym = "sym_"
+    num_versions = 3
+    num_symbols = 3
+    for v_num in range(num_versions):
+        write_requests = []
+        for sym_num in range(num_symbols):
+            write_requests.append(
+                WritePayload(
+                    sym + str(sym_num), pd.DataFrame({"col": [sym_num + v_num, sym_num * v_num, sym_num - v_num]})
+                )
+            )
+        lib.write_batch(write_requests)
+        time.sleep(1)
+
+    data_non_batch = []
+    requests_batch = []
+    original_dataframes = []
+    for sym_num in range(num_symbols):
+        versions = lib.list_versions(sym + str(sym_num))
+        for version_info in versions:
+            date_obj = versions[version_info].date
+            date_obj += +timedelta(seconds=1)
+            as_of = datetime(
+                date_obj.year,
+                date_obj.month,
+                date_obj.day,
+                date_obj.hour,
+                date_obj.minute,
+                date_obj.second,
+                tzinfo=timezone.utc,
+            )
+            data_non_batch.append(lib.read(sym + str(sym_num), as_of=as_of))
+            requests_batch.append(ReadRequest(sym + str(sym_num), as_of=as_of))
+            v_num = version_info.version
+            original_dataframes.append(pd.DataFrame({"col": [sym_num + v_num, sym_num * v_num, sym_num - v_num]}))
+
+    data_batch = lib.read_batch(requests_batch)
+    list_data_to_compate = list(zip(data_non_batch, data_batch, original_dataframes))
+    for d1, d2, d3 in list_data_to_compate:
+        assert_frame_equal(d1.data, d2.data)
+        assert_frame_equal(d2.data, d3)
+
+
 def test_write_with_unpacking(arctic_library):
     """Check the syntactic sugar that lets us unpack WritePayload in `write` calls using *."""
     lib = arctic_library