Fix symbol list bug on compaction (#1324)

- Only adds a symbol list entry on append if there is no previous
  version.
- Fixes `finalize_staged_data` to write to the symbol list.
- Adds a python test for the new behavior.

Author: IvoDD

cpp/arcticdb/version/local_versioned_engine.cpp

@@ -993,19 +993,28 @@ void LocalVersionedEngine::write_parallel_frame(
     write_parallel(store_, stream_id, frame, validate_index);
 }
 
+void LocalVersionedEngine::add_to_symbol_list_on_compaction(
+        const StreamId& stream_id,
+        const CompactIncompleteOptions& options,
+        const UpdateInfo& update_info) {
+    if(cfg_.symbol_list()) {
+        if (!options.append_ || !update_info.previous_index_key_.has_value()) {
+            symbol_list().add_symbol(store_, stream_id, update_info.next_version_id_);
+        }
+    }
+}
+
 VersionedItem LocalVersionedEngine::compact_incomplete_dynamic(
     const StreamId& stream_id,
     const std::optional<arcticdb::proto::descriptors::UserDefinedMetadata>& user_meta,
     const CompactIncompleteOptions& options) {
     log::version().debug("Compacting incomplete symbol {}", stream_id);
 
     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, options, get_write_options());
+    auto versioned_item = compact_incomplete_impl(store_, stream_id, user_meta, update_info, options, get_write_options());
 
     write_version_and_prune_previous(options.prune_previous_versions_, versioned_item.key_, update_info.previous_index_key_);
-
-    if(cfg_.symbol_list())
-        symbol_list().add_symbol(store_, stream_id, update_info.next_version_id_);
+    add_to_symbol_list_on_compaction(stream_id, options, update_info);
 
     return versioned_item;
 }
@@ -1654,6 +1663,7 @@ VersionedItem LocalVersionedEngine::sort_merge_internal(
     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, options);
     write_version_and_prune_previous(options.prune_previous_versions_, versioned_item.key_, update_info.previous_index_key_);
+    add_to_symbol_list_on_compaction(stream_id, options, update_info);
     return versioned_item;
 }
 

cpp/arcticdb/version/local_versioned_engine.hpp

@@ -441,6 +441,7 @@ class LocalVersionedEngine : public VersionedEngine {
 
 private:
     void initialize(const std::shared_ptr<storage::Library>& library);
+    void add_to_symbol_list_on_compaction(const StreamId& stream_id, const CompactIncompleteOptions& options, const UpdateInfo& update_info);
 
     std::shared_ptr<Store> store_;
     arcticdb::proto::storage::VersionStoreConfig cfg_;

python/arcticdb/version_store/library.py

@@ -1109,7 +1109,7 @@ def finalize_staged_data(
         mode: Optional[StagedDataFinalizeMethod] = StagedDataFinalizeMethod.WRITE,
         prune_previous_versions: bool = False,
         metadata: Any = None,
-        validate_index=True,
+        validate_index = True,
     ) -> VersionedItem:
         """
         Finalizes staged data, making it available for reads.
@@ -1157,7 +1157,7 @@ def sort_and_finalize_staged_data(
         symbol: str,
         mode: Optional[StagedDataFinalizeMethod] = StagedDataFinalizeMethod.WRITE,
         prune_previous_versions: bool = False,
-        metadata: Any = None
+        metadata: Any = None,
     ) -> VersionedItem:
         """
         sort_merge will sort and finalize staged data. This differs from `finalize_staged_data` in that it

python/tests/unit/arcticdb/version_store/test_sort_merge.py

@@ -2,6 +2,7 @@
 import numpy as np
 from pandas.testing import assert_frame_equal
 import pytest
+from arcticdb_ext.storage import KeyType
 from arcticdb.version_store.library import StagedDataFinalizeMethod
 from arcticdb.exceptions import UserInputException, SortingException
 
@@ -268,4 +269,37 @@ def test_append_to_missing_symbol(lmdb_library):
     df = pd.DataFrame({"col": [1]}, index=pd.DatetimeIndex([np.datetime64('2023-01-01')], dtype="datetime64[ns]"))
     lib.write("sym", df, staged=True)
     lib.sort_and_finalize_staged_data("sym", mode=StagedDataFinalizeMethod.APPEND)
-    assert_frame_equal(lib.read("sym").data, df)
+    assert_frame_equal(lib.read("sym").data, df)
+
+
+def test_update_symbol_list(lmdb_library):
+    lib = lmdb_library
+    lib_tool = lmdb_library._nvs.library_tool()
+    sym = "sym"
+    sym_2 = "sym_2"
+    df = pd.DataFrame({"col": [1]}, index=pd.DatetimeIndex([np.datetime64('2023-01-01')], dtype="datetime64[ns]"))
+
+    # We always add to the symbol list on write
+    lib.write(sym, df, staged=True)
+    lib.sort_and_finalize_staged_data(sym, mode=StagedDataFinalizeMethod.WRITE)
+    assert lib_tool.count_keys(KeyType.SYMBOL_LIST) == 1
+    assert lib.list_symbols() == [sym]
+
+    # We don't add to the symbol on append when there is an existing version
+    lib.write(sym, df, staged=True)
+    lib.sort_and_finalize_staged_data(sym, mode=StagedDataFinalizeMethod.APPEND)
+    assert lib_tool.count_keys(KeyType.SYMBOL_LIST) == 1
+    assert lib.list_symbols() == [sym]
+
+    # We always add to the symbol list on write, even when there is an existing version
+    lib.write(sym, df, staged=True)
+    lib.sort_and_finalize_staged_data(sym, mode=StagedDataFinalizeMethod.WRITE)
+    assert lib_tool.count_keys(KeyType.SYMBOL_LIST) == 2
+    assert lib.list_symbols() == [sym]
+
+    # We add to the symbol list on append when there is no previous version
+    lib.write(sym_2, df, staged=True)
+    lib.sort_and_finalize_staged_data(sym_2, mode=StagedDataFinalizeMethod.APPEND)
+    assert lib_tool.count_keys(KeyType.SYMBOL_LIST) == 3
+    assert set(lib.list_symbols()) == set([sym, sym_2])
+