feat(flex): support LFIndexer resize (#3351)

<!--
Thanks for your contribution! please review
https://github.com/alibaba/GraphScope/blob/main/CONTRIBUTING.md before
opening an issue.
-->

support resize in LFIndexer

<!-- Please give a short brief about these changes. -->

## Related issue number

<!-- Are there any issues opened that will be resolved by merging this
change? -->

Fixes

flex/utils/id_indexer.h

@@ -197,7 +197,7 @@ void build_lf_indexer(const IdIndexer<KEY_T, INDEX_T>& input,
 template <typename INDEX_T>
 class LFIndexer {
  public:
-  LFIndexer() : num_elements_(0), hasher_(), keys_(nullptr) {}
+  LFIndexer() : num_elements_(0), hasher_(), indices_size_(0), keys_(nullptr) {}
   LFIndexer(LFIndexer&& rhs)
       : keys_(rhs.keys_),
         indices_(rhs.indices_),
@@ -233,6 +233,41 @@ class LFIndexer {
       LOG(FATAL) << "Not support type [" << type << "] as pk type ..";
     }
   }
+  void resize(size_t size) { rehash(std::max(size, num_elements_.load())); }
+
+  void rehash(size_t size) {
+    size = std::max(size, 4ul);
+    keys_->resize(size);
+    size =
+        static_cast<size_t>(std::ceil(size / id_indexer_impl::max_load_factor));
+    if (size == indices_size_) {
+      return;
+    }
+
+    auto new_prime_index = hash_policy_.next_size_over(size);
+    hash_policy_.commit(new_prime_index);
+    size_t num_elements = num_elements_.load();
+
+    indices_.resize(size);
+    indices_size_ = size;
+    for (size_t k = 0; k != size; ++k) {
+      indices_[k] = std::numeric_limits<INDEX_T>::max();
+    }
+    num_slots_minus_one_ = size - 1;
+    for (INDEX_T idx = 0; idx < num_elements; ++idx) {
+      const auto& oid = keys_->get(idx);
+      size_t index =
+          hash_policy_.index_for_hash(hasher_(oid), num_slots_minus_one_);
+      static constexpr INDEX_T sentinel = std::numeric_limits<INDEX_T>::max();
+      while (true) {
+        if (indices_[index] == sentinel) {
+          indices_[index] = idx;
+          break;
+        }
+        index = (index + 1) % (num_slots_minus_one_ + 1);
+      }
+    }
+  }
 
   size_t size() const { return num_elements_.load(); }
   PropertyType get_type() const { return keys_->type(); }

flex/utils/mmap_array.h

@@ -162,15 +162,16 @@ class mmap_array<std::string_view> {
   ~mmap_array() { release(); }
 
   void resize(size_t size) {
+    static constexpr size_t width = 64;
     size_t old_size = string_items_.size();
     if (old_size == size) {
       return;
     }
     if (old_size == 0 && buffer_.size() == 0) {
       string_items_.resize(size);
-      buffer_.resize(size * 1024);
+      buffer_.resize(size * width);
       buffer_loc_.store(0);
-    } else if (buffer_.size() >= (size * 1024)) {
+    } else if (buffer_.size() >= (size * width)) {
       mmap_array<string_item> new_string_items;
       new_string_items.resize(size);
       size_t accum_length = 0;
@@ -186,7 +187,7 @@ class mmap_array<std::string_view> {
       buffer_loc_.store(accum_length);
     } else {
       mmap_array<char> new_buffer;
-      new_buffer.resize(size * 1024);
+      new_buffer.resize(size * width);
       memcpy(new_buffer.data(), buffer_.data(), buffer_loc_.load());
 
       mmap_array<string_item> new_string_items;
@@ -202,6 +203,7 @@ class mmap_array<std::string_view> {
         new_string_items[i] = string_items_[i];
         accum_length += string_items_[i].length;
       }
+      new_string_items.swap(string_items_);
       buffer_loc_.store(accum_length);
       buffer_.swap(new_buffer);
     }