consistent add_mapping

* rename lookup_or_add_mapping to just add_mapping
* change the other add_mapping to have same semantics
* in FastValueIndex, check the result from add_mapping
  each place
* use __builtin_expect to tell both the reader and the
  compiler what the expected path is

2 files changed, 34 insertions, 30 deletions

diff --git a/eval/src/vespa/eval/eval/fast_sparse_map.h b/eval/src/vespa/eval/eval/fast_sparse_map.h
index 904226c3019..19c171cfed8 100644
--- a/eval/src/vespa/eval/eval/fast_sparse_map.h
+++ b/eval/src/vespa/eval/eval/fast_sparse_map.h
@@ -126,19 +126,25 @@ public:
         return h;
     }
 
+    // used to add a mapping, but in the unlikely case
+    // of hash collision it works like a lookup instead.
     template <typename T>
-    void add_mapping(ConstArrayRef<T> addr, uint64_t hash) {
+    uint32_t add_mapping(ConstArrayRef<T> addr, uint64_t hash) {
         uint32_t value = _map.size();
-        for (const auto &label: addr) {
-            _labels.emplace_back(label);
+        auto [iter, did_add] = _map.insert(std::make_pair(Key(hash), value));
+        if (__builtin_expect(did_add, true)) {
+            for (const auto &label: addr) {
+                _labels.emplace_back(label);
+            }
+            return value;
         }
-        _map.insert(std::make_pair(Key(hash), value));
+        return iter->second;
     }
 
     // used to add a mapping, but in the unlikely case
-    // of hash collision it works like lookup instead.
+    // of hash collision it works like a lookup instead.
     template <typename T>
-    size_t lookup_or_add_mapping(ConstArrayRef<T> addr) {
+    uint32_t add_mapping(ConstArrayRef<T> addr) {
         uint64_t hash = 0;
         size_t old_labels_size = _labels.size();
         for (const auto &label: addr) {
@@ -147,7 +153,7 @@ public:
         }
         uint32_t value = _map.size();
         auto [iter, did_add] = _map.insert(std::make_pair(Key(hash), value));
-        if (did_add) {
+        if (__builtin_expect(did_add, true)) {
             return value;
         }
         // undo adding to _labels
@@ -155,17 +161,6 @@ public:
         return iter->second;
     }
 
-    template <typename T>
-    void add_mapping(ConstArrayRef<T> addr) {
-        uint64_t h = 0;
-        uint32_t value = _map.size();
-        for (const auto &label: addr) {
-            _labels.emplace_back(label);
-            h = 31 * h + hash_label(_labels.back());
-        }
-        _map.insert(std::make_pair(Key(h), value));
-    }
-
     size_t lookup(uint64_t hash) const {
         auto pos = _map.find(hash);
         return (pos == _map.end()) ? npos() : pos->second;
diff --git a/eval/src/vespa/eval/eval/fast_value.hpp b/eval/src/vespa/eval/eval/fast_value.hpp
index cad60c833e2..f02d02489b8 100644
--- a/eval/src/vespa/eval/eval/fast_value.hpp
+++ b/eval/src/vespa/eval/eval/fast_value.hpp
@@ -230,8 +230,8 @@ struct FastValue final : Value, ValueBuilder<T> {
     const Value::Index &index() const override { return my_index; }
     TypedCells cells() const override { return TypedCells(my_cells.memory, get_cell_type<T>(), my_cells.size); }
     ArrayRef<T> add_subspace(ConstArrayRef<vespalib::stringref> addr) override {
-        size_t idx = my_index.map.lookup_or_add_mapping(addr) * my_subspace_size;
-        if (idx == my_cells.size) {
+        size_t idx = my_index.map.add_mapping(addr) * my_subspace_size;
+        if (__builtin_expect((idx == my_cells.size), true)) {
             return my_cells.add_cells(my_subspace_size);
         } 
         return ArrayRef<T>(my_cells.get(idx), my_subspace_size);
@@ -307,12 +307,16 @@ FastValueIndex::sparse_full_overlap_join(const ValueType &res_type, const Fun &f
                                          ConstArrayRef<LCT> lhs_cells, ConstArrayRef<RCT> rhs_cells, Stash &stash)
 {
     auto &result = stash.create<FastValue<OCT>>(res_type, lhs.map.num_dims(), 1, lhs.map.size());
+    auto &result_map = result.my_index.map;
     lhs.map.each_map_entry([&](auto lhs_subspace, auto hash)
                            {
                                auto rhs_subspace = rhs.map.lookup(hash);
                                if (rhs_subspace != FastSparseMap::npos()) {
-                                   result.my_index.map.add_mapping(lhs.map.make_addr(lhs_subspace), hash);
-                                   result.my_cells.push_back_fast(fun(lhs_cells[lhs_subspace], rhs_cells[rhs_subspace]));
+                                   auto idx = result_map.add_mapping(lhs.map.make_addr(lhs_subspace), hash);
+                                   if (__builtin_expect((idx == result.my_cells.size), true)) {
+                                       auto cell_value = fun(lhs_cells[lhs_subspace], rhs_cells[rhs_subspace]);
+                                       result.my_cells.push_back_fast(cell_value);
+                                   }
                                }
                            });
     return result;
@@ -329,20 +333,25 @@ FastValueIndex::sparse_only_merge(const ValueType &res_type, const Fun &fun,
     auto &result = stash.create<FastValue<OCT>>(res_type, lhs.map.num_dims(), 1, lhs.map.size()+rhs.map.size());
     lhs.map.each_map_entry([&](auto lhs_subspace, auto hash)
                            {
-                               auto rhs_subspace = rhs.map.lookup(hash);
-                               result.my_index.map.add_mapping(lhs.map.make_addr(lhs_subspace), hash);
-                               if (rhs_subspace != FastSparseMap::npos()) {
-                                   result.my_cells.push_back_fast(fun(lhs_cells[lhs_subspace], rhs_cells[rhs_subspace]));
-                               } else {
-                                   result.my_cells.push_back_fast(lhs_cells[lhs_subspace]);
+                               auto idx = result.my_index.map.add_mapping(lhs.map.make_addr(lhs_subspace), hash);
+                               if (__builtin_expect((idx == result.my_cells.size), true)) {
+                                   auto rhs_subspace = rhs.map.lookup(hash);
+                                   if (rhs_subspace != FastSparseMap::npos()) {
+                                       auto cell_value = fun(lhs_cells[lhs_subspace], rhs_cells[rhs_subspace]);
+                                       result.my_cells.push_back_fast(cell_value);
+                                   } else {
+                                       result.my_cells.push_back_fast(lhs_cells[lhs_subspace]);
+                                   }
                                }
                            });
     rhs.map.each_map_entry([&](auto rhs_subspace, auto hash)
                            {
                                auto lhs_subspace = lhs.map.lookup(hash);
                                if (lhs_subspace == FastSparseMap::npos()) {
-                                   result.my_index.map.add_mapping(rhs.map.make_addr(rhs_subspace), hash);
-                                   result.my_cells.push_back_fast(rhs_cells[rhs_subspace]);
+                                   auto idx = result.my_index.map.add_mapping(rhs.map.make_addr(rhs_subspace), hash);
+                                   if (__builtin_expect((idx == result.my_cells.size), true)) {
+                                       result.my_cells.push_back_fast(rhs_cells[rhs_subspace]);
+                                   }
                                }
                            });