Restore enum store compaction support for enum attributes.

4 files changed, 99 insertions, 41 deletions

diff --git a/vespalib/src/tests/datastore/unique_store/unique_store_test.cpp b/vespalib/src/tests/datastore/unique_store/unique_store_test.cpp
index a1a24247b3f..88a5a05738b 100644
--- a/vespalib/src/tests/datastore/unique_store/unique_store_test.cpp
+++ b/vespalib/src/tests/datastore/unique_store/unique_store_test.cpp
@@ -1,5 +1,6 @@
 // Copyright 2017 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
 #include <vespa/vespalib/datastore/unique_store.hpp>
+#include <vespa/vespalib/datastore/unique_store_remapper.h>
 #include <vespa/vespalib/datastore/unique_store_string_allocator.hpp>
 #include <vespa/vespalib/datastore/unique_store_string_comparator.h>
 #include <vespa/vespalib/gtest/gtest.h>
@@ -100,14 +101,16 @@ struct TestBase : public ::testing::Test {
         store.trimHoldLists(generation);
     }
     void compactWorst() {
-        ICompactionContext::UP ctx = store.compactWorst();
+        auto remapper = store.compact_worst(true, true);
         std::vector<EntryRef> refs;
         for (const auto &elem : refStore) {
             refs.push_back(elem.first);
         }
         refs.push_back(EntryRef());
         std::vector<EntryRef> compactedRefs = refs;
-        ctx->compact(ArrayRef<EntryRef>(compactedRefs));
+        remapper->remap(ArrayRef<EntryRef>(compactedRefs));
+        remapper->done();
+        remapper.reset();
         ASSERT_FALSE(refs.back().valid());
         refs.pop_back();
         ReferenceStore compactedRefStore;
diff --git a/vespalib/src/vespa/vespalib/datastore/unique_store.h b/vespalib/src/vespa/vespalib/datastore/unique_store.h
index 6b85e79d3eb..b4ed3f6d86b 100644
--- a/vespalib/src/vespa/vespalib/datastore/unique_store.h
+++ b/vespalib/src/vespa/vespalib/datastore/unique_store.h
@@ -7,7 +7,6 @@
 #include "datastore.h"
 #include "entry_comparator_wrapper.h"
 #include "entryref.h"
-#include "i_compaction_context.h"
 #include "i_unique_store_dictionary.h"
 #include "unique_store_add_result.h"
 #include "unique_store_allocator.h"
@@ -22,6 +21,9 @@ class UniqueStoreBuilder;
 template <typename RefT>
 class UniqueStoreEnumerator;
 
+template <typename RefT>
+class UniqueStoreRemapper;
+
 /**
  * Datastore for unique values of type EntryT that is accessed via a
  * 32-bit EntryRef.
@@ -35,6 +37,7 @@ public:
     using RefType = RefT;
     using Enumerator = UniqueStoreEnumerator<RefT>;
     using Builder = UniqueStoreBuilder<Allocator>;
+    using Remapper = UniqueStoreRemapper<RefT>;
     using EntryConstRefType = typename Allocator::EntryConstRefType;
 private:
     Allocator _allocator;
@@ -50,7 +53,7 @@ public:
     EntryRef find(EntryConstRefType value);
     EntryConstRefType get(EntryRef ref) const { return _allocator.get(ref); }
     void remove(EntryRef ref);
-    ICompactionContext::UP compactWorst();
+    std::unique_ptr<Remapper> compact_worst(bool compact_memory, bool compact_address_space);
     vespalib::MemoryUsage getMemoryUsage() const;
     vespalib::AddressSpace get_address_space_usage() const;
 
diff --git a/vespalib/src/vespa/vespalib/datastore/unique_store.hpp b/vespalib/src/vespa/vespalib/datastore/unique_store.hpp
index ebd81010612..abac2888a34 100644
--- a/vespalib/src/vespa/vespalib/datastore/unique_store.hpp
+++ b/vespalib/src/vespa/vespalib/datastore/unique_store.hpp
@@ -4,6 +4,7 @@
 
 #include "unique_store.h"
 #include "unique_store_dictionary.h"
+#include "unique_store_remapper.h"
 #include "datastore.hpp"
 #include "unique_store_allocator.hpp"
 #include "unique_store_builder.hpp"
@@ -80,28 +81,26 @@ UniqueStore<EntryT, RefT, Compare, Allocator>::remove(EntryRef ref)
 namespace uniquestore {
 
 template <typename RefT>
-class CompactionContext : public ICompactionContext, public ICompactable {
+class CompactionContext : public UniqueStoreRemapper<RefT>, public ICompactable {
 private:
     using DictionaryTraits = btree::BTreeTraits<32, 32, 7, true>;
     using Dictionary = btree::BTree<EntryRef, uint32_t,
                                     btree::NoAggregated,
                                     EntryComparatorWrapper,
                                     DictionaryTraits>;
+    using UniqueStoreRemapper<RefT>::_compacting_buffer;
+    using UniqueStoreRemapper<RefT>::_mapping;
     DataStoreBase &_dataStore;
     IUniqueStoreDictionary &_dict;
     ICompactable &_store;
     std::vector<uint32_t> _bufferIdsToCompact;
-    std::vector<std::vector<EntryRef>> _mapping;
-
-    bool compactingBuffer(uint32_t bufferId) {
-        return std::find(_bufferIdsToCompact.begin(), _bufferIdsToCompact.end(),
-                         bufferId) != _bufferIdsToCompact.end();
-    }
 
     void allocMapping() {
+        _compacting_buffer.resize(RefT::numBuffers());
         _mapping.resize(RefT::numBuffers());
         for (const auto bufferId : _bufferIdsToCompact) {
             BufferState &state = _dataStore.getBufferState(bufferId);
+            _compacting_buffer[bufferId] = true;
             _mapping[bufferId].resize(state.size());
         }
     }
@@ -109,9 +108,10 @@ private:
     EntryRef move(EntryRef oldRef) override {
         RefT iRef(oldRef);
         assert(iRef.valid());
-        if (compactingBuffer(iRef.bufferId())) {
-            assert(iRef.offset() < _mapping[iRef.bufferId()].size());
-            EntryRef &mappedRef = _mapping[iRef.bufferId()][iRef.offset()];
+        uint32_t buffer_id = iRef.bufferId();
+        if (_compacting_buffer[buffer_id]) {
+            assert(iRef.offset() < _mapping[buffer_id].size());
+            EntryRef &mappedRef = _mapping[buffer_id][iRef.offset()];
             assert(!mappedRef.valid());
             EntryRef newRef = _store.move(oldRef);
             mappedRef = newRef;
@@ -130,48 +130,40 @@ public:
                       IUniqueStoreDictionary &dict,
                       ICompactable &store,
                       std::vector<uint32_t> bufferIdsToCompact)
-        : ICompactionContext(),
+        : UniqueStoreRemapper<RefT>(),
           ICompactable(),
           _dataStore(dataStore),
           _dict(dict),
           _store(store),
-          _bufferIdsToCompact(std::move(bufferIdsToCompact)),
-          _mapping()
+          _bufferIdsToCompact(std::move(bufferIdsToCompact))
     {
+        if (!_bufferIdsToCompact.empty()) {
+            allocMapping();
+            fillMapping();
+        }
     }
-    virtual ~CompactionContext() {
+
+    void done() override {
         _dataStore.finishCompact(_bufferIdsToCompact);
+        _bufferIdsToCompact.clear();
     }
-    virtual void compact(vespalib::ArrayRef<EntryRef> refs) override {
-        if (!_bufferIdsToCompact.empty()) {
-            if (_mapping.empty()) {
-                allocMapping();
-                fillMapping();
-            }
-            for (auto &ref : refs) {
-                if (ref.valid()) {
-                    RefT internalRef(ref);
-                    if (compactingBuffer(internalRef.bufferId())) {
-                        assert(internalRef.offset() < _mapping[internalRef.bufferId()].size());
-                        EntryRef newRef = _mapping[internalRef.bufferId()][internalRef.offset()];
-                        assert(newRef.valid());
-                        ref = newRef;
-                    }
-                }
-            }
-        }
+    ~CompactionContext() override {
+        assert(_bufferIdsToCompact.empty());
     }
 };
 
 }
 
 template <typename EntryT, typename RefT, typename Compare, typename Allocator>
-ICompactionContext::UP
-UniqueStore<EntryT, RefT, Compare, Allocator>::compactWorst()
+std::unique_ptr<typename UniqueStore<EntryT, RefT, Compare, Allocator>::Remapper>
+UniqueStore<EntryT, RefT, Compare, Allocator>::compact_worst(bool compact_memory, bool compact_address_space)
 {
-    std::vector<uint32_t> bufferIdsToCompact = _store.startCompactWorstBuffers(true, true);
-    return std::make_unique<uniquestore::CompactionContext<RefT>>
-        (_store, *_dict, _allocator, std::move(bufferIdsToCompact));
+    std::vector<uint32_t> bufferIdsToCompact = _store.startCompactWorstBuffers(compact_memory, compact_address_space);
+    if (bufferIdsToCompact.empty()) {
+        return std::unique_ptr<Remapper>();
+    } else {
+        return std::make_unique<uniquestore::CompactionContext<RefT>>(_store, *_dict, _allocator, std::move(bufferIdsToCompact));
+    }
 }
 
 template <typename EntryT, typename RefT, typename Compare, typename Allocator>
diff --git a/vespalib/src/vespa/vespalib/datastore/unique_store_remapper.h b/vespalib/src/vespa/vespalib/datastore/unique_store_remapper.h
new file mode 100644
index 00000000000..13a732e6524
--- /dev/null
+++ b/vespalib/src/vespa/vespalib/datastore/unique_store_remapper.h
@@ -0,0 +1,60 @@
+// Copyright 2019 Oath Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+#pragma once
+
+#include "entryref.h"
+#include <vector>
+
+namespace search::datastore {
+
+/**
+ * Remapper for related UniqueStore class, used for adjusting
+ * references to unique store after compaction.
+ */
+template <typename RefT>
+class UniqueStoreRemapper {
+public:
+    using RefType = RefT;
+
+protected:
+    std::vector<bool> _compacting_buffer;
+    std::vector<std::vector<EntryRef>> _mapping;
+public:
+    UniqueStoreRemapper()
+        : _compacting_buffer(),
+          _mapping()
+    {
+    }
+    virtual ~UniqueStoreRemapper() = default;
+
+    EntryRef remap(EntryRef ref) const {
+        if (ref.valid()) {
+            RefType internal_ref(ref);
+            if (!_compacting_buffer[internal_ref.bufferId()]) {
+                // No remapping for references to buffers not being compacted
+                return ref;
+            } else {
+                auto &inner_mapping = _mapping[internal_ref.bufferId()];
+                assert(internal_ref.unscaled_offset() < inner_mapping.size());
+                EntryRef mapped_ref = inner_mapping[internal_ref.unscaled_offset()];
+                assert(mapped_ref.valid());
+                return mapped_ref;
+            }
+        } else {
+            return EntryRef();
+        }
+    }
+
+    void remap(vespalib::ArrayRef<EntryRef> refs) const {
+        for (auto &ref : refs) {
+            auto mapped_ref = remap(ref);
+            if (mapped_ref != ref) {
+                ref = mapped_ref;
+            }
+        }
+    }
+
+    virtual void done() = 0;
+};
+
+}