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

Issue:
Example implementation of enumerate
Example of internal implementation of enumerate


Discussion / hints:

Fix patch:
diff --git a/cpp/arcticdb/pipeline/query.hpp b/cpp/arcticdb/pipeline/query.hpp
index 5d20744500..db5b1717a5 100644
--- a/cpp/arcticdb/pipeline/query.hpp
+++ b/cpp/arcticdb/pipeline/query.hpp
@@ -96,7 +96,6 @@ using VersionQueryType = std::variant<
 
 struct VersionQuery {
     VersionQueryType content_;
-    std::optional<bool> iterate_on_failure_;
 
     void set_snap_name(const std::string& snap_name) {
         content_ = SnapshotVersionQuery{snap_name};
@@ -109,10 +108,6 @@ struct VersionQuery {
     void set_version(SignedVersionId version) {
         content_ = SpecificVersionQuery{version};
     }
-
-    void set_iterate_on_failure(const std::optional<bool>& iterate_on_failure) {
-        iterate_on_failure_ = iterate_on_failure;
-    }
 };
 
 template<typename ContainerType>
diff --git a/cpp/arcticdb/util/constants.hpp b/cpp/arcticdb/util/constants.hpp
index d10f4984c8..4f9779af81 100644
--- a/cpp/arcticdb/util/constants.hpp
+++ b/cpp/arcticdb/util/constants.hpp
@@ -19,7 +19,9 @@ constexpr auto string_nan = std::numeric_limits<position_t>::max() - 1;
 
 constexpr auto NaT = std::numeric_limits<timestamp>::min();
 
-static constexpr decltype(timestamp(0) - timestamp(0)) ONE_SECOND = 1'000'000'000;
+static constexpr decltype(timestamp(0) - timestamp(0)) ONE_MILLISECOND = 1'000'000;
+
+static constexpr decltype(timestamp(0) - timestamp(0)) ONE_SECOND = 1'000 * ONE_MILLISECOND;
 
 static constexpr decltype(timestamp(0) - timestamp(0)) ONE_MINUTE = 60 * ONE_SECOND;
 
diff --git a/cpp/arcticdb/version/local_versioned_engine.cpp b/cpp/arcticdb/version/local_versioned_engine.cpp
index ba7508fe24..5defea827f 100644
--- a/cpp/arcticdb/version/local_versioned_engine.cpp
+++ b/cpp/arcticdb/version/local_versioned_engine.cpp
@@ -208,9 +208,8 @@ std::string LocalVersionedEngine::dump_versions(const StreamId& stream_id) {
 }
 
 std::optional<VersionedItem> LocalVersionedEngine::get_latest_version(
-    const StreamId &stream_id,
-    const VersionQuery& version_query) {
-    auto key = get_latest_undeleted_version(store(), version_map(), stream_id,  version_query);
+    const StreamId &stream_id) {
+    auto key = get_latest_undeleted_version(store(), version_map(), stream_id);
     if (!key) {
         ARCTICDB_DEBUG(log::version(), "get_latest_version didn't find version for stream_id: {}", stream_id);
         return std::nullopt;
@@ -220,16 +219,15 @@ std::optional<VersionedItem> LocalVersionedEngine::get_latest_version(
 
 std::optional<VersionedItem> LocalVersionedEngine::get_specific_version(
     const StreamId &stream_id,
-    SignedVersionId signed_version_id,
-    const VersionQuery& version_query) {
+    SignedVersionId signed_version_id) {
     ARCTICDB_RUNTIME_DEBUG(log::version(), "Command: get_specific_version");
-    auto key = ::arcticdb::get_specific_version(store(), version_map(), stream_id, signed_version_id, version_query);
+    auto key = ::arcticdb::get_specific_version(store(), version_map(), stream_id, signed_version_id);
     if (!key) {
         VersionId version_id;
         if (signed_version_id >= 0) {
             version_id = static_cast<VersionId>(signed_version_id);
         } else {
-            auto [opt_latest_key, deleted] = ::arcticdb::get_latest_version(store(), version_map(), stream_id, version_query);
+            auto [opt_latest_key, deleted] = ::arcticdb::get_latest_version(store(), version_map(), stream_id);
             if (opt_latest_key.has_value()) {
                 auto opt_version_id = get_version_id_negative_index(opt_latest_key->version_id(), signed_version_id);
                 if (opt_version_id.has_value()) {
@@ -261,11 +259,10 @@ std::optional<VersionedItem> LocalVersionedEngine::get_specific_version(
 
 std::optional<VersionedItem> LocalVersionedEngine::get_version_at_time(
     const StreamId& stream_id,
-    timestamp as_of,
-    const VersionQuery& version_query
+    timestamp as_of
     ) {
 
-    auto index_key = load_index_key_from_time(store(), version_map(), stream_id, as_of, version_query);
+    auto index_key = load_index_key_from_time(store(), version_map(), stream_id, as_of);
     if (!index_key) {
         auto index_keys = get_index_keys_in_snapshots(store(), stream_id);
         auto vector_index_keys = std::vector<AtomKey>(index_keys.begin(), index_keys.end());
@@ -309,17 +306,17 @@ std::optional<VersionedItem> LocalVersionedEngine::get_version_to_read(
     const VersionQuery &version_query
     ) {
     return util::variant_match(version_query.content_,
-       [&stream_id, &version_query, this](const SpecificVersionQuery &specific) {
-            return get_specific_version(stream_id, specific.version_id_, version_query);
+       [&stream_id, this](const SpecificVersionQuery &specific) {
+            return get_specific_version(stream_id, specific.version_id_);
         },
         [&stream_id, this](const SnapshotVersionQuery &snapshot) {
             return get_version_from_snapshot(stream_id, snapshot.name_);
         },
-        [&stream_id, &version_query, this](const TimestampVersionQuery &timestamp) {
-            return get_version_at_time(stream_id, timestamp.timestamp_, version_query);
+        [&stream_id, this](const TimestampVersionQuery &timestamp) {
+            return get_version_at_time(stream_id, timestamp.timestamp_);
         },
-        [&stream_id, &version_query, this](const std::monostate &) {
-            return get_latest_version(stream_id, version_query);
+        [&stream_id, this](const std::monostate &) {
+            return get_latest_version(stream_id);
     }
     );
 }
@@ -466,7 +463,7 @@ std::shared_ptr<DeDupMap> LocalVersionedEngine::get_de_dup_map(
     ){
     auto de_dup_map = std::make_shared<DeDupMap>();
     if (write_options.de_duplication) {
-        auto maybe_undeleted_prev = get_latest_undeleted_version(store(), version_map(), stream_id, VersionQuery{});
+        auto maybe_undeleted_prev = get_latest_undeleted_version(store(), version_map(), stream_id);
         if (maybe_undeleted_prev) {
             // maybe_undeleted_prev is index key
             auto data_keys = get_data_keys(store(), {maybe_undeleted_prev.value()}, storage::ReadKeyOpts{});
@@ -491,7 +488,7 @@ std::shared_ptr<DeDupMap> LocalVersionedEngine::get_de_dup_map(
 
 
 VersionedItem LocalVersionedEngine::sort_index(const StreamId& stream_id, bool dynamic_schema) {
-    auto maybe_prev = get_latest_undeleted_version(store(), version_map(), stream_id, VersionQuery{});
+    auto maybe_prev = get_latest_undeleted_version(store(), version_map(), stream_id);
     util::check(maybe_prev.has_value(), "Cannot delete from non-existent symbol {}", stream_id);
     auto version_id = get_next_version_from_key(*maybe_prev);
     auto [index_segment_reader, slice_and_keys] = index::read_index_to_vector(store(), *maybe_prev);
@@ -531,7 +528,7 @@ VersionedItem LocalVersionedEngine::delete_range_internal(
     const StreamId& stream_id,
     const UpdateQuery & query,
     bool dynamic_schema) {
-    auto maybe_prev = get_latest_undeleted_version(store(), version_map(), stream_id, VersionQuery{});
+    auto maybe_prev = get_latest_undeleted_version(store(), version_map(), stream_id);
     util::check(maybe_prev.has_value(), "Cannot delete from non-existent symbol {}", stream_id);
     auto versioned_item = delete_range_impl(store(),
                                             *maybe_prev,
@@ -553,8 +550,7 @@ VersionedItem LocalVersionedEngine::update_internal(
     py::gil_scoped_release release_gil;
     auto update_info = get_latest_undeleted_version_and_next_version_id(store(),
                                                                         version_map(),
-                                                                        stream_id,
-                                                                        VersionQuery{});
+                                                                        stream_id);
     if (update_info.previous_index_key_.has_value()) {
         if (frame->empty()) {
             ARCTICDB_DEBUG(log::version(), "Updating existing data with an empty item has no effect. \n"
@@ -602,8 +598,7 @@ VersionedItem LocalVersionedEngine::write_versioned_metadata_internal(
     ) {
     auto update_info = get_latest_undeleted_version_and_next_version_id(store(),
                                                                         version_map(),
-                                                                        stream_id,
-                                                                        VersionQuery{});
+                                                                        stream_id);
     if(update_info.previous_index_key_.has_value()) {
         ARCTICDB_DEBUG(log::version(), "write_versioned_dataframe for stream_id: {}", stream_id);
         auto index_key = UpdateMetadataTask{store(), update_info, std::move(user_meta)}();
@@ -690,7 +685,7 @@ VersionedItem LocalVersionedEngine::write_versioned_dataframe_internal(
     ARCTICDB_SAMPLE(WriteVersionedDataFrame, 0)
     py::gil_scoped_release release_gil;
     ARCTICDB_RUNTIME_DEBUG(log::version(), "Command: write_versioned_dataframe");
-    auto [maybe_prev, deleted] = ::arcticdb::get_latest_version(store(), version_map(), stream_id, VersionQuery{});
+    auto [maybe_prev, deleted] = ::arcticdb::get_latest_version(store(), version_map(), stream_id);
     auto version_id = get_next_version_from_key(maybe_prev);
     ARCTICDB_DEBUG(log::version(), "write_versioned_dataframe for stream_id: {} , version_id = {}", stream_id, version_id);
     auto write_options = get_write_options();
@@ -720,7 +715,7 @@ std::pair<VersionedItem, TimeseriesDescriptor> LocalVersionedEngine::restore_ver
     auto version_to_restore = get_version_to_read(stream_id, version_query);
     missing_data::check<ErrorCode::E_NO_SUCH_VERSION>(static_cast<bool>(version_to_restore),
                                                  "Unable to restore {}@{}: version not found", stream_id, version_query);
-    auto [maybe_prev, deleted] = ::arcticdb::get_latest_version(store(), version_map(), stream_id, VersionQuery{});
+    auto [maybe_prev, deleted] = ::arcticdb::get_latest_version(store(), version_map(), stream_id);
     ARCTICDB_DEBUG(log::version(), "restore for stream_id: {} , version_id = {}", stream_id, version_to_restore->key_.version_id());
     return AsyncRestoreVersionTask{store(), version_map(), stream_id, version_to_restore->key_, maybe_prev}().get();
 }
@@ -733,7 +728,7 @@ VersionedItem LocalVersionedEngine::write_individual_segment(
     ARCTICDB_SAMPLE(WriteVersionedDataFrame, 0)
 
     ARCTICDB_RUNTIME_DEBUG(log::version(), "Command: write_versioned_dataframe");
-    auto [maybe_prev, deleted] = ::arcticdb::get_latest_version(store(), version_map(), stream_id, VersionQuery{});
+    auto [maybe_prev, deleted] = ::arcticdb::get_latest_version(store(), version_map(), stream_id);
     auto version_id = get_next_version_from_key(maybe_prev);
     ARCTICDB_DEBUG(log::version(), "write_versioned_dataframe for stream_id: {} , version_id = {}", stream_id, version_id);
     auto index = index_type_from_descriptor(segment.descriptor());
@@ -851,10 +846,10 @@ folly::Future<folly::Unit> LocalVersionedEngine::delete_trees_responsibly(
         {
             auto min_versions = min_versions_for_each_stream(orig_keys_to_delete);
             for (const auto& min : min_versions) {
-                auto load_param = load_type == LoadType::LOAD_DOWNTO
-                        ? LoadParameter{load_type, static_cast<SignedVersionId>(min.second)}
-                        : LoadParameter{load_type};
-                const auto entry = version_map()->check_reload(store(), min.first, load_param, __FUNCTION__);
+                auto load_strategy = load_type == LoadType::DOWNTO
+                        ? LoadStrategy{load_type, LoadObjective::INCLUDE_DELETED, static_cast<SignedVersionId>(min.second)}
+                        : LoadStrategy{load_type, LoadObjective::INCLUDE_DELETED};
+                const auto entry = version_map()->check_reload(store(), min.first, load_strategy, __FUNCTION__);
                 entry_map.try_emplace(std::move(min.first), entry);
             }
         }
@@ -987,7 +982,7 @@ VersionedItem LocalVersionedEngine::compact_incomplete_dynamic(
     bool prune_previous_versions) {
     log::version().debug("Compacting incomplete symbol {}", stream_id);
 
-    auto update_info = get_latest_undeleted_version_and_next_version_id(store(), version_map(), stream_id, VersionQuery{});
+    auto update_info = get_latest_undeleted_version_and_next_version_id(store(), version_map(), stream_id);
     auto versioned_item =  compact_incomplete_impl(
             store_, stream_id, user_meta, update_info,
             append, convert_int_to_float, via_iteration, sparsify, get_write_options());
@@ -1003,7 +998,7 @@ VersionedItem LocalVersionedEngine::compact_incomplete_dynamic(
 
 bool LocalVersionedEngine::is_symbol_fragmented(const StreamId& stream_id, std::optional<size_t> segment_size) {
     auto update_info = get_latest_undeleted_version_and_next_version_id(
-            store(), version_map(), stream_id, VersionQuery{});
+            store(), version_map(), stream_id);
     auto options = get_write_options();
     auto pre_defragmentation_info = get_pre_defragmentation_info(
         store(), stream_id, update_info, options, segment_size.value_or(options.segment_row_size));
@@ -1015,7 +1010,7 @@ VersionedItem LocalVersionedEngine::defragment_symbol_data(const StreamId& strea
 
     // Currently defragmentation only for latest version - is there a use-case to allow compaction for older data?
     auto update_info = get_latest_undeleted_version_and_next_version_id(
-        store(), version_map(), stream_id, VersionQuery{});
+        store(), version_map(), stream_id);
 
     auto options = get_write_options();
     auto versioned_item = defragment_symbol_data_impl(
@@ -1365,8 +1360,7 @@ VersionedItem LocalVersionedEngine::append_internal(
     py::gil_scoped_release release_gil;
     auto update_info = get_latest_undeleted_version_and_next_version_id(store(),
                                                                         version_map(),
-                                                                        stream_id,
-                                                                        VersionQuery{});
+                                                                        stream_id);
 
     if(update_info.previous_index_key_.has_value()) {
         if (frame->empty()) {
@@ -1670,7 +1664,7 @@ VersionedItem LocalVersionedEngine::sort_merge_internal(
     bool via_iteration,
     bool sparsify
     ) {
-    auto update_info = get_latest_undeleted_version_and_next_version_id(store(), version_map(), stream_id, VersionQuery{});
+    auto update_info = get_latest_undeleted_version_and_next_version_id(store(), version_map(), stream_id);
     auto versioned_item = sort_merge_impl(store_, stream_id, user_meta, update_info, append, convert_int_to_float, via_iteration, sparsify);
     version_map()->write_version(store(), versioned_item.key_, update_info.previous_index_key_);
     return versioned_item;
@@ -1708,7 +1702,7 @@ timestamp LocalVersionedEngine::latest_timestamp(const std::string& symbol) {
     if(auto latest_incomplete = latest_incomplete_timestamp(store(), symbol); latest_incomplete)
         return *latest_incomplete;
 
-    if(auto latest_key = get_latest_version(symbol, VersionQuery{}); latest_key)
+    if(auto latest_key = get_latest_version(symbol); latest_key)
         return latest_key->key_.end_time();
 
     return -1;
diff --git a/cpp/arcticdb/version/local_versioned_engine.hpp b/cpp/arcticdb/version/local_versioned_engine.hpp
index 04ac6fdd32..5826570325 100644
--- a/cpp/arcticdb/version/local_versioned_engine.hpp
+++ b/cpp/arcticdb/version/local_versioned_engine.hpp
@@ -101,18 +101,15 @@ class LocalVersionedEngine : public VersionedEngine {
     ) override;
 
     std::optional<VersionedItem> get_latest_version(
-        const StreamId &stream_id,
-        const VersionQuery& version_query);
+        const StreamId &stream_id);
 
     std::optional<VersionedItem> get_specific_version(
         const StreamId &stream_id,
-        SignedVersionId signed_version_id,
-        const VersionQuery& version_query);
+        SignedVersionId signed_version_id);
 
     std::optional<VersionedItem> get_version_at_time(
         const StreamId& stream_id,
-        timestamp as_of,
-        const VersionQuery& version_query);
+        timestamp as_of);
 
     std::optional<VersionedItem> get_version_from_snapshot(
         const StreamId& stream_id,
diff --git a/cpp/arcticdb/version/python_bindings.cpp b/cpp/arcticdb/version/python_bindings.cpp
index 3a7cb6172c..ed6795ad5a 100644
--- a/cpp/arcticdb/version/python_bindings.cpp
+++ b/cpp/arcticdb/version/python_bindings.cpp
@@ -98,8 +98,7 @@ void register_bindings(py::module &version, py::exception<arcticdb::ArcticExcept
         .def(py::init())
         .def("set_snap_name", &VersionQuery::set_snap_name)
         .def("set_timestamp", &VersionQuery::set_timestamp)
-        .def("set_version", &VersionQuery::set_version)
-        .def("set_iterate_on_failure", &VersionQuery::set_iterate_on_failure);
+        .def("set_version", &VersionQuery::set_version);
 
     py::class_<ReadOptions>(version, "PythonVersionStoreReadOptions")
         .def(py::init())
@@ -671,10 +670,9 @@ void register_bindings(py::module &version, py::exception<arcticdb::ArcticExcept
                 const std::optional<StreamId> & s_id,
                 const std::optional<SnapshotId> & snap_id,
                 const std::optional<bool>& latest,
-                const std::optional<bool>& iterate_on_failure,
                 const std::optional<bool>& skip_snapshots
                 ){
-                 return v.list_versions(s_id, snap_id, latest, iterate_on_failure, skip_snapshots);
+                 return v.list_versions(s_id, snap_id, latest, skip_snapshots);
              },
              py::call_guard<SingleThreadMutexHolder>(), "List all the version ids for this store.")
         .def("_compact_version_map",
diff --git a/cpp/arcticdb/version/version_functions.hpp b/cpp/arcticdb/version/version_functions.hpp
index f213d884b5..46dda3969e 100644
--- a/cpp/arcticdb/version/version_functions.hpp
+++ b/cpp/arcticdb/version/version_functions.hpp
@@ -14,31 +14,23 @@
 
 namespace arcticdb {
 
-inline void set_load_param_options(LoadParameter& load_param, const pipelines::VersionQuery& version_query) {
-    load_param.iterate_on_failure_ = version_query.iterate_on_failure_.value_or(false);
-}
-
 inline std::optional<AtomKey> get_latest_undeleted_version(
     const std::shared_ptr<Store> &store,
     const std::shared_ptr<VersionMap> &version_map,
-    const StreamId &stream_id,
-    const pipelines::VersionQuery& version_query) {
+    const StreamId &stream_id) {
     ARCTICDB_RUNTIME_SAMPLE(GetLatestUndeletedVersion, 0)
-    LoadParameter load_param{LoadType::LOAD_LATEST_UNDELETED};
-    set_load_param_options(load_param, version_query);
-    const auto entry = version_map->check_reload(store, stream_id, load_param, __FUNCTION__);
+    LoadStrategy load_strategy{LoadType::LATEST, LoadObjective::UNDELETED_ONLY};
+    const auto entry = version_map->check_reload(store, stream_id, load_strategy, __FUNCTION__);
     return entry->get_first_index(false).first;
 }
 
 inline std::pair<std::optional<AtomKey>, bool> get_latest_version(
     const std::shared_ptr<Store> &store,
     const std::shared_ptr<VersionMap> &version_map,
-    const StreamId &stream_id,
-    const pipelines::VersionQuery& version_query) {
+    const StreamId &stream_id) {
     ARCTICDB_SAMPLE(GetLatestVersion, 0)
-    LoadParameter load_param{LoadType::LOAD_LATEST};
-    set_load_param_options(load_param, version_query);
-    auto entry = version_map->check_reload(store, stream_id, load_param, __FUNCTION__);
+    LoadStrategy load_strategy{LoadType::LATEST, LoadObjective::INCLUDE_DELETED};
+    auto entry = version_map->check_reload(store, stream_id, load_strategy, __FUNCTION__);
     return entry->get_first_index(true);
 }
 
@@ -46,12 +38,10 @@ inline std::pair<std::optional<AtomKey>, bool> get_latest_version(
 inline version_store::UpdateInfo get_latest_undeleted_version_and_next_version_id(
         const std::shared_ptr<Store> &store,
         const std::shared_ptr<VersionMap> &version_map,
-        const StreamId &stream_id,
-        const pipelines::VersionQuery& version_query) {
+        const StreamId &stream_id) {
     ARCTICDB_SAMPLE(GetLatestUndeletedVersionAndHighestVersionId, 0)
-    LoadParameter load_param{LoadType::LOAD_LATEST_UNDELETED};
-        set_load_param_options(load_param, version_query);
-    auto entry = version_map->check_reload(store, stream_id, load_param, __FUNCTION__);
+    LoadStrategy load_strategy{LoadType::LATEST, LoadObjective::UNDELETED_ONLY};
+    auto entry = version_map->check_reload(store, stream_id, load_strategy, __FUNCTION__);
     auto latest_version = entry->get_first_index(true).first;
     auto latest_undeleted_version = entry->get_first_index(false).first;
     VersionId next_version_id = latest_version.has_value() ? latest_version->version_id() + 1 : 0;
@@ -61,13 +51,11 @@ inline version_store::UpdateInfo get_latest_undeleted_version_and_next_version_i
 inline std::vector<AtomKey> get_all_versions(
     const std::shared_ptr<Store> &store,
     const std::shared_ptr<VersionMap> &version_map,
-    const StreamId &stream_id,
-    const pipelines::VersionQuery& version_query
+    const StreamId &stream_id
     ) {
     ARCTICDB_SAMPLE(GetAllVersions, 0)
-    LoadParameter load_param{LoadType::LOAD_UNDELETED};
-    set_load_param_options(load_param, version_query);
-    auto entry = version_map->check_reload(store, stream_id, load_param, __FUNCTION__);
+    LoadStrategy load_strategy{LoadType::ALL, LoadObjective::UNDELETED_ONLY};
+    auto entry = version_map->check_reload(store, stream_id, load_strategy, __FUNCTION__);
     return entry->get_indexes(false);
 }
 
@@ -76,11 +64,9 @@ inline std::optional<AtomKey> get_specific_version(
         const std::shared_ptr<VersionMap> &version_map,
         const StreamId &stream_id,
         SignedVersionId signed_version_id,
-        const pipelines::VersionQuery& version_query,
         bool include_deleted = false) {
-    LoadParameter load_param{LoadType::LOAD_DOWNTO, signed_version_id};
-    auto entry = version_map->check_reload(store, stream_id, load_param, __FUNCTION__);
-    set_load_param_options(load_param, version_query);
+    LoadStrategy load_strategy{LoadType::DOWNTO, LoadObjective::UNDELETED_ONLY, signed_version_id};
+    auto entry = version_map->check_reload(store, stream_id, load_strategy, __FUNCTION__);
     VersionId version_id;
     if (signed_version_id >= 0) {
         version_id = static_cast<VersionId>(signed_version_id);
@@ -134,9 +120,7 @@ inline bool has_undeleted_version(
     const std::shared_ptr<Store> &store,
     const std::shared_ptr<VersionMap> &version_map,
     const StreamId &id) {
-    pipelines::VersionQuery version_query;
-    version_query.set_iterate_on_failure(false);
-    auto maybe_undeleted = get_latest_undeleted_version(store, version_map, id, version_query);
+    auto maybe_undeleted = get_latest_undeleted_version(store, version_map, id);
     return static_cast<bool>(maybe_undeleted);
 }
 
@@ -153,11 +137,9 @@ inline void insert_if_undeleted(
 inline std::unordered_map<VersionId, bool> get_all_tombstoned_versions(
     const std::shared_ptr<Store> &store,
     const std::shared_ptr<VersionMap> &version_map,
-    const StreamId &stream_id,
-    const pipelines::VersionQuery& version_query) {
-    LoadParameter load_param{LoadType::LOAD_ALL};
-    set_load_param_options(load_param, version_query);
-    auto entry = version_map->check_reload(store, stream_id, load_param, __FUNCTION__);
+    const StreamId &stream_id) {
+    LoadStrategy load_strategy{LoadType::ALL, LoadObjective::INCLUDE_DELETED};
+    auto entry = version_map->check_reload(store, stream_id, load_strategy, __FUNCTION__);
     std::unordered_map<VersionId, bool> result;
     for (auto key: entry->get_tombstoned_indexes())
             result[key.version_id()] = store->key_exists(key).get();
@@ -170,13 +152,11 @@ inline version_store::TombstoneVersionResult tombstone_version(
     const std::shared_ptr<VersionMap> &version_map,
     const StreamId &stream_id,
     VersionId version_id,
-    const pipelines::VersionQuery& version_query,
     bool allow_tombstoning_beyond_latest_version=false,
     const std::optional<timestamp>& creation_ts=std::nullopt) {
     ARCTICDB_DEBUG(log::version(), "Tombstoning version {} for stream {}", version_id, stream_id);
-    LoadParameter load_param{LoadType::LOAD_UNDELETED};
-    set_load_param_options(load_param, version_query);
-    auto entry = version_map->check_reload(store, stream_id, load_param, __FUNCTION__);
+    LoadStrategy load_strategy{LoadType::ALL, LoadObjective::UNDELETED_ONLY};
+    auto entry = version_map->check_reload(store, stream_id, load_strategy, __FUNCTION__);
     // Might as well do the previous/next version check while we find the required version_id.
     // But if entry is empty, it's possible the load failed (since iterate_on_failure=false above), so set the flag
     // to defer the check to delete_tree() (instead of reloading in case eager delete is disabled).
@@ -188,7 +168,6 @@ inline version_store::TombstoneVersionResult tombstone_version(
             is_live_index_type_key // Entry could be cached with deleted keys even if LOAD_UNDELETED
             );
 
-    AtomKey tombstone;
     if (res.keys_to_delete.empty()) {
         // It is possible to have a tombstone key without a corresponding index_key
         // This scenario can happen in case of DR sync
@@ -196,19 +175,18 @@ inline version_store::TombstoneVersionResult tombstone_version(
             util::raise_rte("Version {} for symbol {} is already deleted", version_id, stream_id);
         } else {
             if (!allow_tombstoning_beyond_latest_version) {
-                auto latest_key = get_latest_version(store, version_map, stream_id, version_query).first;
+                auto latest_key = get_latest_version(store, version_map, stream_id).first;
                 if (!latest_key || latest_key->version_id() < version_id)
                     util::raise_rte("Can't delete version {} for symbol {} - it's higher than the latest version",
                             stream_id, version_id);
             }
             // We will write a tombstone key even when the index_key is not found
-            tombstone = version_map->write_tombstone(store, version_id, stream_id, entry, creation_ts);
+            version_map->write_tombstone(store, version_id, stream_id, entry, creation_ts);
         }
     } else {
-        tombstone = version_map->write_tombstone(store, res.keys_to_delete[0], stream_id, entry, creation_ts);
+        version_map->write_tombstone(store, res.keys_to_delete[0], stream_id, entry, creation_ts);
     }
 
-    entry->tombstones_.try_emplace(version_id, std::move(tombstone));
     if (version_map->validate())
         entry->validate();
 
@@ -238,11 +216,9 @@ inline std::optional<AtomKey> load_index_key_from_time(
     const std::shared_ptr<Store> &store,
     const std::shared_ptr<VersionMap> &version_map,
     const StreamId &stream_id,
-    timestamp from_time,
-    const pipelines::VersionQuery& version_query) {
-    LoadParameter load_param{LoadType::LOAD_FROM_TIME, from_time};
-    set_load_param_options(load_param, version_query);
-    auto entry = version_map->check_reload(store, stream_id, load_param, __FUNCTION__);
+    timestamp from_time) {
+    LoadStrategy load_strategy{LoadType::FROM_TIME, LoadObjective::UNDELETED_ONLY, from_time};
+    auto entry = version_map->check_reload(store, stream_id, load_strategy, __FUNCTION__);
     auto indexes = entry->get_indexes(false);
     return get_index_key_from_time(from_time, indexes);
 }
@@ -250,11 +226,9 @@ inline std::optional<AtomKey> load_index_key_from_time(
 inline std::vector<AtomKey> get_index_and_tombstone_keys(
     const std::shared_ptr<Store> &store,
     const std::shared_ptr<VersionMap> &version_map,
-    const StreamId &stream_id,
-    const pipelines::VersionQuery& version_query) {
-    LoadParameter load_param{LoadType::LOAD_ALL};
-    set_load_param_options(load_param, version_query);
-    const auto entry = version_map->check_reload(store, stream_id, load_param, __FUNCTION__);
+    const StreamId &stream_id) {
+    LoadStrategy load_strategy{LoadType::ALL, LoadObjective::INCLUDE_DELETED};
+    const auto entry = version_map->check_reload(store, stream_id, load_strategy, __FUNCTION__);
     std::vector<AtomKey> res;
     std::copy_if(std::begin(entry->keys_), std::end(entry->keys_), std::back_inserter(res),
                  [&](const auto &key) { return is_index_or_tombstone(key); });
diff --git a/cpp/arcticdb/version/version_map.hpp b/cpp/arcticdb/version/version_map.hpp
index a196ad2130..df1edf062f 100644
--- a/cpp/arcticdb/version/version_map.hpp
+++ b/cpp/arcticdb/version/version_map.hpp
@@ -16,6 +16,7 @@
 #include <unordered_set>
 #include <map>
 #include <deque>
+#include <gtest/gtest_prod.h>
 
 #include <arcticdb/entity/types.hpp>
 #include <arcticdb/entity/atom_key.hpp>
@@ -107,7 +108,7 @@ class VersionMapImpl {
      */
     using MapType =  std::map<StreamId, std::shared_ptr<VersionMapEntry>>;
 
-    static constexpr uint64_t DEFAULT_CLOCK_UNSYNC_TOLERANCE = ONE_SECOND * 2;
+    static constexpr uint64_t DEFAULT_CLOCK_UNSYNC_TOLERANCE = ONE_MILLISECOND * 200;
     static constexpr uint64_t DEFAULT_RELOAD_INTERVAL = ONE_SECOND * 2;
     MapType map_;
     bool validate_ = false;
@@ -145,7 +146,7 @@ class VersionMapImpl {
         const std::shared_ptr<Store>& store,
         const VersionMapEntry& ref_entry,
         const std::shared_ptr<VersionMapEntry>& entry,
-        const LoadParameter& load_params) const {
+        const LoadStrategy& load_strategy) const {
         auto next_key = ref_entry.head_;
         entry->head_ = ref_entry.head_;
 
@@ -158,9 +159,8 @@ class VersionMapImpl {
             cached_penultimate_index = ref_entry.keys_[1];
         }
 
-        if (key_exists_in_ref_entry(load_params, ref_entry, cached_penultimate_index, load_progress)) {
-            load_progress.loaded_until_ = ref_entry.loaded_until_;
-            load_progress.oldest_loaded_index_version_ = ref_entry.loaded_until_;
+        if (key_exists_in_ref_entry(load_strategy, ref_entry, cached_penultimate_index)) {
+            load_progress = ref_entry.load_progress_;
             entry->keys_.push_back(ref_entry.keys_[0]);
             if(cached_penultimate_index)
                 entry->keys_.push_back(*cached_penultimate_index);
@@ -171,20 +171,20 @@ class VersionMapImpl {
                 next_key = read_segment_with_keys(seg, entry, load_progress);
                 set_latest_version(entry, latest_version);
             } while (next_key
-            && !loaded_until_version_id(load_params, load_progress, latest_version)
-            && !loaded_until_timestamp(load_params, load_progress)
-            && load_latest_ongoing(load_params, entry)
-            && looking_for_undeleted(load_params, entry, load_progress));
+            && continue_when_loading_version(load_strategy, load_progress, latest_version)
+            && continue_when_loading_from_time(load_strategy, load_progress)
+            && continue_when_loading_latest(load_strategy, entry)
+            && continue_when_loading_undeleted(load_strategy, entry, load_progress));
         }
-        set_loaded_until(load_progress, entry);
+        entry->load_progress_ = load_progress;
     }
 
     void load_via_ref_key(
         std::shared_ptr<Store> store,
         const StreamId& stream_id,
-        const LoadParameter& load_params,
+        const LoadStrategy& load_strategy,
         const std::shared_ptr<VersionMapEntry>& entry) {
-        load_params.validate();
+        load_strategy.validate();
         static const auto max_trial_config = ConfigsMap::instance()->get_int("VersionMap.MaxReadRefTrials", 2);
         auto max_trials = max_trial_config;
         while (true) {
@@ -194,7 +194,7 @@ class VersionMapImpl {
                 if (ref_entry.empty())
                     return;
 
-                follow_version_chain(store, ref_entry, entry, load_params);
+                follow_version_chain(store, ref_entry, entry, load_strategy);
                 break;
             } catch (const std::exception &err) {
                 if (--max_trials <= 0) {
@@ -235,7 +235,7 @@ class VersionMapImpl {
     }
 
     void write_version(std::shared_ptr<Store> store, const AtomKey &key, const std::optional<AtomKey>& previous_key) {
-        LoadParameter load_param{LoadType::LOAD_LATEST};
+        LoadStrategy load_param{LoadType::LATEST, LoadObjective::INCLUDE_DELETED};
         auto entry = check_reload(store, key.id(), load_param,  __FUNCTION__);
 
         do_write(store, key, entry);
@@ -259,7 +259,7 @@ class VersionMapImpl {
         auto entry = check_reload(
             store,
             stream_id,
-            LoadParameter{LoadType::LOAD_UNDELETED},
+            LoadStrategy{LoadType::ALL, LoadObjective::UNDELETED_ONLY},
             __FUNCTION__);
         auto output = tombstone_from_key_or_all_internal(store, stream_id, first_key_to_tombstone, entry);
 
@@ -273,7 +273,7 @@ class VersionMapImpl {
     }
 
     std::string dump_entry(const std::shared_ptr<Store>& store, const StreamId& stream_id) {
-        const auto entry = check_reload(store, stream_id, LoadParameter{LoadType::LOAD_ALL}, __FUNCTION__);
+        const auto entry = check_reload(store, stream_id, LoadStrategy{LoadType::ALL, LoadObjective::INCLUDE_DELETED}, __FUNCTION__);
         return entry->dump();
     }
 
@@ -285,7 +285,7 @@ class VersionMapImpl {
         auto entry = check_reload(
                 store,
                 key.id(),
-                LoadParameter{LoadType::LOAD_UNDELETED},
+                LoadStrategy{LoadType::ALL, LoadObjective::UNDELETED_ONLY},
                 __FUNCTION__);
         auto [_, result] = tombstone_from_key_or_all_internal(store, key.id(), previous_key, entry);
 
@@ -324,7 +324,7 @@ class VersionMapImpl {
         // This method has no API, and is not tested in the rapidcheck tests, but could easily be enabled there.
         // It compacts the version map but skips any keys which have been deleted (to free up space).
         ARCTICDB_DEBUG(log::version(), "Version map compacting versions for stream {}", stream_id);
-        auto entry = check_reload(store, stream_id, LoadParameter{LoadType::LOAD_ALL}, __FUNCTION__);
+        auto entry = check_reload(store, stream_id, LoadStrategy{LoadType::ALL, LoadObjective::INCLUDE_DELETED}, __FUNCTION__);
         if (!requires_compaction(entry))
             return;
 
@@ -439,7 +439,7 @@ class VersionMapImpl {
 
     void compact(std::shared_ptr<Store> store, const StreamId& stream_id) {
         ARCTICDB_DEBUG(log::version(), "Version map compacting versions for stream {}", stream_id);
-        auto entry = check_reload(store, stream_id, LoadParameter{LoadType::LOAD_ALL}, __FUNCTION__);
+        auto entry = check_reload(store, stream_id, LoadStrategy{LoadType::ALL, LoadObjective::INCLUDE_DELETED}, __FUNCTION__);
         if (entry->empty()) {
             log::version().warn("Entry is empty in compact");
             return;
@@ -461,7 +461,7 @@ class VersionMapImpl {
 
     void overwrite_symbol_tree(
             std::shared_ptr<Store> store, const StreamId& stream_id, const std::vector<AtomKey>& index_keys) {
-        auto entry = check_reload(store, stream_id, LoadParameter{LoadType::LOAD_ALL}, __FUNCTION__);
+        auto entry = check_reload(store, stream_id, LoadStrategy{LoadType::ALL, LoadObjective::INCLUDE_DELETED}, __FUNCTION__);
         auto old_entry = *entry;
         if (!index_keys.empty()) {
             entry->keys_.assign(std::begin(index_keys), std::end(index_keys));
@@ -480,15 +480,15 @@ class VersionMapImpl {
     std::shared_ptr<VersionMapEntry> check_reload(
         std::shared_ptr<Store> store,
         const StreamId& stream_id,
-        const LoadParameter& load_param,
+        const LoadStrategy& load_strategy,
         const char* function ARCTICDB_UNUSED) {
         ARCTICDB_DEBUG(log::version(), "Check reload in function {} for id {}", function, stream_id);
 
-        if (has_cached_entry(stream_id, load_param)) {
+        if (has_cached_entry(stream_id, load_strategy)) {
             return get_entry(stream_id);
         }
 
-        return storage_reload(store, stream_id, load_param, load_param.iterate_on_failure_);
+        return storage_reload(store, stream_id, load_strategy);
     }
 
     /**
@@ -548,13 +548,13 @@ class VersionMapImpl {
      *
      * @param stream_id symbol to check
      * @param load_param the load type
-     * @return whether we have a cached entry suitable for the load type, so do not need to go to storage
+     * @return whether we have a cached entry suitable for the load strategy, so do not need to go to storage
      */
-    bool has_cached_entry(const StreamId &stream_id, const LoadParameter& load_param) const {
-        LoadType requested_load_type = load_param.load_type_;
+    bool has_cached_entry(const StreamId &stream_id, const LoadStrategy& requested_load_strategy) const {
+        LoadType requested_load_type = requested_load_strategy.load_type_;
         util::check(requested_load_type < LoadType::UNKNOWN, "Unexpected load type requested {}", requested_load_type);
 
-        load_param.validate();
+        requested_load_strategy.validate();
         MapType::const_iterator entry_it;
         if(!find_entry(entry_it, stream_id)) {
             return false;
@@ -572,68 +572,40 @@ class VersionMapImpl {
             return false;
         }
 
-        if (requested_load_type == LoadType::NOT_LOADED) {
-            return true;
-        }
+        LoadType cached_load_type = entry->load_strategy_.load_type_;
 
-        LoadType cached_load_type = entry->load_type_;
-
-        switch(cached_load_type) {
+        switch (requested_load_type) {
             case LoadType::NOT_LOADED:
-                break;
-            case LoadType::LOAD_LATEST:
-                // Future: This case and LOAD_LATEST_UNDELETED could be optimized: use cache if request is
-                // LOAD_FROM_TIME for a later time than the cached entry.
-                if (requested_load_type == LoadType::LOAD_LATEST) {
-                    return true;
-                }
-                if (requested_load_type == LoadType::LOAD_DOWNTO) {
-                    return loaded_as_far_as_load_until(*entry, load_param);
-                }
-                break;
-            case LoadType::LOAD_LATEST_UNDELETED:
-                if (requested_load_type == LoadType::LOAD_LATEST_UNDELETED
-                    || requested_load_type == LoadType::LOAD_LATEST) {
-                    return true;
-                }
-
-                if (requested_load_type == LoadType::LOAD_DOWNTO) {
-                    return loaded_as_far_as_load_until(*entry, load_param);
-                }
-                break;
-            case LoadType::LOAD_DOWNTO:
-                if (requested_load_type == LoadType::LOAD_DOWNTO) {
-                    return loaded_as_far_as_load_until(*entry, load_param);
-                }
-
-                if (requested_load_type == LoadType::LOAD_LATEST_UNDELETED) {
-                    auto opt_latest = entry->get_first_index(false).first;
-                    return opt_latest.has_value();
-                }
-
-                if (requested_load_type == LoadType::LOAD_LATEST) {
-                    auto opt_latest = entry->get_first_index(true).first;
-                    return opt_latest.has_value();
+                return true;
+            case LoadType::LATEST: {
+                // If entry has at least one (maybe undeleted) index we have the latest value cached
+
+                // This check can be slow if we have thousands of deleted versions before the first undeleted and we're
+                // looking for an undeleted version. If that is ever a problem we can just store a boolean whether
+                // we have an undeleted version.
+                auto opt_latest = entry->get_first_index(requested_load_strategy.should_include_deleted()).first;
+                return opt_latest.has_value();
+            }
+            case LoadType::DOWNTO:
+                // We check whether the oldest loaded version is before or at the requested one
+                return loaded_as_far_as_version_id(*entry, requested_load_strategy.load_until_version_.value());
+            case LoadType::FROM_TIME: {
+                // We check whether the cached (deleted or undeleted) timestamp is before or at the requested one
+                auto cached_timestamp = requested_load_strategy.should_include_deleted() ?
+                                        entry->load_progress_.earliest_loaded_timestamp_ :
+                                        entry->load_progress_.earliest_loaded_undeleted_timestamp_;
+                return cached_timestamp <= requested_load_strategy.load_from_time_.value();
+            }
+            case LoadType::ALL:
+                // We can use cache when it was populated by a ALL call, in which case it is only unsafe to use
+                // when the cache is of undeleted versions and the request is for all versions
+                if (cached_load_type==LoadType::ALL){
+                    return entry->load_strategy_.should_include_deleted() || !requested_load_strategy.should_include_deleted();
                 }
-
                 return false;
-            case LoadType::LOAD_FROM_TIME:
-                // Future: This case could be optimized: use cache if it is LOAD_FROM_TIME for earlier time or
-                // LOAD_DOWNTO for a version with an earlier timestamp
-
-                // LOAD_FROM_TIME keeps looking till it finds an undeleted version, even that is earlier than the
-                // search time requested
-                return requested_load_type == LoadType::NOT_LOADED
-                    || requested_load_type == LoadType::LOAD_LATEST_UNDELETED
-                    || requested_load_type == LoadType::LOAD_LATEST;
-            case LoadType::LOAD_UNDELETED:
-                return requested_load_type != LoadType::LOAD_ALL;
-            case LoadType::LOAD_ALL:
-                return true;
             default:
                 util::raise_rte("Unexpected load type in cache {}", cached_load_type);
         }
-        return false;
     }
 
 private:
@@ -702,27 +674,27 @@ class VersionMapImpl {
 
     /**
      * Whether entry contains as much of the version map as specified by load_param. Checks whether
-     * loaded_until_ in entry is earlier than that specified in load_param.
+     * oldest_loaded_index_version_ in entry is earlier than that specified in load_param.
      *
      * @param entry the version map state to check
      * @param load_param the load request to test for completeness
      * @return true if and only if entry already contains data at least as far back as load_param requests
      */
-    bool loaded_as_far_as_load_until(const VersionMapEntry& entry, const LoadParameter& load_param) const {
-        if (is_positive_version_query(load_param)) {
-            if (entry.loaded_until_ <= static_cast<VersionId>(load_param.load_until_version_.value())) {
+    bool loaded_as_far_as_version_id(const VersionMapEntry& entry, SignedVersionId requested_version_id) const {
+        if (requested_version_id >= 0) {
+            if (entry.load_progress_.oldest_loaded_index_version_ <= static_cast<VersionId>(requested_version_id)) {
                 ARCTICDB_DEBUG(log::version(), "Loaded as far as required value {}, have {}",
-                               load_param.load_until_version_.value(), entry.loaded_until_);
+                               requested_version_id, entry.load_progress_.oldest_loaded_index_version_);
                 return true;
             }
         } else {
             auto opt_latest = entry.get_first_index(true).first;
             if (opt_latest.has_value()) {
                 auto opt_version_id = get_version_id_negative_index(opt_latest->version_id(),
-                                                                    *load_param.load_until_version_);
-                if (opt_version_id.has_value() && entry.loaded_until_ <= *opt_version_id) {
+                                                                    requested_version_id);
+                if (opt_version_id.has_value() && entry.load_progress_.oldest_loaded_index_version_ <= *opt_version_id) {
                     ARCTICDB_DEBUG(log::version(), "Loaded as far as required value {}, have {} and there are {} total versions",
-                                   load_param.load_until_version_.value(), entry.loaded_until_, opt_latest->version_id());
+                                   requested_version_id, entry.load_progress_.oldest_loaded_index_version_, opt_latest->version_id());
                     return true;
                 }
             }
@@ -770,8 +742,7 @@ class VersionMapImpl {
     std::shared_ptr<VersionMapEntry> storage_reload(
         std::shared_ptr<Store> store,
         const StreamId& stream_id,
-        const LoadParameter& load_param,
-        bool iterate_on_failure) {
+        const LoadStrategy& load_strategy) {
         /*
          * Goes to the storage for a given symbol, and recreates the VersionMapEntry from preferably the ref key
          * structure, and if that fails it then goes and builds that from iterating all keys from storage which can
@@ -783,27 +754,12 @@ class VersionMapImpl {
         const auto clock_unsync_tolerance = ConfigsMap::instance()->get_int("VersionMap.UnsyncTolerance",
                                                                             DEFAULT_CLOCK_UNSYNC_TOLERANCE);
         entry->last_reload_time_ = Clock::nanos_since_epoch() - clock_unsync_tolerance;
-        entry->load_type_ = LoadType::NOT_LOADED; // FUTURE: to make more thread-safe with #368
+        entry->load_strategy_ = LoadStrategy{LoadType::NOT_LOADED, LoadObjective::INCLUDE_DELETED}; // FUTURE: to make more thread-safe with #368
 
-        try {
-            auto temp = std::make_shared<VersionMapEntry>(*entry);
-            load_via_ref_key(store, stream_id, load_param, temp);
-            std::swap(*entry, *temp);
-            entry->load_type_ = load_param.load_type_;
-        }
-        catch (const std::runtime_error &err) {
-            if (iterate_on_failure) {
-                log::version().info(
-                        "Loading versions from storage via ref key failed with error: {}, will load via iteration",
-                        err.what());
-            } else {
-                throw;
-            }
-        }
-        if (iterate_on_failure && entry->empty()) {
-            (void) load_via_iteration(store, stream_id, entry);
-            entry->load_type_ = LoadType::LOAD_ALL;
-        }
+        auto temp = std::make_shared<VersionMapEntry>(*entry);
+        load_via_ref_key(store, stream_id, load_strategy, temp);
+        std::swap(*entry, *temp);
+        entry->load_strategy_ = load_strategy;
 
         util::check(entry->keys_.empty() || entry->head_, "Non-empty VersionMapEntry should set head");
         if (validate_)
@@ -843,7 +799,7 @@ class VersionMapImpl {
         load_via_iteration(store, stream_id, entry_iteration);
         auto maybe_latest_pair = get_latest_key_pair(entry_iteration);
         if (!maybe_latest_pair) {
-            log::version().warn("Latest version not found for {}", stream_id);
+            log::version().warn("LATEST version not found for {}", stream_id);
             return false;
         }
 
@@ -868,7 +824,7 @@ class VersionMapImpl {
 
         try {
             auto entry_ref = std::make_shared<VersionMapEntry>();
-            load_via_ref_key(store, stream_id, LoadParameter{LoadType::LOAD_ALL}, entry_ref);
+            load_via_ref_key(store, stream_id, LoadStrategy{LoadType::ALL, LoadObjective::INCLUDE_DELETED}, entry_ref);
             entry_ref->validate();
         } catch (const std::exception& err) {
             log::version().warn(
@@ -881,7 +837,7 @@ class VersionMapImpl {
 
     bool indexes_sorted(const std::shared_ptr<Store>& store, const StreamId& stream_id) {
         auto entry_ref = std::make_shared<VersionMapEntry>();
-        load_via_ref_key(store, stream_id, LoadParameter{LoadType::LOAD_ALL}, entry_ref);
+        load_via_ref_key(store, stream_id, LoadStrategy{LoadType::ALL, LoadObjective::INCLUDE_DELETED}, entry_ref);
         auto indexes = entry_ref->get_indexes(true);
         return std::is_sorted(std::cbegin(indexes), std::cend(indexes), [] (const auto& l, const auto& r) {
             return l > r;
@@ -962,6 +918,7 @@ class VersionMapImpl {
     }
 
 private:
+    FRIEND_TEST(VersionMap, CacheInvalidationWithTombstoneAllAfterLoad);
     std::pair<VersionId, std::vector<AtomKey>> tombstone_from_key_or_all_internal(
             std::shared_ptr<Store> store,
             const StreamId& stream_id,
@@ -971,7 +928,7 @@ class VersionMapImpl {
             entry = check_reload(
                     store,
                     stream_id,
-                    LoadParameter{LoadType::LOAD_UNDELETED},
+                    LoadStrategy{LoadType::ALL, LoadObjective::UNDELETED_ONLY},
                     __FUNCTION__);
         }
 
@@ -999,6 +956,14 @@ class VersionMapImpl {
         return {version_id, std::move(output)};
     }
 
+    // Invalidates the cached undeleted entry if it got tombstoned either by a tombstone or by a tombstone_all
+    void maybe_invalidate_cached_undeleted(VersionMapEntry& entry){
+        if (entry.is_tombstoned(entry.load_progress_.oldest_loaded_undeleted_index_version_)){
+            entry.load_progress_.oldest_loaded_undeleted_index_version_ = std::numeric_limits<VersionId>::max();
+            entry.load_progress_.earliest_loaded_undeleted_timestamp_ = std::numeric_limits<timestamp>::max();
+        }
+    }
+
     AtomKey write_tombstone_all_key_internal(
             const std::shared_ptr<Store>& store,
             const AtomKey& previous_key,
@@ -1006,6 +971,7 @@ class VersionMapImpl {
         auto tombstone_key = get_tombstone_all_key(previous_key, store->current_timestamp());
         entry->try_set_tombstone_all(tombstone_key);
         do_write(store, tombstone_key, entry);
+        maybe_invalidate_cached_undeleted(*entry);
         return tombstone_key;
     }
 
@@ -1022,6 +988,8 @@ class VersionMapImpl {
             return index_to_tombstone(k, stream_id, creation_ts.value_or(store->current_timestamp()));
         });
         do_write(store, tombstone,  entry);
+        entry->tombstones_.try_emplace(tombstone.version_id(), tombstone);
+        maybe_invalidate_cached_undeleted(*entry);
         if(log_changes_)
             log_tombstone(store, tombstone.id(), tombstone.version_id());
 
diff --git a/cpp/arcticdb/version/version_map_batch_methods.cpp b/cpp/arcticdb/version/version_map_batch_methods.cpp
index 6e18165c40..1818e3ea26 100644
--- a/cpp/arcticdb/version/version_map_batch_methods.cpp
+++ b/cpp/arcticdb/version/version_map_batch_methods.cpp
@@ -20,59 +20,18 @@ void StreamVersionData::react(const pipelines::VersionQuery &version_query) {
 }
 
 void StreamVersionData::do_react(std::monostate) {
-    if (load_param_.load_type_ == LoadType::NOT_LOADED)
-        load_param_ = LoadParameter{LoadType::LOAD_LATEST_UNDELETED};
-
     ++count_;
+    load_strategy_ = union_of_undeleted_strategies(load_strategy_, LoadStrategy{LoadType::LATEST, LoadObjective::UNDELETED_ONLY});
 }
 
 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_version_.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_version_.value() < 0)) {
-            load_param_.load_until_version_ = std::min(load_param_.load_until_version_.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_);
-    }
+    load_strategy_ = union_of_undeleted_strategies(load_strategy_, LoadStrategy{LoadType::DOWNTO, LoadObjective::UNDELETED_ONLY, specific_version.version_id_});
 }
 
 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_);
-    }
+    load_strategy_ = union_of_undeleted_strategies(load_strategy_, LoadStrategy{LoadType::FROM_TIME, LoadObjective::UNDELETED_ONLY, timestamp_query.timestamp_});
 }
 
 void StreamVersionData::do_react(const pipelines::SnapshotVersionQuery &snapshot_query) {
@@ -222,7 +181,7 @@ folly::Future<VersionEntryOrSnapshot> set_up_version_future(
 ) {
     if (version_data.count_ == 1) {
         return async::submit_io_task(CheckReloadTask{store, version_map, symbol,
-                                                     version_data.load_param_}).thenValue(
+                                                     version_data.load_strategy_}).thenValue(
             [](std::shared_ptr<VersionMapEntry> version_map_entry) {
                 return VersionEntryOrSnapshot{std::move(version_map_entry)};
             });
@@ -235,7 +194,7 @@ folly::Future<VersionEntryOrSnapshot> set_up_version_future(
                         CheckReloadTask{store,
                                         version_map,
                                         symbol,
-                                        version_data.load_param_}).thenValue(
+                                        version_data.load_strategy_}).thenValue(
                         [](std::shared_ptr<VersionMapEntry> version_map_entry) {
                             return VersionEntryOrSnapshot{
                                 std::move(version_map_entry)};
diff --git a/cpp/arcticdb/version/version_map_batch_methods.hpp b/cpp/arcticdb/version/version_map_batch_methods.hpp
index e6b77b78b2..a1c1dd1de3 100644
--- a/cpp/arcticdb/version/version_map_batch_methods.hpp
+++ b/cpp/arcticdb/version/version_map_batch_methods.hpp
@@ -64,13 +64,13 @@ inline std::shared_ptr<std::unordered_map<StreamId, SymbolStatus>> batch_check_l
     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};
+    const LoadStrategy load_strategy{LoadType::LATEST, LoadObjective::UNDELETED_ONLY};
     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});
+        [store, version_map, &load_strategy](auto &symbol) {
+          return async::submit_io_task(CheckReloadTask{store, version_map, symbol, load_strategy});
         },
         [output, mutex](const auto& id, const std::shared_ptr<VersionMapEntry> &entry) {
           auto index_key = entry->get_first_index(false).first;
@@ -100,13 +100,13 @@ inline std::shared_ptr<std::unordered_map<StreamId, AtomKey>> batch_get_latest_v
     const std::vector<StreamId> &stream_ids,
     bool include_deleted) {
     ARCTICDB_SAMPLE(BatchGetLatestVersion, 0)
-    const LoadParameter load_param{include_deleted ? LoadType::LOAD_LATEST : LoadType::LOAD_LATEST_UNDELETED};
+    const LoadStrategy load_strategy{LoadType::LATEST, include_deleted ? LoadObjective::INCLUDE_DELETED : LoadObjective::UNDELETED_ONLY};
     auto output = std::make_shared<std::unordered_map<StreamId, AtomKey>>();
     auto mutex = std::make_shared<std::mutex>();
 
     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});
+            [store, version_map, &load_strategy](auto& stream_id) {
+                return async::submit_io_task(CheckReloadTask{store, version_map, stream_id, load_strategy});
             },
             [output, include_deleted, mutex](auto id, auto entry) {
                 auto [index_key, deleted] = entry->get_first_index(include_deleted);
@@ -129,7 +129,7 @@ inline std::vector<folly::Future<std::pair<std::optional<AtomKey>, std::optional
         vector_fut.push_back(async::submit_io_task(CheckReloadTask{store,
                                                                    version_map,
                                                                    stream_id,
-                                                                   LoadParameter{LoadType::LOAD_LATEST_UNDELETED}})
+                                                                   LoadStrategy{LoadType::LATEST, LoadObjective::UNDELETED_ONLY}})
                                  .thenValue([](const std::shared_ptr<VersionMapEntry>& entry){
                                      return std::make_pair(entry->get_first_index(false).first, entry->get_first_index(true).first);
                                  }));
@@ -147,7 +147,7 @@ inline std::vector<folly::Future<version_store::UpdateInfo>> batch_get_latest_un
         vector_fut.push_back(async::submit_io_task(CheckReloadTask{store,
                                                      version_map,
                                                      stream_id,
-                                                     LoadParameter{LoadType::LOAD_LATEST_UNDELETED}})
+                                                     LoadStrategy{LoadType::LATEST, LoadObjective::UNDELETED_ONLY}})
         .thenValue([](auto entry){
             auto latest_version = entry->get_first_index(true).first;
             auto latest_undeleted_version = entry->get_first_index(false).first;
@@ -194,8 +194,8 @@ inline std::shared_ptr<std::unordered_map<StreamId, AtomKey>> batch_get_specific
 
     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});
+            LoadStrategy load_strategy{LoadType::DOWNTO, LoadObjective::UNDELETED_ONLY, static_cast<SignedVersionId>(sym_version.second)};
+            return async::submit_io_task(CheckReloadTask{store, version_map, sym_version.first, load_strategy});
         },
         [output, option, output_mutex, store, tombstoned_vers, tombstoned_vers_mutex]
                     (auto sym_version, const std::shared_ptr<VersionMapEntry>& entry) {
@@ -247,8 +247,8 @@ inline std::shared_ptr<std::unordered_map<std::pair<StreamId, VersionId>, AtomKe
 
     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});
+                LoadStrategy load_strategy{LoadType::DOWNTO, LoadObjective::UNDELETED_ONLY, static_cast<SignedVersionId>(first_version)};
+                return async::submit_io_task(CheckReloadTask{store, version_map, sym_version.first, load_strategy});
             },
 
             [output, &sym_versions, include_deleted, mutex](auto sym_version, const std::shared_ptr<VersionMapEntry>& entry) {
@@ -266,9 +266,13 @@ inline std::shared_ptr<std::unordered_map<std::pair<StreamId, VersionId>, AtomKe
 
     return output;
 }
+
+// [StreamVersionData] is used to combine different [VersionQuery]s for a stream_id into a list of needed snapshots and
+// a single [LoadStrategy] which will query the union of all version queries.
+// It only ever produces load parameters where to_load=UNDELETED_ONLY.
 struct StreamVersionData {
     size_t count_ = 0;
-    LoadParameter load_param_ = LoadParameter{LoadType::NOT_LOADED};
+    LoadStrategy load_strategy_ = LoadStrategy{LoadType::NOT_LOADED, LoadObjective::UNDELETED_ONLY};
     boost::container::small_vector<SnapshotId, 1> snapshots_;
 
     explicit StreamVersionData(const pipelines::VersionQuery& version_query);
diff --git a/cpp/arcticdb/version/version_map_entry.hpp b/cpp/arcticdb/version/version_map_entry.hpp
index 3fce47f1f8..8545227b39 100644
--- a/cpp/arcticdb/version/version_map_entry.hpp
+++ b/cpp/arcticdb/version/version_map_entry.hpp
@@ -22,19 +22,23 @@ using namespace arcticdb::stream;
 
 enum class LoadType : uint32_t {
     NOT_LOADED = 0,
-    LOAD_LATEST,
-    LOAD_LATEST_UNDELETED,
-    LOAD_DOWNTO,
-    LOAD_FROM_TIME,
-    LOAD_UNDELETED,
-    LOAD_ALL,
+    LATEST,
+    DOWNTO,
+    FROM_TIME,
+    ALL,
     UNKNOWN
 };
 
 inline constexpr bool is_partial_load_type(LoadType load_type) {
-    return load_type == LoadType::LOAD_DOWNTO || load_type == LoadType::LOAD_FROM_TIME;
+    return load_type == LoadType::DOWNTO || load_type == LoadType::FROM_TIME;
 }
 
+// Used to specify whether we want to load all or only undeleted versions
+enum class LoadObjective : uint32_t {
+    INCLUDE_DELETED,
+    UNDELETED_ONLY
+};
+
 enum class VersionStatus {
     LIVE,
     TOMBSTONED,
@@ -46,43 +50,105 @@ struct VersionDetails {
     VersionStatus version_status_;
 };
 
-inline constexpr bool is_latest_load_type(LoadType load_type) {
-    return load_type == LoadType::LOAD_LATEST || load_type == LoadType::LOAD_LATEST_UNDELETED;
-}
-
-struct LoadParameter {
-    explicit LoadParameter(LoadType load_type) :
-        load_type_(load_type) {
+// The LoadStrategy describes how to load versions from the version chain. It consists of:
+// load_type: Describes up to which point in the chain we need to go.
+// load_objective: Whether to include tombstoned versions
+struct LoadStrategy {
+    explicit LoadStrategy(LoadType load_type, LoadObjective load_objective) :
+            load_type_(load_type), load_objective_(load_objective) {
     }
 
-    LoadParameter(LoadType load_type, int64_t load_from_time_or_until) :
-        load_type_(load_type) {
+    LoadStrategy(LoadType load_type, LoadObjective load_objective, int64_t load_from_time_or_until) :
+            load_type_(load_type), load_objective_(load_objective) {
         switch(load_type_) {
-            case LoadType::LOAD_FROM_TIME:
+            case LoadType::FROM_TIME:
                 load_from_time_ = load_from_time_or_until;
                 break;
-            case LoadType::LOAD_DOWNTO:
+            case LoadType::DOWNTO:
                 load_until_version_ = load_from_time_or_until;
                 break;
             default:
-                internal::raise<ErrorCode::E_ASSERTION_FAILURE>("LoadParameter constructor with load_from_time_or_until parameter {} provided invalid load_type {}",
+                internal::raise<ErrorCode::E_ASSERTION_FAILURE>("LoadStrategy constructor with load_from_time_or_until parameter {} provided invalid load_type {}",
                                                                 load_from_time_or_until, static_cast<uint32_t>(load_type));
         }
     }
 
     LoadType load_type_ = LoadType::NOT_LOADED;
+    LoadObjective load_objective_ = LoadObjective::INCLUDE_DELETED;
     std::optional<SignedVersionId> load_until_version_ = std::nullopt;
     std::optional<timestamp> load_from_time_ = std::nullopt;
-    bool iterate_on_failure_ = false;
+
+    bool should_include_deleted() const {
+        switch (load_objective_) {
+            case LoadObjective::INCLUDE_DELETED:
+                return true;
+            case LoadObjective::UNDELETED_ONLY:
+                return false;
+            default:
+                util::raise_rte("Invalid load_objective: {}", load_objective_);
+        }
+    }
 
     void validate() const {
-        util::check((load_type_ == LoadType::LOAD_DOWNTO) == load_until_version_.has_value(),
+        internal::check<ErrorCode::E_ASSERTION_FAILURE>((load_type_ == LoadType::DOWNTO) == load_until_version_.has_value(),
                     "Invalid load parameter: load_type {} with load_util {}", int(load_type_), load_until_version_.value_or(VersionId{}));
-        util::check((load_type_ == LoadType::LOAD_FROM_TIME) == load_from_time_.has_value(),
+        internal::check<ErrorCode::E_ASSERTION_FAILURE>((load_type_ == LoadType::FROM_TIME) == load_from_time_.has_value(),
             "Invalid load parameter: load_type {} with load_from_time_ {}", int(load_type_), load_from_time_.value_or(timestamp{}));
     }
 };
 
+
+inline bool is_undeleted_strategy_subset(const LoadStrategy& left, const LoadStrategy& right){
+    switch (left.load_type_) {
+        case LoadType::NOT_LOADED:
+            return true;
+        case LoadType::LATEST:
+            // LATEST is not a subset of DOWNTO because DOWNTO may not reach the latest undeleted version.
+            return right.load_type_ != LoadType::NOT_LOADED && right.load_type_ != LoadType::DOWNTO;
+        case LoadType::DOWNTO:
+            if (right.load_type_ == LoadType::ALL) {
+                return true;
+            }
+            if (right.load_type_ == LoadType::DOWNTO && ((left.load_until_version_.value() >= 0) == (right.load_until_version_.value() >= 0))) {
+                // Left is subset of right only when the [load_until]s have same sign and left's version is >= right's version
+                return left.load_until_version_.value() >= right.load_until_version_.value();
+            }
+            break;
+        case LoadType::FROM_TIME:
+            if (right.load_type_ == LoadType::ALL){
+                return true;
+            }
+            if (right.load_type_ == LoadType::FROM_TIME){
+                return left.load_from_time_.value() >= right.load_from_time_.value();
+            }
+            break;
+        case LoadType::ALL:
+            return right.load_type_ == LoadType::ALL;
+        default:
+            util::raise_rte("Invalid load type: {}", left.load_type_);
+    }
+    return false;
+}
+
+// Returns a strategy which is guaranteed to load all versions requested by left and right.
+// Works only on strategies with include_deleted=false.
+inline LoadStrategy union_of_undeleted_strategies(const LoadStrategy& left, const LoadStrategy& right){
+    internal::check<ErrorCode::E_ASSERTION_FAILURE>(!left.should_include_deleted(), "Trying to produce a union of undeleted strategies but left strategy includes deleted.");
+    internal::check<ErrorCode::E_ASSERTION_FAILURE>(!right.should_include_deleted(), "Trying to produce a union of undeleted strategies but right strategy includes deleted.");
+    if (is_undeleted_strategy_subset(left, right)){
+        return right;
+    }
+    if (is_undeleted_strategy_subset(right, left)){
+        return left;
+    }
+    // If none is subset of the other, then we should load all versions. We can't be less conservative because we can't
+    // know where to load to with strategies which have a different load type. E.g. for FROM_TIME and DOWNTO
+    // we can't know where to read to unless we know the version chain.
+    // A possible workaround for this is to restructure loading the version chain to get a set of LoadStrategies and stop
+    // searching only when all of them are satisfied.
+    return LoadStrategy{LoadType::ALL, LoadObjective::UNDELETED_ONLY};
+}
+
 template<typename T>
 bool deque_is_unique(std::deque<T> vec) {
     sort(vec.begin(), vec.end());
@@ -132,15 +198,22 @@ inline AtomKey get_tombstone_all_key(const AtomKey &latest, timestamp creation_t
         .build(latest.id(), KeyType::TOMBSTONE_ALL);
 }
 
+struct LoadProgress {
+    VersionId oldest_loaded_index_version_ = std::numeric_limits<VersionId>::max();
+    VersionId oldest_loaded_undeleted_index_version_ = std::numeric_limits<VersionId>::max();
+    timestamp earliest_loaded_timestamp_ = std::numeric_limits<timestamp>::max();
+    timestamp earliest_loaded_undeleted_timestamp_ = std::numeric_limits<timestamp>::max();
+};
+
 struct VersionMapEntry {
     /*
       VersionMapEntry is all the data we have in-memory about each stream_id in the version map which in its essence
       is a map of StreamId: VersionMapEntry. It's created from the linked-list-like structure that we have in the
       storage, where the head_ points to the latest version and keys_ are basically all the index/version keys
-      loaded in memory in a deque - based on the load_type.
+      loaded in memory in a deque - based on the load_strategy.
 
-      load_type signifies the current state of the in memory structure vs the state on disk, where LOAD_LATEST will
-      just load the latest version, and LOAD_ALL loads everything in memory by going through the linked list on disk.
+      load_strategy signifies the current state of the in memory structure vs the state on disk, where LATEST will
+      just load the latest version, and ALL loads everything in memory by going through the linked list on disk.
 
       It also contains a map of version_ids and the tombstone key corresponding to it iff it has been pruned or
       explicitly deleted.
@@ -155,6 +228,7 @@ struct VersionMapEntry {
         if (keys_.empty())
             return;
 
+        // Sorting by creation_ts is safe from clock skew because we don't support parallel writes to the same symbol.
         std::sort(std::begin(keys_), std::end(keys_), [](const AtomKey &l, const AtomKey &r) {
             return l.creation_ts() > r.creation_ts();
         });
@@ -162,11 +236,12 @@ struct VersionMapEntry {
 
     void clear() {
         head_.reset();
-        load_type_ = LoadType::NOT_LOADED;
         last_reload_time_ = 0;
         tombstones_.clear();
         tombstone_all_.reset();
         keys_.clear();
+        load_progress_ = LoadProgress{};
+        load_strategy_ = LoadStrategy{LoadType::NOT_LOADED, LoadObjective::INCLUDE_DELETED};
     }
 
     bool empty() const {
@@ -183,7 +258,8 @@ struct VersionMapEntry {
         swap(left.last_reload_time_, right.last_reload_time_);
         swap(left.tombstone_all_, right.tombstone_all_);
         swap(left.head_, right.head_);
-        swap(left.load_type_, right.load_type_);
+        swap(left.load_strategy_, right.load_strategy_);
+        swap(left.load_progress_, right.load_progress_);
     }
 
     // Below four functions used to return optional<AtomKey> of the tombstone, but copying keys is expensive and only
@@ -356,9 +432,9 @@ struct VersionMapEntry {
     }
 
     std::optional<AtomKey> head_;
-    LoadType load_type_ = LoadType::NOT_LOADED;
+    LoadStrategy load_strategy_ = LoadStrategy{LoadType::NOT_LOADED, LoadObjective::INCLUDE_DELETED};
     timestamp last_reload_time_ = 0;
-    VersionId loaded_until_ = std::numeric_limits<uint64_t>::max();
+    LoadProgress load_progress_;
     std::deque<AtomKey> keys_;
     std::unordered_map<VersionId, AtomKey> tombstones_;
     std::optional<AtomKey> tombstone_all_;
@@ -443,24 +519,38 @@ namespace fmt {
         template<typename ParseContext>
         constexpr auto parse(ParseContext &ctx) { return ctx.begin(); }
 
-    template<typename FormatContext>
-    auto format(arcticdb::LoadType l, FormatContext &ctx) const {
-        switch(l) {
-        case arcticdb::LoadType::NOT_LOADED:
-            return fmt::format_to(ctx.out(), "NOT_LOADED");
-        case arcticdb::LoadType::LOAD_LATEST:
-            return fmt::format_to(ctx.out(), "LOAD_LATEST");
-        case arcticdb::LoadType::LOAD_LATEST_UNDELETED:
-            return fmt::format_to(ctx.out(), "LOAD_LATEST_UNDELETED");
-        case arcticdb::LoadType::LOAD_DOWNTO:
-            return fmt::format_to(ctx.out(), "LOAD_DOWNTO");
-        case arcticdb::LoadType::LOAD_UNDELETED:
-            return fmt::format_to(ctx.out(), "LOAD_UNDELETED");
-        case arcticdb::LoadType::LOAD_ALL:
-            return fmt::format_to(ctx.out(), "LOAD_ALL");
-        default:
-            arcticdb::util::raise_rte("Unrecognized load type {}", int(l));
+        template<typename FormatContext>
+        auto format(arcticdb::LoadType l, FormatContext &ctx) const {
+            switch (l) {
+                case arcticdb::LoadType::NOT_LOADED:
+                    return fmt::format_to(ctx.out(), "NOT_LOADED");
+                case arcticdb::LoadType::LATEST:
+                    return fmt::format_to(ctx.out(), "LATEST");
+                case arcticdb::LoadType::DOWNTO:
+                    return fmt::format_to(ctx.out(), "DOWNTO");
+                case arcticdb::LoadType::ALL:
+                    return fmt::format_to(ctx.out(), "ALL");
+                default:
+                    arcticdb::util::raise_rte("Unrecognized load type {}", int(l));
+            }
         }
-    }
-};
+    };
+
+    template<>
+    struct formatter<arcticdb::LoadObjective> {
+        template<typename ParseContext>
+        constexpr auto parse(ParseContext &ctx) { return ctx.begin(); }
+
+        template<typename FormatContext>
+        auto format(arcticdb::LoadObjective l, FormatContext &ctx) const {
+            switch (l) {
+                case arcticdb::LoadObjective::INCLUDE_DELETED:
+                    return fmt::format_to(ctx.out(), "ANY");
+                case arcticdb::LoadObjective::UNDELETED_ONLY:
+                    return fmt::format_to(ctx.out(), "UNDELETED_ONLY");
+                default:
+                    arcticdb::util::raise_rte("Unrecognized to load {}", int(l));
+            }
+        }
+    };
 }
diff --git a/cpp/arcticdb/version/version_store_api.cpp b/cpp/arcticdb/version/version_store_api.cpp
index ff4d4d0ad4..c177717ee9 100644
--- a/cpp/arcticdb/version/version_store_api.cpp
+++ b/cpp/arcticdb/version/version_store_api.cpp
@@ -52,7 +52,7 @@ VersionedItem PythonVersionStore::write_dataframe_specific_version(
     ARCTICDB_SAMPLE(WriteDataFrame, 0)
 
     ARCTICDB_DEBUG(log::version(), "write_dataframe_specific_version stream_id: {} , version_id: {}", stream_id, version_id);
-    if (auto version_key = ::arcticdb::get_specific_version(store(), version_map(), stream_id, version_id, VersionQuery{}); version_key) {
+    if (auto version_key = ::arcticdb::get_specific_version(store(), version_map(), stream_id, version_id); version_key) {
         log::version().warn("Symbol stream_id: {} already exists with version_id: {}", stream_id, version_id);
         return {std::move(*version_key)};
     }
@@ -190,12 +190,11 @@ VersionResultVector get_latest_versions_for_symbols(
     const std::shared_ptr<Store>& store,
     const std::shared_ptr<VersionMap>& version_map,
     const std::set<StreamId>& stream_ids,
-    SymbolVersionToSnapshotMap& snapshots_for_symbol,
-    const VersionQuery& version_query
+    SymbolVersionToSnapshotMap& snapshots_for_symbol
 ) {
     VersionResultVector res;
     for (auto &s_id: stream_ids) {
-            const auto& opt_version_key = get_latest_undeleted_version(store, version_map, s_id, version_query);
+            const auto& opt_version_key = get_latest_undeleted_version(store, version_map, s_id);
             if (opt_version_key) {
                 res.emplace_back(
                     s_id,
@@ -220,7 +219,7 @@ VersionResultVector get_all_versions_for_symbols(
     VersionResultVector res;
     std::unordered_set<std::pair<StreamId, VersionId>> unpruned_versions;
     for (auto &s_id: stream_ids) {
-            auto all_versions = get_all_versions(store, version_map, s_id, VersionQuery{});
+            auto all_versions = get_all_versions(store, version_map, s_id);
             unpruned_versions = {};
             for (const auto &entry: all_versions) {
                 unpruned_versions.emplace(s_id, entry.version_id());
@@ -252,7 +251,6 @@ VersionResultVector PythonVersionStore::list_versions(
     const std::optional<StreamId> &stream_id,
     const std::optional<SnapshotId> &snap_name,
     const std::optional<bool>& latest_only,
-    const std::optional<bool>& iterate_on_failure,
     const std::optional<bool>& skip_snapshots) {
     ARCTICDB_SAMPLE(ListVersions, 0)
     ARCTICDB_RUNTIME_DEBUG(log::version(), "Command: list_versions");
@@ -276,11 +274,8 @@ VersionResultVector PythonVersionStore::list_versions(
             return list_versions_for_snapshot(stream_ids, snap_name, *versions_for_snapshots, snapshots_for_symbol);
     }
 
-    VersionQuery version_query;
-    version_query.set_iterate_on_failure(iterate_on_failure);
-
    if(opt_false(latest_only))
-       return get_latest_versions_for_symbols(store(), version_map(), stream_ids, snapshots_for_symbol, version_query);
+       return get_latest_versions_for_symbols(store(), version_map(), stream_ids, snapshots_for_symbol);
    else
        return get_all_versions_for_symbols(store(), version_map(), stream_ids, snapshots_for_symbol, creation_ts_for_version_symbol);
 }
@@ -511,7 +506,7 @@ VersionedItem PythonVersionStore::write_partitioned_dataframe(
     const std::vector<std::string>& partition_value
     ) {
     ARCTICDB_SAMPLE(WritePartitionedDataFrame, 0)
-    auto [maybe_prev, deleted] = ::arcticdb::get_latest_version(store(), version_map(), stream_id, VersionQuery{});
+    auto [maybe_prev, deleted] = ::arcticdb::get_latest_version(store(), version_map(), stream_id);
     auto version_id = get_next_version_from_key(maybe_prev);
 
     //    TODO: We are not actually partitioning stuff atm, just assuming a single partition is passed for now.
@@ -564,7 +559,7 @@ VersionedItem PythonVersionStore::write_versioned_composite_data(
     ARCTICDB_SAMPLE(WriteVersionedMultiKey, 0)
 
     ARCTICDB_RUNTIME_DEBUG(log::version(), "Command: write_versioned_composite_data");
-    auto [maybe_prev, deleted] = ::arcticdb::get_latest_version(store(), version_map(), stream_id, VersionQuery{});
+    auto [maybe_prev, deleted] = ::arcticdb::get_latest_version(store(), version_map(), stream_id);
     auto version_id = get_next_version_from_key(maybe_prev);
     ARCTICDB_DEBUG(log::version(), "write_versioned_composite_data for stream_id: {} , version_id = {}", stream_id, version_id);
     // TODO: Assuming each sub key is always going to have the same version attached to it.
@@ -742,7 +737,7 @@ void PythonVersionStore::write_parallel(
 }
 
 std::unordered_map<VersionId, bool> PythonVersionStore::get_all_tombstoned_versions(const StreamId &stream_id) {
-    return ::arcticdb::get_all_tombstoned_versions(store(), version_map(), stream_id, VersionQuery{});
+    return ::arcticdb::get_all_tombstoned_versions(store(), version_map(), stream_id);
 }
 
 FrameAndDescriptor create_frame(const StreamId& target_id, SegmentInMemory seg, const ReadOptions& read_options) {
@@ -909,7 +904,7 @@ void PythonVersionStore::delete_version(
         const StreamId& stream_id,
         VersionId version_id) {
     ARCTICDB_RUNTIME_DEBUG(log::version(), "Command: delete_version");
-    auto result = ::arcticdb::tombstone_version(store(), version_map(), stream_id, version_id, VersionQuery{});
+    auto result = ::arcticdb::tombstone_version(store(), version_map(), stream_id, version_id);
 
     if (!result.keys_to_delete.empty() && !cfg().write_options().delayed_deletes()) {
         delete_tree(result.keys_to_delete, result);
@@ -923,7 +918,7 @@ void PythonVersionStore::delete_version(
 void PythonVersionStore::fix_symbol_trees(const std::vector<StreamId>& symbols) {
     auto snaps = get_master_snapshots_map(store());
     for (const auto& sym : symbols) {
-        auto index_keys_from_symbol_tree = get_all_versions(store(), version_map(), sym, VersionQuery{});
+        auto index_keys_from_symbol_tree = get_all_versions(store(), version_map(), sym);
         for(const auto& [key, map] : snaps[sym]) {
             index_keys_from_symbol_tree.push_back(key);
         }
@@ -941,7 +936,7 @@ void PythonVersionStore::prune_previous_versions(const StreamId& stream_id) {
     const std::shared_ptr<VersionMapEntry>& entry = version_map()->check_reload(
             store(),
             stream_id,
-            LoadParameter{LoadType::LOAD_UNDELETED},
+            LoadStrategy{LoadType::ALL, LoadObjective::UNDELETED_ONLY},
             __FUNCTION__);
     storage::check<ErrorCode::E_SYMBOL_NOT_FOUND>(!entry->empty(), "Symbol {} is not found", stream_id);
     auto [latest, deleted] = entry->get_first_index(false);
@@ -952,7 +947,7 @@ void PythonVersionStore::prune_previous_versions(const StreamId& stream_id) {
         return;
     }
 
-    auto previous = ::arcticdb::get_specific_version(store(), version_map(), stream_id, *prev_id, VersionQuery{});
+    auto previous = ::arcticdb::get_specific_version(store(), version_map(), stream_id, *prev_id);
     auto [_, pruned_indexes] = version_map()->tombstone_from_key_or_all(store(), stream_id, previous);
     delete_unreferenced_pruned_indexes(pruned_indexes, *latest).get();
 }
@@ -1109,7 +1104,7 @@ ReadResult PythonVersionStore::read_index(
 }
 
 std::vector<AtomKey> PythonVersionStore::get_version_history(const StreamId& stream_id) {
-    return get_index_and_tombstone_keys(store(), version_map(), stream_id, VersionQuery{});
+    return get_index_and_tombstone_keys(store(), version_map(), stream_id);
 }
 
 void PythonVersionStore::_compact_version_map(const StreamId& id) {
@@ -1125,6 +1120,7 @@ std::vector<SliceAndKey> PythonVersionStore::list_incompletes(const StreamId& st
 }
 
 void PythonVersionStore::clear(const bool continue_on_error) {
+    version_map()->flush();
     if (store()->fast_delete()) {
         // Most storage backends have a fast deletion method for a db/collection equivalent, eg. drop() for mongo and
         // lmdb and iterating each key is always going to be suboptimal.
diff --git a/cpp/arcticdb/version/version_store_api.hpp b/cpp/arcticdb/version/version_store_api.hpp
index 326d5d6124..69389d7324 100644
--- a/cpp/arcticdb/version/version_store_api.hpp
+++ b/cpp/arcticdb/version/version_store_api.hpp
@@ -260,7 +260,6 @@ class PythonVersionStore : public LocalVersionedEngine {
         const std::optional<StreamId> &stream_id,
         const std::optional<SnapshotId>& snap_name,
         const std::optional<bool> &latest_only,
-        const std::optional<bool>& iterate_on_failure,
         const std::optional<bool>& skip_snapshots);
 
     // Batch methods
diff --git a/cpp/arcticdb/version/version_store_objects.hpp b/cpp/arcticdb/version/version_store_objects.hpp
index 2272861974..2fb85534c5 100644
--- a/cpp/arcticdb/version/version_store_objects.hpp
+++ b/cpp/arcticdb/version/version_store_objects.hpp
@@ -43,8 +43,8 @@ struct PreDeleteChecks {
     std::unordered_set<IndexTypeKey> could_share_data {};
 
     LoadType calc_load_type() const {
-        if (prev_version) return LoadType::LOAD_ALL;
-        if (version_visible | next_version) return LoadType::LOAD_DOWNTO;
+        if (prev_version) return LoadType::ALL;
+        if (version_visible | next_version) return LoadType::DOWNTO;
         return LoadType::NOT_LOADED;
     }
 };
diff --git a/cpp/arcticdb/version/version_tasks.hpp b/cpp/arcticdb/version/version_tasks.hpp
index b44c2dfed9..e383c91e9c 100644
--- a/cpp/arcticdb/version/version_tasks.hpp
+++ b/cpp/arcticdb/version/version_tasks.hpp
@@ -108,21 +108,21 @@ struct CheckReloadTask : async::BaseTask {
     const std::shared_ptr<Store> store_;
     const std::shared_ptr<VersionMap> version_map_;
     const StreamId stream_id_;
-    const LoadParameter load_param_;
+    const LoadStrategy load_strategy_;
 
     CheckReloadTask(
         std::shared_ptr<Store> store,
         std::shared_ptr<VersionMap> version_map,
         StreamId stream_id,
-        LoadParameter load_param) :
+        LoadStrategy load_strategy) :
         store_(std::move(store)),
         version_map_(std::move(version_map)),
         stream_id_(std::move(stream_id)),
-        load_param_(load_param) {
+        load_strategy_(load_strategy) {
     }
 
     std::shared_ptr<VersionMapEntry> operator()() const {
-        return version_map_->check_reload(store_, stream_id_, load_param_, __FUNCTION__);
+        return version_map_->check_reload(store_, stream_id_, load_strategy_, __FUNCTION__);
     }
 };
 
diff --git a/cpp/arcticdb/version/version_utils.hpp b/cpp/arcticdb/version/version_utils.hpp
index e21229280a..f8878f23bf 100644
--- a/cpp/arcticdb/version/version_utils.hpp
+++ b/cpp/arcticdb/version/version_utils.hpp
@@ -66,12 +66,6 @@ inline entity::VariantKey write_multi_index_entry(
     return multi_key_fut.wait().value();
 }
 
-struct LoadProgress {
-    VersionId loaded_until_ = std::numeric_limits<VersionId>::max();
-    VersionId oldest_loaded_index_version_ = std::numeric_limits<VersionId>::max();
-    timestamp earliest_loaded_timestamp_ = std::numeric_limits<timestamp>::max();
-};
-
 inline std::optional<AtomKey> read_segment_with_keys(
     const SegmentInMemory &seg,
     VersionMapEntry &entry,
@@ -79,7 +73,9 @@ inline std::optional<AtomKey> read_segment_with_keys(
     ssize_t row = 0;
     std::optional<AtomKey> next;
     VersionId oldest_loaded_index = std::numeric_limits<VersionId>::max();
+    VersionId oldest_loaded_undeleted_index = std::numeric_limits<VersionId>::max();
     timestamp earliest_loaded_timestamp = std::numeric_limits<timestamp>::max();
+    timestamp earliest_loaded_undeleted_timestamp = std::numeric_limits<timestamp>::max();
 
     for (; row < ssize_t(seg.row_count()); ++row) {
         auto key = read_key_row(seg, row);
@@ -88,12 +84,12 @@ inline std::optional<AtomKey> read_segment_with_keys(
         if (is_index_key_type(key.type())) {
             entry.keys_.push_back(key);
             oldest_loaded_index = std::min(oldest_loaded_index, key.version_id());
+            earliest_loaded_timestamp = std::min(earliest_loaded_timestamp, key.creation_ts());
 
-            // Note that LOAD_FROM_TIME is implicitly loading undeleted. If there was a requirement
-            // to load from a particular time disregarding whether symbols are deleted or not, there would
-            // need to be an additional termination condition
-            if(!entry.is_tombstoned(key))
-                earliest_loaded_timestamp = std::min(earliest_loaded_timestamp, key.creation_ts());
+            if(!entry.is_tombstoned(key)) {
+                oldest_loaded_undeleted_index = std::min(oldest_loaded_undeleted_index, key.version_id());
+                earliest_loaded_undeleted_timestamp = std::min(earliest_loaded_timestamp, key.creation_ts());
+            }
 
         } else if (key.type() == KeyType::TOMBSTONE) {
             entry.tombstones_.try_emplace(key.version_id(), key);
@@ -111,9 +107,10 @@ inline std::optional<AtomKey> read_segment_with_keys(
         }
     }
     util::check(row == ssize_t(seg.row_count()), "Unexpected ordering in journal segment");
-    load_progress.loaded_until_ = oldest_loaded_index;
-    load_progress.oldest_loaded_index_version_ = std::min(load_progress.oldest_loaded_index_version_, load_progress.loaded_until_);
+    load_progress.oldest_loaded_index_version_ = std::min(load_progress.oldest_loaded_index_version_, oldest_loaded_index);
+    load_progress.oldest_loaded_undeleted_index_version_ = std::min(load_progress.oldest_loaded_undeleted_index_version_, oldest_loaded_undeleted_index);
     load_progress.earliest_loaded_timestamp_ = std::min(load_progress.earliest_loaded_timestamp_, earliest_loaded_timestamp);
+    load_progress.earliest_loaded_undeleted_timestamp_ = std::min(load_progress.earliest_loaded_undeleted_timestamp_, earliest_loaded_undeleted_timestamp);
     return next;
 }
 
@@ -196,7 +193,7 @@ inline void read_symbol_ref(const std::shared_ptr<StreamSource>& store, const St
 
     LoadProgress load_progress;
     entry.head_ = read_segment_with_keys(key_seg_pair.second, entry, load_progress);
-    entry.loaded_until_ = load_progress.loaded_until_;
+    entry.load_progress_ = load_progress;
 }
 
 inline void write_symbol_ref(std::shared_ptr<StreamSink> store,
@@ -234,35 +231,39 @@ inline std::optional<VersionId> get_version_id_negative_index(VersionId latest,
 
 std::unordered_map<StreamId, size_t> get_num_version_entries(const std::shared_ptr<Store> &store, size_t batch_size);
 
-inline bool is_positive_version_query(const LoadParameter& load_params) {
-    return load_params.load_until_version_.value() >= 0;
+inline bool is_positive_version_query(const LoadStrategy& load_strategy) {
+    return load_strategy.load_until_version_.value() >= 0;
 }
 
-inline bool loaded_until_version_id(const LoadParameter &load_params, const LoadProgress& load_progress, const std::optional<VersionId>& latest_version) {
-    if (!load_params.load_until_version_)
-        return false;
+inline bool continue_when_loading_version(const LoadStrategy& load_strategy, const LoadProgress& load_progress, const std::optional<VersionId>& latest_version) {
+    if (!load_strategy.load_until_version_)
+        // Should continue when not loading down to a version
+        return true;
 
-    if (is_positive_version_query(load_params)) {
-        if (load_progress.loaded_until_ > static_cast<VersionId>(*load_params.load_until_version_)) {
-            return false;
+    if (is_positive_version_query(load_strategy)) {
+        if (load_progress.oldest_loaded_index_version_ > static_cast<VersionId>(*load_strategy.load_until_version_)) {
+            // Should continue when version was not reached
+            return true;
         }
     } else {
         if (latest_version.has_value()) {
-            if (auto opt_version_id = get_version_id_negative_index(*latest_version, *load_params.load_until_version_);
-                opt_version_id && load_progress.loaded_until_ > *opt_version_id) {
-                    return false;
+            if (auto opt_version_id = get_version_id_negative_index(*latest_version, *load_strategy.load_until_version_);
+                opt_version_id && load_progress.oldest_loaded_index_version_ > *opt_version_id) {
+                    // Should continue when version was not reached
+                    return true;
             }
         } else {
-            return false;
+            // Should continue if not yet reached any index key
+            return true;
         }
     }
     ARCTICDB_DEBUG(log::version(),
                    "Exiting load downto because loaded to version {} for request {} with {} total versions",
-                   load_progress.loaded_until_,
-                   *load_params.load_until_version_,
+                   load_progress.oldest_loaded_index_version_,
+                   *load_strategy.load_until_version_,
                    latest_version.value()
                   );
-    return true;
+    return false;
 }
 
 inline void set_latest_version(const std::shared_ptr<VersionMapEntry>& entry, std::optional<VersionId>& latest_version) {
@@ -277,69 +278,74 @@ static constexpr timestamp nanos_to_seconds(timestamp nanos) {
     return nanos / timestamp(10000000000);
 }
 
-inline bool loaded_until_timestamp(const LoadParameter &load_params, const LoadProgress& load_progress) {
-    if (!load_params.load_from_time_ || load_progress.earliest_loaded_timestamp_ > *load_params.load_from_time_)
-        return false;
+inline bool continue_when_loading_from_time(const LoadStrategy &load_strategy, const LoadProgress& load_progress) {
+    if (!load_strategy.load_from_time_)
+        return true;
+
+    auto loaded_deleted_or_undeleted_timestamp = load_strategy.should_include_deleted() ? load_progress.earliest_loaded_timestamp_ : load_progress.earliest_loaded_undeleted_timestamp_;
+
+    if (loaded_deleted_or_undeleted_timestamp > *load_strategy.load_from_time_)
+        return true;
 
     ARCTICDB_DEBUG(log::version(),
                    "Exiting load from timestamp because request {} <= {}",
-                   *load_params.load_from_time_,
-                   load_progress.earliest_loaded_timestamp_);
-    return true;
+                   loaded_deleted_or_undeleted_timestamp,
+                   *load_strategy.load_from_time_);
+    return false;
 }
 
-inline bool load_latest_ongoing(const LoadParameter &load_params, const std::shared_ptr<VersionMapEntry> &entry) {
-    if (!(load_params.load_type_ == LoadType::LOAD_LATEST_UNDELETED && entry->get_first_index(false).first) &&
-        !(load_params.load_type_ == LoadType::LOAD_LATEST && entry->get_first_index(true).first))
+inline bool continue_when_loading_latest(const LoadStrategy& load_strategy, const std::shared_ptr<VersionMapEntry> &entry) {
+    if (!(load_strategy.load_type_ == LoadType::LATEST && entry->get_first_index(load_strategy.should_include_deleted()).first))
         return true;
 
-    ARCTICDB_DEBUG(log::version(), "Exiting because we found a non-deleted index in load latest");
+    ARCTICDB_DEBUG(log::version(), "Exiting because we found the latest version with include_deleted: {}", load_strategy.should_include_deleted());
     return false;
 }
 
-inline bool looking_for_undeleted(const LoadParameter& load_params, const std::shared_ptr<VersionMapEntry>& entry, const LoadProgress& load_progress) {
-    if(!(load_params.load_type_ == LoadType::LOAD_UNDELETED || load_params.load_type_ == LoadType::LOAD_LATEST_UNDELETED)) {
+inline bool continue_when_loading_undeleted(const LoadStrategy& load_strategy, const std::shared_ptr<VersionMapEntry>& entry, const LoadProgress& load_progress) {
+    if (load_strategy.should_include_deleted()){
         return true;
     }
 
     if(entry->tombstone_all_) {
-        const bool is_deleted_by_tombstone_all = entry->tombstone_all_->version_id() >= load_progress.oldest_loaded_index_version_;
-        if(is_deleted_by_tombstone_all) {
+        // We need the check below because it is possible to have a tombstone_all which doesn't cover all version keys after it.
+        // For example when we use prune_previous_versions (without write) we write a tombstone_all key which applies to keys
+        // before the previous one. So it's possible the version chain can look like:
+        // v0 <- v1 <- v2 <- tombstone_all(version=1)
+        // In this case we need to terminate at v1.
+        const bool is_deleted_by_tombstone_all =
+                entry->tombstone_all_->version_id() >= load_progress.oldest_loaded_index_version_;
+        if (is_deleted_by_tombstone_all) {
             ARCTICDB_DEBUG(
-                log::version(),
-                "Exiting because tombstone all key deletes all versions beyond: {} and the oldest loaded index has version: {}",
-                entry->tombstone_all_->version_id(),
-                load_progress.oldest_loaded_index_version_);
+                    log::version(),
+                    "Exiting because tombstone all key deletes all versions beyond: {} and the oldest loaded index has version: {}",
+                    entry->tombstone_all_->version_id(),
+                    load_progress.oldest_loaded_index_version_);
             return false;
-        } else {
-            return true;
         }
-    } else {
-        return true;
     }
+    return true;
 }
 
-inline bool penultimate_key_contains_required_version_id(const AtomKey& key, const LoadParameter& load_params) {
-    if(is_positive_version_query(load_params)) {
-        return key.version_id() <= static_cast<VersionId>(load_params.load_until_version_.value());
+inline bool penultimate_key_contains_required_version_id(const AtomKey& key, const LoadStrategy& load_strategy) {
+    if(is_positive_version_query(load_strategy)) {
+        return key.version_id() <= static_cast<VersionId>(load_strategy.load_until_version_.value());
     } else {
-        return *load_params.load_until_version_ == -1;
+        return *load_strategy.load_until_version_ == -1;
     }
 }
 
-inline bool key_exists_in_ref_entry(const LoadParameter& load_params, const VersionMapEntry& ref_entry, std::optional<AtomKey>& cached_penultimate_key, LoadProgress& load_progress) {
-    if (is_latest_load_type(load_params.load_type_) && is_index_key_type(ref_entry.keys_[0].type()))
+inline bool key_exists_in_ref_entry(const LoadStrategy& load_strategy, const VersionMapEntry& ref_entry, std::optional<AtomKey>& cached_penultimate_key) {
+    if (load_strategy.load_type_ == LoadType::LATEST && is_index_key_type(ref_entry.keys_[0].type()))
         return true;
 
-    if(cached_penultimate_key && is_partial_load_type(load_params.load_type_)) {
-        load_params.validate();
-        if(load_params.load_type_ == LoadType::LOAD_DOWNTO && penultimate_key_contains_required_version_id(*cached_penultimate_key, load_params)) {
-            load_progress.loaded_until_ = cached_penultimate_key->version_id();
+    if(cached_penultimate_key && is_partial_load_type(load_strategy.load_type_)) {
+        load_strategy.validate();
+        if(load_strategy.load_type_ == LoadType::DOWNTO && penultimate_key_contains_required_version_id(*cached_penultimate_key, load_strategy)) {
             return true;
         }
 
-        if(load_params.load_type_ == LoadType::LOAD_FROM_TIME &&cached_penultimate_key->creation_ts() <= load_params.load_from_time_.value()) {
-            load_progress.loaded_until_ = cached_penultimate_key->version_id();
+        if(load_strategy.load_type_ == LoadType::FROM_TIME && cached_penultimate_key->creation_ts() <= load_strategy.load_from_time_.value()) {
             return true;
         }
     }
@@ -347,10 +353,6 @@ inline bool key_exists_in_ref_entry(const LoadParameter& load_params, const Vers
     return false;
 }
 
-inline void set_loaded_until(const LoadProgress& load_progress, const std::shared_ptr<VersionMapEntry>& entry) {
-    entry->loaded_until_ = load_progress.loaded_until_;
-}
-
 inline SortedValue deduce_sorted(SortedValue existing_frame, SortedValue input_frame) {
     using namespace arcticdb;
     constexpr auto UNKNOWN = SortedValue::UNKNOWN;
diff --git a/python/arcticdb/version_store/_store.py b/python/arcticdb/version_store/_store.py
index 96144cc407..f2b9c8c140 100644
--- a/python/arcticdb/version_store/_store.py
+++ b/python/arcticdb/version_store/_store.py
@@ -1484,7 +1484,6 @@ def batch_restore_version(
 
     def _get_version_query(self, as_of: VersionQueryInput, **kwargs):
         version_query = _PythonVersionStoreVersionQuery()
-        version_query.set_iterate_on_failure(_assume_false("iterate_on_failure", kwargs))
 
         if isinstance(as_of, str):
             version_query.set_snap_name(as_of)
@@ -1984,7 +1983,7 @@ def list_versions(
             Only include the latest version for each returned symbol. Has no effect if `snapshot` argument is also
             specified.
         iterate_on_failure: `bool`
-            Iterate the type in the storage if the top-level key isn't present.
+            DEPRECATED: Passing this doesn't change behavior
         skip_snapshots: `bool`
             Don't populate version list with snapshot information.
             Can improve performance significantly if there are many snapshots.
@@ -2011,11 +2010,14 @@ def list_versions(
         `List[Dict]`
             List of dictionaries describing the discovered versions in the library.
         """
+        if iterate_on_failure:
+            log.warning("The iterate_on_failure argument is deprecated and will soon be removed. It's safe to remove since it doesn't change behavior.")
+
         if latest_only and snapshot and not NativeVersionStore._warned_about_list_version_latest_only_and_snapshot:
             log.warning("latest_only has no effect when snapshot is specified")
             NativeVersionStore._warned_about_list_version_latest_only_and_snapshot = True
 
-        result = self.version_store.list_versions(symbol, snapshot, latest_only, iterate_on_failure, skip_snapshots)
+        result = self.version_store.list_versions(symbol, snapshot, latest_only, skip_snapshots)
         return [
             {
                 "symbol": version_result[0],
@@ -2280,16 +2282,18 @@ def has_symbol(
         as_of : `Optional[VersionQueryInput]`, default=None
             See documentation of `read` method for more details.
         iterate_on_failure: `Optional[bool]`, default=False
-            Iterate the type in the storage if the top-level key isn;t present
+            DEPRECATED: Passing this doesn't change behavior
 
         Returns
         -------
         `bool`
             True if the symbol exists as_of the specified revision, False otherwise.
         """
+        if iterate_on_failure:
+            log.warning("The iterate_on_failure argument is deprecated and will soon be removed. It's safe to remove since it doesn't change behavior.")
 
         return (
-            self._find_version(symbol, as_of=as_of, raise_on_missing=False, iterate_on_failure=iterate_on_failure)
+            self._find_version(symbol, as_of=as_of, raise_on_missing=False)
             is not None
         )
 
diff --git a/python/arcticdb/version_store/library.py b/python/arcticdb/version_store/library.py
index f3eb1a55f9..f46b64d12d 100644
--- a/python/arcticdb/version_store/library.py
+++ b/python/arcticdb/version_store/library.py
@@ -1509,7 +1509,6 @@ def list_versions(
             symbol=symbol,
             snapshot=snapshot,
             latest_only=latest_only,
-            iterate_on_failure=False,
             skip_snapshots=skip_snapshots,
         )
         return {


Test patch:
diff --git a/cpp/arcticdb/version/test/rapidcheck_version_map.cpp b/cpp/arcticdb/version/test/rapidcheck_version_map.cpp
index 2bc9886bf5..26aa884aa0 100644
--- a/cpp/arcticdb/version/test/rapidcheck_version_map.cpp
+++ b/cpp/arcticdb/version/test/rapidcheck_version_map.cpp
@@ -23,7 +23,7 @@
 template <typename Model>
 void check_latest_versions(const Model&  s0, MapStorePair &sut, std::string symbol) {
     using namespace arcticdb;
-    auto prev = get_latest_version(sut.store_,sut.map_, symbol, pipelines::VersionQuery{}).first;
+    auto prev = get_latest_version(sut.store_,sut.map_, symbol).first;
     auto sut_version_id = prev ? prev->version_id() : 0;
     auto model_prev = s0.get_latest_version(symbol);
     auto model_version_id = model_prev ? model_prev.value() : 0;
@@ -33,9 +33,7 @@ void check_latest_versions(const Model&  s0, MapStorePair &sut, std::string symb
 template <typename Model>
 void check_latest_undeleted_versions(const Model&  s0, MapStorePair &sut, std::string symbol) {
     using namespace arcticdb;
-    pipelines::VersionQuery version_query;
-    version_query.set_iterate_on_failure(true);
-    auto prev = get_latest_undeleted_version(sut.store_, sut.map_, symbol, version_query);
+    auto prev = get_latest_undeleted_version(sut.store_, sut.map_, symbol);
     auto sut_version_id = prev ? prev->version_id() : 0;
     auto model_prev = s0.get_latest_undeleted_version(symbol);
     auto model_version_id = model_prev ? model_prev.value() : 0;
@@ -224,7 +222,7 @@ struct GetAllVersions : rc::state::Command<Model, MapStorePair> {
     void run(const Model& s0, MapStorePair &sut) const override {
         auto model_versions = s0.get_all_versions(symbol_);
         using namespace arcticdb;
-        auto sut_version = get_all_versions(sut.store_, sut.map_, symbol_, pipelines::VersionQuery{});
+        auto sut_version = get_all_versions(sut.store_, sut.map_, symbol_);
         RC_ASSERT(model_versions.size() == sut_version.size());
 
         for(auto i = size_t{0}; i < model_versions.size(); ++i)
diff --git a/cpp/arcticdb/version/test/test_stream_version_data.cpp b/cpp/arcticdb/version/test/test_stream_version_data.cpp
index f51934001c..6419104a1a 100644
--- a/cpp/arcticdb/version/test/test_stream_version_data.cpp
+++ b/cpp/arcticdb/version/test/test_stream_version_data.cpp
@@ -7,25 +7,27 @@ TEST(StreamVersionData, SpecificVersion) {
     using namespace arcticdb::pipelines;
 
     StreamVersionData stream_version_data;
-    VersionQuery query_1{SpecificVersionQuery{VersionId(12)}, false};
+    VersionQuery query_1{SpecificVersionQuery{VersionId(12)}};
     stream_version_data.react(query_1);
-    VersionQuery query_2{SpecificVersionQuery{VersionId(4)}, false};
+    VersionQuery query_2{SpecificVersionQuery{VersionId(4)}};
     stream_version_data.react(query_2);
     ASSERT_EQ(stream_version_data.count_, 2);
-    ASSERT_EQ(stream_version_data.load_param_.load_type_, LoadType::LOAD_DOWNTO);
-    ASSERT_EQ(stream_version_data.load_param_.load_until_version_, 4);
+    ASSERT_EQ(stream_version_data.load_strategy_.load_type_, LoadType::DOWNTO);
+    ASSERT_EQ(stream_version_data.load_strategy_.load_objective_, LoadObjective::UNDELETED_ONLY);
+    ASSERT_EQ(stream_version_data.load_strategy_.load_until_version_, 4);
 }
 
 TEST(StreamVersionData, SpecificVersionReversed) {
     using namespace arcticdb;
     using namespace arcticdb::pipelines;
 
-    StreamVersionData stream_version_data(VersionQuery{SpecificVersionQuery{VersionId(4)}, false});
-    VersionQuery query_2{SpecificVersionQuery{VersionId(12)}, false};
+    StreamVersionData stream_version_data(VersionQuery{SpecificVersionQuery{VersionId(4)}});
+    VersionQuery query_2{SpecificVersionQuery{VersionId(12)}};
     stream_version_data.react(query_2);
     ASSERT_EQ(stream_version_data.count_, 2);
-    ASSERT_EQ(stream_version_data.load_param_.load_type_, LoadType::LOAD_DOWNTO);
-    ASSERT_EQ(stream_version_data.load_param_.load_until_version_, 4);
+    ASSERT_EQ(stream_version_data.load_strategy_.load_type_, LoadType::DOWNTO);
+    ASSERT_EQ(stream_version_data.load_strategy_.load_objective_, LoadObjective::UNDELETED_ONLY);
+    ASSERT_EQ(stream_version_data.load_strategy_.load_until_version_, 4);
 }
 
 TEST(StreamVersionData, Timestamp) {
@@ -33,13 +35,14 @@ TEST(StreamVersionData, Timestamp) {
     using namespace arcticdb::pipelines;
 
     StreamVersionData stream_version_data;
-    VersionQuery query_1{TimestampVersionQuery{timestamp(12)}, false};
+    VersionQuery query_1{TimestampVersionQuery{timestamp(12)}};
     stream_version_data.react(query_1);
-    VersionQuery query_2{TimestampVersionQuery{timestamp(4)}, false};
+    VersionQuery query_2{TimestampVersionQuery{timestamp(4)}};
     stream_version_data.react(query_2);
     ASSERT_EQ(stream_version_data.count_, 2);
-    ASSERT_EQ(stream_version_data.load_param_.load_type_, LoadType::LOAD_FROM_TIME);
-    ASSERT_EQ(stream_version_data.load_param_.load_from_time_, 4);
+    ASSERT_EQ(stream_version_data.load_strategy_.load_type_, LoadType::FROM_TIME);
+    ASSERT_EQ(stream_version_data.load_strategy_.load_objective_, LoadObjective::UNDELETED_ONLY);
+    ASSERT_EQ(stream_version_data.load_strategy_.load_from_time_, 4);
 }
 
 TEST(StreamVersionData, TimestampUnordered) {
@@ -47,15 +50,16 @@ TEST(StreamVersionData, TimestampUnordered) {
     using namespace arcticdb::pipelines;
 
     StreamVersionData stream_version_data;
-    VersionQuery query_1{TimestampVersionQuery{timestamp(3)}, false};
+    VersionQuery query_1{TimestampVersionQuery{timestamp(3)}};
     stream_version_data.react(query_1);
-    VersionQuery query_2{TimestampVersionQuery{timestamp(7)}, false};
+    VersionQuery query_2{TimestampVersionQuery{timestamp(7)}};
     stream_version_data.react(query_2);
-    VersionQuery query_3{TimestampVersionQuery{timestamp(4)}, false};
+    VersionQuery query_3{TimestampVersionQuery{timestamp(4)}};
     stream_version_data.react(query_3);
     ASSERT_EQ(stream_version_data.count_, 3);
-    ASSERT_EQ(stream_version_data.load_param_.load_type_, LoadType::LOAD_FROM_TIME);
-    ASSERT_EQ(stream_version_data.load_param_.load_from_time_, 3);
+    ASSERT_EQ(stream_version_data.load_strategy_.load_type_, LoadType::FROM_TIME);
+    ASSERT_EQ(stream_version_data.load_strategy_.load_objective_, LoadObjective::UNDELETED_ONLY);
+    ASSERT_EQ(stream_version_data.load_strategy_.load_from_time_, 3);
 }
 
 TEST(StreamVersionData, Latest) {
@@ -63,11 +67,12 @@ TEST(StreamVersionData, Latest) {
     using namespace arcticdb::pipelines;
 
     StreamVersionData stream_version_data;
-    VersionQuery query_1{std::monostate{}, false};
+    VersionQuery query_1{std::monostate{}};
     stream_version_data.react(query_1);
     ASSERT_EQ(stream_version_data.count_, 1);
-    ASSERT_EQ(stream_version_data.load_param_.load_type_, LoadType::LOAD_LATEST_UNDELETED);
-    ASSERT_EQ(stream_version_data.load_param_.load_until_version_.has_value(), false);
+    ASSERT_EQ(stream_version_data.load_strategy_.load_type_, LoadType::LATEST);
+    ASSERT_EQ(stream_version_data.load_strategy_.load_objective_, LoadObjective::UNDELETED_ONLY);
+    ASSERT_EQ(stream_version_data.load_strategy_.load_until_version_.has_value(), false);
 }
 
 TEST(StreamVersionData, SpecificToTimestamp) {
@@ -75,14 +80,15 @@ TEST(StreamVersionData, SpecificToTimestamp) {
     using namespace arcticdb::pipelines;
 
     StreamVersionData stream_version_data;
-    VersionQuery query_1{SpecificVersionQuery{VersionId(12)}, false};
+    VersionQuery query_1{SpecificVersionQuery{VersionId(12)}};
     stream_version_data.react(query_1);
-    VersionQuery query_2{TimestampVersionQuery{timestamp(3)}, false};
+    VersionQuery query_2{TimestampVersionQuery{timestamp(3)}};
     stream_version_data.react(query_2);
     ASSERT_EQ(stream_version_data.count_, 2);
-    ASSERT_EQ(stream_version_data.load_param_.load_type_, LoadType::LOAD_UNDELETED);
-    ASSERT_EQ(stream_version_data.load_param_.load_until_version_.has_value(), false);
-    ASSERT_EQ(stream_version_data.load_param_.load_from_time_.has_value(), false);
+    ASSERT_EQ(stream_version_data.load_strategy_.load_type_, LoadType::ALL);
+    ASSERT_EQ(stream_version_data.load_strategy_.load_objective_, LoadObjective::UNDELETED_ONLY);
+    ASSERT_EQ(stream_version_data.load_strategy_.load_until_version_.has_value(), false);
+    ASSERT_EQ(stream_version_data.load_strategy_.load_from_time_.has_value(), false);
 }
 
 TEST(StreamVersionData, TimestampToSpecific) {
@@ -90,12 +96,13 @@ TEST(StreamVersionData, TimestampToSpecific) {
     using namespace arcticdb::pipelines;
 
     StreamVersionData stream_version_data;
-    VersionQuery query_1{TimestampVersionQuery{timestamp(3)}, false};
+    VersionQuery query_1{TimestampVersionQuery{timestamp(3)}};
     stream_version_data.react(query_1);
-    VersionQuery query_2{SpecificVersionQuery{VersionId(12)}, false};
+    VersionQuery query_2{SpecificVersionQuery{VersionId(12)}};
     stream_version_data.react(query_2);
     ASSERT_EQ(stream_version_data.count_, 2);
-    ASSERT_EQ(stream_version_data.load_param_.load_type_, LoadType::LOAD_UNDELETED);
-    ASSERT_EQ(stream_version_data.load_param_.load_until_version_.has_value(), false);
-    ASSERT_EQ(stream_version_data.load_param_.load_from_time_.has_value(), false);
+    ASSERT_EQ(stream_version_data.load_strategy_.load_type_, LoadType::ALL);
+    ASSERT_EQ(stream_version_data.load_strategy_.load_objective_, LoadObjective::UNDELETED_ONLY);
+    ASSERT_EQ(stream_version_data.load_strategy_.load_until_version_.has_value(), false);
+    ASSERT_EQ(stream_version_data.load_strategy_.load_from_time_.has_value(), false);
 }
\ No newline at end of file
diff --git a/cpp/arcticdb/version/test/test_version_map.cpp b/cpp/arcticdb/version/test/test_version_map.cpp
index 38d9a5c95b..cbad577f2c 100644
--- a/cpp/arcticdb/version/test/test_version_map.cpp
+++ b/cpp/arcticdb/version/test/test_version_map.cpp
@@ -72,19 +72,17 @@ TEST(VersionMap, WithPredecessors) {
     auto version_map = std::make_shared<VersionMap>();
     version_map->set_validate(true);
     version_map->write_version(store, key1, std::nullopt);
-    pipelines::VersionQuery version_query;
-    version_query.set_iterate_on_failure(false);
-    auto latest = get_latest_undeleted_version(store, version_map, id, version_query);
+    auto latest = get_latest_undeleted_version(store, version_map, id);
     ASSERT_TRUE(latest);
     ASSERT_EQ(latest.value(), key1);
     version_map->write_version(store, key2, key1);
 
-    latest = get_latest_undeleted_version(store, version_map, id, version_query);
+    latest = get_latest_undeleted_version(store, version_map, id);
     ASSERT_EQ(latest.value(), key2);
     version_map->write_version(store, key3, key2);
 
     std::vector<AtomKey> expected{ key3, key2, key1};
-    auto result = get_all_versions(store, version_map, id, version_query);
+    auto result = get_all_versions(store, version_map, id);
     ASSERT_EQ(result, expected);
 
 }
@@ -96,52 +94,49 @@ TEST(VersionMap, TombstoneDelete) {
     auto key4 = atom_key_builder().version_id(4).creation_ts(PilotedClock::nanos_since_epoch()).content_hash(6).start_index(
         7).end_index(8).build(id, KeyType::TABLE_INDEX);
 
-    pipelines::VersionQuery version_query;
-    version_query.set_iterate_on_failure(false);
-
     auto version_map = std::make_shared<VersionMap>();
     version_map->set_validate(true);
     version_map->write_version(store, key1, std::nullopt);
-    auto latest = get_latest_undeleted_version(store, version_map, id, version_query);
+    auto latest = get_latest_undeleted_version(store, version_map, id);
     ASSERT_TRUE(latest);
     ASSERT_EQ(latest.value(), key1);
     version_map->write_version(store, key2, key1);
 
-    latest = get_latest_undeleted_version(store, version_map, id, version_query);
+    latest = get_latest_undeleted_version(store, version_map, id);
     ASSERT_EQ(latest.value(), key2);
     version_map->write_version(store, key3, key2);
 
-    latest = get_latest_undeleted_version(store, version_map, id, version_query);
+    latest = get_latest_undeleted_version(store, version_map, id);
     ASSERT_EQ(latest.value(), key3);
     version_map->write_version(store, key4, key3);
 
-    auto del_res = tombstone_version(store, version_map, id, VersionId{2}, pipelines::VersionQuery{});
+    auto del_res = tombstone_version(store, version_map, id, VersionId{2});
 
     ASSERT_FALSE(del_res.no_undeleted_left);
     ASSERT_EQ(del_res.keys_to_delete.front(), key2);
     ASSERT_THAT(del_res.could_share_data, UnorderedElementsAre(key1, key3));
 
     std::vector<AtomKey> expected{key4, key3, key1};
-    auto result = get_all_versions(store, version_map, id, version_query);
+    auto result = get_all_versions(store, version_map, id);
     ASSERT_EQ(result, expected);
 
-    del_res = tombstone_version(store, version_map, id, VersionId{3}, pipelines::VersionQuery{});
+    del_res = tombstone_version(store, version_map, id, VersionId{3});
     ASSERT_FALSE(del_res.no_undeleted_left);
     ASSERT_EQ(del_res.keys_to_delete.front(), key3);
     ASSERT_THAT(del_res.could_share_data, UnorderedElementsAre(key1, key4));
 
-    latest = get_latest_undeleted_version(store, version_map, id, version_query);
+    latest = get_latest_undeleted_version(store, version_map, id);
     ASSERT_EQ(latest.value(), key4);
 
-    del_res = tombstone_version(store, version_map, id, VersionId{4}, pipelines::VersionQuery{});
+    del_res = tombstone_version(store, version_map, id, VersionId{4});
     ASSERT_FALSE(del_res.no_undeleted_left);
     ASSERT_EQ(del_res.keys_to_delete.front(), key4);
     ASSERT_EQ(*del_res.could_share_data.begin(), key1);
 
-    latest = get_latest_undeleted_version(store, version_map, id, version_query);
+    latest = get_latest_undeleted_version(store, version_map, id);
     ASSERT_EQ(latest.value(), key1);
 
-    del_res = tombstone_version(store, version_map, id, VersionId{1}, pipelines::VersionQuery{});
+    del_res = tombstone_version(store, version_map, id, VersionId{1});
     ASSERT_TRUE(del_res.no_undeleted_left);
     ASSERT_EQ(del_res.keys_to_delete.front(), key1);
     ASSERT_TRUE(del_res.could_share_data.empty());
@@ -161,7 +156,7 @@ TEST(VersionMap, PingPong) {
         4).end_index(5).build(id, KeyType::TABLE_INDEX);
 
     left->write_version(store, key1, std::nullopt);
-    auto latest = get_latest_undeleted_version(store, right, id, pipelines::VersionQuery{});
+    auto latest = get_latest_undeleted_version(store, right, id);
     ASSERT_EQ(latest.value(), key1);
 
     auto key2 = atom_key_builder().version_id(2).creation_ts(3).content_hash(4).start_index(
@@ -175,9 +170,9 @@ TEST(VersionMap, PingPong) {
     left->write_version(store, key3, key2);
 
     std::vector<AtomKey> expected{ key3, key2, key1};
-    auto left_result = get_all_versions(store, left, id, pipelines::VersionQuery{});
+    auto left_result = get_all_versions(store, left, id);
     ASSERT_EQ(left_result, expected);
-    auto right_result = get_all_versions(store, right, id, pipelines::VersionQuery{});
+    auto right_result = get_all_versions(store, right, id);
     ASSERT_EQ(right_result, expected);
 }
 
@@ -202,14 +197,14 @@ TEST(VersionMap, TestLoadsRefAndIteration) {
     ScopedConfig reload_interval("VersionMap.ReloadInterval", 0); // always reload
 
     std::vector<AtomKey> expected{ key3, key2, key1};
-    auto result = get_all_versions(store, version_map, id, pipelines::VersionQuery{});
+    auto result = get_all_versions(store, version_map, id);
     ASSERT_EQ(result, expected);
 
     auto entry_iteration = std::make_shared<VersionMapEntry>();
     version_map->load_via_iteration(store, id, entry_iteration);
 
     auto entry_ref = std::make_shared<VersionMapEntry>();
-    version_map->load_via_ref_key(store, id, LoadParameter{LoadType::LOAD_ALL}, entry_ref);
+    version_map->load_via_ref_key(store, id, LoadStrategy{LoadType::ALL, LoadObjective::INCLUDE_DELETED}, entry_ref);
 
     ASSERT_EQ(entry_iteration->head_, entry_ref->head_);
     ASSERT_EQ(entry_iteration->keys_.size(), entry_ref->keys_.size());
@@ -240,7 +235,7 @@ TEST(VersionMap, TestCompact) {
     ASSERT_EQ(store->num_ref_keys(), 1);
 
     std::vector<AtomKey> expected{ key3, key2, key1};
-    auto result = get_all_versions(store, version_map, id, pipelines::VersionQuery{});
+    auto result = get_all_versions(store, version_map, id);
     ASSERT_EQ(result, expected);
 }
 
@@ -254,7 +249,7 @@ TEST(VersionMap, TestCompactWithDelete) {
     version_map->write_version(store, key1, std::nullopt);
     version_map->write_version(store, key2, key1);
     version_map->write_version(store, key3, key2);
-    tombstone_version(store, version_map, id, 2, pipelines::VersionQuery{});
+    tombstone_version(store, version_map, id, 2);
 
     ScopedConfig max_blocks("VersionMap.MaxVersionBlocks", 1);
     ScopedConfig reload_interval("VersionMap.ReloadInterval", 0); // always reload
@@ -264,7 +259,7 @@ TEST(VersionMap, TestCompactWithDelete) {
     ASSERT_EQ(store->num_ref_keys(), 1);
 
     std::vector<AtomKey> expected{ key3, key1};
-    auto result = get_all_versions(store, version_map, id, pipelines::VersionQuery{});
+    auto result = get_all_versions(store, version_map, id);
     ASSERT_EQ(result, expected);
 }
 
@@ -280,8 +275,8 @@ TEST(VersionMap, TestLatestVersionWithDeleteTombstones) {
     version_map->write_version(store, key1, std::nullopt);
     version_map->write_version(store, key2, key1);
     version_map->write_version(store, key3, key2);
-    tombstone_version(store, version_map, id, 2, pipelines::VersionQuery{});
-    auto [maybe_prev, deleted] = get_latest_version(store, version_map, id, pipelines::VersionQuery{});
+    tombstone_version(store, version_map, id, 2);
+    auto [maybe_prev, deleted] = get_latest_version(store, version_map, id);
     auto version_id = get_next_version_from_key(maybe_prev);
     ASSERT_EQ(version_id, 4);
 }
@@ -296,14 +291,14 @@ TEST(VersionMap, TestCompactWithDeleteTombstones) {
     version_map->write_version(store, key1, std::nullopt);
     version_map->write_version(store, key2, key1);
     version_map->write_version(store, key3, key2);
-    tombstone_version(store, version_map, id, 2, pipelines::VersionQuery{});
+    tombstone_version(store, version_map, id, 2);
 
     ScopedConfig max_blocks("VersionMap.MaxVersionBlocks", 1);
     ScopedConfig reload_interval("VersionMap.ReloadInterval", 0); // always reload
     version_map->compact(store, id);
 
     std::vector<AtomKey> expected{ key3, key1};
-    auto result = get_all_versions(store, version_map, id, pipelines::VersionQuery{});
+    auto result = get_all_versions(store, version_map, id);
     ASSERT_EQ(result, expected);
 }
 
@@ -335,35 +330,6 @@ void write_old_style_journal_entry(const AtomKey &key, std::shared_ptr<StreamSin
     journal_agg.commit();
 }
 
-TEST(VersionMap, IterateOnFailure) {
-    auto store = std::make_shared<InMemoryStore>();
-    StreamId id{"test1"};
-
-    auto version_map = std::make_shared<VersionMap>();
-    version_map->set_validate(true);
-    auto key1 =
-        atom_key_builder().version_id(1).creation_ts(PilotedClock::nanos_since_epoch()).content_hash(3).start_index(
-            4).end_index(5).build(id, KeyType::TABLE_INDEX);
-    version_map->write_version(store, key1, std::nullopt);
-    auto key2 =
-        atom_key_builder().version_id(2).creation_ts(PilotedClock::nanos_since_epoch()).content_hash(4).start_index(
-            5).end_index(6).build(id, KeyType::TABLE_INDEX);
-    version_map->write_version(store, key2, key1);
-    auto key3 =
-        atom_key_builder().version_id(3).creation_ts(PilotedClock::nanos_since_epoch()).content_hash(5).start_index(
-            6).end_index(7).build(id, KeyType::TABLE_INDEX);
-    version_map->write_version(store, key3, key2);
-
-    RefKey ref_key{id, KeyType::VERSION_REF};
-    store->remove_key_sync(ref_key, storage::RemoveOpts{});
-
-    std::vector<AtomKey> expected{ key3, key2, key1};
-    pipelines::VersionQuery version_query;
-    version_query.set_iterate_on_failure(true);
-    auto result = get_all_versions(store, version_map, id, version_query);
-    ASSERT_EQ(result, expected);
-}
-
 TEST(VersionMap, GetNextVersionInEntry) {
     using namespace arcticdb;
 
@@ -426,7 +392,7 @@ TEST(VersionMap, FixRefKey) {
     ASSERT_TRUE(version_map->check_ref_key(store, id));
 
     std::vector<AtomKey> expected{key3, key2, key1};
-    auto result = get_all_versions(store, version_map, id, pipelines::VersionQuery{});
+    auto result = get_all_versions(store, version_map, id);
     ASSERT_EQ(result, expected);
 }
 
@@ -452,7 +418,7 @@ TEST(VersionMap, FixRefKeyTombstones) {
     auto key5 = atom_key_with_version(id, 1, 1696590624590123209);
     version_map->write_version(store, key5, key4);
     auto key6 = atom_key_with_version(id, 0, 1696590624612743245);
-    auto entry = version_map->check_reload(store, id, LoadParameter{LoadType::LOAD_LATEST},  __FUNCTION__);
+    auto entry = version_map->check_reload(store, id, LoadStrategy{LoadType::LATEST, LoadObjective::INCLUDE_DELETED}, __FUNCTION__);
     version_map->journal_single_key(store, key5, entry->head_.value());
 
     auto valid = version_map->check_ref_key(store, id);
@@ -485,11 +451,11 @@ TEST(VersionMap, RewriteVersionKeys) {
     ASSERT_TRUE(version_map->check_ref_key(store, id));
 
     // This will just write tombstone key
-    arcticdb::tombstone_version(store, version_map, id, 2, pipelines::VersionQuery{});
-    arcticdb::tombstone_version(store, version_map, id, 1, pipelines::VersionQuery{});
+    arcticdb::tombstone_version(store, version_map, id, 2);
+    arcticdb::tombstone_version(store, version_map, id, 1);
 
-    auto index_key1 = arcticdb::get_specific_version(store, version_map, id, 1, pipelines::VersionQuery{});
-    auto index_key2 = arcticdb::get_specific_version(store, version_map, id, 2, pipelines::VersionQuery{});
+    auto index_key1 = arcticdb::get_specific_version(store, version_map, id, 1);
+    auto index_key2 = arcticdb::get_specific_version(store, version_map, id, 2);
 
     // will be null since they were tombstoned (but not actually deleted
     ASSERT_FALSE(index_key1.has_value());
@@ -497,15 +463,15 @@ TEST(VersionMap, RewriteVersionKeys) {
 
     version_map->remove_and_rewrite_version_keys(store, id);
 
-    auto final_index_key1 = arcticdb::get_specific_version(store, version_map, id, 1, pipelines::VersionQuery{});
-    auto final_index_key2 = arcticdb::get_specific_version(store, version_map, id, 2, pipelines::VersionQuery{});
+    auto final_index_key1 = arcticdb::get_specific_version(store, version_map, id, 1);
+    auto final_index_key2 = arcticdb::get_specific_version(store, version_map, id, 2);
 
     // will not be null anymore since the data actually existed
     ASSERT_TRUE(final_index_key1.has_value());
     ASSERT_TRUE(final_index_key2.has_value());
 
     std::vector<AtomKey> expected{key3, key2, key1};
-    auto result = get_all_versions(store, version_map, id, pipelines::VersionQuery{});
+    auto result = get_all_versions(store, version_map, id);
     ASSERT_EQ(result, expected);
 }
 
@@ -529,11 +495,11 @@ TEST(VersionMap, RecoverDeleted) {
 
     deleted = version_map->find_deleted_version_keys(store, id);
     ASSERT_EQ(deleted.size(), 3);
-    EXPECT_THROW({ get_all_versions(store, version_map, id, pipelines::VersionQuery{}); }, std::runtime_error);
+    EXPECT_THROW({ get_all_versions(store, version_map, id); }, std::runtime_error);
     version_map->recover_deleted(store, id);
 
     std::vector<AtomKey> expected{ key3, key2, key1};
-    auto result = get_all_versions(store, version_map, id, pipelines::VersionQuery{});
+    auto result = get_all_versions(store, version_map, id);
     ASSERT_EQ(result, expected);
 }
 
@@ -548,9 +514,9 @@ TEST(VersionMap, StorageLogging) {
     version_map->write_version(store, key2, key1);
     version_map->write_version(store, key3, key2);
 
-    tombstone_version(store, version_map, id, key1.version_id(), pipelines::VersionQuery{});
-    tombstone_version(store, version_map, id, key3.version_id(), pipelines::VersionQuery{});
-    tombstone_version(store, version_map, id, key2.version_id(), pipelines::VersionQuery{});
+    tombstone_version(store, version_map, id, key1.version_id());
+    tombstone_version(store, version_map, id, key3.version_id());
+    tombstone_version(store, version_map, id, key2.version_id());
 
     std::unordered_set<AtomKey> log_keys;
 
@@ -578,7 +544,7 @@ std::shared_ptr<VersionMapEntry> write_two_versions(std::shared_ptr<InMemoryStor
     auto entry = version_map->check_reload(
             store,
             id,
-            LoadParameter{LoadType::NOT_LOADED},
+            LoadStrategy{LoadType::NOT_LOADED, LoadObjective::INCLUDE_DELETED},
             __FUNCTION__);
 
     auto key1 = atom_key_with_version(id, 0, 0);
@@ -586,336 +552,324 @@ std::shared_ptr<VersionMapEntry> write_two_versions(std::shared_ptr<InMemoryStor
     write_symbol_ref(store, key1, std::nullopt, entry->head_.value());
     auto key2 = atom_key_with_version(id, 1, 1);
     version_map->do_write(store, key2, entry);
+    // We override the symbol ref without a prev_key on purpose. This way we'll only load the version=1 from the ref key
     write_symbol_ref(store, key2, std::nullopt, entry->head_.value());
 
     return entry;
 }
 
-TEST(VersionMap, CacheInvalidationLoadDownTo) {
-    using namespace arcticdb;
-    // Given - symbol with 2 versions - load downto version 1
-    // never time-invalidate the cache so we can test our other cache invalidation logic
-    ScopedConfig sc("VersionMap.ReloadInterval", std::numeric_limits<int64_t>::max());
-    auto store = std::make_shared<InMemoryStore>();
-
-    auto version_map = std::make_shared<VersionMap>();
-
-    StreamId id{"test"};
-    write_two_versions(store, version_map, id);
-
+// Produces the following version chain: v0 <- tombstone_all <- v1 <- v2 <- tombstone
+void write_alternating_deleted_undeleted(std::shared_ptr<InMemoryStore> store, std::shared_ptr<VersionMap> version_map, StreamId id) {
     auto entry = version_map->check_reload(
             store,
             id,
-            LoadParameter{LoadType::LOAD_DOWNTO, static_cast<SignedVersionId>(1)},
+            LoadStrategy{LoadType::NOT_LOADED, LoadObjective::INCLUDE_DELETED},
             __FUNCTION__);
 
-    ASSERT_TRUE(version_map->has_cached_entry(id, LoadParameter{LoadType::LOAD_LATEST_UNDELETED}));
+    auto key1 = atom_key_with_version(id, 0, 0);
+    auto key2 = atom_key_with_version(id, 1, 1);
+    auto key3 = atom_key_with_version(id, 2, 2);
 
-    // When - we delete version 1 and reload
-    uint64_t ver_to_delete = 1;
-    auto tombstone = version_map->write_tombstone(store, ver_to_delete, id, entry);
-    entry->tombstones_.try_emplace(ver_to_delete, std::move(tombstone));
+    // Write version 0
+    version_map->do_write(store, key1, entry);
+    write_symbol_ref(store, key1, std::nullopt, entry->head_.value());
+
+    // Tombstone_all on version 0
+    version_map->delete_all_versions(store, id);
 
-    ASSERT_FALSE(version_map->has_cached_entry(id, LoadParameter{LoadType::LOAD_LATEST_UNDELETED}));
+    // Write version 1
+    version_map->do_write(store, key2, entry);
+    write_symbol_ref(store, key2, std::nullopt, entry->head_.value());
 
-    LoadParameter load_param{LoadType::LOAD_LATEST_UNDELETED};
-    const auto latest_undeleted_entry = version_map->check_reload(store, id, load_param, __FUNCTION__);
+    // Write version 2
+    version_map->do_write(store, key3, entry);
+    write_symbol_ref(store, key3, std::nullopt, entry->head_.value());
 
-    // Then - version 0 should be returned
-    ASSERT_TRUE(latest_undeleted_entry->get_first_index(false).first.has_value());
+    // Tombstone version 2
+    version_map->write_tombstone(store, VersionId{2}, id, entry, timestamp{3});
 }
 
-TEST(VersionMap, CacheInvalidationLoadDownToFurther) {
-    using namespace arcticdb;
-    // Given - symbol with 2 versions - load downto version 1
-    // never time-invalidate the cache so we can test our other cache invalidation logic
-    ScopedConfig sc("VersionMap.ReloadInterval", std::numeric_limits<int64_t>::max());
+TEST(VersionMap, FollowingVersionChain){
+    // Set up the version chain v0(tombstone_all) <- v1 <- v2(tombstoned)
     auto store = std::make_shared<InMemoryStore>();
-
     auto version_map = std::make_shared<VersionMap>();
-
     StreamId id{"test"};
-    write_two_versions(store, version_map, id);
-
-    auto entry = version_map->check_reload(
-            store,
-            id,
-            LoadParameter{LoadType::LOAD_DOWNTO, static_cast<SignedVersionId>(1)},
-            __FUNCTION__);
-
-    ASSERT_FALSE(version_map->has_cached_entry(id,
-                                               LoadParameter{LoadType::LOAD_DOWNTO, static_cast<SignedVersionId>(0)}));
-    ASSERT_EQ(entry->get_indexes(false).size(), 1);
-
-    // When - we load down to version 0
-    const auto reloaded = version_map->check_reload(
-            store,
-            id,
-            LoadParameter{LoadType::LOAD_DOWNTO, static_cast<SignedVersionId>(0)},
-            __FUNCTION__);
-
-    // Then - version 0 should be returned
-    ASSERT_EQ(entry->get_indexes(false).size(), 2);
-    ASSERT_TRUE(
-            version_map->has_cached_entry(id, LoadParameter{LoadType::LOAD_DOWNTO, static_cast<SignedVersionId>(0)}));
+    write_alternating_deleted_undeleted(store, version_map, id);
+
+    auto check_strategy_loads_to = [&](LoadStrategy load_strategy, VersionId should_load_to){
+        auto ref_entry = VersionMapEntry{};
+        read_symbol_ref(store, id, ref_entry);
+        auto follow_result = std::make_shared<VersionMapEntry>();
+
+        version_map->follow_version_chain(store, ref_entry, follow_result, load_strategy);
+        EXPECT_EQ(follow_result->load_progress_.oldest_loaded_index_version_, VersionId{should_load_to});
+    };
+
+    check_strategy_loads_to(LoadStrategy{LoadType::DOWNTO, LoadObjective::INCLUDE_DELETED, static_cast<SignedVersionId>(0)}, 0);
+    check_strategy_loads_to(LoadStrategy{LoadType::DOWNTO, LoadObjective::INCLUDE_DELETED, static_cast<SignedVersionId>(-2)}, 1);
+    // DOWN_TO will not skip through tombstoned versions even when include_deleted=false
+    check_strategy_loads_to(LoadStrategy{LoadType::DOWNTO, LoadObjective::UNDELETED_ONLY, static_cast<SignedVersionId>(-1)}, 2);
+    check_strategy_loads_to(LoadStrategy{LoadType::DOWNTO, LoadObjective::UNDELETED_ONLY, static_cast<SignedVersionId>(0)}, 0);
+
+    // FROM_TIME when include_deleted=false will skip through deleted versions to go to the latest undeleted version before the timestamp.
+    check_strategy_loads_to(LoadStrategy{LoadType::FROM_TIME, LoadObjective::UNDELETED_ONLY, static_cast<timestamp>(10)}, 1);
+    check_strategy_loads_to(LoadStrategy{LoadType::FROM_TIME, LoadObjective::UNDELETED_ONLY, static_cast<timestamp>(0)}, 0);
+    check_strategy_loads_to(LoadStrategy{LoadType::FROM_TIME, LoadObjective::INCLUDE_DELETED, static_cast<timestamp>(2)}, 2);
+    check_strategy_loads_to(LoadStrategy{LoadType::FROM_TIME, LoadObjective::INCLUDE_DELETED, static_cast<timestamp>(0)}, 0);
+
+    check_strategy_loads_to(LoadStrategy{LoadType::LATEST, LoadObjective::INCLUDE_DELETED}, 2);
+    check_strategy_loads_to(LoadStrategy{LoadType::LATEST, LoadObjective::UNDELETED_ONLY}, 1);
+
+    check_strategy_loads_to(LoadStrategy{LoadType::ALL, LoadObjective::INCLUDE_DELETED}, 0);
+    check_strategy_loads_to(LoadStrategy{LoadType::ALL, LoadObjective::UNDELETED_ONLY}, 0);
 }
 
-TEST(VersionMap, CacheInvalidationLoadDownToAfterLoadLatest) {
-    using namespace arcticdb;
-    // Given - symbol with 2 versions - load down to version 1
-    // never time-invalidate the cache so we can test our other cache invalidation logic
+TEST(VersionMap, FollowingVersionChainWithCaching){
     ScopedConfig sc("VersionMap.ReloadInterval", std::numeric_limits<int64_t>::max());
+    // Set up the version chain v0(tombstone_all) <- v1 <- v2(tombstoned)
     auto store = std::make_shared<InMemoryStore>();
-    StreamId id{"test"};
-
     auto version_map = std::make_shared<VersionMap>();
-
+    StreamId id{"test"};
+    write_alternating_deleted_undeleted(store, version_map, id);
+    // We create an empty version map after populating the versions
+    version_map = std::make_shared<VersionMap>();
+
+    auto check_loads_versions = [&](LoadStrategy load_strategy, uint32_t should_load_any, uint32_t should_load_undeleted){
+        auto loaded = version_map->check_reload(store, id, load_strategy, __FUNCTION__);
+        EXPECT_EQ(loaded->get_indexes(true).size(), should_load_any);
+        EXPECT_EQ(loaded->get_indexes(false).size(), should_load_undeleted);
+    };
+
+    check_loads_versions(LoadStrategy{LoadType::DOWNTO, LoadObjective::INCLUDE_DELETED, static_cast<SignedVersionId>(-1)}, 1, 0);
+    // FROM_TIME should not be cached by the DOWNTO and should reload from storage up to the latest undeleted version, hence loading 2 versions, 1 of which is undeleted.
+    check_loads_versions(LoadStrategy{LoadType::FROM_TIME, LoadObjective::UNDELETED_ONLY, static_cast<timestamp>(10)}, 2, 1);
+    // LATEST should be cached by the FROM_TIME, so we still have the same 2 loaded versions
+    check_loads_versions(LoadStrategy{LoadType::LATEST, LoadObjective::INCLUDE_DELETED}, 2, 1);
+    // This FROM_TIME should still use the cached 2 versions
+    check_loads_versions(LoadStrategy{LoadType::FROM_TIME, LoadObjective::INCLUDE_DELETED, static_cast<timestamp>(1)}, 2, 1);
+
+    // We just get the entry to use for the tombstone and the write
     auto entry = version_map->check_reload(
             store,
             id,
-            LoadParameter{LoadType::LOAD_LATEST},
+            LoadStrategy{LoadType::NOT_LOADED, LoadObjective::INCLUDE_DELETED},
             __FUNCTION__);
+    // We delete the only undeleted key
+    version_map->write_tombstone(store, VersionId{1}, id, entry, timestamp{4});
 
-    auto key = atom_key_with_version(id, 0, 0);
-    version_map->do_write(store, key, entry);
-    write_symbol_ref(store, key, std::nullopt, entry->head_.value());
+    // LATEST should still be cached, but the cached entry now needs to have no undeleted keys
+    check_loads_versions(LoadStrategy{LoadType::LATEST, LoadObjective::INCLUDE_DELETED}, 2, 0);
+    // FROM_TIME UNDELETED_ONLY should no longer be cached even though we used the same request before because the undeleted key it went to got deleted. So it will load the entire version chain
+    check_loads_versions(LoadStrategy{LoadType::FROM_TIME, LoadObjective::UNDELETED_ONLY, static_cast<timestamp>(10)}, 3, 0);
 
-    // When - we load latest then down to version 0
-    auto reloaded = version_map->check_reload(
-            store,
-            id,
-            LoadParameter{LoadType::LOAD_DOWNTO, static_cast<SignedVersionId>(0)},
-            __FUNCTION__);
+    // We add a new undeleted key
+    auto key4 = atom_key_with_version(id, 3, 5);
+    version_map->do_write(store, key4, entry);
+    write_symbol_ref(store, key4, std::nullopt, entry->head_.value());
 
-    // Then - version 0 should be returned
-    ASSERT_EQ(reloaded->get_indexes(false).size(), 1);
-    ASSERT_TRUE(version_map->has_cached_entry(id, LoadParameter{LoadType::LOAD_DOWNTO, static_cast<SignedVersionId>(0)}));
-}
+    // LATEST should still be cached, but the cached entry now needs to have one more undeleted version
+    check_loads_versions(LoadStrategy{LoadType::LATEST, LoadObjective::INCLUDE_DELETED}, 4, 1);
 
-TEST(VersionMap, CacheInvalidationAfterLoadLatestUndeleted) {
-    using namespace arcticdb;
-    ScopedConfig sc("VersionMap.ReloadInterval", std::numeric_limits<int64_t>::max());
-    auto store = std::make_shared<InMemoryStore>();
-
-    auto version_map = std::make_shared<VersionMap>();
-
-    StreamId id{"test"};
-    write_two_versions(store, version_map, id);
-
-    version_map->check_reload(
-            store,
-            id,
-            LoadParameter{LoadType::LOAD_LATEST_UNDELETED},
-            __FUNCTION__);
-
-    ASSERT_TRUE(version_map->has_cached_entry(id, LoadParameter{LoadType::LOAD_LATEST_UNDELETED}));
-    ASSERT_TRUE(version_map->has_cached_entry(id, LoadParameter{LoadType::LOAD_LATEST}));
-    ASSERT_TRUE(version_map->has_cached_entry(id, LoadParameter{LoadType::NOT_LOADED}));
-    ASSERT_FALSE(version_map->has_cached_entry(id, LoadParameter{LoadType::LOAD_ALL}));
-    ASSERT_FALSE(version_map->has_cached_entry(id, LoadParameter{LoadType::LOAD_FROM_TIME, static_cast<timestamp>(1234)}));
-    ASSERT_FALSE(version_map->has_cached_entry(id, LoadParameter{LoadType::LOAD_UNDELETED}));
-    ASSERT_TRUE(version_map->has_cached_entry(id, LoadParameter{LoadType::LOAD_DOWNTO, static_cast<SignedVersionId>(1)}));
-    ASSERT_FALSE(version_map->has_cached_entry(id, LoadParameter{LoadType::LOAD_DOWNTO, static_cast<SignedVersionId>(0)}));
-
-    const auto reloaded = version_map->check_reload(
-            store,
-            id,
-            LoadParameter{LoadType::LOAD_DOWNTO, static_cast<SignedVersionId>(0)},
-            __FUNCTION__);
+    // We delete everything with a tombstone_all
+    version_map->delete_all_versions(store, id);
 
-    ASSERT_EQ(reloaded->get_indexes(false).size(), 2);
-    ASSERT_TRUE(version_map->has_cached_entry(id, LoadParameter{LoadType::LOAD_DOWNTO, static_cast<SignedVersionId>(0)}));
+    // LATEST should still be cached, but now have no undeleted versions
+    check_loads_versions(LoadStrategy{LoadType::LATEST, LoadObjective::INCLUDE_DELETED}, 4, 0);
 }
 
-TEST(VersionMap, CacheInvalidationAfterLoadFromTime) {
-    using namespace arcticdb;
-    ScopedConfig sc("VersionMap.ReloadInterval", std::numeric_limits<int64_t>::max());
+TEST(VersionMap, FollowingVersionChainEndEarlyOnTombstoneAll) {
     auto store = std::make_shared<InMemoryStore>();
-
     auto version_map = std::make_shared<VersionMap>();
-
     StreamId id{"test"};
-    write_two_versions(store, version_map, id);
 
-    version_map->check_reload(
-            store,
-            id,
-            LoadParameter{LoadType::LOAD_FROM_TIME, static_cast<timestamp>(123)},
-            __FUNCTION__);
-
-    ASSERT_TRUE(version_map->has_cached_entry(id, LoadParameter{LoadType::LOAD_LATEST_UNDELETED}));
-    ASSERT_TRUE(version_map->has_cached_entry(id, LoadParameter{LoadType::LOAD_LATEST}));
-    ASSERT_TRUE(version_map->has_cached_entry(id, LoadParameter{LoadType::NOT_LOADED}));
-    ASSERT_FALSE(version_map->has_cached_entry(id, LoadParameter{LoadType::LOAD_ALL}));
-    ASSERT_FALSE(version_map->has_cached_entry(id, LoadParameter{LoadType::LOAD_FROM_TIME, static_cast<timestamp>(1234)}));
-    ASSERT_FALSE(version_map->has_cached_entry(id, LoadParameter{LoadType::LOAD_UNDELETED}));
-    ASSERT_FALSE(version_map->has_cached_entry(id, LoadParameter{LoadType::LOAD_DOWNTO, static_cast<SignedVersionId>(1)}));
+    write_two_versions(store, version_map, id);
+    // Deleting should add a TOMBSTONE_ALL which should end searching for undeleted versions early.
+    version_map->delete_all_versions(store, id);
 
-    const auto reloaded = version_map->check_reload(
-            store,
-            id,
-            LoadParameter{LoadType::LOAD_DOWNTO, static_cast<SignedVersionId>(0)},
-            __FUNCTION__);
+    auto ref_entry = VersionMapEntry{};
+    read_symbol_ref(store, id, ref_entry);
+    auto follow_result = std::make_shared<VersionMapEntry>();
+
+    for (auto load_strategy: {
+        LoadStrategy{LoadType::DOWNTO, LoadObjective::UNDELETED_ONLY, static_cast<SignedVersionId>(0)},
+        LoadStrategy{LoadType::FROM_TIME, LoadObjective::UNDELETED_ONLY, static_cast<timestamp>(0)},
+        LoadStrategy{LoadType::ALL, LoadObjective::UNDELETED_ONLY},
+        LoadStrategy{LoadType::LATEST, LoadObjective::UNDELETED_ONLY}
+    }) {
+        follow_result->clear();
+        version_map->follow_version_chain(store, ref_entry, follow_result, load_strategy);
+        // When loading with any of the specified load strategies with include_deleted=false we should end following the version chain early
+        // at version 1 because that's when we encounter the TOMBSTONE_ALL.
+        EXPECT_EQ(follow_result->load_progress_.oldest_loaded_index_version_, VersionId{1});
+    }
 
-    ASSERT_EQ(reloaded->get_indexes(false).size(), 2);
+    for (auto load_strategy: {
+            LoadStrategy{LoadType::DOWNTO, LoadObjective::INCLUDE_DELETED, static_cast<SignedVersionId>(0)},
+            LoadStrategy{LoadType::FROM_TIME, LoadObjective::INCLUDE_DELETED, static_cast<timestamp>(0)},
+            LoadStrategy{LoadType::ALL, LoadObjective::INCLUDE_DELETED}
+    }) {
+        follow_result->clear();
+        version_map->follow_version_chain(store, ref_entry, follow_result, load_strategy);
+        // When loading with any of the specified load strategies with include_deleted=true we should continue to the beginning
+        // at version 0 even though it was deleted.
+        EXPECT_EQ(follow_result->load_progress_.oldest_loaded_index_version_, VersionId{0});
+    }
 }
 
-TEST(VersionMap, CacheInvalidationAfterLoadFromTimeTombstoned) {
-    using namespace arcticdb;
+TEST(VersionMap, CacheInvalidation) {
     ScopedConfig sc("VersionMap.ReloadInterval", std::numeric_limits<int64_t>::max());
+    // Set up the version chain v0(tombstone_all) <- v1 <- v2(tombstoned)
     auto store = std::make_shared<InMemoryStore>();
-
     auto version_map = std::make_shared<VersionMap>();
-
     StreamId id{"test"};
-    auto entry = write_two_versions(store, version_map, id);
-
-    uint64_t ver_to_delete = 1;
-    auto tombstone = version_map->write_tombstone(store, ver_to_delete, id, entry);
-
-    version_map->check_reload(
-            store,
-            id,
-            LoadParameter{LoadType::LOAD_FROM_TIME, static_cast<timestamp>(2)},
-            __FUNCTION__);
-
-    // LOAD_FROM_TIME always goes at least as far back as the latest undeleted version, even if that is earlier than
-    // the specified load_from_time_or_until
-    ASSERT_TRUE(version_map->has_cached_entry(id, LoadParameter{LoadType::LOAD_LATEST_UNDELETED}));
-    ASSERT_TRUE(version_map->has_cached_entry(id, LoadParameter{LoadType::LOAD_LATEST}));
+    write_alternating_deleted_undeleted(store, version_map, id);
+
+    auto check_caching = [&](LoadStrategy to_load, LoadStrategy to_check_if_cached, bool expected_outcome){
+        auto clean_version_map = std::make_shared<VersionMap>();
+        // Load to_load inside the clean version map cache
+        clean_version_map->check_reload(store, id, to_load, __FUNCTION__);
+        // Check whether to_check_if_cached is being cached by to_load
+        EXPECT_EQ(clean_version_map->has_cached_entry(id, to_check_if_cached), expected_outcome);
+    };
+
+    auto check_all_caching = [&](const std::vector<LoadStrategy>& to_load, const std::vector<LoadStrategy>& to_check_if_cached, bool expected_result){
+        for (auto to_load_strategy : to_load) {
+            for (auto to_check_if_cached_param : to_check_if_cached){
+                check_caching(to_load_strategy, to_check_if_cached_param, expected_result);
+            }
+        }
+    };
+
+    auto load_all_param = LoadStrategy{LoadType::ALL, LoadObjective::INCLUDE_DELETED};
+    auto load_all_undeleted_param = LoadStrategy{LoadType::ALL, LoadObjective::UNDELETED_ONLY};
+    check_caching(load_all_param, load_all_undeleted_param, true);
+    check_caching(load_all_undeleted_param, load_all_param, false);
+
+    constexpr auto num_versions = 3u;
+    std::vector<LoadStrategy> should_load_to_v[num_versions] = {
+        // Different parameters which should all load to v0
+        std::vector<LoadStrategy>{
+            LoadStrategy{LoadType::DOWNTO, LoadObjective::INCLUDE_DELETED, static_cast<SignedVersionId>(0)},
+            LoadStrategy{LoadType::DOWNTO, LoadObjective::INCLUDE_DELETED, static_cast<SignedVersionId>(-3)},
+            LoadStrategy{LoadType::FROM_TIME, LoadObjective::INCLUDE_DELETED, static_cast<timestamp>(0)},
+        },
+
+        // Different parameters which should all load to v1
+        std::vector<LoadStrategy>{
+            LoadStrategy{LoadType::DOWNTO, LoadObjective::INCLUDE_DELETED, static_cast<SignedVersionId>(1)},
+            LoadStrategy{LoadType::DOWNTO, LoadObjective::INCLUDE_DELETED, static_cast<SignedVersionId>(-2)},
+            LoadStrategy{LoadType::FROM_TIME, LoadObjective::INCLUDE_DELETED, static_cast<timestamp>(1)},
+            LoadStrategy{LoadType::FROM_TIME, LoadObjective::UNDELETED_ONLY,
+                         static_cast<timestamp>(2)}, // when include_deleted=false FROM_TIME searches for an undeleted version
+            LoadStrategy{LoadType::LATEST, LoadObjective::UNDELETED_ONLY},
+        },
+
+        // Different parameters which should all load to v2
+        std::vector<LoadStrategy>{
+            LoadStrategy{LoadType::DOWNTO, LoadObjective::INCLUDE_DELETED, static_cast<SignedVersionId>(2)},
+            LoadStrategy{LoadType::DOWNTO, LoadObjective::INCLUDE_DELETED, static_cast<SignedVersionId>(-1)},
+            LoadStrategy{LoadType::FROM_TIME, LoadObjective::INCLUDE_DELETED, static_cast<timestamp>(2)},
+            LoadStrategy{LoadType::LATEST, LoadObjective::INCLUDE_DELETED},
+        }
+    };
 
-    const auto reloaded = version_map->check_reload(
-            store,
-            id,
-            LoadParameter{LoadType::LOAD_LATEST_UNDELETED},
-            __FUNCTION__);
+    for (auto i=0u; i<num_versions; ++i){
+        for (auto j=0u; j<num_versions; ++j){
+            // For every two versions we check that all load params for earlier versions cache load paramse for later versions:
+            check_all_caching(should_load_to_v[i], should_load_to_v[j], i<=j);
+        }
 
-    ASSERT_EQ(reloaded->get_indexes(false).size(), 1);
-    ASSERT_EQ(reloaded->get_indexes(true).size(), 2);
+        // ALL and LOAD_UNDELETED because they both load to v0 (UNDELETED_ONLY will load v0 because only there it will load
+        check_all_caching({load_all_param, load_all_undeleted_param}, should_load_to_v[i], true);
+        check_all_caching(should_load_to_v[i], {load_all_param, load_all_undeleted_param}, false);
+    }
 }
 
-TEST(VersionMap, CacheInvalidationLoadAllAfterLoadFromTime) {
+
+TEST(VersionMap, CacheInvalidationWithTombstoneAfterLoad) {
     using namespace arcticdb;
+    // Given - symbol with 2 versions - load downto version 1
+    // never time-invalidate the cache so we can test our other cache invalidation logic
     ScopedConfig sc("VersionMap.ReloadInterval", std::numeric_limits<int64_t>::max());
     auto store = std::make_shared<InMemoryStore>();
 
     auto version_map = std::make_shared<VersionMap>();
-
     StreamId id{"test"};
     write_two_versions(store, version_map, id);
 
-    auto reloaded = version_map->check_reload(
-            store,
-            id,
-            LoadParameter{LoadType::LOAD_FROM_TIME, static_cast<timestamp>(1)},
-            __FUNCTION__);
-
-    ASSERT_FALSE(version_map->has_cached_entry(id, LoadParameter{LoadType::LOAD_ALL}));
-    ASSERT_EQ(reloaded->get_indexes(false).size(), 1);
-
-    reloaded = version_map->check_reload(
-            store,
-            id,
-            LoadParameter{LoadType::LOAD_ALL},
-            __FUNCTION__);
-
-    ASSERT_EQ(reloaded->get_indexes(false).size(), 2);
-}
-
-TEST(VersionMap, CacheInvalidationAfterLoadUndeleted) {
-    using namespace arcticdb;
-    ScopedConfig sc("VersionMap.ReloadInterval", std::numeric_limits<int64_t>::max());
-    auto store = std::make_shared<InMemoryStore>();
-    StreamId id{"test"};
-
-    auto version_map = std::make_shared<VersionMap>();
+    // Use a clean version_map
+    version_map = std::make_shared<VersionMap>();
 
     auto entry = version_map->check_reload(
             store,
             id,
-            LoadParameter{LoadType::NOT_LOADED},
+            LoadStrategy{LoadType::DOWNTO, LoadObjective::INCLUDE_DELETED, static_cast<SignedVersionId>(1)},
             __FUNCTION__);
 
-    auto key1 = atom_key_with_version(id, 0, 0);
-    version_map->do_write(store, key1, entry);
-    write_symbol_ref(store, key1, std::nullopt, entry->head_.value());
-    auto key2 = atom_key_with_version(id, 1, 1);
-    version_map->do_write(store, key2, entry);
-    write_symbol_ref(store, key2, std::nullopt, entry->head_.value());
-    auto key3 = atom_key_with_version(id, 2, 2);
-    version_map->do_write(store, key3, entry);
-    write_symbol_ref(store, key3, std::nullopt, entry->head_.value());
+    ASSERT_TRUE(version_map->has_cached_entry(id, LoadStrategy{LoadType::LATEST, LoadObjective::UNDELETED_ONLY}));
+    ASSERT_TRUE(version_map->has_cached_entry(id, LoadStrategy{LoadType::FROM_TIME, LoadObjective::UNDELETED_ONLY, static_cast<timestamp>(1)}));
+    ASSERT_FALSE(version_map->has_cached_entry(id, LoadStrategy{LoadType::FROM_TIME, LoadObjective::UNDELETED_ONLY, static_cast<timestamp>(0)}));
+    ASSERT_TRUE(version_map->has_cached_entry(id, LoadStrategy{LoadType::DOWNTO, LoadObjective::UNDELETED_ONLY, static_cast<SignedVersionId>(-1)}));
+    ASSERT_FALSE(version_map->has_cached_entry(id, LoadStrategy{LoadType::DOWNTO, LoadObjective::UNDELETED_ONLY, static_cast<SignedVersionId>(-2)}));
 
-    auto tombstone_all = get_tombstone_all_key(key2, 3);
-    entry->try_set_tombstone_all(tombstone_all);
-    version_map->do_write(store, tombstone_all, entry);
-    write_symbol_ref(store, key2, std::nullopt, entry->head_.value());
+    // When - we delete version 1 and reload
+    version_map->write_tombstone(store, VersionId{1}, id, entry);
 
-    entry = version_map->check_reload(
-            store,
-            id,
-            LoadParameter{LoadType::LOAD_UNDELETED},
-            __FUNCTION__);
+    // Now when the cached version is deleted, we should invalidate the cache for load parameters which look for undeleted.
+    ASSERT_FALSE(version_map->has_cached_entry(id, LoadStrategy{LoadType::LATEST, LoadObjective::UNDELETED_ONLY}));
+    ASSERT_FALSE(version_map->has_cached_entry(id, LoadStrategy{LoadType::FROM_TIME, LoadObjective::UNDELETED_ONLY, static_cast<timestamp>(1)}));
+    ASSERT_TRUE(version_map->has_cached_entry(id, LoadStrategy{LoadType::LATEST, LoadObjective::INCLUDE_DELETED}));
+    ASSERT_TRUE(version_map->has_cached_entry(id, LoadStrategy{LoadType::FROM_TIME, LoadObjective::INCLUDE_DELETED, static_cast<timestamp>(1)}));
+    ASSERT_TRUE(version_map->has_cached_entry(id, LoadStrategy{LoadType::DOWNTO, LoadObjective::UNDELETED_ONLY, static_cast<SignedVersionId>(-1)}));
 
-    ASSERT_EQ(entry->get_indexes(true).size(), 2);
-    ASSERT_TRUE(version_map->has_cached_entry(id, LoadParameter{LoadType::LOAD_LATEST_UNDELETED}));
-    ASSERT_TRUE(version_map->has_cached_entry(id, LoadParameter{LoadType::LOAD_LATEST}));
-    ASSERT_TRUE(version_map->has_cached_entry(id, LoadParameter{LoadType::NOT_LOADED}));
-    ASSERT_FALSE(version_map->has_cached_entry(id, LoadParameter{LoadType::LOAD_ALL}));
-    ASSERT_TRUE(version_map->has_cached_entry(id, LoadParameter{LoadType::LOAD_FROM_TIME, static_cast<timestamp>(1234)}));
-    ASSERT_TRUE(version_map->has_cached_entry(id, LoadParameter{LoadType::LOAD_UNDELETED}));
-    ASSERT_TRUE(version_map->has_cached_entry(id, LoadParameter{LoadType::LOAD_DOWNTO, static_cast<SignedVersionId>(1)}));
-
-    // Should not rely on the cache upon (eg) a LOAD_ALL here hence pick up an extra INDEX
-    auto reloaded = version_map->check_reload(
-            store,
-            id,
-            LoadParameter{LoadType::LOAD_ALL},
-            __FUNCTION__);
+    LoadStrategy load_strategy{LoadType::LATEST, LoadObjective::UNDELETED_ONLY};
+    const auto latest_undeleted_entry = version_map->check_reload(store, id, load_strategy, __FUNCTION__);
 
-    ASSERT_EQ(reloaded->get_indexes(true).size(), 3);
+    // Then - version 0 should be returned
+    ASSERT_TRUE(latest_undeleted_entry->get_first_index(false).first.has_value());
 }
 
-TEST(VersionMap, CacheInvalidationAfterLoadAll) {
+TEST(VersionMap, CacheInvalidationWithTombstoneAllAfterLoad) {
     using namespace arcticdb;
+    // Given - symbol with 2 versions - load downto version 0
+    // never time-invalidate the cache so we can test our other cache invalidation logic
     ScopedConfig sc("VersionMap.ReloadInterval", std::numeric_limits<int64_t>::max());
     auto store = std::make_shared<InMemoryStore>();
-    StreamId id{"test"};
 
     auto version_map = std::make_shared<VersionMap>();
+    StreamId id{"test"};
+    write_two_versions(store, version_map, id);
+
+    // Use a clean version_map
+    version_map = std::make_shared<VersionMap>();
 
     auto entry = version_map->check_reload(
             store,
             id,
-            LoadParameter{LoadType::LOAD_ALL},
+            LoadStrategy{LoadType::DOWNTO, LoadObjective::INCLUDE_DELETED, static_cast<SignedVersionId>(0)},
             __FUNCTION__);
 
-    ASSERT_TRUE(version_map->has_cached_entry(id, LoadParameter{LoadType::LOAD_LATEST_UNDELETED}));
-    ASSERT_TRUE(version_map->has_cached_entry(id, LoadParameter{LoadType::LOAD_LATEST}));
-    ASSERT_TRUE(version_map->has_cached_entry(id, LoadParameter{LoadType::NOT_LOADED}));
-    ASSERT_TRUE(version_map->has_cached_entry(id, LoadParameter{LoadType::LOAD_ALL}));
-    ASSERT_TRUE(version_map->has_cached_entry(id, LoadParameter{LoadType::LOAD_FROM_TIME, static_cast<timestamp>(1234)}));
-    ASSERT_TRUE(version_map->has_cached_entry(id, LoadParameter{LoadType::LOAD_UNDELETED}));
-    ASSERT_TRUE(version_map->has_cached_entry(id, LoadParameter{LoadType::LOAD_DOWNTO, static_cast<SignedVersionId>(1)}));
-
-    // Write a key to storage but not the in memory version map
-    auto key = atom_key_with_version(id, 0, 0);
-    auto other_version_map = std::make_shared<VersionMap>();
-    auto other_entry = other_version_map->check_reload(
-            store,
-            id,
-            LoadParameter{LoadType::LOAD_ALL},
-            __FUNCTION__);
-    other_version_map->do_write(store, key, other_entry);
+    ASSERT_TRUE(version_map->has_cached_entry(id, LoadStrategy{LoadType::LATEST, LoadObjective::UNDELETED_ONLY}));
+    ASSERT_TRUE(version_map->has_cached_entry(id, LoadStrategy{LoadType::FROM_TIME, LoadObjective::UNDELETED_ONLY, static_cast<timestamp>(1)}));
+    ASSERT_TRUE(version_map->has_cached_entry(id, LoadStrategy{LoadType::FROM_TIME, LoadObjective::UNDELETED_ONLY, static_cast<timestamp>(0)}));
+    ASSERT_TRUE(version_map->has_cached_entry(id, LoadStrategy{LoadType::DOWNTO, LoadObjective::UNDELETED_ONLY, static_cast<SignedVersionId>(-1)}));
+    ASSERT_TRUE(version_map->has_cached_entry(id, LoadStrategy{LoadType::DOWNTO, LoadObjective::UNDELETED_ONLY, static_cast<SignedVersionId>(-2)}));
 
-    // Then - it should not be visible to the original map as we rely on the cache
-    auto reloaded = version_map->check_reload(
-            store,
-            id,
-            LoadParameter{LoadType::LOAD_UNDELETED},
-            __FUNCTION__);
+    // When - we delete version 1
+    auto tombstone_key = version_map->write_tombstone(store, VersionId{1}, id, entry);
+
+    // We should not invalidate the cache because the version we loaded to is still undeleted
+    ASSERT_TRUE(version_map->has_cached_entry(id, LoadStrategy{LoadType::LATEST, LoadObjective::UNDELETED_ONLY}));
+    ASSERT_TRUE(version_map->has_cached_entry(id, LoadStrategy{LoadType::FROM_TIME, LoadObjective::UNDELETED_ONLY, static_cast<timestamp>(1)}));
+    ASSERT_TRUE(version_map->has_cached_entry(id, LoadStrategy{LoadType::FROM_TIME, LoadObjective::UNDELETED_ONLY, static_cast<timestamp>(0)}));
+
+    // When - we delete all versions without reloading
+    version_map->write_tombstone_all_key_internal(store, tombstone_key, entry);
 
-    ASSERT_TRUE(reloaded->get_indexes(false).empty());
+    // We should invalidate cached undeleted checks
+    ASSERT_FALSE(version_map->has_cached_entry(id, LoadStrategy{LoadType::LATEST, LoadObjective::UNDELETED_ONLY}));
+    ASSERT_FALSE(version_map->has_cached_entry(id, LoadStrategy{LoadType::FROM_TIME, LoadObjective::UNDELETED_ONLY, static_cast<timestamp>(1)}));
+    ASSERT_FALSE(version_map->has_cached_entry(id, LoadStrategy{LoadType::FROM_TIME, LoadObjective::UNDELETED_ONLY, static_cast<timestamp>(0)}));
 }
 
 #define GTEST_COUT std::cerr << "[          ] [ INFO ]"
diff --git a/cpp/arcticdb/version/test/test_version_map_batch.cpp b/cpp/arcticdb/version/test/test_version_map_batch.cpp
index 761aa1c01c..9713741824 100644
--- a/cpp/arcticdb/version/test/test_version_map_batch.cpp
+++ b/cpp/arcticdb/version/test/test_version_map_batch.cpp
@@ -57,7 +57,7 @@ TEST_F(VersionMapBatchStore, SimpleVersionIdQueries) {
         auto stream = fmt::format("stream_{}", i);
         for(uint64_t j = 0; j < num_versions_per_stream; j++){
             stream_ids.emplace_back(stream);
-            version_queries.emplace_back(VersionQuery{SpecificVersionQuery{static_cast<SignedVersionId>(j)}, false});
+            version_queries.emplace_back(VersionQuery{SpecificVersionQuery{static_cast<SignedVersionId>(j)}});
         }
     }
 
@@ -99,7 +99,7 @@ TEST_F(VersionMapBatchStore, SimpleTimestampQueries) {
         auto stream = fmt::format("stream_{}", i);
         for(uint64_t j = 0; j < num_versions_per_stream; j++){
             stream_ids.emplace_back(stream);
-            version_queries.emplace_back(VersionQuery{SpecificVersionQuery{static_cast<SignedVersionId>(j)}, false});
+            version_queries.emplace_back(VersionQuery{SpecificVersionQuery{static_cast<SignedVersionId>(j)}});
         }
     }
 
@@ -111,7 +111,7 @@ TEST_F(VersionMapBatchStore, SimpleTimestampQueries) {
     for(uint64_t i = 0; i < num_streams; i++){
         for(uint64_t j = 0; j < num_versions_per_stream; j++){
             uint64_t idx = i * num_versions_per_stream + j;
-            version_queries.emplace_back(VersionQuery{TimestampVersionQuery{timestamp(versions[idx]->creation_ts())}, false});
+            version_queries.emplace_back(VersionQuery{TimestampVersionQuery{timestamp(versions[idx]->creation_ts())}});
         }
     }
 
@@ -149,7 +149,7 @@ TEST_F(VersionMapBatchStore, MultipleVersionsSameSymbolVersionIdQueries) {
     // Add queries
     for(uint64_t i = 0; i < num_versions; i++){
         stream_ids.emplace_back("stream_0");
-        version_queries.emplace_back(VersionQuery{SpecificVersionQuery{static_cast<SignedVersionId>(i)}, false});
+        version_queries.emplace_back(VersionQuery{SpecificVersionQuery{static_cast<SignedVersionId>(i)}});
     }
 
     // Do query
@@ -182,7 +182,7 @@ TEST_F(VersionMapBatchStore, MultipleVersionsSameSymbolTimestampQueries) {
     // Add queries
     for(uint64_t i = 0; i < num_versions; i++){
         stream_ids.emplace_back("stream_0");
-        version_queries.emplace_back(VersionQuery{SpecificVersionQuery{static_cast<SignedVersionId>(i)}, false});
+        version_queries.emplace_back(VersionQuery{SpecificVersionQuery{static_cast<SignedVersionId>(i)}});
     }
 
     // Do query
@@ -191,7 +191,7 @@ TEST_F(VersionMapBatchStore, MultipleVersionsSameSymbolTimestampQueries) {
     //Secondly, once we have the timestamps in hand, we are going to query them
     version_queries.clear();
     for(uint64_t i = 0; i < num_versions; i++){
-        version_queries.emplace_back(VersionQuery{TimestampVersionQuery{timestamp(versions[i]->creation_ts())}, false});
+        version_queries.emplace_back(VersionQuery{TimestampVersionQuery{timestamp(versions[i]->creation_ts())}});
     }
 
     // Now we can perform the actual batch query per timestamps
@@ -229,7 +229,7 @@ TEST_F(VersionMapBatchStore, CombinedQueries) {
         auto stream = fmt::format("stream_{}", i);
         for(uint64_t j = 0; j < num_versions_per_stream; j++){
             stream_ids.emplace_back(stream);
-            version_queries.emplace_back(VersionQuery{SpecificVersionQuery{static_cast<SignedVersionId>(j)}, false});
+            version_queries.emplace_back(VersionQuery{SpecificVersionQuery{static_cast<SignedVersionId>(j)}});
         }
     }
 
@@ -244,11 +244,11 @@ TEST_F(VersionMapBatchStore, CombinedQueries) {
         for(uint64_t j = 0; j < num_versions_per_stream; j++){
             uint64_t idx = i * num_versions_per_stream + j;
             stream_ids.emplace_back(stream);
-            version_queries.emplace_back(VersionQuery{SpecificVersionQuery{static_cast<SignedVersionId>(j)}, false});
+            version_queries.emplace_back(VersionQuery{SpecificVersionQuery{static_cast<SignedVersionId>(j)}});
             stream_ids.emplace_back(stream);
-            version_queries.emplace_back(VersionQuery{TimestampVersionQuery{timestamp(versions[idx]->creation_ts())}, false});
+            version_queries.emplace_back(VersionQuery{TimestampVersionQuery{timestamp(versions[idx]->creation_ts())}});
             stream_ids.emplace_back(stream);
-            version_queries.emplace_back(VersionQuery{std::monostate{}, false});
+            version_queries.emplace_back(VersionQuery{std::monostate{}});
         }
     }
 
diff --git a/cpp/arcticdb/version/test/test_version_store.cpp b/cpp/arcticdb/version/test/test_version_store.cpp
index 8ba24ab280..c01385a6b6 100644
--- a/cpp/arcticdb/version/test/test_version_store.cpp
+++ b/cpp/arcticdb/version/test/test_version_store.cpp
@@ -136,7 +136,7 @@ TEST(PythonVersionStore, IterationVsRefWrite) {
     auto ref_entry = std::make_shared<VersionMapEntry>();
 
     version_map->load_via_iteration(mock_store, stream_id, iter_entry);
-    version_map->load_via_ref_key(mock_store, stream_id, LoadParameter{LoadType::LOAD_ALL}, ref_entry);
+    version_map->load_via_ref_key(mock_store, stream_id, LoadStrategy{LoadType::ALL, LoadObjective::INCLUDE_DELETED}, ref_entry);
 
     EXPECT_EQ(std::string(iter_entry->head_.value().view()), std::string(ref_entry->head_.value().view()));
     ASSERT_EQ(iter_entry->keys_.size(), ref_entry->keys_.size());
@@ -151,7 +151,7 @@ TEST(PythonVersionStore, IterationVsRefWrite) {
     auto ref_entry_compact = std::make_shared<VersionMapEntry>();
 
     version_map->load_via_iteration(mock_store, stream_id, iter_entry_compact);
-    version_map->load_via_ref_key(mock_store, stream_id, LoadParameter{LoadType::LOAD_ALL}, ref_entry_compact);
+    version_map->load_via_ref_key(mock_store, stream_id, LoadStrategy{LoadType::ALL, arcticdb::LoadObjective::INCLUDE_DELETED}, ref_entry_compact);
 
     EXPECT_EQ(std::string(iter_entry_compact->head_.value().view()), std::string(ref_entry_compact->head_.value().view()));
     ASSERT_EQ(iter_entry_compact->keys_.size(), ref_entry_compact->keys_.size());
@@ -376,21 +376,21 @@ TEST(VersionStore, TestReadTimestampAt) {
 
   auto version_map = version_store._test_get_version_map();
   version_map->write_version(mock_store, key1, std::nullopt);
-  auto key = load_index_key_from_time(mock_store, version_map, id, timestamp(0), pipelines::VersionQuery{});
+  auto key = load_index_key_from_time(mock_store, version_map, id, timestamp(0));
   ASSERT_EQ(key.value().content_hash(), 3);
 
   version_map->write_version(mock_store, key2, key1);
-  key = load_index_key_from_time(mock_store, version_map, id, timestamp(0), pipelines::VersionQuery{});
+  key = load_index_key_from_time(mock_store, version_map, id, timestamp(0));
   ASSERT_EQ(key.value().content_hash(), 3);
-  key = load_index_key_from_time(mock_store, version_map, id, timestamp(1), pipelines::VersionQuery{});
+  key = load_index_key_from_time(mock_store, version_map, id, timestamp(1));
   ASSERT_EQ(key.value().content_hash(), 4);
 
   version_map->write_version(mock_store, key3, key2);
-  key = load_index_key_from_time(mock_store, version_map, id, timestamp(0), pipelines::VersionQuery{});
+  key = load_index_key_from_time(mock_store, version_map, id, timestamp(0));
   ASSERT_EQ(key.value().content_hash(), 3);
-  key = load_index_key_from_time(mock_store, version_map, id, timestamp(1), pipelines::VersionQuery{});
+  key = load_index_key_from_time(mock_store, version_map, id, timestamp(1));
   ASSERT_EQ(key.value().content_hash(), 4);
-  key = load_index_key_from_time(mock_store, version_map, id, timestamp(2), pipelines::VersionQuery{});
+  key = load_index_key_from_time(mock_store, version_map, id, timestamp(2));
   ASSERT_EQ(key.value().content_hash(), 5);
 }
 
@@ -408,7 +408,7 @@ TEST(VersionStore, TestReadTimestampAtInequality) {
 
   auto version_map = version_store._test_get_version_map();
   version_map->write_version(mock_store, key1, std::nullopt);
-  auto key = load_index_key_from_time(mock_store, version_map, id, timestamp(1), pipelines::VersionQuery{});
+  auto key = load_index_key_from_time(mock_store, version_map, id, timestamp(1));
   ASSERT_EQ(static_cast<bool>(key), true);
   ASSERT_EQ(key.value().content_hash(), 3);
 }
diff --git a/cpp/arcticdb/version/test/version_backwards_compat.hpp b/cpp/arcticdb/version/test/version_backwards_compat.hpp
index a132052697..41c63b7adb 100644
--- a/cpp/arcticdb/version/test/version_backwards_compat.hpp
+++ b/cpp/arcticdb/version/test/version_backwards_compat.hpp
@@ -33,7 +33,7 @@ std::deque<AtomKey> backwards_compat_delete_all_versions(
     const StreamId& stream_id
     ) {
     std::deque<AtomKey> output;
-    auto entry = version_map->check_reload(store, stream_id, LoadParameter{LoadType::LOAD_ALL}, __FUNCTION__);
+    auto entry = version_map->check_reload(store, stream_id, LoadStrategy{LoadType::ALL, LoadObjective::INCLUDE_DELETED}, __FUNCTION__);
     auto indexes = entry->get_indexes(false);
     output.assign(std::begin(indexes), std::end(indexes));
 
@@ -49,7 +49,7 @@ std::vector<AtomKey> backwards_compat_write_and_prune_previous(std::shared_ptr<S
     log::version().debug("Version map pruning previous versions for stream {}", key.id());
 
     std::vector<AtomKey> output;
-    auto entry = version_map->check_reload(store, key.id(), LoadParameter{LoadType::LOAD_ALL},  __FUNCTION__);
+    auto entry = version_map->check_reload(store, key.id(), LoadStrategy{LoadType::ALL, LoadObjective::INCLUDE_DELETED}, __FUNCTION__);
 
     auto old_entry = *entry;
     entry->clear();
diff --git a/cpp/arcticdb/version/test/version_map_model.hpp b/cpp/arcticdb/version/test/version_map_model.hpp
index cf6be722d5..83aaf036d2 100644
--- a/cpp/arcticdb/version/test/version_map_model.hpp
+++ b/cpp/arcticdb/version/test/version_map_model.hpp
@@ -36,7 +36,7 @@ struct MapStorePair {
 
     void write_version(const std::string &id) {
         log::version().info("MapStorePair, write version {}", id);
-        auto prev = get_latest_version(store_, map_, id, pipelines::VersionQuery{}).first;
+        auto prev = get_latest_version(store_, map_, id).first;
         auto version_id = prev ? prev->version_id() + 1 : 0;
         map_->write_version(store_, make_test_index_key(id, version_id, KeyType::TABLE_INDEX), prev);
     }
@@ -51,7 +51,7 @@ struct MapStorePair {
 
     void write_and_prune_previous(const std::string &id) {
         log::version().info("MapStorePair, write_and_prune_previous version {}", id);
-        auto prev = get_latest_version(store_, map_, id, pipelines::VersionQuery{}).first;
+        auto prev = get_latest_version(store_, map_, id).first;
         auto version_id = prev ? prev->version_id() + 1 : 0;
 
         if(tombstones_)