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

Issue:
batch_write, batch_append, and batch_write_metadata orphan data when prune_previous_version=True
The version map is updated correctly, but the index and data keys are never deleted. Steps to reproduce given a `NativeVersionStore` `lib`:
```
import pandas as pd
lib_tool = lib.library_tool()
sym = "test_symbol"
df0 = pd.DataFrame({"col_0": ["a", "b"]}, index=pd.date_range("2000-01-01", periods=2))
df1 = pd.DataFrame({"col_0": ["c", "d"]}, index=pd.date_range("2000-01-03", periods=2))

lib.write(sym, df0)

lib.batch_write([sym], [df1], prune_previous_version=True) # Or batch_append

print(lib.list_versions(sym)) # Correctly shows only version 1 is live
print(lib_tool.find_keys(KeyType.TABLE_INDEX)) # Shows index keys for versions 0 and 1
print(lib_tool.find_keys(KeyType.TABLE_DATA)) # Shows data keys for versions 0 and 1
```


Discussion / hints:

Fix patch:
diff --git a/cpp/arcticdb/async/async_store.hpp b/cpp/arcticdb/async/async_store.hpp
index 7a938a7b30..4cce5a8c78 100644
--- a/cpp/arcticdb/async/async_store.hpp
+++ b/cpp/arcticdb/async/async_store.hpp
@@ -253,6 +253,13 @@ class AsyncStore : public Store {
                async::submit_io_task(RemoveBatchTask{keys, library_, opts});
     }
 
+    folly::Future<std::vector<RemoveKeyResultType>> remove_keys(std::vector<entity::VariantKey> &&keys,
+                                                                storage::RemoveOpts opts) override {
+        return keys.empty() ?
+               std::vector<RemoveKeyResultType>() :
+               async::submit_io_task(RemoveBatchTask{std::move(keys), library_, opts});
+    }
+
     void copy_to_results(
         std::vector<folly::Future<storage::KeySegmentPair>> &batch,
         std::vector<storage::KeySegmentPair> &res
diff --git a/cpp/arcticdb/stream/stream_sink.hpp b/cpp/arcticdb/stream/stream_sink.hpp
index 0b1a89ca18..6a7b85e760 100644
--- a/cpp/arcticdb/stream/stream_sink.hpp
+++ b/cpp/arcticdb/stream/stream_sink.hpp
@@ -121,6 +121,9 @@ struct StreamSink {
     [[nodiscard]] virtual folly::Future<std::vector<RemoveKeyResultType>> remove_keys(
         const std::vector<entity::VariantKey> &keys, storage::RemoveOpts opts = storage::RemoveOpts{}) = 0;
 
+    [[nodiscard]] virtual folly::Future<std::vector<RemoveKeyResultType>> remove_keys(
+        std::vector<entity::VariantKey> &&keys, storage::RemoveOpts opts = storage::RemoveOpts{}) = 0;
+
     virtual timestamp current_timestamp() = 0;
 };
 
diff --git a/cpp/arcticdb/version/local_versioned_engine.cpp b/cpp/arcticdb/version/local_versioned_engine.cpp
index 679a9b65ac..31aa26ba01 100644
--- a/cpp/arcticdb/version/local_versioned_engine.cpp
+++ b/cpp/arcticdb/version/local_versioned_engine.cpp
@@ -44,7 +44,7 @@ LocalVersionedEngine::LocalVersionedEngine(
     }
 }
 
-void LocalVersionedEngine::delete_unreferenced_pruned_indexes(
+folly::Future<folly::Unit> LocalVersionedEngine::delete_unreferenced_pruned_indexes(
         const std::vector<AtomKey> &pruned_indexes,
         const AtomKey& key_to_keep
 ) {
@@ -56,12 +56,13 @@ void LocalVersionedEngine::delete_unreferenced_pruned_indexes(
                     pruned_indexes);
             in_snaps.insert(key_to_keep);
             PreDeleteChecks checks{false, false, false, false, in_snaps};
-            delete_trees_responsibly(not_in_snaps, {}, {}, checks);
+            return delete_trees_responsibly(not_in_snaps, {}, {}, checks);
         }
     } catch (const std::exception &ex) {
         // Best-effort so deliberately swallow
         log::version().warn("Could not prune previous versions due to: {}", ex.what());
     }
