Reduce symbol list peak heap usage during load and compaction

cpp/arcticdb/CMakeLists.txt

@@ -1234,7 +1234,8 @@ if(${TEST})
             processing/test/benchmark_common.cpp
             processing/test/benchmark_ternary.cpp
             util/test/benchmark_bitset.cpp
-            version/test/benchmark_write.cpp)
+            version/test/benchmark_write.cpp
+            version/test/benchmark_symbol_list.cpp)
 
     add_executable(benchmarks ${benchmark_srcs})
 

cpp/arcticdb/version/symbol_list.hpp

@@ -21,8 +21,6 @@ struct SymbolListEntry;
 struct SymbolEntryData;
 
 using MapType = std::unordered_map<StreamId, std::vector<SymbolEntryData>>;
-using Compaction = std::vector<AtomKey>::const_iterator;
-using MaybeCompaction = std::optional<Compaction>;
 using CollectionType = std::vector<SymbolListEntry>;
 
 enum class WillAttemptCompaction : uint8_t {
@@ -32,12 +30,10 @@ enum class WillAttemptCompaction : uint8_t {
 };
 
 struct LoadResult {
-    std::vector<AtomKey> symbol_list_keys_;
-    MaybeCompaction maybe_previous_compaction;
+    std::optional<AtomKey> compaction_key_;
     CollectionType symbols_;
-    timestamp timestamp_ = 0L;
-
-    std::vector<AtomKey>&& detach_symbol_list_keys() { return std::move(symbol_list_keys_); }
+    size_t total_key_count_ = 0;
+    std::vector<VariantKey> symbol_list_keys_;
 };
 
 struct SymbolListData {
@@ -154,12 +150,20 @@ class SymbolList {
                 return WillAttemptCompaction::NO_INSUFFICIENT_PERMISSIONS;
             return WillAttemptCompaction::YES;
         }();
+
         LoadResult load_result = ExponentialBackoff<StorageException>(100, 2000).go(
                 [this, &version_map, &store, will_attempt_compaction]() {
                     return attempt_load(version_map, store, data_, will_attempt_compaction);
                 }
         );
 
+        // Build output before compaction — compact_internal frees symbols_ after writing them to storage
+        R output;
+        for (const auto& entry : load_result.symbols_) {
+            if (entry.action_ == ActionType::ADD)
+                output.insert(entry.stream_id_);
+        }
+
         if (will_attempt_compaction == WillAttemptCompaction::YES && needs_compaction(load_result)) {
             ARCTICDB_RUNTIME_DEBUG(log::symbol(), "Compaction necessary. Obtaining lock...");
             try {
@@ -180,12 +184,6 @@ class SymbolList {
             }
         }
 
-        R output;
-        for (const auto& entry : load_result.symbols_) {
-            if (entry.action_ == ActionType::ADD)
-                output.insert(entry.stream_id_);
-        }
-
         return output;
     }
 
@@ -215,9 +213,7 @@ class SymbolList {
     [[nodiscard]] bool needs_compaction(const LoadResult& load_result) const;
 };
 
-std::vector<Store::RemoveKeyResultType> delete_keys(
-        const std::shared_ptr<Store>& store, std::vector<AtomKey>&& remove, const AtomKey& exclude
-);
+void delete_keys(const std::shared_ptr<Store>& store, std::vector<AtomKey>&& remove, const AtomKey& exclude);
 
 struct WriteSymbolTask : async::BaseTask {
     const std::shared_ptr<Store> store_;