Merge pull request #80 from hendrik-cliqz/dictionary-merger

Dictionary merger optimizations

Author: hendrikmuhs

keyvi/src/cpp/dictionary/dictionary_compiler.h

@@ -121,9 +121,8 @@ class DictionaryCompiler
     void Add(const std::string& input_key, typename ValueStoreT::value_t value =
                  ValueStoreT::no_value) {
 
-      sorter_.push(key_value_t(input_key, RegisterValue(value)));
-
       size_of_keys_ += input_key.size();
+      sorter_.push(key_value_t(std::move(input_key), RegisterValue(value)));
     }
 
 #ifdef Py_PYTHON_H
@@ -157,7 +156,7 @@ class DictionaryCompiler
 
           TRACE("adding to generator: %s", key_value.key.c_str());
 
-          generator_->Add(key_value.key, key_value.value);
+          generator_->Add(std::move(key_value.key), key_value.value);
           ++added_key_values_;
           if (progress_callback && (added_key_values_ % callback_trigger_ == 0)){
             progress_callback(added_key_values_, number_of_items_, user_data);

keyvi/src/cpp/dictionary/dictionary_merger.h

@@ -50,11 +50,14 @@ final {
     int priority;
     bool operator<(const SegmentEntryForMerge& rhs) const
     {
-      if (entry_iterator.GetKey() == rhs.entry_iterator.GetKey()) {
-        return priority < rhs.priority;
-      }
+      // very important difference in semantics: we have to ensure that in case of equal key,
+      // the iterator with the higher priority is taken
+
+      if (priority < rhs.priority) {
+        return entry_iterator > rhs.entry_iterator;
+       }
 
-      return entry_iterator > rhs.entry_iterator;
+      return rhs.entry_iterator < entry_iterator;
     }
   };
 
@@ -86,20 +89,23 @@ final {
 
     fsa::Generator<PersistenceT, ValueStoreT> generator(1073741824, fsa::generator_param_t(), value_store);
 
+    std::string top_key;
+
     while(!pqueue.empty()){
       auto entry_it = pqueue.top();
       pqueue.pop();
 
-      auto key = entry_it.entry_iterator.GetKey();
+      top_key = entry_it.entry_iterator.GetKey();
 
       // check for same keys and merge only the most recent one
-      while (!pqueue.empty() and pqueue.top().entry_iterator.GetKey() == key) {
+      while (!pqueue.empty() and pqueue.top().entry_iterator.operator==(top_key)) {
 
         auto to_inc = pqueue.top();
         TRACE("removing element with prio %d (in favor of %d)", to_inc.priority, entry_it.priority);
 
         pqueue.pop();
         if (++to_inc.entry_iterator != end_it) {
+          TRACE("push iterator");
           pqueue.push(to_inc);
         }
       }
@@ -109,18 +115,21 @@ final {
 
       // Todo: if inner weights are used update them
       //handle.weight = value_store_->GetWeightValue(value);
+      handle.weight = 0;
 
       handle.value_idx = value_store->GetValue(entry_it.entry_iterator.GetFsa()->GetValueStore()->GetValueStorePayload(),
                                                entry_it.entry_iterator.GetValueId(),
                                                handle.no_minimization);
 
-      TRACE("Add key: %s", key.c_str());
-      generator.Add(key, handle);
+      TRACE("Add key: %s", top_key.c_str());
+      generator.Add(std::move(top_key), handle);
 
       if (++entry_it.entry_iterator != end_it) {
         pqueue.push(entry_it);
       }
     }
+    TRACE("finished iterating, do final compile.");
+
     generator.CloseFeeding();
 
     generator.WriteToFile(filename);

keyvi/src/cpp/dictionary/fsa/entry_iterator.h

@@ -40,9 +40,9 @@ class EntryIterator final{
 
  public:
   EntryIterator()
-      : current_state_(0),
-        current_value_(0),
-        fsa_(nullptr) {
+      : fsa_(nullptr),
+        current_state_(0),
+        current_value_(0) {
   }
 
   EntryIterator(automata_t f) : EntryIterator(f, f->GetStartState()) {}
@@ -107,6 +107,14 @@ class EntryIterator final{
     return !(operator==(other));
   }
 
+  bool operator==(const std::string& other_key) const {
+    if (GetDepth() != other_key.size()) {
+      return false;
+    }
+
+    return std::memcmp(other_key.c_str(), (const char*) traversal_stack_.data(), other_key.size() == 0);
+  }
+
   bool operator<(const EntryIterator& other) const {
 
     int compare_result = memcmp((const char*) other.traversal_stack_.data(),

keyvi/src/cpp/dictionary/fsa/generator.h

@@ -236,7 +236,7 @@ final {
       size_t commonPrefixLength = get_common_prefix_length(last_key_.c_str(), key);
 
       if (commonPrefixLength == input_key.size() && last_key_.size() == input_key.size()) {
-        last_key_ = key;
+        last_key_ = std::move(input_key);
         return;
       }
 
@@ -256,7 +256,7 @@ final {
         stack_->UpdateWeights(0, input_key.size() + 1, handle.weight);
       }
 
-      last_key_ = key;
+      last_key_ = std::move(input_key);
       state_ = generator_state::FEEDING;
     }
 

keyvi/src/cpp/dictionary/fsa/generator_adapter.h

@@ -61,11 +61,11 @@ class GeneratorAdapter final: public GeneratorAdapterInterface<PersistenceT, Val
 
   void Add(const std::string& input_key, typename ValueStoreT::value_t value =
                  ValueStoreT::no_value) {
-    generator_.Add(input_key, value);
+    generator_.Add(std::move(input_key), value);
   }
 
   void Add(const std::string& input_key, const fsa::ValueHandle& value) {
-    generator_.Add(input_key, value);
+    generator_.Add(std::move(input_key), value);
   }
 
   size_t GetFsaSize() const {

keyvi/src/cpp/dictionary/fsa/internal/json_value_store.h

@@ -219,22 +219,47 @@ class JsonValueStore final : public IValueStoreWriter {
    friend class ::keyvi::dictionary::DictionaryMerger;
 
    uint64_t GetValue(const char* payload, uint64_t fsa_value, bool& no_minimization){
-     std::string packed_string = util::decodeVarintString(payload + fsa_value);
+     size_t buffer_size;
 
-     return GetValue(util::DecodeJsonValue(packed_string), no_minimization);
+     const char* full_buf = payload + fsa_value;
+     const char* buf_ptr = util::decodeVarintString(full_buf, &buffer_size);
+
+     const RawPointerForCompare<MemoryMapManager> stp(buf_ptr, buffer_size,
+                                                          values_extern_);
+     const RawPointer p = hash_.Get(stp);
+
+     if (!p.IsEmpty()) {
+       // found the same value again, minimize
+       TRACE("Minimized value");
+       return p.GetOffset();
+     } // else persist string value
+
+     no_minimization = true;
+     TRACE("New unique value");
+     ++number_of_unique_values_;
+
+     uint64_t pt = static_cast<uint64_t>(values_buffer_size_);
+     size_t full_buf_size = (buf_ptr-full_buf) + buffer_size;
+
+     values_extern_->Append((void*)full_buf,
+                                full_buf_size);
+     values_buffer_size_ += full_buf_size;
+
+     hash_.Add(RawPointer(pt, stp.GetHashcode(), buffer_size));
+
+     return pt;
    }
 
   uint64_t AddValue(){
     uint64_t pt = static_cast<uint64_t>(values_buffer_size_);
+    size_t length;
 
-    dictionary::util::encodeVarint(string_buffer_.size(), *values_extern_);
-    values_buffer_size_ += util::getVarintLength(string_buffer_.size());
-
+    dictionary::util::encodeVarint(string_buffer_.size(), *values_extern_, &length);
+    values_buffer_size_ += length;
     values_extern_->Append((void*)string_buffer_.data(),
                            string_buffer_.size());
     values_buffer_size_ += string_buffer_.size();
 
-    TRACE("add value to hash at %d, length %d", pt, string_buffer_.size());
     return pt;
   }
 

keyvi/src/cpp/dictionary/fsa/internal/sparse_array_builder.h

@@ -362,7 +362,7 @@ class SparseArrayBuilder<SparseArrayPersistence<uint16_t>, OffsetTypeT, HashCode
 
     // else overflow encoding
     uint16_t vshort_pointer[8];
-    uint8_t vshort_size = 0;
+    size_t vshort_size = 0;
 
     util::encodeVarshort(transitionPointer_high, vshort_pointer, &vshort_size);
 
@@ -441,7 +441,7 @@ class SparseArrayBuilder<SparseArrayPersistence<uint16_t>, OffsetTypeT, HashCode
                                   FINAL_OFFSET_CODE, 0);
 
     uint16_t vshort_pointer[8];
-    uint8_t vshort_size = 0;
+    size_t vshort_size = 0;
 
     util::encodeVarshort(value, vshort_pointer, &vshort_size);
 

keyvi/src/cpp/dictionary/util/vint.h

@@ -40,8 +40,8 @@ namespace util {
  * @parma outputSizePtr A pointer to a single integer that is set to the number of bytes used in the output memory.
  */
 template<typename int_t = uint64_t>
-void encodeVarint(int_t value, uint8_t* output, uint8_t* outputSizePtr) {
-  uint8_t outputSize = 0;
+void encodeVarint(int_t value, uint8_t* output, size_t* outputSizePtr) {
+  size_t outputSize = 0;
   //While more than 7 bits of data are left, occupy the last output byte
   // and set the next byte flag
   while (value > 127) {
@@ -63,8 +63,8 @@ void encodeVarint(int_t value, uint8_t* output, uint8_t* outputSizePtr) {
  * @parma outputSizePtr A pointer to a single integer that is set to the number of bytes used in the output memory.
  */
 template<typename int_t = uint64_t>
-void encodeVarshort(int_t value, uint16_t* output, uint8_t* outputSizePtr) {
-  uint8_t outputSize = 0;
+void encodeVarshort(int_t value, uint16_t* output, size_t* outputSizePtr) {
+  size_t outputSize = 0;
   //While more than 15 bits of data are left, occupy the last output byte
   // and set the next byte flag
   while (value > 32767) {
@@ -115,16 +115,19 @@ size_t getVarshortLength(int_t value){
  * @param output the buffer to write to
  */
 template<typename int_t = uint64_t, typename buffer_t>
-void encodeVarint(int_t value, buffer_t& output) {
+void encodeVarint(int_t value, buffer_t& output, size_t* written_bytes) {
   //While more than 7 bits of data are left, occupy the last output byte
   // and set the next byte flag
+  size_t length = 0;
   while (value > 127) {
     //|128: Set the next byte flag
     output.push_back(((uint8_t) (value & 127)) | 128);
     //Remove the seven bits we just wrote
     value >>= 7;
+    ++length;
   }
   output.push_back(((uint8_t) value) & 127);
+  *written_bytes = ++length;
 }
 
 /**
@@ -178,7 +181,7 @@ inline size_t skipVarint(const char* input) {
 
 inline std::string decodeVarintString(const char* input){
   uint64_t length = 0;
-  uint8_t i = 0;
+  size_t i = 0;
 
   for (;; i++) {
     length |= (input[i] & 127) << (7 * i);
@@ -192,6 +195,24 @@ inline std::string decodeVarintString(const char* input){
   return std::string(input + i + 1, length);
 }
 
+inline const char* decodeVarintString(const char* input, size_t* length_ptr) {
+  size_t length = 0;
+  size_t i = 0;
+
+    for (;; i++) {
+      length |= (input[i] & 127) << (7 * i);
+
+      //If the next-byte flag is set
+      if (!(input[i] & 128)) {
+        break;
+      }
+    }
+
+    *length_ptr = length;
+    return input + i + 1;
+
+}
+
 
 } /* namespace util */
 } /* namespace dictionary */

keyvi/tests/cpp/dictionary/dictionary_merger_test.cpp

@@ -136,6 +136,7 @@ BOOST_AUTO_TEST_CASE ( MergeIntegerDictsValueMerge) {
 
   std::vector<std::pair<std::string, uint32_t>> test_data2 = {
              { "abc", 25 },
+             { "abbc", 42 },
              { "abcd", 21 },
              { "abbc", 30 },
          };
@@ -178,6 +179,8 @@ BOOST_AUTO_TEST_CASE ( MergeIntegerDictsValueMerge) {
 
   BOOST_CHECK_EQUAL("22", d2->operator[]("abc").GetValueAsString());
   BOOST_CHECK_EQUAL("21", d2->operator[]("abcd").GetValueAsString());
+
+  // overwritten by 2nd
   BOOST_CHECK_EQUAL("24", d2->operator[]("abbc").GetValueAsString());
 
   std::remove(filename.c_str());
@@ -218,6 +221,7 @@ BOOST_AUTO_TEST_CASE ( MergeStringDicts) {
 
   std::vector<std::pair<std::string, std::string>> test_data2 = {
              { "abbe", "d" },
+             { "abbc", "z" },
              { "abcd", "a" },
              { "bbacd", "f" },
          };
@@ -241,7 +245,9 @@ BOOST_AUTO_TEST_CASE ( MergeStringDicts) {
   BOOST_CHECK(d->Contains("bba"));
 
   BOOST_CHECK_EQUAL("a", d->operator[]("abc").GetValueAsString());
-  BOOST_CHECK_EQUAL("b", d->operator[]("abbc").GetValueAsString());
+
+  // overwritten by 2nd
+  BOOST_CHECK_EQUAL("z", d->operator[]("abbc").GetValueAsString());
   BOOST_CHECK_EQUAL("c", d->operator[]("abbcd").GetValueAsString());
   BOOST_CHECK_EQUAL("a", d->operator[]("abcde").GetValueAsString());
   BOOST_CHECK_EQUAL("b", d->operator[]("abdd").GetValueAsString());
@@ -271,6 +277,7 @@ BOOST_AUTO_TEST_CASE ( MergeJsonDicts) {
 
   std::vector<std::pair<std::string, std::string>> test_data2 = {
              { "abbe", "{d:4}" },
+             { "abbc", "{b:3}" },
              { "abcd", "{a:1}" },
              { "bbacd", "{f:5}" },
          };
@@ -294,7 +301,9 @@ BOOST_AUTO_TEST_CASE ( MergeJsonDicts) {
   BOOST_CHECK(d->Contains("bba"));
 
   BOOST_CHECK_EQUAL("\"{a:1}\"", d->operator[]("abc").GetValueAsString());
-  BOOST_CHECK_EQUAL("\"{b:2}\"", d->operator[]("abbc").GetValueAsString());
+
+  // overwritten by 2nd
+  BOOST_CHECK_EQUAL("\"{b:3}\"", d->operator[]("abbc").GetValueAsString());
   BOOST_CHECK_EQUAL("\"{c:3}\"", d->operator[]("abbcd").GetValueAsString());
   BOOST_CHECK_EQUAL("\"{a:1}\"", d->operator[]("abcde").GetValueAsString());
   BOOST_CHECK_EQUAL("\"{b:2}\"", d->operator[]("abdd").GetValueAsString());