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

Issue:
Port `batch_read` with snapshots fix to `4.0.x` branch



Discussion / hints:

Fix patch:
diff --git a/cpp/arcticdb/CMakeLists.txt b/cpp/arcticdb/CMakeLists.txt
index 2d50f04f05..6e7ede90d5 100644
--- a/cpp/arcticdb/CMakeLists.txt
+++ b/cpp/arcticdb/CMakeLists.txt
@@ -428,7 +428,9 @@ set(arcticdb_srcs
         version/version_core.cpp
         version/version_store_api.cpp
         version/version_utils.cpp
-        version/symbol_list.cpp)
+        version/symbol_list.cpp
+        version/version_map_batch_methods.cpp
+        )
 
 # Exclude vendored sources when using conda.
 if(NOT ${ARCTICDB_USING_CONDA})
diff --git a/cpp/arcticdb/async/task_scheduler.hpp b/cpp/arcticdb/async/task_scheduler.hpp
index 0cea307e45..acfecd1c57 100644
--- a/cpp/arcticdb/async/task_scheduler.hpp
+++ b/cpp/arcticdb/async/task_scheduler.hpp
@@ -47,11 +47,11 @@ struct TaskSchedulerPtrWrapper{
         ptr_ = ptr;
     }
 
-    TaskScheduler* operator->() {
+    TaskScheduler* operator->() const {
         return ptr_;
     }
 
-    TaskScheduler& operator*() {
+    TaskScheduler& operator*() const {
         return *ptr_;
     }
 };
@@ -61,6 +61,7 @@ class InstrumentedNamedFactory : public folly::ThreadFactory{
     explicit InstrumentedNamedFactory(folly::StringPiece prefix) : named_factory_(prefix){}
 
     std::thread newThread(folly::Func&& func) override {
+        std::lock_guard lock{mutex_};
         return named_factory_.newThread(
                 [func = std::move(func)]() mutable {
                 ARCTICDB_SAMPLE_THREAD();
@@ -76,6 +77,7 @@ class InstrumentedNamedFactory : public folly::ThreadFactory{
 #endif
 
 private:
+    std::mutex mutex_;
     folly::NamedThreadFactory named_factory_;
 };
 
@@ -102,8 +104,7 @@ struct SchedulerWrapper : public SchedulerType {
 };
 
 inline int64_t get_cgroup_value(const std::string& cgroup_file) {
-    const auto path = std::filesystem::path{fmt::format("/sys/fs/cgroup/{}",cgroup_file)};
-    if(std::filesystem::exists(path)){
+    if(const auto path = std::filesystem::path{fmt::format("/sys/fs/cgroup/{}",cgroup_file)}; std::filesystem::exists(path)){
         std::ifstream strm(path.string());
         util::check(static_cast<bool>(strm), "Failed to open cgroups cpu file for read at path '{}': {}", path.string(), std::strerror(errno));
         std::string str;
@@ -118,20 +119,14 @@ inline auto get_default_num_cpus() {
     #ifdef _WIN32
         return static_cast<int64_t>(cpu_count);
     #else
-        auto quota_count = 0;
-        auto limit_count = 0;
+        auto quota_count = 0UL;
         auto quota = get_cgroup_value("cpu/cpu.cfs_quota_us");
         auto period = get_cgroup_value("cpu/cpu.cfs_period_us");
 
-        if (quota > -1 && period > 0) {
-            quota_count = ceil(static_cast<double>(quota) / static_cast<double>(period));
-        }
+        if (quota > -1 && period > 0)
+            quota_count = static_cast<int64_t>(ceil(static_cast<double>(quota) / static_cast<double>(period)));
 
-        if (quota_count != 0) {
-            limit_count = quota_count;
-        } else {
-            limit_count = cpu_count;
-        }
+        auto limit_count = quota_count != 0 ? quota_count : cpu_count;
         return std::min(static_cast<int64_t>(cpu_count), static_cast<int64_t>(limit_count));
     #endif
 }
@@ -146,6 +141,10 @@ inline auto get_default_num_cpus() {
  * 4/ Throttling: (similar to priority) how to absorb work spikes and apply memory backpressure
  */
 
+inline int64_t default_io_thread_count(uint64_t cpu_count) {
+    return std::min(int64_t(100L), static_cast<int64_t>(static_cast<double>(cpu_count) * 1.5));
+}
+
 class TaskScheduler {
   public:
     using CPUSchedulerType = folly::FutureExecutor<folly::CPUThreadPoolExecutor>;
@@ -156,22 +155,27 @@ class TaskScheduler {
         io_thread_count_(io_thread_count ? io_thread_count.value() : ConfigsMap::instance()->get_int("VersionStore.NumIOThreads", std::min(100, (int) (cpu_thread_count_ * 1.5)))),
         cpu_exec_(cpu_thread_count_, std::make_shared<InstrumentedNamedFactory>("CPUPool")) ,
         io_exec_(io_thread_count_,  std::make_shared<InstrumentedNamedFactory>("IOPool")){
+        util::check(cpu_thread_count_ > 0 && io_thread_count_ > 0, "Zero IO or CPU threads: {} {}", io_thread_count_, cpu_thread_count_);
         ARCTICDB_RUNTIME_DEBUG(log::schedule(), "Task scheduler created with {:d} {:d}", cpu_thread_count_, io_thread_count_);
     }
 
     ~TaskScheduler() = default;
 
     template<class Task>
-    auto submit_cpu_task(Task &&task) {
+    auto submit_cpu_task(Task &&t) {
+        auto task = std::forward<decltype(t)>(t);
         static_assert(std::is_base_of_v<BaseTask, std::decay_t<Task>>, "Only supports Task derived from BaseTask");
         ARCTICDB_DEBUG(log::schedule(), "{} Submitting CPU task {}: {} of {}", uintptr_t(this), typeid(task).name(), cpu_exec_.getTaskQueueSize(), cpu_exec_.kDefaultMaxQueueSize);
+        std::lock_guard lock{cpu_mutex_};
         return cpu_exec_.addFuture(std::move(task));
     }
 
     template<class Task>
-    auto submit_io_task(Task &&task) {
+    auto submit_io_task(Task &&t) {
+        auto task = std::forward<decltype(t)>(t);
         static_assert(std::is_base_of_v<BaseTask, std::decay_t<Task>>, "Only support Tasks derived from BaseTask");
         ARCTICDB_DEBUG(log::schedule(), "{} Submitting IO task {}: {}", uintptr_t(this), typeid(task).name(), io_exec_.getPendingTaskCount());
+        std::lock_guard lock{io_mutex_};
         return io_exec_.addFuture(std::move(task));
     }
 
@@ -248,6 +252,8 @@ class TaskScheduler {
     size_t io_thread_count_;
     SchedulerWrapper<CPUSchedulerType> cpu_exec_;
     SchedulerWrapper<IOSchedulerType> io_exec_;
+    std::mutex cpu_mutex_;
+    std::mutex io_mutex_;
 };
 
 
@@ -261,40 +267,13 @@ inline auto& io_executor() {
 
 template <typename Task>
 inline auto submit_cpu_task(Task&& task) {
-    return TaskScheduler::instance()->submit_cpu_task(std::move(task));
+    return TaskScheduler::instance()->submit_cpu_task(std::forward<decltype(task)>(task));
 }
 
 
 template <typename Task>
 inline auto submit_io_task(Task&& task) {
-    return TaskScheduler::instance()->submit_io_task(std::move(task));
-}
-
-template <typename Inputs, typename TaskSubmitter, typename ResultHandler>
-inline void submit_tasks_for_range(const Inputs& inputs, TaskSubmitter submitter, ResultHandler result_handler) {
-    using TaskReturn = decltype(submitter(std::declval<std::add_const_t<std::add_lvalue_reference_t<typename Inputs::value_type>>>()));
-
-    std::vector<TaskReturn> futs;
-    futs.reserve(inputs.size());
-    try {
-        for (const auto& input : inputs) {
-            futs.emplace_back(submitter(input));
-        }
-    } catch(...) { // Clean up the Futures that have already been submitted
-        folly::collectAll(futs).wait();
-        throw;
-    }
-
-    auto fut_itr = futs.begin();
-    try {
-        for(const auto& input : inputs) {
-            auto&& resolved = std::move(*(fut_itr++)).get();
-            result_handler(input, std::move(resolved));
-        }
-    } catch(...) {
-        folly::collectAll(fut_itr, futs.end()).wait();
-        throw;
-    }
+    return TaskScheduler::instance()->submit_io_task(std::forward<decltype(task)>(task));
 }
 
 void print_scheduler_stats();
diff --git a/cpp/arcticdb/column_store/string_pool.hpp b/cpp/arcticdb/column_store/string_pool.hpp
index 8469d1449f..61517671e6 100644
--- a/cpp/arcticdb/column_store/string_pool.hpp
+++ b/cpp/arcticdb/column_store/string_pool.hpp
@@ -112,6 +112,8 @@ class StringBlock {
 
     [[nodiscard]] const StringHead* const_head_at(position_t pos) const {
         auto data = data_.buffer().internal_ptr_cast<uint8_t>(pos, sizeof(StringHead));
+        auto head = reinterpret_cast<const StringHead *>(data);
+        data_.buffer().assert_size(pos + StringHead::calc_size(head->size()));
         return reinterpret_cast<const StringHead *>(data);
     }
 
diff --git a/cpp/arcticdb/entity/data_error.hpp b/cpp/arcticdb/entity/data_error.hpp
index 5b94e3f09b..c4a6dcd62d 100644
--- a/cpp/arcticdb/entity/data_error.hpp
+++ b/cpp/arcticdb/entity/data_error.hpp
@@ -24,12 +24,12 @@ enum class VersionRequestType: uint32_t {
 
 class DataError {
 public:
-    DataError(const StreamId& symbol,
+    DataError(StreamId symbol,
               std::string&& exception_string,
               const std::optional<pipelines::VersionQueryType>& version_query_type=std::nullopt,
               std::optional<ErrorCode> error_code=std::nullopt) :
-        symbol_(symbol),
-        exception_string_(exception_string),
+        symbol_(std::move(symbol)),
+        exception_string_(std::move(exception_string)),
         error_code_(error_code){
         if (version_query_type.has_value()){
             util::variant_match(
@@ -69,7 +69,7 @@ class DataError {
         return version_request_type_;
     }
 
-    std::optional<std::variant<std::monostate, int64_t, std::string>> version_request_data() const {
+    std::optional<std::variant<std::monostate, int64_t, SnapshotId>> version_request_data() const {
         return version_request_data_;
     }
 
@@ -89,22 +89,25 @@ class DataError {
         std::string version_request_explanation = "UNKNOWN";
         if (version_request_type_.has_value()) {
             switch (*version_request_type_) {
-                case VersionRequestType::SNAPSHOT:
-                    version_request_explanation = fmt::format("in snapshot '{}'", std::get<std::string>(*version_request_data_));
-                    break;
-                case VersionRequestType::TIMESTAMP:
-                    version_request_explanation = fmt::format("at time '{}'", std::get<int64_t>(*version_request_data_));
-                    break;
-                case VersionRequestType::SPECIFIC:
-                    version_request_explanation = fmt::format("with specified version '{}'", std::get<int64_t>(*version_request_data_));
-                    break;
-                case VersionRequestType::LATEST:
-                    version_request_explanation = fmt::format("with specified version 'latest'");
-                    break;
-                default:
-                    internal::raise<ErrorCode::E_ASSERTION_FAILURE>(
-                            "Unexpected enum value in DataError::to_string: {}",
-                            static_cast<uint32_t>(*version_request_type_));
+            case VersionRequestType::SNAPSHOT:
+                version_request_explanation = fmt::format("in snapshot '{}'",
+                                                          std::get<SnapshotId>(*version_request_data_));
+                break;
+            case VersionRequestType::TIMESTAMP:
+                version_request_explanation = fmt::format("at time '{}'",
+                                                          std::get<int64_t>(*version_request_data_));
+                break;
+            case VersionRequestType::SPECIFIC:
+                version_request_explanation = fmt::format("with specified version '{}'",
+                                                          std::get<int64_t>(*version_request_data_));
+                break;
+            case VersionRequestType::LATEST:
+                version_request_explanation = fmt::format("with specified version 'latest'");
+                break;
+            default:
+                internal::raise<ErrorCode::E_ASSERTION_FAILURE>(
+                    "Unexpected enum value in DataError::to_string: {}",
+                    static_cast<uint32_t>(*version_request_type_));
             }
         }
         return fmt::format(
@@ -118,7 +121,7 @@ class DataError {
     StreamId symbol_;
     std::optional<VersionRequestType> version_request_type_;
     // int64_t for timestamp and SignedVersionId
-    std::optional<std::variant<std::monostate, int64_t, std::string>> version_request_data_;
+    std::optional<std::variant<std::monostate, int64_t, SnapshotId>> version_request_data_;
     std::string exception_string_;
     std::optional<ErrorCode> error_code_;
 };
diff --git a/cpp/arcticdb/entity/types.hpp b/cpp/arcticdb/entity/types.hpp
index 9818f6cd4c..7573ec2245 100644
--- a/cpp/arcticdb/entity/types.hpp
+++ b/cpp/arcticdb/entity/types.hpp
@@ -124,7 +124,7 @@ enum class ValueType : uint8_t {
     // For instance, Pandas empty series whose types has not been specified is mapping to EMPTY.
     // When data is appended, the column type is inferred from the data and the column is promoted to the inferred type.
     EMPTY = 13,
-    COUNT // Not a real value type, should not be added to proto descriptor. Used to count the number of items in the enum
+    COUNT = 14 // Not a real value type, should not be added to proto descriptor. Used to count the number of items in the enum
 };
 
 // Sequence types are composed of more than one element
@@ -172,7 +172,7 @@ enum class SizeBits : uint8_t {
     S16 = 2,
     S32 = 3,
     S64 = 4,
-    COUNT
+    COUNT = 5
 };
 
 constexpr SizeBits get_size_bits(uint8_t size) {
@@ -193,7 +193,6 @@ constexpr uint8_t combine_val_bits(ValueType v, SizeBits b = SizeBits::UNKNOWN_S
 } // namespace anonymous
 
 enum class DataType : uint8_t {
-#define DT_COMBINE(TYPE, SIZE) TYPE = combine_val_bits(ValueType::TYPE, SizeBits::SIZE)
     UINT8 = detail::combine_val_bits(ValueType::UINT, SizeBits::S8),
     UINT16 = detail::combine_val_bits(ValueType::UINT, SizeBits::S16),
     UINT32 = detail::combine_val_bits(ValueType::UINT, SizeBits::S32),
@@ -211,7 +210,6 @@ enum class DataType : uint8_t {
     UTF_FIXED64 = detail::combine_val_bits(ValueType::UTF8_FIXED, SizeBits::S64),
     UTF_DYNAMIC64 = detail::combine_val_bits(ValueType::UTF_DYNAMIC, SizeBits::S64),
     EMPTYVAL = detail::combine_val_bits(ValueType::EMPTY, SizeBits::S64),
-#undef DT_COMBINE
     UNKNOWN = 0,
 };
 
@@ -301,7 +299,7 @@ constexpr bool is_empty_type(DataType v){
     return is_empty_type(slice_value_type(v));
 }
 
-static_assert(slice_value_type((DataType::UINT16)) == ValueType(1));
+static_assert(slice_value_type(DataType::UINT16) == ValueType(1));
 static_assert(get_type_size(DataType::UINT32) == 4);
 static_assert(get_type_size(DataType::UINT64) == 8);
 
@@ -514,8 +512,8 @@ inline TypeDescriptor type_desc_from_proto(const arcticdb::proto::descriptors::T
     };
 }
 
-inline DataType data_type_from_proto(const arcticdb::proto::descriptors::TypeDescriptor type_desc) {
-    return type_desc_from_proto((type_desc)).data_type();
+inline DataType data_type_from_proto(const arcticdb::proto::descriptors::TypeDescriptor& type_desc) {
+    return type_desc_from_proto(type_desc).data_type();
 }
 
 template <typename DTT>
@@ -536,31 +534,7 @@ inline arcticdb::proto::descriptors::StreamDescriptor_FieldDescriptor field_prot
 
     return output;
 }
-/*
-inline auto scalar_field (DataType dt, std::string_view name) {
-    return field_proto<Dimension::Dim0>(dt, name);
-}
-
-inline auto scalar_field(TypeDescriptor::Proto&& proto, std::string_view name = std::string_view()) {
-    using namespace pb2::descriptors_pb2;
-    StreamDescriptor_FieldDescriptor output;
-    if(!name.empty())
-        output.set_name(name.data(), name.size());
-
-    *output.mutable_type_desc() = std::move(proto);
-    return output;
-}
-
-inline auto scalar_field_proto(const TypeDescriptor::Proto& proto, std::string_view name = std::string_view()) {
-    using namespace pb2::descriptors_pb2;
-    StreamDescriptor_FieldDescriptor output;
-    if(!name.empty())
-        output.set_name(name.data(), name.size());
 
-    output.mutable_type_desc()->CopyFrom(proto);
-    return output;
-}
-*/
 struct IndexDescriptor {
         using Proto = arcticdb::proto::descriptors::IndexDescriptor;
 
@@ -580,8 +554,8 @@ struct IndexDescriptor {
         data_.set_field_count(static_cast<uint32_t>(field_count));
     }
 
-    explicit IndexDescriptor(const arcticdb::proto::descriptors::IndexDescriptor& data)
-        : data_(data) {
+    explicit IndexDescriptor(arcticdb::proto::descriptors::IndexDescriptor data)
+        : data_(std::move(data)) {
     }
 
     bool uninitialized() const {
@@ -617,7 +591,7 @@ constexpr IndexDescriptor::TypeChar to_type_char(IndexDescriptor::Type type) {
     case IndexDescriptor::ROWCOUNT:return 'R';
     case IndexDescriptor::STRING:return 'S';
     case IndexDescriptor::UNKNOWN:return 'U';
-    default:util::raise_rte("Unknown index type: {}", type);
+    default:util::raise_rte("Unknown index type: {}", int(type));
     }
 }
 
@@ -627,7 +601,7 @@ constexpr IndexDescriptor::Type from_type_char(IndexDescriptor::TypeChar type) {
     case 'R': return IndexDescriptor::ROWCOUNT;
     case 'S': return IndexDescriptor::STRING;
     case 'U': return IndexDescriptor::UNKNOWN;
-    default:util::raise_rte("Unknown index type: {}", type);
+    default:util::raise_rte("Unknown index type: {}", int(type));
     }
 }
 
@@ -635,11 +609,11 @@ struct FieldRef {
     TypeDescriptor type_;
     std::string_view name_;
 
-    TypeDescriptor type() const {
+    [[nodiscard]] TypeDescriptor type() const {
         return type_;
     }
 
-    std::string_view name() const {
+    [[nodiscard]] std::string_view name() const {
         return name_;
     }
 
@@ -671,7 +645,7 @@ struct Field {
     using Proto = arcticdb::proto::descriptors::StreamDescriptor_FieldDescriptor;
 
 private:
-    Field(const FieldRef& ref) {
+    explicit Field(const FieldRef& ref) {
         set(ref.type_, ref.name_);
     }
 
@@ -683,19 +657,15 @@ struct Field {
         new (ptr) Field(type, name);
     }
 
-    static Field from_ref(const FieldRef& ref) {
-        return Field(ref);
-    }
-
     static size_t calc_size(std::string_view name) {
       return sizeof(type_) + sizeof(size_) + std::max(NameSize, name.size());
     }
 
-    std::string_view name() const {
+    [[nodiscard]] std::string_view name() const {
         return {name_, size_};
     }
 
-    const TypeDescriptor& type() const {
+    [[nodiscard]] const TypeDescriptor& type() const {
         return type_;
     }
 
@@ -703,7 +673,7 @@ struct Field {
         return type_;
     }
 
-    FieldRef ref() const {
+    [[nodiscard]] FieldRef ref() const {
         return {type_, name()};
     }
 
@@ -786,8 +756,6 @@ struct less<arcticdb::entity::StreamId> {
 };
 }
 
-
-
 namespace fmt {
 
 using namespace arcticdb::entity;
@@ -799,10 +767,8 @@ struct formatter<FieldRef> {
     constexpr auto parse(ParseContext &ctx) { return ctx.begin(); }
 
     template<typename FormatContext>
-    auto format(const FieldRef f, FormatContext &ctx) const {
-        auto out = ctx.out();
+    auto format(const FieldRef& f, FormatContext &ctx) const {
         return format_to(ctx.out(), "{}: {}", f.type_, f.name_);
-        return out;
     }
 };
 
diff --git a/cpp/arcticdb/log/log.cpp b/cpp/arcticdb/log/log.cpp
index 1867f1cfbc..0a5a9b740e 100644
--- a/cpp/arcticdb/log/log.cpp
+++ b/cpp/arcticdb/log/log.cpp
@@ -65,6 +65,14 @@ spdlog::logger &message() {
     return Loggers::instance()->message();
 }
 
+spdlog::logger &symbol() {
+    return Loggers::instance()->symbol();
+}
+
+spdlog::logger &snapshot() {
+    return Loggers::instance()->snapshot();
+}
+
 namespace fs = std::filesystem;
 
 using SinkConf = arcticdb::proto::logger::SinkConfig;
@@ -117,6 +125,14 @@ spdlog::logger &Loggers::message() {
     return logger_ref(message_);
 }
 
+spdlog::logger &Loggers::symbol() {
+    return logger_ref(symbol_);
+}
+
+spdlog::logger &Loggers::snapshot() {
+    return logger_ref(snapshot_);
+}
+
 spdlog::logger &Loggers::root() {
     return logger_ref(root_);
 }
@@ -132,6 +148,8 @@ void Loggers::flush_all() {
     lock().flush();
     schedule().flush();
     message().flush();
+    symbol().flush();
+    snapshot().flush();
 }
 
 
@@ -245,6 +263,9 @@ bool Loggers::configure(const arcticdb::proto::logger::LoggersConfig &conf, bool
     check_and_configure("lock", "root", lock_);
     check_and_configure("schedule", "root", schedule_);
     check_and_configure("message", "root", message_);
+    check_and_configure("symbol", "root", symbol_);
+    check_and_configure("snapshot", "root", snapshot_);
+
 
     auto flush_sec = util::as_opt(conf.flush_interval_seconds()).value_or(1);
     if (flush_sec != 0) {
diff --git a/cpp/arcticdb/log/log.hpp b/cpp/arcticdb/log/log.hpp
index 7698556efa..da8cffd26a 100644
--- a/cpp/arcticdb/log/log.hpp
+++ b/cpp/arcticdb/log/log.hpp
@@ -58,6 +58,8 @@ class Loggers : public std::enable_shared_from_this<Loggers> {
     spdlog::logger &lock();
     spdlog::logger &schedule();
     spdlog::logger &message();
+    spdlog::logger &symbol();
+    spdlog::logger &snapshot();
 
     void flush_all();
 
@@ -82,6 +84,8 @@ class Loggers : public std::enable_shared_from_this<Loggers> {
     std::unique_ptr<spdlog::logger> lock_;
     std::unique_ptr<spdlog::logger> schedule_;
     std::unique_ptr<spdlog::logger> message_;
+    std::unique_ptr<spdlog::logger> symbol_;
+    std::unique_ptr<spdlog::logger> snapshot_;
     std::shared_ptr<spdlog::details::thread_pool> thread_pool_;
     std::unique_ptr<spdlog::details::periodic_worker> periodic_worker_;
 };
@@ -98,6 +102,8 @@ spdlog::logger &timings();
 spdlog::logger &lock();
 spdlog::logger &schedule();
 spdlog::logger &message();
+spdlog::logger &symbol();
+spdlog::logger &snapshot();
 
 inline std::unordered_map<std::string, spdlog::logger*> get_loggers_by_name() {
     return {
@@ -110,7 +116,9 @@ inline std::unordered_map<std::string, spdlog::logger*> get_loggers_by_name() {
         {"timings", &timings()},
         {"lock", &lock()},
         {"schedule", &schedule()},
-        {"message", &message()}
+        {"message", &message()},
+        {"symbol", &symbol()},
+        {"snapshot", &snapshot()}
     };
 }
 
diff --git a/cpp/arcticdb/pipeline/input_tensor_frame.hpp b/cpp/arcticdb/pipeline/input_tensor_frame.hpp
index 78db481054..e706699148 100644
--- a/cpp/arcticdb/pipeline/input_tensor_frame.hpp
+++ b/cpp/arcticdb/pipeline/input_tensor_frame.hpp
@@ -62,12 +62,16 @@ struct InputTensorFrame {
     void set_index_range() {
             // Fill index range
             // Note RowCountIndex will normally have an index field count of 0
-            if (desc.index().field_count() == 1) {
+            if(num_rows == 0) {
+                index_range.start_  = IndexValue{0};
+                index_range.end_ = IndexValue{0};
+            } else if (desc.index().field_count() == 1) {
                 visit_field(desc.field(0), [&](auto &&tag) {
                     using DT = std::decay_t<decltype(tag)>;
                     using RawType = typename DT::DataTypeTag::raw_type;
                     if constexpr (std::is_integral_v<RawType> || std::is_floating_point_v<RawType>) {
                         util::check(static_cast<bool>(index_tensor), "Got null index tensor in set_index_range");
+                        util::check(index_tensor->nbytes() > 0, "Empty index tensor");
                         auto &tensor = index_tensor.value();
                         auto start_t = tensor.ptr_cast<RawType>(0);
                         auto end_t = tensor.ptr_cast<RawType>(static_cast<size_t>(tensor.shape(0) - 1));
diff --git a/cpp/arcticdb/pipeline/query.hpp b/cpp/arcticdb/pipeline/query.hpp
index f86d6a8a0c..4997851545 100644
--- a/cpp/arcticdb/pipeline/query.hpp
+++ b/cpp/arcticdb/pipeline/query.hpp
@@ -69,7 +69,7 @@ struct ReadQuery {
 };
 
 struct SnapshotVersionQuery {
-    std::string name_;
+    SnapshotId name_;
 };
 
 struct TimestampVersionQuery {
diff --git a/cpp/arcticdb/pipeline/read_frame.cpp b/cpp/arcticdb/pipeline/read_frame.cpp
index b2bf926e02..a037629ea9 100644
--- a/cpp/arcticdb/pipeline/read_frame.cpp
+++ b/cpp/arcticdb/pipeline/read_frame.cpp
@@ -739,7 +739,12 @@ class DynamicStringReducer : public StringReducer {
 
     struct UnicodeFromUnicodeCreator {
         static PyObject* create(std::string_view sv, bool) {
-            const auto actual_length = std::min(sv.size() / UNICODE_WIDTH, wcslen(reinterpret_cast<const wchar_t *>(sv.data())));
+            const auto size = sv.size() + 4;
+            auto* buffer = reinterpret_cast<char*>(alloca(size));
+            memset(buffer, 0, size);
+            memcpy(buffer, sv.data(), sv.size());
+
+            const auto actual_length = std::min(sv.size() / UNICODE_WIDTH, wcslen(reinterpret_cast<const wchar_t *>(buffer)));
             return PyUnicode_FromKindAndData(PyUnicode_4BYTE_KIND, reinterpret_cast<const UnicodeType*>(sv.data()), actual_length);
         }
     };
diff --git a/cpp/arcticdb/pipeline/string_pool_utils.cpp b/cpp/arcticdb/pipeline/string_pool_utils.cpp
index 09153229d0..ffbe509af7 100644
--- a/cpp/arcticdb/pipeline/string_pool_utils.cpp
+++ b/cpp/arcticdb/pipeline/string_pool_utils.cpp
@@ -13,10 +13,4 @@ size_t first_context_row(const pipelines::SliceAndKey& slice_and_key, size_t fir
     return slice_and_key.slice_.row_range.first - first_row_in_frame;
 }
 
-position_t get_offset_string(const pipelines::PipelineContextRow& context_row, ChunkedBuffer &src, std::size_t first_row_in_frame) {
-    auto offset = first_context_row(context_row.slice_and_key(), first_row_in_frame);
-    auto offset_val = get_offset_string_at(offset, src);
-    util::check(offset_val != nan_placeholder() && offset_val != not_a_string(), "NaN or None placeholder in get_offset_string");
-    return offset_val;
-}
 }
\ No newline at end of file
diff --git a/cpp/arcticdb/processing/operation_dispatch_binary.hpp b/cpp/arcticdb/processing/operation_dispatch_binary.hpp
index 2aad279ece..9580ebe8c1 100644
--- a/cpp/arcticdb/processing/operation_dispatch_binary.hpp
+++ b/cpp/arcticdb/processing/operation_dispatch_binary.hpp
@@ -317,19 +317,19 @@ VariantData binary_comparator(const ColumnWithStrings& column_with_strings, cons
             using DataTypeTag =  typename decltype(value_desc_tag)::DataTypeTag;
             if constexpr(is_sequence_type(ColumnTagType::data_type) && is_sequence_type(DataTypeTag::data_type)) {
                 std::optional<std::string> utf32_string;
-                std::optional<std::string_view> value_string;
+                std::string value_string;
                 if constexpr(is_fixed_string_type(ColumnTagType::data_type)) {
                     auto width = column_with_strings.get_fixed_width_string_size();
                     if (width.has_value()) {
                         utf32_string = ascii_to_padded_utf32(std::string_view(*val.str_data(), val.len()), *width);
                         if (utf32_string.has_value()) {
-                            value_string = std::string_view(*utf32_string);
+                            value_string = *utf32_string;
                         }
                     }
                 } else {
-                    value_string = std::string_view(*val.str_data(), val.len());
+                    value_string = std::string(*val.str_data(), val.len());
                 }
-                auto value_offset = column_with_strings.string_pool_->get_offset_for_column(*value_string, *column_with_strings.column_);
+                auto value_offset = column_with_strings.string_pool_->get_offset_for_column(value_string, *column_with_strings.column_);
                 auto column_data = column_with_strings.column_->data();
 
                 util::BitSet::bulk_insert_iterator inserter(*output);
diff --git a/cpp/arcticdb/python/python_module.cpp b/cpp/arcticdb/python/python_module.cpp
index 44b48d69c2..b539c5ebb0 100644
--- a/cpp/arcticdb/python/python_module.cpp
+++ b/cpp/arcticdb/python/python_module.cpp
@@ -44,7 +44,9 @@ enum class LoggerId {
     MEMORY,
     TIMINGS,
     LOCK,
-    SCHEDULE
+    SCHEDULE,
+    SYMBOL,
+    SNAPSHOT
 };
 
 void initialize_folly() {
@@ -77,6 +79,8 @@ void register_log(py::module && log) {
              .value("TIMINGS", LoggerId::TIMINGS)
              .value("LOCK", LoggerId::LOCK)
              .value("SCHEDULE", LoggerId::SCHEDULE)
+             .value("SYMBOL", LoggerId::SYMBOL)
+             .value("SNAPSHOT", LoggerId::SNAPSHOT)
              .export_values()
     ;
     auto choose_logger = [&](LoggerId log_id) -> decltype(arcticdb::log::storage()) /* logger ref */{
@@ -99,6 +103,10 @@ void register_log(py::module && log) {
                 return arcticdb::log::lock();
             case LoggerId::SCHEDULE:
                 return arcticdb::log::schedule();
+            case LoggerId::SYMBOL:
+                return arcticdb::log::symbol();
+            case LoggerId::SNAPSHOT:
+                return arcticdb::log::snapshot();
             default:
                 arcticdb::util::raise_rte("Unsupported logger id");
         }
diff --git a/cpp/arcticdb/python/python_to_tensor_frame.cpp b/cpp/arcticdb/python/python_to_tensor_frame.cpp
index 4cbc287c6c..dc1acae214 100644
--- a/cpp/arcticdb/python/python_to_tensor_frame.cpp
+++ b/cpp/arcticdb/python/python_to_tensor_frame.cpp
@@ -107,7 +107,8 @@ NativeTensor obj_to_tensor(PyObject *ptr) {
                 all_nans = true;
                 util::check(c_style, "Non contiguous columns with first element as None not supported yet.");
                 PyObject** current_object = obj;
-                while(current_object < obj + size) {
+                const auto* end = obj + size;
+                while(current_object < end) {
                     if(*current_object == none.ptr()) {
                         all_nans = false;
                     } else if(is_py_nan(*current_object)) {
@@ -119,7 +120,8 @@ NativeTensor obj_to_tensor(PyObject *ptr) {
                     }
                     ++current_object;
                 }
-                sample = *current_object;
+                if(current_object != end)
+                    sample = *current_object;
             }
             if (empty) {
                 val_type = ValueType::EMPTY;
diff --git a/cpp/arcticdb/storage/protobuf_mappings.hpp b/cpp/arcticdb/storage/protobuf_mappings.hpp
index 25d898170c..bb4972723b 100644
--- a/cpp/arcticdb/storage/protobuf_mappings.hpp
+++ b/cpp/arcticdb/storage/protobuf_mappings.hpp
@@ -63,7 +63,7 @@ inline storage::LibraryDescriptor decode_library_descriptor(const arcticdb::prot
             case arcticdb::proto::storage::LibraryDescriptor::StoreTypeCase::STORE_TYPE_NOT_SET:
             // nothing to do, the variant is a monostate (empty struct) by default
             break;
-        default: util::raise_rte("Unsupported store config type {}", protobuf_descriptor.store_type_case());
+        default: util::raise_rte("Unsupported store config type {}", int(protobuf_descriptor.store_type_case()));
     }
 
     return output;
diff --git a/cpp/arcticdb/stream/index.hpp b/cpp/arcticdb/stream/index.hpp
index ac111f25ec..9ed4fc764f 100644
--- a/cpp/arcticdb/stream/index.hpp
+++ b/cpp/arcticdb/stream/index.hpp
@@ -306,7 +306,7 @@ inline Index index_type_from_descriptor(const StreamDescriptor &desc) {
         return TableIndex::make_from_descriptor(desc);
     case IndexDescriptor::ROWCOUNT:
         return RowCountIndex{};
-    default:util::raise_rte("Data obtained from storage refers to an index type that this build of ArcticDB doesn't understand ({}).", desc.index().proto().kind());
+    default:util::raise_rte("Data obtained from storage refers to an index type that this build of ArcticDB doesn't understand ({}).", int(desc.index().proto().kind()));
     }
 }
 
@@ -319,7 +319,7 @@ inline Index default_index_type_from_descriptor(const IndexDescriptor::Proto &de
     case IndexDescriptor::ROWCOUNT:
         return RowCountIndex::default_index();
     default:
-        util::raise_rte("Unknown index type {} trying to generate index type", desc.kind());
+        util::raise_rte("Unknown index type {} trying to generate index type", int(desc.kind()));
     }
 }
 
@@ -333,7 +333,7 @@ inline Index variant_index_from_type(const IndexDescriptor::Type &type) {
     case IndexDescriptor::ROWCOUNT:
         return RowCountIndex{};
     default:
-        util::raise_rte("Unknown index type {} trying to generate index type", type);
+        util::raise_rte("Unknown index type {} trying to generate index type", int(type));
     }
 }
 
diff --git a/cpp/arcticdb/toolbox/library_tool.cpp b/cpp/arcticdb/toolbox/library_tool.cpp
index e4f866a39d..97870ecc6b 100644
--- a/cpp/arcticdb/toolbox/library_tool.cpp
+++ b/cpp/arcticdb/toolbox/library_tool.cpp
@@ -24,6 +24,7 @@ using namespace arcticdb::entity;
 Segment LibraryTool::read_to_segment(const VariantKey& key) {
     auto kv = std::visit([lib=lib_](const auto &k) { return lib->read(k); }, key);
     util::check(kv.has_segment(), "Failed to read key: {}", key);
+    kv.segment().force_own_buffer();
     return kv.segment();
 }
 
diff --git a/cpp/arcticdb/version/snapshot.cpp b/cpp/arcticdb/version/snapshot.cpp
index 2b4df11513..d73965476c 100644
--- a/cpp/arcticdb/version/snapshot.cpp
+++ b/cpp/arcticdb/version/snapshot.cpp
@@ -9,6 +9,8 @@
 #include <arcticdb/storage/storage.hpp>
 #include <arcticdb/version/version_log.hpp>
 
+#include <algorithm>
+
 using namespace arcticdb::entity;
 using namespace arcticdb::stream;
 
@@ -23,16 +25,18 @@ void write_snapshot_entry(
         KeyType key_type
 ) {
     ARCTICDB_SAMPLE(WriteJournalEntry, 0)
-    ARCTICDB_RUNTIME_DEBUG(log::version(), "Command: write snapshot entry");
-    IndexAggregator <RowCountIndex> snapshot_agg(snapshot_id, [&](auto &&segment) {
+    ARCTICDB_RUNTIME_DEBUG(log::snapshot(), "Command: write snapshot entry");
+    IndexAggregator <RowCountIndex> snapshot_agg(snapshot_id, [&store, key_type, &snapshot_id](SegmentInMemory&& segment) {
         store->write(key_type, snapshot_id, std::move(segment)).get();
     });
 
+    ARCTICDB_DEBUG(log::snapshot(), "Constructing snapshot {}", snapshot_id);
     // Most of the searches in snapshot are for a given symbol, this helps us do a binary search on the segment
     // on read time.
     std::sort(keys.begin(), keys.end(), [](const AtomKey &l, const AtomKey &r) {return l.id() < r.id(); });
 
     for (const auto &key: keys) {
+        ARCTICDB_DEBUG(log::snapshot(), "Adding key {}", key);
         snapshot_agg.add_key(key);
     }
     // Serialize and store the python metadata in the journal entry for snapshot.
@@ -51,7 +55,7 @@ void write_snapshot_entry(
 }
 
 void tombstone_snapshot(
-    std::shared_ptr<StreamSink> store,
+    const std::shared_ptr<StreamSink>& store,
     const RefKey& key,
     SegmentInMemory&& segment_in_memory,
     bool log_changes
@@ -66,7 +70,7 @@ void tombstone_snapshot(
 }
 
 void tombstone_snapshot(
-        std::shared_ptr<StreamSink> store,
+        const std::shared_ptr<StreamSink>& store,
         storage::KeySegmentPair&& key_segment_pair,
         bool log_changes
         ) {
@@ -80,39 +84,23 @@ void tombstone_snapshot(
     store->write_compressed(std::move(key_segment_pair)).get();
 }
 
-// Returns true if snapshot with given key is successfully tombstoned, or false if the snapshot with this key doesn't exist
-void tombstone_snapshot(
-        std::shared_ptr<Store> store,
-        const RefKey& key,
-        bool log_changes
-        ) {
-    try {
-        auto key_segment_pair = store->read_compressed(key).get();
-        tombstone_snapshot(store, std::move(key_segment_pair), log_changes);
-    } catch (const storage::KeyNotFoundException& e) {
-        log::version().info("Cannot tombstone snapshot {}, key does not exist on the store", key);
-    } catch (const std::exception& e) {
-        log::version().error("Cannot tombstone snapshot {}: {}", key, e.what());
-    }
-}
-
-void iterate_snapshots(std::shared_ptr <Store> store, folly::Function<void(entity::VariantKey & )> visitor) {
+void iterate_snapshots(const std::shared_ptr<Store>& store, folly::Function<void(entity::VariantKey & )> visitor) {
     ARCTICDB_SAMPLE(IterateSnapshots, 0)
 
     std::vector<VariantKey> snap_variant_keys;
     std::unordered_set<SnapshotId> seen;
 
-    store->iterate_type(KeyType::SNAPSHOT_REF, [&](VariantKey &&vk) {
+    store->iterate_type(KeyType::SNAPSHOT_REF, [&snap_variant_keys, &seen](VariantKey &&vk) {
         util::check(std::holds_alternative<RefKey>(vk), "Expected shapshot ref to be reference type, got {}", variant_key_view(vk));
-        auto ref_key = std::get<RefKey>(vk);
+        auto ref_key = std::get<RefKey>(std::move(vk));
         seen.insert(ref_key.id());
         snap_variant_keys.emplace_back(ref_key);
     });
 
-    store->iterate_type(KeyType::SNAPSHOT, [&](VariantKey &&vk) {
-        const auto &key = to_atom(vk);
+    store->iterate_type(KeyType::SNAPSHOT, [&snap_variant_keys, &seen](VariantKey &&vk) {
+        auto key = to_atom(std::move(vk));
         if (seen.find(key.id()) == seen.end()) {
-            snap_variant_keys.push_back(key);
+            snap_variant_keys.emplace_back(key);
         }
     });
 
@@ -128,8 +116,10 @@ void iterate_snapshots(std::shared_ptr <Store> store, folly::Function<void(entit
     }
 }
 
-std::optional<size_t>
-row_id_for_stream_in_snapshot_segment(SegmentInMemory &seg, bool using_ref_key, StreamId stream_id) {
+std::optional<size_t> row_id_for_stream_in_snapshot_segment(
+    SegmentInMemory &seg,
+    bool using_ref_key,
+    const StreamId& stream_id) {
     if (using_ref_key) {
         // With ref keys we are sure the snapshot segment has the index atom keys sorted by stream_id.
         auto lb = std::lower_bound(std::begin(seg), std::end(seg), stream_id,
@@ -148,7 +138,7 @@ row_id_for_stream_in_snapshot_segment(SegmentInMemory &seg, bool using_ref_key,
     }
     // Fall back to linear search for old atom key snapshots.
     for (size_t idx = 0; idx < seg.row_count(); idx++) {
-        auto row_stream_id = stream_id_from_segment<pipelines::index::Fields>(seg, idx);
+        auto row_stream_id = stream_id_from_segment<pipelines::index::Fields>(seg, static_cast<ssize_t>(idx));
         if (row_stream_id == stream_id) {
             return idx;
         }
@@ -156,38 +146,15 @@ row_id_for_stream_in_snapshot_segment(SegmentInMemory &seg, bool using_ref_key,
     return std::nullopt;
 }
 
-std::unordered_map<VariantKey, SnapshotRefData> get_snapshots_and_index_keys(
-        const std::shared_ptr<Store>& store
-        ) {
-    ARCTICDB_SAMPLE(GetSnapshotsAndIndexKeys, 0)
-    std::unordered_map<VariantKey, SnapshotRefData> res;
-    iterate_snapshots(store, [&store, &res](const VariantKey &vk) {
-        try {
-            auto segment = store->read_sync(vk).second;
-            std::unordered_set<VariantKey> index_keys;
-            for (size_t idx = 0; idx < segment.row_count(); idx++) {
-                auto key = read_key_row(segment, idx);
-                if (is_index_key_type(key.type())) {
-                    index_keys.emplace(std::move(key));
-                }
-            }
-            SnapshotRefData snapshot_ref_data{std::move(segment), std::move(index_keys)};
-            res.try_emplace(vk, std::move(snapshot_ref_data));
-        } catch (const storage::KeyNotFoundException& e) {
-            log::version().info("Failed to read snapshot key: {}", e.what());
-        }
-    });
-    return res;
-}
-
 std::unordered_set<entity::AtomKey> get_index_keys_in_snapshots(
-        std::shared_ptr <Store> store,
+        const std::shared_ptr<Store>& store,
         const StreamId &stream_id) {
     ARCTICDB_SAMPLE(GetIndexKeysInSnapshot, 0)
 
     std::unordered_set<entity::AtomKey> index_keys_in_snapshots{};
 
-    iterate_snapshots(store, [&](VariantKey &vk) {
+    iterate_snapshots(store, [&index_keys_in_snapshots, &store, &stream_id](const VariantKey &vk) {
+        ARCTICDB_DEBUG(log::snapshot(), "Reading snapshot {}", vk);
         bool snapshot_using_ref = variant_key_type(vk) == KeyType::SNAPSHOT_REF;
         SegmentInMemory snapshot_segment = store->read_sync(vk).second;
         if (snapshot_segment.row_count() == 0) {
@@ -199,8 +166,11 @@ std::unordered_set<entity::AtomKey> get_index_keys_in_snapshots(
         auto opt_idx_for_stream_id = row_id_for_stream_in_snapshot_segment(
                 snapshot_segment, snapshot_using_ref, stream_id);
         if (opt_idx_for_stream_id) {
-            auto stream_idx = opt_idx_for_stream_id.value();
-            index_keys_in_snapshots.insert(read_key_row(snapshot_segment, stream_idx));
+            ARCTICDB_DEBUG(log::snapshot(), "Found index key for {} at {}", stream_id, *opt_idx_for_stream_id);
+            auto stream_idx = *opt_idx_for_stream_id;
+            index_keys_in_snapshots.insert(read_key_row(snapshot_segment, static_cast<ssize_t>(stream_idx)));
+        } else {
+            ARCTICDB_DEBUG(log::snapshot(), "Failed to find index key for {}", stream_id);
         }
     });
 
@@ -211,7 +181,7 @@ std::unordered_set<entity::AtomKey> get_index_keys_in_snapshots(
  * Returned pair has first: keys not in snapshots, second: keys in snapshots.
  */
 std::pair<std::vector<AtomKey>, std::unordered_set<AtomKey>> get_index_keys_partitioned_by_inclusion_in_snapshots(
-        std::shared_ptr <Store> store,
+        const std::shared_ptr<Store>& store,
         const StreamId& stream_id,
         const std::vector<entity::AtomKey> &all_index_keys
 ) {
@@ -228,20 +198,23 @@ std::pair<std::vector<AtomKey>, std::unordered_set<AtomKey>> get_index_keys_part
     return std::make_pair(index_keys_not_in_snapshot, index_keys_in_snapshot);
 }
 
-std::optional<VariantKey> get_snapshot_key(std::shared_ptr <Store> store, const SnapshotId &snap_name) {
+VariantKey get_ref_key(const SnapshotId& snap_name) {
+    return RefKey{snap_name, KeyType::SNAPSHOT_REF};
+}
+
+std::optional<VariantKey> get_snapshot_key(const std::shared_ptr<Store>& store, const SnapshotId &snap_name) {
     ARCTICDB_SAMPLE(getSnapshot, 0)
 
-    auto maybe_ref_key = RefKey{std::move(snap_name), KeyType::SNAPSHOT_REF};
-    if(store->key_exists_sync(maybe_ref_key))
+    if(auto maybe_ref_key = get_ref_key(snap_name); store->key_exists_sync(maybe_ref_key))
         return maybe_ref_key;
 
     // Fall back to iteration
     ARCTICDB_DEBUG(log::version(), "Ref key not found for snapshot, falling back to slow path: {}", snap_name);
-    std::optional<std::pair<VariantKey, SegmentInMemory>> opt_segment = std::nullopt;
+    std::optional<std::pair<VariantKey, SegmentInMemory>> opt_segment;
 
     std::optional<VariantKey> ret;
     store->iterate_type(KeyType::SNAPSHOT, [&ret, &snap_name](VariantKey &&vk) {
-        auto snap_key = to_atom(vk);
+        auto snap_key = to_atom(std::move(vk));
         if (snap_key.id() == snap_name) {
             ret = snap_key;
         }
@@ -249,7 +222,65 @@ std::optional<VariantKey> get_snapshot_key(std::shared_ptr <Store> store, const
     return ret;
 }
 
-std::optional<std::pair<VariantKey, SegmentInMemory>> get_snapshot(std::shared_ptr <Store> store, const SnapshotId &snap_name) {
+std::unordered_map<SnapshotId, std::optional<VariantKey>> all_ref_keys(
+    const std::vector<SnapshotId>& snap_names,
+    const std::vector<VariantKey>& ref_keys
+    ) {
+    std::unordered_map<SnapshotId, std::optional<VariantKey>> output;
+    output.reserve(snap_names.size());
+    for(auto name : folly::enumerate(snap_names))
+        output.try_emplace(*name, ref_keys[name.index]);
+
+    return output;
+}
+
+std::unordered_map<SnapshotId, std::optional<VariantKey>> get_snapshot_keys_via_iteration(
+    const std::vector<bool>& ref_key_exists,
+    const std::vector<SnapshotId>& snap_names,
+    const std::vector<VariantKey>& ref_keys,
+    const std::shared_ptr<Store>& store
+    ){
+    std::unordered_map<SnapshotId, std::optional<VariantKey>> output;
+    for (auto snap : folly::enumerate(snap_names)) {
+        if (!ref_key_exists[snap.index])
+            output.try_emplace(*snap, std::nullopt);
+    }
+
+    store->iterate_type(KeyType::SNAPSHOT, [&output](VariantKey &&vk) {
+        if (auto it = output.find(variant_key_id(vk)); it != output.end())
+            it->second = std::move(vk);
+    });
+
+    for (auto snap : folly::enumerate(snap_names)) {
+        if (ref_key_exists[snap.index])
+            output.try_emplace(*snap, ref_keys[snap.index]);
+    }
+    return output;
+}
+
+
+std::unordered_map<SnapshotId, std::optional<VariantKey>> get_keys_for_snapshots(
+    const std::shared_ptr<Store>& store,
+    const std::vector<SnapshotId>& snap_names) {
+    std::vector<VariantKey> ref_keys;
+    ref_keys.resize(snap_names.size());
+    std::transform(std::begin(snap_names), std::end(snap_names), std::begin(ref_keys), [] (const auto& name) { return get_ref_key(name); });
+
+    auto found_keys = folly::collect(store->batch_key_exists(ref_keys)).via(&async::io_executor()).thenValue(
+        [&snap_names, &ref_keys, store] (std::vector<bool> ref_key_exists) {
+            if(std::all_of(std::begin(ref_key_exists), std::end(ref_key_exists), [] (bool b) { return b; })) {
+                return all_ref_keys(snap_names, ref_keys);
+            } else {
+                return get_snapshot_keys_via_iteration(ref_key_exists, snap_names, ref_keys, store);
+            }
+        });
+
+    return std::move(found_keys).get();
+}
+
+std::optional<std::pair<VariantKey, SegmentInMemory>> get_snapshot(
+    const std::shared_ptr<Store>& store,
+    const SnapshotId &snap_name) {
     ARCTICDB_SAMPLE(getSnapshot, 0)
     auto opt_snap_key = get_snapshot_key(store, snap_name);
     if(!opt_snap_key)
@@ -268,42 +299,27 @@ std::set<StreamId> list_streams_in_snapshot(
     if (!opt_snap_key)
         throw storage::NoDataFoundException(snap_name);
 
-    auto& snapshot_segment = opt_snap_key.value().second;
+    const auto& snapshot_segment = opt_snap_key->second;
 
     for (size_t idx = 0; idx < snapshot_segment.row_count(); idx++) {
-        auto stream_index = read_key_row(snapshot_segment, idx);
+        auto stream_index = read_key_row(snapshot_segment, static_cast<ssize_t>(idx));
         res.insert(stream_index.id());
     }
     return res;
 }
 
-namespace {
+
 std::vector<AtomKey> get_versions_from_segment(
     const SegmentInMemory& snapshot_segment
     ) {
     std::vector<AtomKey> res;
     for (size_t idx = 0; idx < snapshot_segment.row_count(); idx++) {
-        auto stream_index = read_key_row(snapshot_segment, idx);
+        auto stream_index = read_key_row(snapshot_segment, static_cast<ssize_t>(idx));
         res.push_back(stream_index);
     }
     return res;
 }
 
-py::object get_metadata_from_segment(
-    const SegmentInMemory& snapshot_segment
-    ) {
-    auto metadata_proto = snapshot_segment.metadata();
-    py::object pyobj;
-    if (metadata_proto) {
-        arcticdb::proto::descriptors::UserDefinedMetadata user_meta_proto;
-        metadata_proto->UnpackTo(&user_meta_proto);
-        pyobj = python_util::pb_to_python(user_meta_proto);
-    } else {
-        pyobj = pybind11::none();
-    }
-    return pyobj;
-}
-
 std::vector<AtomKey> get_versions_from_snapshot(
     const std::shared_ptr<Store>& store,
     const VariantKey& vk) {
@@ -311,15 +327,6 @@ std::vector<AtomKey> get_versions_from_snapshot(
     auto snapshot_segment = store->read_sync(vk).second;
     return get_versions_from_segment(snapshot_segment);
 }
-} //namespace
-
-std::pair<std::vector<AtomKey>, py::object> get_versions_and_metadata_from_snapshot(
-    const std::shared_ptr<Store>& store,
-    const VariantKey& vk
-    ) {
-    auto snapshot_segment = store->read_sync(vk).second;
-    return {get_versions_from_segment(snapshot_segment), get_metadata_from_segment(snapshot_segment)};
-}
 
 SnapshotMap get_versions_from_snapshots(
         const std::shared_ptr<Store>& store
@@ -327,7 +334,7 @@ SnapshotMap get_versions_from_snapshots(
     ARCTICDB_SAMPLE(GetVersionsFromSnapshot, 0)
     SnapshotMap res;
 
-    iterate_snapshots(store, [&](VariantKey &vk) {
+    iterate_snapshots(store, [&res, &store](const VariantKey &vk) {
         SnapshotId snapshot_id{fmt::format("{}", variant_key_id(vk))};
         res[snapshot_id] = get_versions_from_snapshot(store, vk);
     });
@@ -335,22 +342,16 @@ SnapshotMap get_versions_from_snapshots(
     return res;
 }
 
-//TODO I don't love having py:object here, return the protobuf and convert at the outer edge
-py::object get_metadata_for_snapshot(std::shared_ptr <Store> store, VariantKey& snap_key) {
-   auto seg = store->read_sync(snap_key).second;
-   return get_metadata_from_segment(seg);
-}
-
 MasterSnapshotMap get_master_snapshots_map(
         std::shared_ptr<Store> store,
         const std::optional<const std::tuple<const SnapshotVariantKey&, std::vector<IndexTypeKey>&>>& get_keys_in_snapshot
 ) {
     MasterSnapshotMap out;
-    iterate_snapshots(store, [&](VariantKey &sk) {
+    iterate_snapshots(store, [&get_keys_in_snapshot, &out, &store](const VariantKey &sk) {
         auto snapshot_id = variant_key_id(sk);
         auto snapshot_segment = store->read_sync(sk).second;
         for (size_t idx = 0; idx < snapshot_segment.row_count(); idx++) {
-            auto stream_index = read_key_row(snapshot_segment, idx);
+            auto stream_index = read_key_row(snapshot_segment, static_cast<ssize_t>(idx));
             out[stream_index.id()][stream_index].insert(snapshot_id);
             if (get_keys_in_snapshot) {
                 auto[wanted_snap_key, sink] = get_keys_in_snapshot.value();
diff --git a/cpp/arcticdb/version/snapshot.hpp b/cpp/arcticdb/version/snapshot.hpp
index 231e8a8df9..889726d4f1 100644
--- a/cpp/arcticdb/version/snapshot.hpp
+++ b/cpp/arcticdb/version/snapshot.hpp
@@ -19,25 +19,12 @@
 #include <arcticdb/storage/store.hpp>
 #include <arcticdb/util/variant.hpp>
 
-using namespace arcticdb::entity;
-using namespace arcticdb::stream;
 
 namespace arcticdb {
 
-struct SnapshotRefData {
-    explicit SnapshotRefData(SegmentInMemory&& snapshot_segment,
-                             std::unordered_set<VariantKey>&& index_keys):
-                             snapshot_segment_(std::move(snapshot_segment)),
-                             index_keys_(std::move(index_keys)) {
-
-    }
-    SegmentInMemory snapshot_segment_;
-    std::unordered_set<VariantKey> index_keys_;
-};
-
 void write_snapshot_entry(
-    std::shared_ptr <StreamSink> store,
-    std::vector <AtomKey> &keys,
+    std::shared_ptr<stream::StreamSink> store,
+    std::vector<AtomKey> &keys,
     const SnapshotId &snapshot_id,
     const py::object &user_meta,
     bool log_changes,
@@ -45,54 +32,46 @@ void write_snapshot_entry(
 );
 
 void tombstone_snapshot(
-    std::shared_ptr<StreamSink> store,
+    const std::shared_ptr<stream::StreamSink>& store,
     const RefKey& key,
     SegmentInMemory&& segment_in_memory,
     bool log_changes
 );
 
 void tombstone_snapshot(
-        std::shared_ptr<StreamSink> store,
+        const std::shared_ptr<stream::StreamSink>& store,
         storage::KeySegmentPair&& key_segment_pair,
         bool log_changes
         );
 
-void tombstone_snapshot(
-        std::shared_ptr<Store> store,
-        const RefKey& key,
-        bool log_changes
-        );
-
-
-void iterate_snapshots(std::shared_ptr <Store> store, folly::Function<void(entity::VariantKey & )> visitor);
+void iterate_snapshots(const std::shared_ptr<Store>& store, folly::Function<void(entity::VariantKey & )> visitor);
 
 std::optional<size_t> row_id_for_stream_in_snapshot_segment(
     SegmentInMemory &seg,
     bool using_ref_key,
-    StreamId stream_id);
+    const StreamId& stream_id);
 
 // Get a set of the index keys of a particular symbol that exist in any snapshot
 std::unordered_set<entity::AtomKey> get_index_keys_in_snapshots(
-    std::shared_ptr <Store> store,
+    const std::shared_ptr<Store>& store,
     const StreamId &stream_id);
 
 std::pair<std::vector<AtomKey>, std::unordered_set<AtomKey>> get_index_keys_partitioned_by_inclusion_in_snapshots(
-    std::shared_ptr <Store> store,
+    const std::shared_ptr<Store>& store,
     const StreamId& stream_id,
     const std::vector<entity::AtomKey> &all_index_keys
 );
 
-std::pair<std::vector<AtomKey>, py::object> get_versions_and_metadata_from_snapshot(
-    const std::shared_ptr<Store>& store,
-    const VariantKey& vk
-    );
+std::vector<AtomKey> get_versions_from_segment(
+    const SegmentInMemory& snapshot_segment
+);
 
 std::optional<VariantKey> get_snapshot_key(
-    std::shared_ptr <Store> store,
+    const std::shared_ptr<Store>& store,
     const SnapshotId &snap_name);
 
 std::optional<std::pair<VariantKey, SegmentInMemory>> get_snapshot(
-    std::shared_ptr <Store> store,
+    const std::shared_ptr<Store>& store,
     const SnapshotId &snap_name);
 
 std::set<StreamId> list_streams_in_snapshot(
@@ -105,12 +84,9 @@ SnapshotMap get_versions_from_snapshots(
     const std::shared_ptr<Store>& store
     );
 
-// Returns a map from snapshot keys to it's associated segment, and the [multi-]index keys in that segment
-std::unordered_map<VariantKey, SnapshotRefData> get_snapshots_and_index_keys(
-    const std::shared_ptr<Store>& store
-    );
-
-py::object get_metadata_for_snapshot(std::shared_ptr <Store> store, VariantKey& snap_key);
+std::unordered_map<SnapshotId, std::optional<VariantKey>> get_keys_for_snapshots(
+    const std::shared_ptr<Store>& store,
+    const std::vector<SnapshotId>& snap_names);
 
 /**
  * Stream id (symbol) -> all index keys in snapshots -> which snapshot contained that key.
diff --git a/cpp/arcticdb/version/version_functions.hpp b/cpp/arcticdb/version/version_functions.hpp
index bc41d484f2..c8f1729208 100644
--- a/cpp/arcticdb/version/version_functions.hpp
+++ b/cpp/arcticdb/version/version_functions.hpp
@@ -144,7 +144,8 @@ inline bool has_undeleted_version(
     pipelines::VersionQuery version_query;
     version_query.set_skip_compat(true),
     version_query.set_iterate_on_failure(false);
-    return static_cast<bool>(get_latest_undeleted_version(store, version_map, id, version_query, ReadOptions{}));
+    auto maybe_undeleted = get_latest_undeleted_version(store, version_map, id, version_query, ReadOptions{});
+    return static_cast<bool>(maybe_undeleted);
 }
 
 inline void insert_if_undeleted(
diff --git a/cpp/arcticdb/version/version_map.hpp b/cpp/arcticdb/version/version_map.hpp
index 0e9486f2da..e71ae82152 100644
--- a/cpp/arcticdb/version/version_map.hpp
+++ b/cpp/arcticdb/version/version_map.hpp
@@ -474,8 +474,9 @@ class VersionMapImpl {
         const char* function ARCTICDB_UNUSED) {
         ARCTICDB_DEBUG(log::version(), "Check reload in function {}", function);
 
-        if (has_cached_entry(stream_id, load_param))
+        if (has_cached_entry(stream_id, load_param)) {
             return get_entry(stream_id);
+        }
 
         if (!load_param.skip_compat_ && !has_stored_entry(store, stream_id))
             do_backwards_compat_check(store, stream_id);
diff --git a/cpp/arcticdb/version/version_map_batch_methods.cpp b/cpp/arcticdb/version/version_map_batch_methods.cpp
new file mode 100644
index 0000000000..206c2d191b
--- /dev/null
+++ b/cpp/arcticdb/version/version_map_batch_methods.cpp
@@ -0,0 +1,338 @@
+/* Copyright 2023 Man Group Operations Limited
+ *
+ * Use of this software is governed by the Business Source License 1.1 included in the file licenses/BSL.txt.
+ *
+ * As of the Change Date specified in that file, in accordance with the Business Source License, use of this software will be governed by the Apache License, version 2.0.
+ */
+
+#include <arcticdb/version/version_map_batch_methods.hpp>
+
+namespace arcticdb {
+
+StreamVersionData::StreamVersionData(const pipelines::VersionQuery &version_query) {
+    react(version_query);
+}
+
+void StreamVersionData::react(const pipelines::VersionQuery &version_query) {
+    util::variant_match(version_query.content_, [this](const auto &query) {
+        do_react(query);
+    });
+}
+
+void StreamVersionData::do_react(std::monostate) {
+    if (load_param_.load_type_ == LoadType::NOT_LOADED)
+        load_param_ = LoadParameter{LoadType::LOAD_LATEST_UNDELETED};
+
+    ++count_;
+}
+
+void StreamVersionData::do_react(const pipelines::SpecificVersionQuery &specific_version) {
+    ++count_;
+    switch (load_param_.load_type_) {
+    case LoadType::NOT_LOADED:
+        [[fallthrough]];
+    case LoadType::LOAD_LATEST_UNDELETED:
+        load_param_ = LoadParameter{LoadType::LOAD_DOWNTO, specific_version.version_id_};
+        break;
+    case LoadType::LOAD_DOWNTO:
+        util::check(load_param_.load_until_.has_value(),
+                    "Expect LOAD_DOWNTO to have version specified");
+        if ((specific_version.version_id_ >= 0 && is_positive_version_query(load_param_)) ||
+            (specific_version.version_id_ < 0 && load_param_.load_until_.value() < 0)) {
+            load_param_.load_until_ = std::min(load_param_.load_until_.value(), specific_version.version_id_);
+        } else {
+            load_param_ = LoadParameter{LoadType::LOAD_UNDELETED};
+        }
+        break;
+    case LoadType::LOAD_FROM_TIME:
+        [[fallthrough]];
+    case LoadType::LOAD_UNDELETED:
+        load_param_ = LoadParameter{LoadType::LOAD_UNDELETED};
+        break;
+    default:util::raise_rte("Unexpected load state {} applying specific version query", load_param_.load_type_);
+    }
+}
+
+void StreamVersionData::do_react(const pipelines::TimestampVersionQuery &timestamp_query) {
+    ++count_;
+    switch (load_param_.load_type_) {
+    case LoadType::NOT_LOADED:
+        [[fallthrough]];
+    case LoadType::LOAD_LATEST_UNDELETED:
+        load_param_ = LoadParameter{LoadType::LOAD_FROM_TIME, timestamp_query.timestamp_};
+        break;
+    case LoadType::LOAD_FROM_TIME:
+        util::check(load_param_.load_from_time_.has_value(),
+                    "Expect LOAD_TO_TIME to have timestamp specified");
+        load_param_.load_from_time_ = std::min(load_param_.load_from_time_.value(), timestamp_query.timestamp_);
+        break;
+    case LoadType::LOAD_DOWNTO:
+        [[fallthrough]];
+    case LoadType::LOAD_UNDELETED:
+        load_param_ = LoadParameter{LoadType::LOAD_UNDELETED};
+        break;
+    default:util::raise_rte("Unexpected load state {} applying specific version query", load_param_.load_type_);
+    }
+}
+
+void StreamVersionData::do_react(const pipelines::SnapshotVersionQuery &snapshot_query) {
+    snapshots_.push_back(snapshot_query.name_);
+}
+
+std::optional<AtomKey> get_specific_version_from_entry(
+    const std::shared_ptr<VersionMapEntry>& version_map_entry,
+    const pipelines::SpecificVersionQuery& specific_version
+    ) {
+    auto signed_version_id = specific_version.version_id_;
+    VersionId version_id;
+    if (signed_version_id >= 0) {
+        version_id = static_cast<VersionId>(signed_version_id);
+    } else {
+        auto opt_latest = version_map_entry->get_first_index(true);
+        if (opt_latest.has_value()) {
+            auto opt_version_id = get_version_id_negative_index(opt_latest->version_id(),
+                                                                signed_version_id);
+            if (opt_version_id.has_value()) {
+                version_id = *opt_version_id;
+            } else {
+                return std::nullopt;
+            }
+        } else {
+            return std::nullopt;
+        }
+    }
+    return find_index_key_for_version_id(version_id, version_map_entry);
+}
+
+std::optional<AtomKey> get_version_map_entry_by_timestamp(
+    const std::shared_ptr<VersionMapEntry>& version_map_entry,
+    const pipelines::TimestampVersionQuery &timestamp_version
+    ) {
+    auto version_key = get_index_key_from_time(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;
+    }
+}
+
+inline std::optional<AtomKey> get_key_for_version_query(
+    const std::shared_ptr<VersionMapEntry> &version_map_entry,
+    const pipelines::VersionQuery &version_query) {
+    return util::variant_match(version_query.content_,
+        [&version_map_entry](const pipelines::SpecificVersionQuery &specific_version) {
+            return get_specific_version_from_entry(version_map_entry, specific_version);
+        },
+        [&version_map_entry](const pipelines::TimestampVersionQuery &timestamp_version) {
+            return get_version_map_entry_by_timestamp(version_map_entry, timestamp_version);
+        },
+        [&version_map_entry](const std::monostate &) {
+           return version_map_entry->get_first_index(false);
+        },
+        [](const auto &) -> std::optional<AtomKey> {
+           util::raise_rte("Unsupported version query type");
+        });
+}
+
+struct SnapshotCountMap {
+    std::unordered_map<SnapshotId, size_t> snapshot_counts_;
+
+    explicit SnapshotCountMap(const robin_hood::unordered_flat_map<StreamId, StreamVersionData> &version_data) {
+        for (const auto &[_, info] : version_data) {
+            for (const auto &snapshot : info.snapshots_) {
+                const auto it = snapshot_counts_.find(snapshot);
+                if(it == std::end(snapshot_counts_))
+                    snapshot_counts_.emplace(snapshot, 1);
+                else
+                    ++it->second;
+            }
+        }
+    }
+
+    std::vector<SnapshotId> snapshots() const {
+        std::vector<SnapshotId> output;
+        output.reserve(snapshot_counts_.size());
+        for(const auto& [snapshot, _] : snapshot_counts_)
+            output.emplace_back(snapshot);
+
+        return output;
+    }
+
+    size_t get_size(const SnapshotId &snapshot) {
+        const auto it = snapshot_counts_.find(std::cref(snapshot));
+        util::check(it != snapshot_counts_.end(), "Missing snapshot data for snapshot {}", snapshot);
+        return it->second;
+    }
+};
+
+using SnapshotPair = std::pair<VariantKey, SegmentInMemory>;
+using VersionEntryOrSnapshot = std::variant<std::shared_ptr<VersionMapEntry>, std::optional<SnapshotPair>>;
+using SplitterType = folly::FutureSplitter<VersionEntryOrSnapshot>;
+using SnapshotKeyMap = std::unordered_map<SnapshotId, std::optional<VariantKey>>;
+
+folly::Future<VersionEntryOrSnapshot> set_up_snapshot_future(
+    robin_hood::unordered_flat_map<StreamId, SplitterType> &snapshot_futures,
+    const std::shared_ptr<SnapshotCountMap> &snapshot_count_map,
+    const std::shared_ptr<SnapshotKeyMap> &snapshot_key_map,
+    const pipelines::SnapshotVersionQuery &snapshot_query,
+    const std::shared_ptr<Store> &store
+) {
+    auto num_snaps = snapshot_count_map->get_size(snapshot_query.name_);
+    const auto snapshot_key = snapshot_key_map->find(snapshot_query.name_);
+    util::check(snapshot_key != std::end(*snapshot_key_map),
+                "Missing snapshot data for snapshot {}",
+                snapshot_query.name_);
+    if (!snapshot_key->second) {
+        return folly::makeFuture(std::make_optional<SnapshotPair>());
+    } else {
+        if (num_snaps == 1) {
+            return store->read(*snapshot_key->second).thenValue(
+                [](SnapshotPair &&snapshot_pair) {
+                    return VersionEntryOrSnapshot{std::move(snapshot_pair)};
+                });
+        } else {
+            auto fut = snapshot_futures.find(snapshot_query.name_);
+            if (fut == snapshot_futures.end()) {
+                auto [splitter, _] = snapshot_futures.emplace(
+                    snapshot_query.name_,
+                    folly::FutureSplitter{
+                        store->read(*snapshot_key->second).thenValue(
+                            [snap_key =
+                            *snapshot_key->second](std::pair<VariantKey, SegmentInMemory> snapshot_output) mutable -> VersionEntryOrSnapshot {
+                                return SnapshotPair{std::move(snap_key), std::move(snapshot_output.second)};
+                            })});
+
+                return splitter->second.getFuture();
+            } else {
+                return fut->second.getFuture();
+            }
+        }
+    }
+}
+
+folly::Future<VersionEntryOrSnapshot> set_up_version_future(
+    const StreamId &symbol,
+    const StreamVersionData &version_data,
+    robin_hood::unordered_flat_map<StreamId, SplitterType> &version_futures,
+    const std::shared_ptr<Store> &store,
+    const std::shared_ptr<VersionMap> &version_map
+) {
+    if (version_data.count_ == 1) {
+        return async::submit_io_task(CheckReloadTask{store, version_map, symbol,
+                                                     version_data.load_param_}).thenValue(
+            [](std::shared_ptr<VersionMapEntry> version_map_entry) {
+                return VersionEntryOrSnapshot{std::move(version_map_entry)};
+            });
+    } else {
+        auto maybe_fut = version_futures.find(symbol);
+        if (maybe_fut == version_futures.end()) {
+            auto [splitter, inserted] = version_futures.emplace(symbol,
+                folly::FutureSplitter{
+                    async::submit_io_task(
+                        CheckReloadTask{store,
+                                        version_map,
+                                        symbol,
+                                        version_data.load_param_}).thenValue(
+                        [](std::shared_ptr<VersionMapEntry> version_map_entry) {
+                            return VersionEntryOrSnapshot{
+                                std::move(version_map_entry)};
+                        })});
+
+            return splitter->second.getFuture();
+        } else {
+            return maybe_fut->second.getFuture();
+        }
+    }
+}
+
+std::vector<folly::Future<std::optional<AtomKey>>> batch_get_versions_async(
+    const std::shared_ptr<Store> &store,
+    const std::shared_ptr<VersionMap> &version_map,
+    const std::vector<StreamId> &symbols,
+    const std::vector<pipelines::VersionQuery> &version_queries,
+    const std::optional<bool> &use_previous_on_error) {
+    ARCTICDB_SAMPLE(BatchGetVersion, 0)
+    util::check(symbols.size() == version_queries.size(),
+                "Symbol and version query list mismatch: {} != {}",
+                symbols.size(),
+                version_queries.size());
+
+    robin_hood::unordered_flat_map<StreamId, StreamVersionData> version_data;
+    for (const auto &symbol : folly::enumerate(symbols)) {
+        auto it = version_data.find(*symbol);
+        if (it == version_data.end()) {
+            version_data.insert(robin_hood::pair<StreamId, StreamVersionData>(
+                *symbol,
+                StreamVersionData{version_queries[symbol.index]}));
+        } else {
+            it->second.react(version_queries[symbol.index]);
+        }
+    }
+
+    auto snapshot_count_map = std::make_shared<SnapshotCountMap>(version_data);
+    auto snapshot_key_map = std::make_shared<SnapshotKeyMap>(get_keys_for_snapshots(store, snapshot_count_map->snapshots()));
+
+    robin_hood::unordered_flat_map<StreamId, SplitterType> snapshot_futures;
+    robin_hood::unordered_flat_map<StreamId, SplitterType> version_futures;
+
+    std::vector<folly::Future<std::optional<AtomKey>>> output;
+    output.reserve(symbols.size());
+    for (const auto &symbol : folly::enumerate(symbols)) {
+        auto version_query = version_queries[symbol.index];
+        auto version_entry_fut = folly::Future<VersionEntryOrSnapshot>::makeEmpty();
+        util::variant_match(version_query.content_,
+            [&version_entry_fut, &snapshot_count_map, &snapshot_key_map, &snapshot_futures, &store](
+                const pipelines::SnapshotVersionQuery &snapshot_query) {
+                version_entry_fut = set_up_snapshot_future(
+                    snapshot_futures,
+                    snapshot_count_map,
+                    snapshot_key_map,
+                    snapshot_query,
+                    store
+                );
+            },
+            [&version_entry_fut, &version_data, &symbol, &version_futures, use_previous_on_error, &store, &version_map](
+                const auto &) {
+                const auto it = version_data.find(*symbol);
+                util::check(it != version_data.end(), "Missing version data for symbol {}", *symbol);
+
+                if (use_previous_on_error.value_or(false))
+                    it->second.load_param_.use_previous_ = true;
+
+                version_entry_fut = set_up_version_future(
+                    *symbol,
+                    it->second,
+                    version_futures,
+                    store,
+                    version_map
+                );
+            });
+
+        output.push_back(std::move(version_entry_fut)
+             .thenValue([vq = version_query, sid = *symbol](auto version_or_snapshot) {
+                 return util::variant_match(version_or_snapshot,
+                    [&vq](const std::shared_ptr<VersionMapEntry> &version_map_entry) {
+                        return get_key_for_version_query(version_map_entry, vq);
+                    },
+                    [&sid](std::optional<SnapshotPair> snapshot) {
+                        if (!snapshot)
+                            return std::make_optional<AtomKey>();
+
+                        auto [snap_key, snap_segment] = std::move(*snapshot);
+                        auto opt_id = row_id_for_stream_in_snapshot_segment(
+                            snap_segment,
+                            std::holds_alternative<RefKey>(snap_key),
+                            sid);
+
+                        return opt_id
+                               ? std::make_optional<AtomKey>(read_key_row(snap_segment, static_cast<ssize_t>(opt_id.value())))
+                               : std::nullopt;
+                    });
+             }));
+    }
+    return output;
+}
+
+} //namespace arcticdb
\ No newline at end of file
diff --git a/cpp/arcticdb/version/version_map_batch_methods.hpp b/cpp/arcticdb/version/version_map_batch_methods.hpp
index b480f5528d..bbda0bffb6 100644
--- a/cpp/arcticdb/version/version_map_batch_methods.hpp
+++ b/cpp/arcticdb/version/version_map_batch_methods.hpp
@@ -13,10 +13,81 @@
 #include <arcticdb/version/version_store_objects.hpp>
 #include <arcticdb/pipeline/query.hpp>
 #include <arcticdb/version/version_functions.hpp>
+#include <arcticdb/version/snapshot.hpp>
+
 #include <folly/futures/FutureSplitter.h>
 
 namespace arcticdb {
 
+struct SymbolStatus {
+    const VersionId version_id_ = 0;
+    const bool exists_ = false;
+    const timestamp timestamp_ = 0;
+
+    SymbolStatus(VersionId version_id, bool exists, timestamp ts) :
+        version_id_(version_id),
+        exists_(exists),
+        timestamp_(ts) {
+    }
+};
+
+template <typename Inputs, typename TaskSubmitter, typename ResultHandler>
+inline void submit_tasks_for_range(const Inputs& inputs, TaskSubmitter submitter, ResultHandler result_handler) {
+    const auto window_size = async::TaskScheduler::instance()->io_thread_count() * 2;
+
+    auto futures = folly::window(inputs, [&submitter, &result_handler](const auto &input) {
+        return submitter(input).thenValue([&result_handler, &input](auto &&r) {
+            auto result = std::forward<decltype(r)>(r);
+            result_handler(input, std::move(result));
+            return folly::Unit{};
+        });
+    }, window_size);
+
+    auto collected_futs = folly::collectAll(futures).get();
+    std::optional<std::string> all_exceptions;
+    for (auto&& collected_fut: collected_futs) {
+        if (!collected_fut.hasValue()) {
+            all_exceptions = all_exceptions.value_or("").append(collected_fut.exception().what().toStdString()).append("\n");
+        }
+    }
+    internal::check<ErrorCode::E_RUNTIME_ERROR>(!all_exceptions.has_value(), all_exceptions.value_or(""));
+}
+
+inline std::shared_ptr<std::unordered_map<StreamId, SymbolStatus>> batch_check_latest_id_and_status(
+    const std::shared_ptr<Store> &store,
+    const std::shared_ptr<VersionMap> &version_map,
+    const std::shared_ptr<std::vector<StreamId>> &symbols) {
+    ARCTICDB_SAMPLE(BatchGetLatestVersion, 0)
+    const LoadParameter load_param{LoadType::LOAD_LATEST_UNDELETED};
+    auto output = std::make_shared<std::unordered_map<StreamId, SymbolStatus>>();
+    auto mutex = std::make_shared<std::mutex>();
+
+    submit_tasks_for_range(*symbols,
+        [store, version_map, &load_param](auto &symbol) {
+          return async::submit_io_task(CheckReloadTask{store, version_map, symbol, load_param});
+        },
+        [output, mutex](const auto& id, const std::shared_ptr<VersionMapEntry> &entry) {
+          auto index_key = entry->get_first_index(false);
+          if (index_key) {
+              std::lock_guard lock{*mutex};
+              output->insert(std::make_pair<StreamId, SymbolStatus>(StreamId{id}, {index_key->version_id(), true, index_key->creation_ts()}));
+          } else {
+              index_key = entry->get_first_index(true);
+              if (index_key) {
+                  std::lock_guard lock{*mutex};
+                  output->insert(std::make_pair<StreamId, SymbolStatus>(StreamId{id}, {index_key->version_id(), false, index_key->creation_ts()}));
+              } else {
+                  if (entry->head_ && entry->head_->type() == KeyType::TOMBSTONE_ALL) {
+                      const auto& head = *entry->head_;
+                      std::lock_guard lock{*mutex};
+                      output->insert(std::make_pair<StreamId, SymbolStatus>(StreamId{id}, {head.version_id(), false, head.creation_ts()}));
+                  }
+              }
+          }});
+
+    return output;
+}
+
 inline std::shared_ptr<std::unordered_map<StreamId, AtomKey>> batch_get_latest_version(
     const std::shared_ptr<Store> &store,
     const std::shared_ptr<VersionMap> &version_map,
@@ -25,21 +96,41 @@ inline std::shared_ptr<std::unordered_map<StreamId, AtomKey>> batch_get_latest_v
     ARCTICDB_SAMPLE(BatchGetLatestVersion, 0)
     const LoadParameter load_param{include_deleted ? LoadType::LOAD_LATEST : LoadType::LOAD_LATEST_UNDELETED};
     auto output = std::make_shared<std::unordered_map<StreamId, AtomKey>>();
+    auto mutex = std::make_shared<std::mutex>();
 
-    async::submit_tasks_for_range(stream_ids,
+    submit_tasks_for_range(stream_ids,
             [store, version_map, &load_param](auto& stream_id) {
                 return async::submit_io_task(CheckReloadTask{store, version_map, stream_id, load_param});
             },
-            [output, include_deleted](auto& id, auto&& entry) {
+            [output, include_deleted, mutex](auto id, auto entry) {
                 auto index_key = entry->get_first_index(include_deleted);
-                if (index_key)
+                if (index_key) {
+                    std::lock_guard lock{*mutex};
                     (*output)[id] = *index_key;
+                }
             });
 
     return output;
 }
 
-// The logic here is the same as get_latest_undeleted_version_and_next_version_id
+inline std::vector<folly::Future<std::pair<std::optional<AtomKey>, std::optional<AtomKey>>>> batch_get_latest_undeleted_and_latest_versions_async(
+    const std::shared_ptr<Store> &store,
+    const std::shared_ptr<VersionMap> &version_map,
+    const std::vector<StreamId> &stream_ids) {
+    ARCTICDB_SAMPLE(BatchGetLatestUndeletedVersionAndNextVersionId, 0)
+    std::vector<folly::Future<std::pair<std::optional<AtomKey>, std::optional<AtomKey>>>> vector_fut;
+    for (auto& stream_id: stream_ids){
+        vector_fut.push_back(async::submit_io_task(CheckReloadTask{store,
+                                                                   version_map,
+                                                                   stream_id,
+                                                                   LoadParameter{LoadType::LOAD_LATEST_UNDELETED}})
+                                 .thenValue([](const std::shared_ptr<VersionMapEntry>& entry){
+                                     return std::make_pair(entry->get_first_index(false), entry->get_first_index(true));
+                                 }));
+    }
+    return vector_fut;
+}
+
 inline std::vector<folly::Future<version_store::UpdateInfo>> batch_get_latest_undeleted_version_and_next_version_id_async(
         const std::shared_ptr<Store> &store,
         const std::shared_ptr<VersionMap> &version_map,
@@ -51,7 +142,7 @@ inline std::vector<folly::Future<version_store::UpdateInfo>> batch_get_latest_un
                                                      version_map,
                                                      stream_id,
                                                      LoadParameter{LoadType::LOAD_LATEST_UNDELETED}})
-        .thenValue([](auto&& entry){
+        .thenValue([](auto entry){
             auto latest_version = entry->get_first_index(true);
             auto latest_undeleted_version = entry->get_first_index(false);
             VersionId next_version_id = latest_version.has_value() ? latest_version->version_id() + 1 : 0;
@@ -61,21 +152,46 @@ inline std::vector<folly::Future<version_store::UpdateInfo>> batch_get_latest_un
     return vector_fut;
 }
 
+template <typename MapType>
+struct MapRandomAccessWrapper {
+    const MapType& map_;
+
+    using iterator = typename MapType::iterator;
+
+    explicit MapRandomAccessWrapper(const MapType& map) :
+        map_(map) {
+    }
+
+    [[nodiscard]] size_t size() const {
+        return map_.size();
+    }
+
+    auto& operator[](size_t pos) const {
+        auto it = std::begin(map_);
+        std::advance(it, pos);
+        return *it;
+    }
+};
+
+// This version assumes that there is only one version per symbol, so no need for the state machine below
 inline std::shared_ptr<std::unordered_map<StreamId, AtomKey>> batch_get_specific_version(
     const std::shared_ptr<Store>& store,
     const std::shared_ptr<VersionMap>& version_map,
-    std::map<StreamId, VersionId>& sym_versions) {
+    const std::map<StreamId, VersionId>& sym_versions,
+    bool include_deleted = true) {
     ARCTICDB_SAMPLE(BatchGetLatestVersion, 0)
     auto output = std::make_shared<std::unordered_map<StreamId, AtomKey>>();
+    auto mutex = std::make_shared<std::mutex>();
 
-    async::submit_tasks_for_range(sym_versions,
-                                  [store, version_map](auto& sym_version) {
-        LoadParameter load_param{LoadType::LOAD_DOWNTO, static_cast<SignedVersionId>(sym_version.second)};
-        return async::submit_io_task(CheckReloadTask{store, version_map, sym_version.first, load_param});
+    MapRandomAccessWrapper wrapper{sym_versions};
+    submit_tasks_for_range(wrapper, [store, version_map](auto& sym_version) {
+            LoadParameter load_param{LoadType::LOAD_DOWNTO, static_cast<SignedVersionId>(sym_version.second)};
+            return async::submit_io_task(CheckReloadTask{store, version_map, sym_version.first, load_param});
         },
-        [output](auto& sym_version, auto&& entry) {
-        auto index_key = find_index_key_for_version_id(sym_version.second, entry);
+        [output, include_deleted, mutex](auto sym_version, const std::shared_ptr<VersionMapEntry>& entry) {
+        auto index_key = find_index_key_for_version_id(sym_version.second, entry, include_deleted);
         if (index_key) {
+            std::lock_guard lock{*mutex};
             (*output)[sym_version.first] = *index_key;
         }
     });
@@ -92,23 +208,27 @@ using VersionVectorType = std::vector<VersionId>;
 inline std::shared_ptr<std::unordered_map<std::pair<StreamId, VersionId>, AtomKey>> batch_get_specific_versions(
         const std::shared_ptr<Store>& store,
         const std::shared_ptr<VersionMap>& version_map,
-        std::map<StreamId, VersionVectorType>& sym_versions) {
+        std::map<StreamId, VersionVectorType>& sym_versions,
+        bool include_deleted = true) {
     ARCTICDB_SAMPLE(BatchGetLatestVersion, 0)
     auto output = std::make_shared<std::unordered_map<std::pair<StreamId, VersionId>, AtomKey>>();
+    MapRandomAccessWrapper wrapper{sym_versions};
+    auto mutex = std::make_shared<std::mutex>();
 
-    async::submit_tasks_for_range(sym_versions,
-            [store, version_map](auto& sym_version) {
+    submit_tasks_for_range(wrapper, [store, version_map](auto sym_version) {
                 auto first_version = *std::min_element(std::begin(sym_version.second), std::end(sym_version.second));
                 LoadParameter load_param{LoadType::LOAD_DOWNTO, static_cast<SignedVersionId>(first_version)};
                 return async::submit_io_task(CheckReloadTask{store, version_map, sym_version.first, load_param});
             },
-            [output, &sym_versions](auto& sym_version, auto&& entry) {
+
+            [output, &sym_versions, include_deleted, mutex](auto sym_version, const std::shared_ptr<VersionMapEntry>& entry) {
                 auto sym_it = sym_versions.find(sym_version.first);
                 util::check(sym_it != sym_versions.end(), "Failed to find versions for symbol {}", sym_version.first);
                 const auto& versions = sym_it->second;
                 for(auto version : versions) {
-                    auto index_key = find_index_key_for_version_id(version, entry);
+                    auto index_key = find_index_key_for_version_id(version, entry, include_deleted);
                     if (index_key) {
+                        std::lock_guard lock{*mutex};
                         (*output)[std::pair(sym_version.first, version)] = *index_key;
                     }
                 }
@@ -116,166 +236,27 @@ inline std::shared_ptr<std::unordered_map<std::pair<StreamId, VersionId>, AtomKe
 
     return output;
 }
-
 struct StreamVersionData {
-    explicit StreamVersionData(const pipelines::VersionQuery& version_query) {
-        react(version_query);
-    }
-
-    StreamVersionData() = default;
-
     size_t count_ = 0;
-    LoadParameter load_param_ = LoadParameter{LoadType::LOAD_LATEST_UNDELETED};
-
-    void react(const pipelines::VersionQuery& version_query) {
-        util::variant_match(version_query.content_, [that = this](const auto &query) {
-            that->do_react(query);
-        });
-    }
-
-    void do_react(std::monostate) {
-        ++count_;
-    }
-
-    void do_react(const pipelines::SpecificVersionQuery& specific_version) {
-        ++count_;
-        switch(load_param_.load_type_) {
-        case LoadType::LOAD_LATEST_UNDELETED:
-            load_param_ = LoadParameter{LoadType::LOAD_DOWNTO, specific_version.version_id_};
-            break;
-        case LoadType::LOAD_DOWNTO:
-            util::check(load_param_.load_until_.has_value(), "Expect LOAD_DOWNTO to have version specificed");
-            if ((specific_version.version_id_ >= 0 && is_positive_version_query(load_param_)) ||
-                    (specific_version.version_id_ < 0 && load_param_.load_until_.value() < 0)) {
-                load_param_.load_until_ = std::min(load_param_.load_until_.value(), specific_version.version_id_);
-            } else {
-                load_param_ = LoadParameter{LoadType::LOAD_UNDELETED};
-            }
-            break;
-        case LoadType::LOAD_FROM_TIME:
-        case LoadType::LOAD_UNDELETED:
-            load_param_ = LoadParameter{LoadType::LOAD_UNDELETED};
-            break;
-        default:
-            util::raise_rte("Unexpected load state {} applying specific version query", load_param_.load_type_);
-        }
-    }
-
-    void do_react(const pipelines::TimestampVersionQuery& timestamp_query) {
-        ++count_;
-        switch(load_param_.load_type_) {
-        case LoadType::LOAD_LATEST_UNDELETED:
-            load_param_ = LoadParameter{LoadType::LOAD_FROM_TIME, timestamp_query.timestamp_};
-            break;
-        case LoadType::LOAD_FROM_TIME:
-            util::check(load_param_.load_from_time_.has_value(), "Expect LOAD_TO_TIME to have timestamp specificed");
-            load_param_.load_from_time_ = std::min(load_param_.load_from_time_.value(), timestamp_query.timestamp_);
-            break;
-        case LoadType::LOAD_DOWNTO:
-        case LoadType::LOAD_UNDELETED:
-            load_param_ = LoadParameter{LoadType::LOAD_UNDELETED};
-            break;
-        default:
-            util::raise_rte("Unexpected load state {} applying specific version query", load_param_.load_type_);
-        }
-    }
+    LoadParameter load_param_ = LoadParameter{LoadType::NOT_LOADED};
+    folly::small_vector<SnapshotId, 1> snapshots_;
 
-    void do_react(const pipelines::SnapshotVersionQuery&) {
-        util::raise_rte("Snapshot not currently supported in generic batch read version");
-    }
+    explicit StreamVersionData(const pipelines::VersionQuery& version_query);
+    StreamVersionData() = default;
+    void react(const pipelines::VersionQuery& version_query);
+private:
+    void do_react(std::monostate);
+    void do_react(const pipelines::SpecificVersionQuery& specific_version);
+    void do_react(const pipelines::TimestampVersionQuery& timestamp_query);
+    void do_react(const pipelines::SnapshotVersionQuery& snapshot_query);
 };
 
-inline std::optional<AtomKey> get_key_for_version_query(
-    const std::shared_ptr<VersionMapEntry>& version_map_entry, 
-    const pipelines::VersionQuery& version_query) {
-    return util::variant_match(version_query.content_,
-        [&version_map_entry] (const pipelines::SpecificVersionQuery& specific_version) -> std::optional<AtomKey> {
-            auto signed_version_id = specific_version.version_id_;
-            VersionId version_id;
-            if (signed_version_id >= 0) {
-                version_id = static_cast<VersionId>(signed_version_id);
-            } else {
-                auto opt_latest = version_map_entry->get_first_index(true);
-                if (opt_latest.has_value()) {
-                    auto opt_version_id = get_version_id_negative_index(opt_latest->version_id(), signed_version_id);
-                    if (opt_version_id.has_value()) {
-                        version_id = *opt_version_id;
-                    } else {
-                        return std::nullopt;
-                    }
-                } else {
-                    return std::nullopt;
-                }
-            }
-            return find_index_key_for_version_id(version_id, version_map_entry);
-        },
-        [&version_map_entry] (const pipelines::TimestampVersionQuery& timestamp_version) -> std::optional<AtomKey> {
-            auto version_key = get_index_key_from_time(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);
-        },
-        [] (const auto&) -> std::optional<AtomKey> {
-            util::raise_rte("Unsupported version query type");
-        });
-}
-
-inline std::vector<folly::Future<std::optional<AtomKey>>> batch_get_versions_async(
+std::vector<folly::Future<std::optional<AtomKey>>> batch_get_versions_async(
     const std::shared_ptr<Store>& store,
     const std::shared_ptr<VersionMap>& version_map,
     const std::vector<StreamId>& symbols,
     const std::vector<pipelines::VersionQuery>& version_queries,
-    const std::optional<bool>& use_previous_on_error) {
-    ARCTICDB_SAMPLE(BatchGetVersion, 0)
-    util::check(symbols.size() == version_queries.size(), "Symbol and version query list mismatch: {} != {}", symbols.size(), version_queries.size());
-
-    robin_hood::unordered_flat_map<StreamId, StreamVersionData> version_data;
-    for(const auto& symbol : folly::enumerate(symbols)) {
-        auto it = version_data.find(*symbol);
-        if(it == version_data.end())
-            version_data.insert(robin_hood::pair<StreamId, StreamVersionData>(*symbol, StreamVersionData{version_queries[symbol.index]}));
-        else
-            it->second.react(version_queries[symbol.index]);
-    }
-
-    using SplitterType = folly::FutureSplitter<std::shared_ptr<VersionMapEntry>>;
-    robin_hood::unordered_flat_map<StreamId, SplitterType> shared_futures;
-    std::vector<folly::Future<std::optional<AtomKey>>> output;
-    output.reserve(symbols.size());
-    for(const auto& symbol : folly::enumerate(symbols)) {
-        const auto it = version_data.find(*symbol);
-        util::check(it != version_data.end(), "Missing version data for symbol {}", *symbol);
-        auto version_entry_fut = folly::Future<std::shared_ptr<VersionMapEntry>>::makeEmpty();
-
-        if(use_previous_on_error.value_or(false))
-            it->second.load_param_.use_previous_ = true;
-
-        if(it->second.count_ == 1) {
-            version_entry_fut = async::submit_io_task(CheckReloadTask{store, version_map, *symbol, it->second.load_param_});
-        } else {
-            auto fut = shared_futures.find(*symbol);
-            if(fut == shared_futures.end()) {
-                auto [splitter, inserted] = shared_futures.emplace(*symbol, folly::FutureSplitter(async::submit_io_task(CheckReloadTask{store, version_map, *symbol, it->second.load_param_})));
-
-                version_entry_fut = splitter->second.getFuture();
-            } else {
-                version_entry_fut = fut->second.getFuture();
-            }
-        }
-        output.push_back(std::move(version_entry_fut)
-        .thenValue([version_query = version_queries[symbol.index]](auto version_map_entry) {
-            return get_key_for_version_query(version_map_entry, version_query);
-        }));
-    }
-
-    return output;
-}
+    const std::optional<bool>& use_previous_on_error);
 
 inline std::vector<folly::Future<folly::Unit>> batch_write_version(
     const std::shared_ptr<Store> &store,
diff --git a/cpp/arcticdb/version/version_store_api.cpp b/cpp/arcticdb/version/version_store_api.cpp
index 3620548d71..d6907a5940 100644
--- a/cpp/arcticdb/version/version_store_api.cpp
+++ b/cpp/arcticdb/version/version_store_api.cpp
@@ -28,6 +28,7 @@
 #include <arcticdb/version/version_utils.hpp>
 #include <arcticdb/pipeline/pipeline_utils.hpp>
 #include <arcticdb/pipeline/frame_utils.hpp>
+#include <arcticdb/version/snapshot.hpp>
 
 #include <regex>
 
@@ -303,7 +304,40 @@ VersionResultVector PythonVersionStore::list_versions(
        return get_all_versions_for_symbols(store(), version_map(), stream_ids, snapshots_for_symbol, creation_ts_for_version_symbol);
 }
 
-std::vector<std::pair<SnapshotId, py::object>> PythonVersionStore::list_snapshots(const std::optional<bool> load_metadata) {
+namespace {
+
+py::object get_metadata_from_segment(
+    const SegmentInMemory& segment
+) {
+    auto metadata_proto = segment.metadata();
+    py::object pyobj;
+    if (metadata_proto) {
+        arcticdb::proto::descriptors::UserDefinedMetadata user_meta_proto;
+        metadata_proto->UnpackTo(&user_meta_proto);
+        pyobj = python_util::pb_to_python(user_meta_proto);
+    } else {
+        pyobj = pybind11::none();
+    }
+    return pyobj;
+}
+
+py::object get_metadata_for_snapshot(const std::shared_ptr<Store> &store, const VariantKey &snap_key) {
+    auto seg = store->read_sync(snap_key).second;
+    return get_metadata_from_segment(seg);
+}
+
+std::pair<std::vector<AtomKey>, py::object> get_versions_and_metadata_from_snapshot(
+    const std::shared_ptr<Store>& store,
+    const VariantKey& vk
+) {
+    auto snapshot_segment = store->read_sync(vk).second;
+    return {get_versions_from_segment(snapshot_segment), get_metadata_from_segment(snapshot_segment)};
+}
+
+} //namespace
+
+
+std::vector<std::pair<SnapshotId, py::object>> PythonVersionStore::list_snapshots(std::optional<bool> load_metadata) {
     ARCTICDB_RUNTIME_DEBUG(log::version(), "Command: list_snapshots");
     auto snap_ids = std::vector<std::pair<SnapshotId, py::object>>();
     auto fetch_metadata = opt_false(load_metadata);
@@ -436,6 +470,7 @@ void PythonVersionStore::snapshot(
         // User did not provide explicit versions to snapshot, get latest of all filtered symbols.
         auto skip_symbols_set = std::set<StreamId>(skip_symbols.begin(), skip_symbols.end());
         auto all_symbols = list_streams();
+
         std::vector<StreamId> filtered_symbols;
         std::set_difference(all_symbols.begin(), all_symbols.end(), skip_symbols_set.begin(),
                 skip_symbols_set.end(), std::back_inserter(filtered_symbols));
diff --git a/python/arcticdb/log.py b/python/arcticdb/log.py
index a4d034d801..67c1d226e6 100644
--- a/python/arcticdb/log.py
+++ b/python/arcticdb/log.py
@@ -49,6 +49,8 @@ def exception(self, msg, *args, **kwargs):
     "timings": _Logger(_LoggerId.TIMINGS),
     "lock": _Logger(_LoggerId.LOCK),
     "schedule": _Logger(_LoggerId.SCHEDULE),
+    "symbol": _Logger(_LoggerId.SYMBOL),
+    "snapshot": _Logger(_LoggerId.SNAPSHOT)
 }
 
 for key, value in logger_by_name.items():


Test patch:
diff --git a/cpp/arcticdb/async/test/test_async.cpp b/cpp/arcticdb/async/test/test_async.cpp
index ed6aec53fd..b81a8999b6 100644
--- a/cpp/arcticdb/async/test/test_async.cpp
+++ b/cpp/arcticdb/async/test/test_async.cpp
@@ -46,7 +46,7 @@ TEST(Async, SinkBasic) {
         ac::entity::KeyType::GENERATION, 6, 123, 456, 457, 999, std::move(seg), codec_opt, ac::EncodingVersion::V2
     };
 
-    auto v = sched.submit_cpu_task(enc).via(&aa::io_executor()).thenValue(aa::WriteSegmentTask{lib}).get();
+    auto v = sched.submit_cpu_task(std::move(enc)).via(&aa::io_executor()).thenValue(aa::WriteSegmentTask{lib}).get();
 
     ac::HashAccum h;
     auto default_content_hash = h.digest();
diff --git a/cpp/arcticdb/entity/test/test_key_serialization.cpp b/cpp/arcticdb/entity/test/test_key_serialization.cpp
index 8cc7a515aa..6f570d6dcf 100644
--- a/cpp/arcticdb/entity/test/test_key_serialization.cpp
+++ b/cpp/arcticdb/entity/test/test_key_serialization.cpp
@@ -16,7 +16,7 @@ namespace arcticdb {
                            key.version_id(),
                            key.creation_ts(),
                            key.content_hash(),
-                           stream::get_index_value_type(key),
+                           int(stream::get_index_value_type(key)),
                            key.start_index(),
                            key.end_index());
     }
diff --git a/cpp/arcticdb/stream/test/test_types.cpp b/cpp/arcticdb/stream/test/test_types.cpp
index fbdcbeefc1..d519bc67cb 100644
--- a/cpp/arcticdb/stream/test/test_types.cpp
+++ b/cpp/arcticdb/stream/test/test_types.cpp
@@ -14,6 +14,29 @@
 #include <iostream>
 #include <type_traits>
 
+#define GTEST_COUT std::cerr << "[          ] [ INFO ]"
+#define PRINT_TYPE(TYPE) GTEST_COUT << fmt::format("{}: {}", datatype_to_str(DataType::TYPE), static_cast<int>(DataType::TYPE)) << std::endl;
+
+TEST(Types, Print) {
+    using namespace arcticdb;
+    PRINT_TYPE(UINT8)
+    PRINT_TYPE(UINT16)
+    PRINT_TYPE(UINT32)
+    PRINT_TYPE(UINT64)
+    PRINT_TYPE(INT8)
+    PRINT_TYPE(INT16)
+    PRINT_TYPE(INT32)
+    PRINT_TYPE(INT64)
+    PRINT_TYPE(FLOAT32)
+    PRINT_TYPE(FLOAT64)
+    PRINT_TYPE(BOOL8)
+    PRINT_TYPE(NANOSECONDS_UTC64)
+    PRINT_TYPE(ASCII_FIXED64)
+    PRINT_TYPE(ASCII_DYNAMIC64)
+    PRINT_TYPE(UTF_FIXED64)
+    PRINT_TYPE(UTF_DYNAMIC64)
+}
+
 TEST(TickStreamDesc, FromFields) {
     using namespace arcticdb::entity;
     using namespace arcticdb;
@@ -33,7 +56,7 @@ TEST(DataTypeVisit, VisitTag) {
     using namespace arcticdb;
     TypeDescriptor td(DataType::UINT8, 1);
     td.visit_tag([&](auto type_desc_tag) {
-        TypeDescriptor td2 = static_cast<TypeDescriptor>(type_desc_tag);
+        auto td2 = static_cast<TypeDescriptor>(type_desc_tag);
         ASSERT_EQ(td, td2);
         using TD=TypeDescriptorTag<DataTypeTag<DataType::UINT8>, DimensionTag<Dimension::Dim1>>;
         bool b = std::is_same_v<TD, std::decay_t<decltype(type_desc_tag)>>;
diff --git a/cpp/arcticdb/util/test/rapidcheck_generators.hpp b/cpp/arcticdb/util/test/rapidcheck_generators.hpp
index 5ef8c65cd0..dbe1bde819 100644
--- a/cpp/arcticdb/util/test/rapidcheck_generators.hpp
+++ b/cpp/arcticdb/util/test/rapidcheck_generators.hpp
@@ -38,11 +38,11 @@ inline rc::Gen <arcticdb::entity::Dimension> gen_dimension() {
 }
 
 struct TestDataFrame {
-    size_t num_columns_;
-    size_t num_rows_;
+    size_t num_columns_ = 0UL;
+    size_t num_rows_ = 0UL;
     std::vector<arcticdb::entity::TypeDescriptor> types_;
     std::vector<std::vector<int>> data_;
-    arcticdb::entity::timestamp start_ts_;
+    arcticdb::entity::timestamp start_ts_ = 0U;
     std::vector<uint8_t> timestamp_increments_;
     std::vector<std::string> column_names_;
 };
@@ -71,7 +71,7 @@ struct Arbitrary<arcticdb::entity::StreamDescriptor> {
         for (const auto& field_name: field_names) {
             field_descriptors.add_field(arcticdb::entity::scalar_field(*gen_numeric_datatype(), field_name));
         }
-        auto desc =stream_descriptor(arcticdb::entity::StreamId{id}, RowCountIndex{}, arcticdb::fields_from_range(field_descriptors));
+        auto desc =stream_descriptor(arcticdb::entity::StreamId{id}, arcticdb::stream::RowCountIndex{}, arcticdb::fields_from_range(field_descriptors));
         return gen::build<StreamDescriptor>(
             gen::set(&StreamDescriptor::data_, gen::just(desc.data_)),
             gen::set(&StreamDescriptor::fields_, gen::just(desc.fields_))
@@ -107,22 +107,22 @@ struct Arbitrary<TestDataFrame> {
 
 }
 
+namespace ac = arcticdb;
 namespace as = arcticdb::stream;
-namespace ac = arcticdb::entity;
 
-inline as::FixedSchema schema_from_test_frame(const TestDataFrame &data_frame, StreamId stream_id) {
+inline as::FixedSchema schema_from_test_frame(const TestDataFrame &data_frame, ac::StreamId stream_id) {
     arcticdb::FieldCollection fields;
     for (size_t i = 0; i < data_frame.num_columns_; ++i)
         fields.add_field(arcticdb::entity::scalar_field(data_frame.types_[i].data_type(), data_frame.column_names_[i]));
 
     const auto index = as::TimeseriesIndex::default_index();
     return as::FixedSchema{
-        index.create_stream_descriptor(stream_id, fields), index
+        index.create_stream_descriptor(std::move(stream_id), fields), index
     };
 }
 
-inline IndexRange test_frame_range(const TestDataFrame &data_frame) {
-    return IndexRange{data_frame.start_ts_, std::accumulate(data_frame.timestamp_increments_.begin(),
+inline ac::IndexRange test_frame_range(const TestDataFrame &data_frame) {
+    return ac::IndexRange{data_frame.start_ts_, std::accumulate(data_frame.timestamp_increments_.begin(),
                                                             data_frame.timestamp_increments_.end(),
                                                             data_frame.start_ts_)};
 }
@@ -150,19 +150,19 @@ folly::Future<arcticdb::entity::VariantKey> write_frame_data(const TestDataFrame
     return writer.commit();
 }
 
-inline folly::Future<arcticdb::entity::VariantKey> write_test_frame(StreamId stream_id,
-                                                                const TestDataFrame &data_frame,
-                                                                std::shared_ptr<StreamSink> store) {
-    auto schema = schema_from_test_frame(data_frame, stream_id);
+inline folly::Future<arcticdb::entity::VariantKey> write_test_frame(ac::StreamId stream_id,
+                                                                    const TestDataFrame &data_frame,
+                                                                std::shared_ptr<ac::StreamSink> store) {
+    auto schema = schema_from_test_frame(data_frame, std::move(stream_id));
     auto start_end = test_frame_range(data_frame);
     auto gen_id = arcticdb::VersionId(0);
 
-    StreamWriter<TimeseriesIndex, FixedSchema> writer{
+    ac::StreamWriter<ac::TimeseriesIndex, ac::FixedSchema> writer{
         std::move(schema),
-        store,
+        std::move(store),
         gen_id,
         start_end,
-        RowCountSegmentPolicy{4}
+        ac::RowCountSegmentPolicy{4}
     };
 
     return write_frame_data(data_frame, writer);
@@ -178,14 +178,14 @@ bool check_read_frame(const TestDataFrame &data_frame, ReaderType &reader, std::
         timestamp += data_frame.timestamp_increments_[row];
         for (size_t col = 0; col < data_frame.num_columns_; ++col) {
             data_frame.types_[col].visit_tag([&](auto type_desc_tag) {
-                arcticdb::entity::DataType dt = TypeDescriptor(type_desc_tag).data_type();
+                arcticdb::entity::DataType dt = ac::TypeDescriptor(type_desc_tag).data_type();
                 arcticdb::entity::DataType
                     stored_dt = row_ref.segment().column_descriptor(col + 1).type().data_type();
                 if (dt != stored_dt) {
                     errors.push_back(fmt::format("Type mismatch {} != {} at pos {}:{}", dt, stored_dt, col, row));
                     success = false;
                 }
-                auto dimension = static_cast<uint64_t>(TypeDescriptor(type_desc_tag).dimension());
+                auto dimension = static_cast<uint64_t>(ac::TypeDescriptor(type_desc_tag).dimension());
                 auto stored_dimension = row_ref.segment().column_descriptor(col + 1).type().dimension();
                 if (dimension != static_cast<uint64_t>(stored_dimension)) {
                     errors.push_back(fmt::format("Dimension mismatch {} != {} at pos {}:{}",
@@ -216,11 +216,11 @@ bool check_read_frame(const TestDataFrame &data_frame, ReaderType &reader, std::
 
 inline bool check_test_frame(const TestDataFrame &data_frame,
                              const arcticdb::entity::AtomKey &key,
-                             std::shared_ptr<StreamSource> store,
+                             std::shared_ptr<ac::StreamSource> store,
                              std::vector<std::string> &errors) {
-    StreamReader<arcticdb::entity::AtomKey, folly::Function<std::vector<arcticdb::entity::AtomKey>()>, arcticdb::SegmentInMemory::Row> stream_reader{
+    ac::StreamReader<arcticdb::entity::AtomKey, folly::Function<std::vector<arcticdb::entity::AtomKey>()>, arcticdb::SegmentInMemory::Row> stream_reader{
         [&]() { return std::vector<arcticdb::entity::AtomKey>{key}; },
-        store
+        std::move(store)
     };
 
     return check_read_frame(data_frame, stream_reader, errors);
diff --git a/cpp/arcticdb/version/test/test_version_store.cpp b/cpp/arcticdb/version/test/test_version_store.cpp
index 0e85aed4f8..d630d06fd0 100644
--- a/cpp/arcticdb/version/test/test_version_store.cpp
+++ b/cpp/arcticdb/version/test/test_version_store.cpp
@@ -31,8 +31,8 @@ struct VersionStoreTest : arcticdb::TestStore {
 };
 
 auto write_version_frame(
-    const StreamId& stream_id,
-    VersionId v_id,
+    const arcticdb::StreamId& stream_id,
+    arcticdb::VersionId v_id,
     arcticdb::version_store::PythonVersionStore& pvs,
     size_t rows = 1000000,
     bool update_version_map = false,
diff --git a/python/tests/conftest.py b/python/tests/conftest.py
index 377eb2302c..f9935c10fc 100644
--- a/python/tests/conftest.py
+++ b/python/tests/conftest.py
@@ -583,7 +583,7 @@ def lmdb_version_store_dynamic_schema_v2(version_store_factory, lib_name):
 
 @pytest.fixture
 def lmdb_version_store_dynamic_schema(
-    lmdb_version_store_dynamic_schema_v1, lmdb_version_store_dynamic_schema_v2, encoding_version
+        lmdb_version_store_dynamic_schema_v1, lmdb_version_store_dynamic_schema_v2, encoding_version
 ):
     if encoding_version == EncodingVersion.V1:
         return lmdb_version_store_dynamic_schema_v1
diff --git a/python/tests/integration/arcticdb/test_arctic_batch.py b/python/tests/integration/arcticdb/test_arctic_batch.py
index 583557d286..01a67ee1ab 100644
--- a/python/tests/integration/arcticdb/test_arctic_batch.py
+++ b/python/tests/integration/arcticdb/test_arctic_batch.py
@@ -18,7 +18,9 @@
 from arcticdb.arctic import Arctic
 from arcticdb.options import LibraryOptions
 from arcticdb import QueryBuilder, DataError
+from arcticdb_ext.version_store import AtomKey, RefKey
 
+import time
 import pytest
 import pandas as pd
 from datetime import datetime, date, timezone, timedelta
@@ -1305,49 +1307,122 @@ def test_read_description_batch_empty_nat(arctic_library):
         assert np.isnat(results_list[sym].date_range[1]) == True
 
 
-def test_mutlithread_read(library_factory):
-    set_config_int("VersionStore.NumCPUThreads", 2)
-    set_config_int("VersionStore.NumIOThreads", 2)
-    num_days = 2
+def test_read_batch_mixed_with_snapshots(arctic_library):
     num_symbols = 10
-    dt = datetime(2019, 4, 8, 0, 0, 0)
-    num_columns = 10
-    num_rows_per_day = 5
-    write_requests = []
-    read_requests = []
-    list_dataframes = {}
-    columns = random_strings_of_length(num_columns, num_columns, True)
-    df = generate_dataframe(random.sample(columns, num_columns), dt, num_days, num_rows_per_day)
-    for sym in range(num_symbols):
-        write_requests.append(WritePayload("sym_" + str(sym), df, metadata="great_metadata" + str(sym)))
-        read_requests.append("sym_" + str(sym))
-        list_dataframes[sym] = df
-    lib = library_factory(LibraryOptions(rows_per_segment=10, dedup=True))
-    lib.write_batch(write_requests)
-    q = queue.Queue()
-
-    def read_batch():
-        try:
-            read_batch_result = lib.read_batch(read_requests)
-            for sym in range(num_symbols):
-                assert read_batch_result[sym].metadata == "great_metadata" + str(sym)
-                assert_frame_equal(read_batch_result[sym].data, list_dataframes[sym])
-        except Exception as e:
-            q.put(e)
-
-    def read():
-        try:
-            for sym in range(num_symbols):
-                assert_frame_equal(lib.read("sym_" + str(sym)).data, list_dataframes[sym])
-        except Exception as e:
-            q.put(e)
-
-    read_th_count = 1
-    batch_read_th_count = 1
-    thread_list = []
-    [thread_list.append(threading.Thread(target=read)) for _ in range(read_th_count)]
-    [thread_list.append(threading.Thread(target=read_batch)) for _ in range(batch_read_th_count)]
-    [thread.start() for thread in thread_list]
-    [thread.join() for thread in thread_list]
-    if not q.empty():
-        raise Exception([e for e in q.queue])
+    num_versions = 10
+
+    def dataframe_for_offset(version_num, symbol_num):
+        offset = (version_num * num_versions) + symbol_num
+        return pd.DataFrame({"x": np.arange(offset, offset + 10)})
+
+    def dataframe_and_symbol(version_num, symbol_num):
+        symbol_name = "symbol_{}".format(symbol_num)
+        dataframe = dataframe_for_offset(version_num, symbol_num)
+        return symbol_name, dataframe
+
+    lib = arctic_library
+    version_write_times = []
+
+    for version in range(num_versions):
+        version_write_times.append(pd.Timestamp.now())
+        time.sleep(1)
+        for sym in range(num_symbols):
+            symbol, df = dataframe_and_symbol(version, sym)
+            lib.write(symbol, df)
+
+        lib.snapshot("snap_{}".format(version))
+
+    read_requests = [
+        ReadRequest("symbol_1", as_of=None),
+        ReadRequest("symbol_1", as_of=4),
+        ReadRequest("symbol_1", as_of="snap_7"),
+        ReadRequest("symbol_2", as_of="snap_7"),
+        ReadRequest("symbol_2", as_of=None),
+        ReadRequest("symbol_2", as_of=4),
+        ReadRequest("symbol_3", as_of="snap_7"),
+        ReadRequest("symbol_3", as_of="snap_3"),
+        ReadRequest("symbol_3", as_of="snap_1"),
+        ReadRequest("symbol_3", as_of=None),
+        ReadRequest("symbol_3", as_of=3),
+    ]
+
+    vits = lib.read_batch(read_requests)
+
+    expected = dataframe_for_offset(9, 1)
+    assert_frame_equal(vits[0].data, expected)
+    expected = dataframe_for_offset(4, 1)
+    assert_frame_equal(vits[1].data, expected)
+    expected = dataframe_for_offset(7, 1)
+    assert_frame_equal(vits[2].data, expected)
+
+    expected = dataframe_for_offset(7, 2)
+    assert_frame_equal(vits[3].data, expected)
+    expected = dataframe_for_offset(9, 2)
+    assert_frame_equal(vits[4].data, expected)
+    expected = dataframe_for_offset(4, 2)
+    assert_frame_equal(vits[5].data, expected)
+
+    expected = dataframe_for_offset(7, 3)
+    assert_frame_equal(vits[6].data, expected)
+    expected = dataframe_for_offset(3, 3)
+    assert_frame_equal(vits[7].data, expected)
+    expected = dataframe_for_offset(1, 3)
+    assert_frame_equal(vits[8].data, expected)
+    expected = dataframe_for_offset(9, 3)
+    assert_frame_equal(vits[9].data, expected)
+    expected = dataframe_for_offset(3, 3)
+    assert_frame_equal(vits[10].data, expected)
+
+    # Trigger iteration of old-style snapshot keys
+    library_tool = lib._nvs.library_tool()
+    snap_key = RefKey("snap_8", KeyType.SNAPSHOT_REF)
+    snap_segment = library_tool.read_to_segment(snap_key)
+    old_style_snap_key = AtomKey("snap_8", 1, 2, 3, 4, 5, KeyType.SNAPSHOT)
+    library_tool.write(old_style_snap_key, snap_segment)
+    assert library_tool.count_keys(KeyType.SNAPSHOT_REF) == 10
+    library_tool.remove(snap_key)
+    assert library_tool.count_keys(KeyType.SNAPSHOT_REF) == 9
+    assert library_tool.count_keys(KeyType.SNAPSHOT) == 1
+
+    read_requests = [
+        ReadRequest("symbol_7", as_of=None),
+        ReadRequest("symbol_8", as_of=4),
+        ReadRequest("symbol_2", as_of="snap_8"),
+        ReadRequest("symbol_3", as_of="snap_7"),
+        ReadRequest("symbol_1", as_of=4),
+    ]
+
+    vits = lib.read_batch(read_requests)
+    expected = dataframe_for_offset(9, 7)
+    assert_frame_equal(vits[0].data, expected)
+    expected = dataframe_for_offset(4, 8)
+    assert_frame_equal(vits[1].data, expected)
+    expected = dataframe_for_offset(8, 2)
+    assert_frame_equal(vits[2].data, expected)
+    expected = dataframe_for_offset(7, 3)
+    assert_frame_equal(vits[3].data, expected)
+    expected = dataframe_for_offset(4, 1)
+    assert_frame_equal(vits[4].data, expected)
+
+    read_requests = [
+        ReadRequest("symbol_6", as_of="snap_1"),
+        ReadRequest("symbol_6", as_of="snap_3"),
+        ReadRequest("symbol_6", as_of="snap_2"),
+        ReadRequest("symbol_6", as_of="snap_7"),
+        ReadRequest("symbol_6", as_of="snap_9"),
+        ReadRequest("symbol_6", as_of="snap_4"),
+    ]
+
+    vits = lib.read_batch(read_requests)
+    expected = dataframe_for_offset(1, 6)
+    assert_frame_equal(vits[0].data, expected)
+    expected = dataframe_for_offset(3, 6)
+    assert_frame_equal(vits[1].data, expected)
+    expected = dataframe_for_offset(2, 6)
+    assert_frame_equal(vits[2].data, expected)
+    expected = dataframe_for_offset(7, 6)
+    assert_frame_equal(vits[3].data, expected)
+    expected = dataframe_for_offset(9, 6)
+    assert_frame_equal(vits[4].data, expected)
+    expected = dataframe_for_offset(4, 6)
+    assert_frame_equal(vits[5].data, expected)
diff --git a/python/tests/integration/arcticdb/version_store/test_snapshot.py b/python/tests/integration/arcticdb/version_store/test_snapshot.py
index c8335848dc..b1f0f8d2a2 100644
--- a/python/tests/integration/arcticdb/version_store/test_snapshot.py
+++ b/python/tests/integration/arcticdb/version_store/test_snapshot.py
@@ -101,15 +101,17 @@ def test_list_symbols_with_snaps(object_version_store):
 
 
 def test_list_versions(object_version_store):
+    lib = object_version_store
     original_data = [1, 2, 3]
 
-    object_version_store.write("t1", original_data)
-    object_version_store.write("t2", original_data)
+    lib.write("t1", original_data)
+    lib.write("t2", original_data)
 
-    object_version_store.snapshot("snap_versions")
-    object_version_store.write("t1", original_data)
+    lib.snapshot("snap_versions")
+    lib.write("t1", original_data)
 
-    all_versions = object_version_store.list_versions()
+    all_versions = lib.list_versions()
+    print(all_versions)
 
     assert sorted([v["version"] for v in all_versions if v["symbol"] == "t1" and not v["deleted"]]) == sorted([0, 1])
 
diff --git a/python/tests/stress/arcticdb/version_store/test_stress_version_map_compact.py b/python/tests/stress/arcticdb/version_store/test_stress_version_map_compact.py
index 01e7c0dd99..42e92ffde8 100644
--- a/python/tests/stress/arcticdb/version_store/test_stress_version_map_compact.py
+++ b/python/tests/stress/arcticdb/version_store/test_stress_version_map_compact.py
@@ -15,6 +15,10 @@
 from arcticdb_ext import set_config_int
 from arcticdb import log
 
+from arcticdb.config import set_log_level
+
+# set_log_level("DEBUG")
+
 
 def write_data(lib, sym, done, error):
     set_config_int("VersionMap.ReloadInterval", 1)
@@ -27,9 +31,11 @@ def write_data(lib, sym, done, error):
             for idx2 in range(40):
                 if idx2 % 4 == 3:
                     lib.delete_version(sym, delete_version_id)
+                    print("Doing delete {}/{}".format(idx1, idx2))
                     delete_version_id += 1
                 else:
                     number_of_writes += 1
+                    print("Doing write {}/{}".format(idx1, idx2))
                     lib.write(sym, idx2)
             vs = set([v["version"] for v in lib.list_versions(sym)])
             assert len(vs) == number_of_writes - delete_version_id
@@ -41,6 +47,8 @@ def write_data(lib, sym, done, error):
     except Exception as e:
         print(e)
         error.value = 1
+
+    print("Setting done")
     done.value = 1
 
 
diff --git a/python/tests/unit/arcticdb/test_string.py b/python/tests/unit/arcticdb/test_string.py
index 82f633fb80..1dc2c6bdd0 100644
--- a/python/tests/unit/arcticdb/test_string.py
+++ b/python/tests/unit/arcticdb/test_string.py
@@ -202,3 +202,12 @@ def test_string_encoding_error_message(lmdb_version_store_tiny_segment):
         lib.update("sym_update", df, dynamic_strings=True)
     exception_message = str(e.value)
     assert all(string in exception_message for string in ["broken_column", "row 2", "float"])
+
+
+def test_write_dynamic_simple(lmdb_version_store_v2):
+    row = pd.Series(["Aaba", "A", "B", "C", "Baca", "CABA", "dog", "cat", "here is a very long one"])
+    df = pd.DataFrame({"x": row})
+    lmdb_version_store_v2.write("strings", df, dynamic_strings=True)
+    vit = lmdb_version_store_v2.read("strings")
+    assert_frame_equal(df, vit.data)
+