+    return folly::Unit();
 }
 
 void LocalVersionedEngine::create_column_stats_internal(
@@ -555,7 +556,7 @@ VersionedItem LocalVersionedEngine::update_internal(
                                           dynamic_schema);
         if (prune_previous_versions) {
             auto pruned_indexes = version_map()->write_and_prune_previous(store(), versioned_item.key_, update_info.previous_index_key_);
-            delete_unreferenced_pruned_indexes(pruned_indexes, versioned_item.key_);
+            delete_unreferenced_pruned_indexes(pruned_indexes, versioned_item.key_).get();
         } else {
             version_map()->write_version(store(), versioned_item.key_);
         }
@@ -602,7 +603,7 @@ VersionedItem LocalVersionedEngine::write_versioned_metadata_internal(
     if(prune_previous_versions) {
         auto versioned_item = VersionedItem{index_key};
         auto pruned_indexes = version_map()->write_and_prune_previous(store(), index_key, update_info.previous_index_key_);
-        delete_unreferenced_pruned_indexes(pruned_indexes, versioned_item.key_);
+        delete_unreferenced_pruned_indexes(pruned_indexes, versioned_item.key_).get();
         return versioned_item;
     }
     else {
@@ -638,7 +639,7 @@ VersionedItem LocalVersionedEngine::write_versioned_dataframe_internal(
 
     if(prune_previous_versions) {
         auto pruned_indexes = version_map()->write_and_prune_previous(store(), versioned_item.key_, maybe_prev);
-        delete_unreferenced_pruned_indexes(pruned_indexes, versioned_item.key_);
+        delete_unreferenced_pruned_indexes(pruned_indexes, versioned_item.key_).get();
         return versioned_item;
     }
     else {
@@ -743,7 +744,7 @@ std::unordered_map<StreamId, VersionId> min_versions_for_each_stream(const std::
     return out;
 }
 
-void LocalVersionedEngine::delete_trees_responsibly(
+folly::Future<folly::Unit> LocalVersionedEngine::delete_trees_responsibly(
         const std::vector<IndexTypeKey>& orig_keys_to_delete,
         const arcticdb::MasterSnapshotMap& snapshot_map,
         const std::optional<SnapshotId>& snapshot_being_deleted,
@@ -790,19 +791,14 @@ void LocalVersionedEngine::delete_trees_responsibly(
     if (load_type != LoadType::NOT_LOADED) {
         std::unordered_map<StreamId, std::shared_ptr<VersionMapEntry>> entry_map;
         {
-            auto min_vers = min_versions_for_each_stream(orig_keys_to_delete);
-
-            async::submit_tasks_for_range(min_vers,
-                    [store=store(), version_map=version_map(), load_type](auto& sym_version) {
-                        auto load_param = load_type == LoadType::LOAD_DOWNTO
-                                ? LoadParameter{load_type, static_cast<SignedVersionId>(sym_version.second)}
-                                : LoadParameter{load_type};
-                        return async::submit_io_task(CheckReloadTask{store, version_map, sym_version.first,
-                                                                     load_param, true});
-                    },
-                    [&entry_map](auto& sym_version, auto&& entry) {
-                        entry_map.emplace(std::move(sym_version.first), entry);
-                    });
+            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, true, false, __FUNCTION__);
+                entry_map.emplace(std::move(min.first), entry);
+            }
         }
 
         keys_to_delete.remove_if([&entry_map, &check, &not_to_delete](const auto& key) {
@@ -841,39 +837,52 @@ void LocalVersionedEngine::delete_trees_responsibly(
     auto data_keys_not_to_be_deleted = get_data_keys_set(store(), *not_to_delete, read_opts);
     not_to_delete.clear();
     log::version().debug("Forbidden: {} total of data keys", data_keys_not_to_be_deleted.size());
-
     RemoveOpts remove_opts;
     remove_opts.ignores_missing_key_ = true;
+    
+    std::vector<entity::VariantKey> vks_column_stats;
+    std::transform(keys_to_delete->begin(),
+                    keys_to_delete->end(),
+                    std::back_inserter(vks_column_stats),
+                    [](const IndexTypeKey& index_key) {
+        return index_key_to_column_stats_key(index_key);
+    });
+    log::version().debug("Number of Column Stats keys to be deleted: {}", vks_column_stats.size());
 
-    std::vector<entity::VariantKey> vks;
-    // Delete any associated column stats keys first
-    if (!dry_run) {
-        std::transform(keys_to_delete->begin(),
-                       keys_to_delete->end(),
-                       std::back_inserter(vks),
-                       [](const IndexTypeKey& index_key) {
-            return index_key_to_column_stats_key(index_key);
-        });
-        store()->remove_keys(vks, remove_opts).get();
-    }
-
-    vks.clear();
-    if (!dry_run) {
-        std::copy(keys_to_delete->begin(), keys_to_delete->end(), std::back_inserter(vks));
-        store()->remove_keys(vks, remove_opts).get();
-    }
+    std::vector<entity::VariantKey> vks_to_delete;
+    std::copy(std::make_move_iterator(keys_to_delete->begin()),
+                std::make_move_iterator(keys_to_delete->end()),
+                std::back_inserter(vks_to_delete));
+    log::version().debug("Number of Index keys to be deleted: {}", vks_to_delete.size());
 
-    vks.clear();
+    std::vector<entity::VariantKey> vks_data_to_delete;
     std::copy_if(std::make_move_iterator(data_keys_to_be_deleted.begin()),
-                 std::make_move_iterator(data_keys_to_be_deleted.end()),
-                 std::back_inserter(vks),
-                 [&](const auto& k) {return !data_keys_not_to_be_deleted.count(k);});
-    log::version().debug("Index keys deleted. Proceeding with {} number of data keys", vks.size());
+                std::make_move_iterator(data_keys_to_be_deleted.end()),
+                std::back_inserter(vks_data_to_delete),
+                [&](const auto& k) {return !data_keys_not_to_be_deleted.count(k);});
+    log::version().debug("Number of Data keys to be deleted: {}", vks_data_to_delete.size());
+
+    folly::Future<folly::Unit> remove_keys_fut;
     if (!dry_run) {
-        store()->remove_keys(vks, remove_opts).get();
+        // Delete any associated column stats keys first
+        remove_keys_fut = store()->remove_keys(std::move(vks_column_stats), remove_opts)
+        .thenValue([this, vks_to_delete = std::move(vks_to_delete), remove_opts](auto&& ) mutable {
+            log::version().debug("Column Stats keys deleted.");
+            return store()->remove_keys(std::move(vks_to_delete), remove_opts);
+        })
+        .thenValue([this, vks_data_to_delete = std::move(vks_data_to_delete), remove_opts](auto&&) mutable {
+            log::version().debug("Index keys deleted.");
+            return store()->remove_keys(std::move(vks_data_to_delete), remove_opts);
+        })
+        .thenValue([](auto&&){
+            log::version().debug("Data keys deleted.");
+            return folly::Unit();
+        });
     }
+    return remove_keys_fut;
 }
 
+
 void LocalVersionedEngine::remove_incomplete(
     const StreamId& stream_id
     ) {
@@ -1147,6 +1156,28 @@ std::vector<std::variant<ReadVersionOutput, DataError>> LocalVersionedEngine::ba
     return read_versions_or_errors;
 }
 
+std::vector<folly::Future<folly::Unit>> LocalVersionedEngine::batch_write_version_and_prune_if_needed(
+    const std::vector<AtomKey>& index_keys,
+    const std::vector<UpdateInfo>& stream_update_info_vector,
+    bool prune_previous_versions) {
+    if(prune_previous_versions) {
+        std::vector<folly::Future<folly::Unit>> pruned_keys_futures;
+        auto pruned_versions_futures = batch_write_and_prune_previous(store(), version_map(), index_keys, stream_update_info_vector);
+        for(auto&& [idx, pruned_version_fut] : folly::enumerate(pruned_versions_futures)) {
+            pruned_keys_futures.push_back(
+                std::move(pruned_version_fut)
+                .thenValue([this, &index_key = index_keys[idx]](auto&& atom_key_vec){
+                    return delete_unreferenced_pruned_indexes(std::move(atom_key_vec), index_key);
+                })
+            );
+        }
+        return pruned_keys_futures;
+    }
+    else {
+        return batch_write_version(store(), version_map(), index_keys);
+    }
+}
+
 folly::Future<VersionedItem> LocalVersionedEngine::write_index_key_to_version_map_async(
     const std::shared_ptr<VersionMap> &version_map,
     const AtomKey&& index_key,
@@ -1156,7 +1187,10 @@ folly::Future<VersionedItem> LocalVersionedEngine::write_index_key_to_version_ma
     folly::Future<folly::Unit> write_version_fut;
 
     if(prune_previous_versions) {
-        write_version_fut = async::submit_io_task(WriteAndPrunePreviousTask{store(), version_map, index_key, stream_update_info.previous_index_key_});
+        write_version_fut = async::submit_io_task(WriteAndPrunePreviousTask{store(), version_map, index_key, stream_update_info.previous_index_key_})
+        .thenValue([this, index_key](auto&& atom_key_vec){
+            return delete_unreferenced_pruned_indexes(std::move(atom_key_vec), index_key);
+        });
     } else {
         write_version_fut = async::submit_io_task(WriteVersionTask{store(), version_map, index_key});
     }
@@ -1272,7 +1306,7 @@ VersionedItem LocalVersionedEngine::append_internal(
                                           validate_index);
         if (prune_previous_versions) {
             auto pruned_indexes = version_map()->write_and_prune_previous(store(), versioned_item.key_, update_info.previous_index_key_);
-            delete_unreferenced_pruned_indexes(pruned_indexes, versioned_item.key_);
+            delete_unreferenced_pruned_indexes(pruned_indexes, versioned_item.key_).get();
         } else {
             version_map()->write_version(store(), versioned_item.key_);
         }
diff --git a/cpp/arcticdb/version/local_versioned_engine.hpp b/cpp/arcticdb/version/local_versioned_engine.hpp
index e8797640a3..2fec960161 100644
--- a/cpp/arcticdb/version/local_versioned_engine.hpp
+++ b/cpp/arcticdb/version/local_versioned_engine.hpp
@@ -146,7 +146,7 @@ class LocalVersionedEngine : public VersionedEngine {
         const PreDeleteChecks& checks = default_pre_delete_checks
     ) override {
         auto snapshot_map = get_master_snapshots_map(store());
-        delete_trees_responsibly(idx_to_be_deleted, snapshot_map, std::nullopt, checks);
+        delete_trees_responsibly(idx_to_be_deleted, snapshot_map, std::nullopt, checks).get();
     };
 
     /**
@@ -157,7 +157,7 @@ class LocalVersionedEngine : public VersionedEngine {
      * to exclude it from shared data check
      * @param dry_run Only do the check, but don't actually delete anything.
      */
-    void delete_trees_responsibly(
+    folly::Future<folly::Unit> delete_trees_responsibly(
         const std::vector<IndexTypeKey>& idx_to_be_deleted,
         const arcticdb::MasterSnapshotMap& snapshot_map,
         const std::optional<SnapshotId>& snapshot_being_deleted = std::nullopt,
@@ -351,6 +351,11 @@ class LocalVersionedEngine : public VersionedEngine {
         const UpdateInfo& stream_update_info,
         bool prune_previous_versions);
 
+    std::vector<folly::Future<folly::Unit>> batch_write_version_and_prune_if_needed(
+        const std::vector<AtomKey>& index_keys,
+        const std::vector<UpdateInfo>& stream_update_info_vector,
+        bool prune_previous_versions);
+
     std::vector<VersionedItem> batch_write_versioned_dataframe_internal(
         const std::vector<StreamId>& stream_ids,
         std::vector<InputTensorFrame>&& frames,
@@ -387,7 +392,7 @@ class LocalVersionedEngine : public VersionedEngine {
      *
      * @param pruned_indexes Must all share the same id() and should be tombstoned.
      */
-    void delete_unreferenced_pruned_indexes(
+    folly::Future<folly::Unit> delete_unreferenced_pruned_indexes(
             const std::vector<AtomKey> &pruned_indexes,
             const AtomKey& key_to_keep
     );
diff --git a/cpp/arcticdb/version/version_map_batch_methods.hpp b/cpp/arcticdb/version/version_map_batch_methods.hpp
index 7421e58512..a932aaf79d 100644
--- a/cpp/arcticdb/version/version_map_batch_methods.hpp
+++ b/cpp/arcticdb/version/version_map_batch_methods.hpp
@@ -255,7 +255,7 @@ inline std::vector<folly::Future<std::optional<AtomKey>>> batch_get_versions_asy
     return output;
 }
 
-inline void batch_write_version(
+inline std::vector<folly::Future<folly::Unit>> batch_write_version(
     const std::shared_ptr<Store> &store,
     const std::shared_ptr<VersionMap> &version_map,
     const std::vector<AtomKey> &keys) {
@@ -265,21 +265,21 @@ inline void batch_write_version(
         results.emplace_back(async::submit_io_task(WriteVersionTask{store, version_map, key}));
     }
 
-    folly::collect(results).wait();
+    return results;
 }
 
-inline void batch_write_and_prune_previous(
+inline std::vector<folly::Future<std::vector<AtomKey>>> batch_write_and_prune_previous(
     const std::shared_ptr<Store> &store,
     const std::shared_ptr<VersionMap> &version_map,
     const std::vector<AtomKey> &keys,
     const std::vector<version_store::UpdateInfo>& stream_update_info_vector) {
-    std::vector<folly::Future<folly::Unit>> results;
+    std::vector<folly::Future<std::vector<AtomKey>>> results;
     results.reserve(keys.size());
     for(auto key : folly::enumerate(keys)){
         auto previous_index_key = stream_update_info_vector[key.index].previous_index_key_;
         results.emplace_back(async::submit_io_task(WriteAndPrunePreviousTask{store, version_map, *key, previous_index_key}));
     }
-
-    folly::collect(results).wait();
+    
+    return results;
 }
 } //namespace arcticdb
diff --git a/cpp/arcticdb/version/version_store_api.cpp b/cpp/arcticdb/version/version_store_api.cpp
index 7dc19ab464..a4203596a5 100644
--- a/cpp/arcticdb/version/version_store_api.cpp
+++ b/cpp/arcticdb/version/version_store_api.cpp
@@ -90,11 +90,7 @@ std::vector<VersionedItem> PythonVersionStore::batch_write_index_keys_to_version
     const std::vector<UpdateInfo>& stream_update_info_vector,
     bool prune_previous_versions) {
 
-    if(prune_previous_versions)
-        batch_write_and_prune_previous(store(), version_map(), index_keys, stream_update_info_vector);
-    else
-        batch_write_version(store(), version_map(), index_keys);
-
+    folly::collect(batch_write_version_and_prune_if_needed(index_keys, stream_update_info_vector, prune_previous_versions)).get();
     std::vector<VersionedItem> output(index_keys.size());
     for(auto key : folly::enumerate(index_keys))
         output[key.index] = *key;
@@ -377,7 +373,7 @@ void PythonVersionStore::add_to_snapshot(
     if(variant_key_type(snap_key) == KeyType::SNAPSHOT_REF && cfg().write_options().delayed_deletes()) {
         tombstone_snapshot(store(), to_ref(snap_key), std::move(snap_segment), version_map()->log_changes());
     } else {
-        delete_trees_responsibly(deleted_keys, get_master_snapshots_map(store()), snap_name);
+        delete_trees_responsibly(deleted_keys, get_master_snapshots_map(store()), snap_name).get();
         if (version_map()->log_changes()) {
             log_delete_snapshot(store(), snap_name);
         }
@@ -418,7 +414,7 @@ void PythonVersionStore::remove_from_snapshot(
     if(variant_key_type(snap_key) == KeyType::SNAPSHOT_REF && cfg().write_options().delayed_deletes()) {
         tombstone_snapshot(store(), to_ref(snap_key), std::move(snap_segment), version_map()->log_changes());
     } else {
-        delete_trees_responsibly(deleted_keys, get_master_snapshots_map(store()), snap_name);
+        delete_trees_responsibly(deleted_keys, get_master_snapshots_map(store()), snap_name).get();
         if (version_map()->log_changes()) {
             log_delete_snapshot(store(), snap_name);
         }
@@ -842,7 +838,7 @@ void PythonVersionStore::delete_snapshot_sync(const SnapshotId& snap_name, const
         delete_trees_responsibly(
             index_keys_in_current_snapshot,
             snap_map,
-            snap_name);
+            snap_name).get();
         ARCTICDB_DEBUG(log::version(), "Deleted orphaned index keys in snapshot {}", snap_name);
     } catch(const std::exception &ex) {
         log::version().warn("Garbage collection of unreachable deleted index keys failed due to: {}", ex.what());
@@ -940,7 +936,7 @@ void PythonVersionStore::prune_previous_versions(const StreamId& stream_id) {
 
     auto previous = ::arcticdb::get_specific_version(store(), version_map(), stream_id, prev_id.value(), true, false);
     auto pruned_indexes = version_map()->tombstone_from_key_or_all(store(), stream_id, previous);
-    delete_unreferenced_pruned_indexes(pruned_indexes, latest.value());
+    delete_unreferenced_pruned_indexes(pruned_indexes, latest.value()).get();
 }
 
 void PythonVersionStore::delete_all_versions(const StreamId& stream_id) {
@@ -1032,12 +1028,7 @@ std::vector<VersionedItem> PythonVersionStore::batch_write_metadata(
                                                                    std::move(user_meta_proto)}));
     }
     auto index_keys = folly::collect(fut_vec).get();
-    if(prune_previous_versions) {
-        batch_write_and_prune_previous(store(), version_map(), index_keys, stream_update_info_vector);
-    } else {
-        batch_write_version(store(), version_map(), index_keys);
-    }
-
+    folly::collect(batch_write_version_and_prune_if_needed(index_keys, stream_update_info_vector, prune_previous_versions)).get();
     std::vector<VersionedItem> output(index_keys.size());
     for(auto key : folly::enumerate(index_keys))
         output[key.index] = std::move(*key);
diff --git a/cpp/arcticdb/version/version_tasks.hpp b/cpp/arcticdb/version/version_tasks.hpp
index 19f17cf2dc..790a4bb81a 100644
--- a/cpp/arcticdb/version/version_tasks.hpp
+++ b/cpp/arcticdb/version/version_tasks.hpp
@@ -166,10 +166,9 @@ struct WriteAndPrunePreviousTask : async::BaseTask {
         maybe_prev_(std::move(maybe_prev)) {
     }
 
-    folly::Unit operator()() {
+    folly::Future<std::vector<AtomKey>> operator()() {
         ScopedLock lock(version_map_->get_lock_object(key_.id()));
-        version_map_->write_and_prune_previous(store_, key_, maybe_prev_);
-        return folly::Unit{};
+        return version_map_->write_and_prune_previous(store_, key_, maybe_prev_);
     }
 };
 


Test patch:
diff --git a/cpp/arcticdb/storage/test/in_memory_store.hpp b/cpp/arcticdb/storage/test/in_memory_store.hpp
index 5d7181f6f7..552f43d6d9 100644
--- a/cpp/arcticdb/storage/test/in_memory_store.hpp
+++ b/cpp/arcticdb/storage/test/in_memory_store.hpp
@@ -304,6 +304,16 @@ namespace arcticdb {
             return output;
         }
 
+        folly::Future<std::vector<RemoveKeyResultType>>
+        remove_keys(std::vector<entity::VariantKey> &&keys, storage::RemoveOpts opts) override {
+            std::vector<RemoveKeyResultType> output;
+            for (const auto &key: keys) {
+                output.emplace_back(remove_key_sync(key, opts));
+            }
+
+            return output;
+        }
+
         void insert_atom_key(const AtomKey &key) {
             seg_by_atom_key_.insert(std::make_pair(key, std::make_unique<SegmentInMemory>()));
         }
diff --git a/cpp/arcticdb/stream/test/mock_stores.hpp b/cpp/arcticdb/stream/test/mock_stores.hpp
index 03f5ce7b4f..3908d978d4 100644
--- a/cpp/arcticdb/stream/test/mock_stores.hpp
+++ b/cpp/arcticdb/stream/test/mock_stores.hpp
@@ -112,6 +112,10 @@ class NullStore :
         util::raise_rte("Not implemented");
     }
 
+    folly::Future<std::vector<RemoveKeyResultType>> remove_keys(std::vector<entity::VariantKey>&&) override {
+        util::raise_rte("Not implemented");
+    }
+
     folly::Future<std::vector<AtomKey>> batch_write(
         std::vector<std::pair<PartialKey, SegmentInMemory>> &&,
         const std::unordered_map<ContentHash, AtomKey>&,
diff --git a/cpp/arcticdb/version/test/test_version_map.cpp b/cpp/arcticdb/version/test/test_version_map.cpp
index 5594f22541..e1bf5e9dd3 100644
--- a/cpp/arcticdb/version/test/test_version_map.cpp
+++ b/cpp/arcticdb/version/test/test_version_map.cpp
@@ -598,7 +598,7 @@ TEST_F(VersionMapStore, StressTestBatchSameSymbol) {
 
     auto store = test_store_->_test_get_store();
     auto version_map = std::make_shared<VersionMap>();
-    batch_write_version(store, version_map, keys);
+    folly::collect(batch_write_version(store, version_map, keys)).get();
 }
 
 TEST_F(VersionMapStore, StressTestBatchWrite) {
@@ -616,6 +616,6 @@ TEST_F(VersionMapStore, StressTestBatchWrite) {
 
     auto store = test_store_->_test_get_store();
     auto version_map = std::make_shared<VersionMap>();
-    batch_write_version(store, version_map, keys);
+    folly::collect(batch_write_version(store, version_map, keys)).get();
 }
 } // namespace arcticdb
diff --git a/python/tests/integration/arcticdb/version_store/test_basic_version_store.py b/python/tests/integration/arcticdb/version_store/test_basic_version_store.py
index 0d5f63d5e5..ca3bedbcde 100644
--- a/python/tests/integration/arcticdb/version_store/test_basic_version_store.py
+++ b/python/tests/integration/arcticdb/version_store/test_basic_version_store.py
@@ -210,6 +210,99 @@ def test_prune_previous_versions_multiple_times(lmdb_version_store, symbol):
     assert len([ver for ver in lmdb_version_store.list_versions() if not ver["deleted"]]) == 1
 
 
+def test_prune_previous_versions_write_batch(lmdb_version_store):
+    """Verify that the batch write method correctly prunes previous versions when the corresponding option is specified.
+    """
+    # Given
+    lib = lmdb_version_store
+    lib_tool = lib.library_tool()
+    sym1 = "test_symbol1"
+    sym2 = "test_symbol2"
+    df0 = pd.DataFrame({"col_0": ["a", "b"]}, index=pd.date_range("2000-01-01", periods=2))
+    df1 = pd.DataFrame({"col_0": ["c", "d"]}, index=pd.date_range("2000-01-03", periods=2))
+
+    # When
+    lib.batch_write([sym1, sym2], [df0, df0])
+    lib.batch_write([sym1, sym2], [df1, df1], prune_previous_version=True)
+
+    # Then - only latest version and keys should survive
+    assert len(lib.list_versions(sym1)) == 1
+    assert len(lib.list_versions(sym2)) == 1
+    assert len(lib_tool.find_keys(KeyType.TABLE_INDEX)) == 2
+    assert len(lib_tool.find_keys(KeyType.TABLE_DATA)) == 2
+
+    # Then - we got 3 version keys per symbol: version 0, version 0 tombstone, version 1
+    keys_for_sym1 = lib_tool.find_keys_for_id(KeyType.VERSION, sym1)
+    keys_for_sym2 = lib_tool.find_keys_for_id(KeyType.VERSION, sym2)
+
+    assert len(keys_for_sym1) == 3
+    assert len(keys_for_sym2) == 3
+    # Then - we got 4 symbol keys: 2 for each of the writes
+    assert len(lib_tool.find_keys(KeyType.SYMBOL_LIST)) == 4
+
+
+def test_prune_previous_versions_batch_write_metadata(lmdb_version_store):
+    """Verify that the batch write metadata method correctly prunes previous versions when the corresponding option is specified.
+    """
+    # Given
+    lib = lmdb_version_store
+    lib_tool = lib.library_tool()
+    sym1 = "test_symbol1"
+    sym2 = "test_symbol2"
+    meta0 = {"a": 0}
+    meta1 = {"a": 1}
+
+    # When
+    lib.batch_write([sym1, sym2], [None, None], metadata_vector=[meta0, meta0])
+    lib.batch_write_metadata([sym1, sym2], [meta1, meta1], prune_previous_version=True)
+
+    # Then - only latest version and keys should survive
+    assert len(lib.list_versions(sym1)) == 1
+    assert len(lib.list_versions(sym2)) == 1
+    assert len(lib_tool.find_keys(KeyType.TABLE_INDEX)) == 2
+    assert len(lib_tool.find_keys(KeyType.TABLE_DATA)) == 2
+
+    # Then - we got 3 version keys per symbol: version 0, version 0 tombstone, version 1
+    keys_for_sym1 = lib_tool.find_keys_for_id(KeyType.VERSION, sym1)
+    keys_for_sym2 = lib_tool.find_keys_for_id(KeyType.VERSION, sym2)
+
+    assert len(keys_for_sym1) == 3
+    assert len(keys_for_sym2) == 3
+    # Then - we got 2 symbol keys: 1 for each of the writes
+    assert len(lib_tool.find_keys(KeyType.SYMBOL_LIST)) == 2
+
+
+def test_prune_previous_versions_append_batch(lmdb_version_store):
+    """Verify that the batch append method correctly prunes previous versions when the corresponding option is specified.
+    """
+    # Given
+    lib = lmdb_version_store
+    lib_tool = lib.library_tool()
+    sym1 = "test_symbol1"
+    sym2 = "test_symbol2"
+    df0 = pd.DataFrame({"col_0": ["a", "b"]}, index=pd.date_range("2000-01-01", periods=2))
+    df1 = pd.DataFrame({"col_0": ["c", "d"]}, index=pd.date_range("2000-01-03", periods=2))
+
+    # When
+    lib.batch_write([sym1, sym2], [df0, df0])
+    lib.batch_append([sym1, sym2], [df1, df1], prune_previous_version=True)
+
+    # Then - only latest version and index keys should survive. Data keys remain the same
+    assert len(lib.list_versions(sym1)) == 1
+    assert len(lib.list_versions(sym2)) == 1
+    assert len(lib_tool.find_keys(KeyType.TABLE_INDEX)) == 2
+    assert len(lib_tool.find_keys(KeyType.TABLE_DATA)) == 4
+
+    # Then - we got 3 version keys per symbol: version 0, version 0 tombstone, version 1
+    keys_for_sym1 = lib_tool.find_keys_for_id(KeyType.VERSION, sym1)
+    keys_for_sym2 = lib_tool.find_keys_for_id(KeyType.VERSION, sym2)
+
+    assert len(keys_for_sym1) == 3
+    assert len(keys_for_sym2) == 3
+    # Then - we got 4 symbol keys: 2 for each of the writes
+    assert len(lib_tool.find_keys(KeyType.SYMBOL_LIST)) == 4
+
+
 def test_deleting_unknown_symbol(lmdb_version_store, symbol):
     df = sample_dataframe()