Address Ivo's comments

Author: alexowens90

cpp/arcticdb/entity/atom_key.hpp

@@ -89,8 +89,12 @@ class AtomKeyImpl {
     friend bool operator!=(const AtomKeyImpl& l, const AtomKeyImpl& r) { return !(l == r); }
 
     friend bool operator<(const AtomKeyImpl& l, const AtomKeyImpl& r) {
-        const auto lt = std::tie(l.id_, l.version_id_, l.index_start_, l.index_end_, l.creation_ts_);
-        const auto rt = std::tie(r.id_, r.version_id_, r.index_start_, r.index_end_, r.creation_ts_);
+        const auto lt = std::tie(
+                l.id_, l.version_id_, l.index_start_, l.index_end_, l.creation_ts_, l.key_type_, l.content_hash_
+        );
+        const auto rt = std::tie(
+                r.id_, r.version_id_, r.index_start_, r.index_end_, r.creation_ts_, r.key_type_, r.content_hash_
+        );
         return lt < rt;
     }
 
@@ -268,8 +272,10 @@ struct AtomKeyPacked {
     friend bool operator!=(const AtomKeyPacked& l, const AtomKeyPacked& r) { return !(l == r); }
 
     friend bool operator<(const AtomKeyPacked& l, const AtomKeyPacked& r) {
-        const auto lt = std::tie(l.version_id_, l.index_start_, l.index_end_, l.creation_ts_);
-        const auto rt = std::tie(r.version_id_, r.index_start_, r.index_end_, r.creation_ts_);
+        const auto lt =
+                std::tie(l.version_id_, l.index_start_, l.index_end_, l.creation_ts_, l.key_type_, l.content_hash_);
+        const auto rt =
+                std::tie(r.version_id_, r.index_start_, r.index_end_, r.creation_ts_, r.key_type_, r.content_hash_);
         return lt < rt;
     }
 

cpp/arcticdb/version/version_map.hpp

@@ -30,6 +30,7 @@
 #include <arcticdb/util/configs_map.hpp>
 #include <arcticdb/storage/store.hpp>
 #include <arcticdb/storage/storage.hpp>
+#include <arcticdb/storage/storage_utils.hpp>
 #include <arcticdb/util/constants.hpp>
 #include <arcticdb/util/key_utils.hpp>
 #include <arcticdb/version/version_map_entry.hpp>
@@ -38,8 +39,6 @@
 #include <arcticdb/version/version_utils.hpp>
 #include <arcticdb/util/lock_table.hpp>
 
-#include "storage/storage_utils.hpp"
-
 namespace arcticdb {
 
 template<class Clock = util::SysClock>
@@ -137,6 +136,12 @@ class VersionMapImpl {
             const std::shared_ptr<Store>& store, const VersionMapEntry& ref_entry,
             const std::shared_ptr<VersionMapEntry>& entry, const LoadStrategy& load_strategy
     ) const {
+        util::check(
+                ref_entry.stream_id_ == entry->stream_id_,
+                "follow_version_chain called with mismatching stream ids {} != {}",
+                ref_entry.stream_id_,
+                entry->stream_id_
+        );
         auto next_key = ref_entry.head_;
         entry->head_ = ref_entry.head_;
 
@@ -380,8 +385,15 @@ class VersionMapImpl {
                     else {
                         if (tombstone->type() == KeyType::TOMBSTONE_ALL)
                             new_entry->try_set_tombstone_all(*tombstone);
-                        else
-                            new_entry->tombstones_.insert(std::make_pair(key.version_id(), *tombstone));
+                        else {
+                            util::check(
+                                    key.id() == new_entry->stream_id_,
+                                    "compact_and_remove_deleted_indexes found mismatching stream ids {} != {}",
+                                    key.id(),
+                                    new_entry->stream_id_
+                            );
+                            new_entry->try_set_tombstone(*tombstone);
+                        }
 
                         new_entry->keys_.push_back(key);
                     }
@@ -671,7 +683,7 @@ class VersionMapImpl {
         auto tombstone_version_id = tombstones.front().version_id();
         do_write(store, tombstone_version_id, stream_id, std::span{tombstones}, entry);
         for (const auto& key : tombstones) {
-            entry->tombstones_.try_emplace(key.version_id(), key);
+            entry->try_set_tombstone(key);
         }
         maybe_invalidate_cached_undeleted(*entry);
 

cpp/arcticdb/version/version_map_entry.hpp

@@ -314,7 +314,15 @@ struct VersionMapEntry {
         return strm.str();
     }
 
-    void unshift_key(const AtomKey& key) { keys_.push_front(key); }
+    void unshift_key(const AtomKey& key) {
+        util::check(
+                key.id() == stream_id_,
+                "VersionMapEntry::unshift_key called with key with stream id {} != {}",
+                key.id(),
+                stream_id_
+        );
+        keys_.push_front(key);
+    }
 
     std::vector<AtomKeyPacked> get_indexes(bool include_deleted) const {
         std::vector<AtomKeyPacked> output;
@@ -418,7 +426,23 @@ struct VersionMapEntry {
         util::check_rte(head_->id() == stream_id_, "Multiple symbols in keys: {} != {}", head_->id(), stream_id_);
     }
 
+    void try_set_tombstone(const AtomKey& key) {
+        util::check(
+                key.id() == stream_id_,
+                "VersionMapEntry::try_set_tombstone called with key with stream id {} != {}",
+                key.id(),
+                stream_id_
+        );
+        tombstones_.try_emplace(key.version_id(), key);
+    }
+
     void try_set_tombstone_all(const AtomKey& key) {
+        util::check(
+                key.id() == stream_id_,
+                "VersionMapEntry::try_set_tombstone_all called with key with stream id {} != {}",
+                key.id(),
+                stream_id_
+        );
         util::check(key.type() == KeyType::TOMBSTONE_ALL, "Can't set tombstone all key with key {}", key);
         if (!tombstone_all_ || tombstone_all_->version_id() < key.version_id())
             tombstone_all_ = key;
@@ -453,8 +477,10 @@ struct VersionMapEntry {
     timestamp last_reload_time_ = 0;
     LoadProgress load_progress_;
     std::deque<AtomKeyPacked> keys_;
-    ankerl::unordered_dense::map<VersionId, AtomKeyPacked> tombstones_;
     std::optional<AtomKey> tombstone_all_;
+
+  private:
+    ankerl::unordered_dense::map<VersionId, AtomKeyPacked> tombstones_;
 };
 
 inline bool is_live_index_type_key(const AtomKeyPacked& key, const std::shared_ptr<VersionMapEntry>& entry) {

cpp/arcticdb/version/version_utils.hpp

@@ -39,7 +39,12 @@ inline std::optional<AtomKey> read_segment_with_keys(
     for (; row < ssize_t(seg.row_count()); ++row) {
         auto key = read_key_row(seg, row);
         ARCTICDB_TRACE(log::version(), "Reading key {}", key);
-
+        util::check(
+                key.id() == entry.stream_id_,
+                "read_segment_with_keys found mismatching stream ids {} != {}",
+                key.id(),
+                entry.stream_id_
+        );
         if (is_index_key_type(key.type())) {
             entry.keys_.push_back(key);
             oldest_loaded_index = std::min(oldest_loaded_index, key.version_id());
@@ -51,7 +56,7 @@ inline std::optional<AtomKey> read_segment_with_keys(
             }
 
         } else if (key.type() == KeyType::TOMBSTONE) {
-            entry.tombstones_.try_emplace(key.version_id(), key);
+            entry.try_set_tombstone(key);
             entry.keys_.push_back(key);
         } else if (key.type() == KeyType::TOMBSTONE_ALL) {
             entry.try_set_tombstone_all(key);
@@ -96,11 +101,17 @@ std::shared_ptr<VersionMapEntry> build_version_map_entry_with_predicate_iteratio
     for (auto key_type : key_types) {
         store->iterate_type(
                 key_type,
-                [&predicate, &read_keys, &store, &output, &perform_read_segment_with_keys](VariantKey&& vk) {
+                [&stream_id, &predicate, &read_keys, &store, &output, &perform_read_segment_with_keys](VariantKey&& vk
+                ) {
                     const auto& key = to_atom(std::move(vk));
                     if (!predicate(key))
                         return;
-
+                    util::check(
+                            key.id() == stream_id,
+                            "build_version_map_entry_with_predicate_iteration found mismatching stream ids {} != {}",
+                            key.id(),
+                            stream_id
+                    );
                     read_keys.push_back(key);
                     ARCTICDB_DEBUG(log::storage(), "Version map iterating key {}", key);
                     if (perform_read_segment_with_keys) {