id_indexer.h

Go to the documentation of this file.

 
#ifndef GRAPHSCOPE_GRAPH_ID_INDEXER_H_
#define GRAPHSCOPE_GRAPH_ID_INDEXER_H_
 
#include <atomic>
#include <cassert>
#include <cmath>
#include <memory>
#include <string_view>
#include <thread>
#include <type_traits>
#include <vector>
 
#include "flat_hash_map/flat_hash_map.hpp"
#include "flex/utils/mmap_array.h"
#include "flex/utils/property/column.h"
#include "flex/utils/property/types.h"
#include "flex/utils/string_view_vector.h"
#include "glog/logging.h"
#include "grape/io/local_io_adaptor.h"
#include "grape/serialization/in_archive.h"
#include "grape/serialization/out_archive.h"
 
namespace gs {
 
namespace id_indexer_impl {
 
static constexpr int8_t min_lookups = 4;
static constexpr double max_load_factor = 0.5f;
 
inline int8_t log2(size_t value) {
 static constexpr int8_t table[64] = {
 63, 0, 58, 1, 59, 47, 53, 2, 60, 39, 48, 27, 54, 33, 42, 3,
 61, 51, 37, 40, 49, 18, 28, 20, 55, 30, 34, 11, 43, 14, 22, 4,
 62, 57, 46, 52, 38, 26, 32, 41, 50, 36, 17, 19, 29, 10, 13, 21,
 56, 45, 25, 31, 35, 16, 9, 12, 44, 24, 15, 8, 23, 7, 6, 5};
 value |= value >> 1;
 value |= value >> 2;
 value |= value >> 4;
 value |= value >> 8;
 value |= value >> 16;
 value |= value >> 32;
 return table[((value - (value >> 1)) * 0x07EDD5E59A4E28C2) >> 58];
}
 
template <typename T>
struct KeyBuffer {
 using type = std::vector<T>;
 
 template <typename IOADAPTOR_T>
 static void serialize(std::unique_ptr<IOADAPTOR_T>& writer,
 const type& buffer) {
 size_t size = buffer.size();
 CHECK(writer->Write(&size, sizeof(size_t)));
 if (size > 0) {
 CHECK(writer->Write(buffer.data(), size * sizeof(T)));
 }
 }
 
 template <typename IOADAPTOR_T>
 static void deserialize(std::unique_ptr<IOADAPTOR_T>& reader, type& buffer) {
 size_t size;
 CHECK(reader->Read(&size, sizeof(size_t)));
 if (size > 0) {
 buffer.resize(size);
 CHECK(reader->Read(buffer.data(), size * sizeof(T)));
 }
 }
};
 
template <>
struct KeyBuffer<std::string> {
 using type = std::vector<std::string>;
 
 template <typename IOADAPTOR_T>
 static void serialize(std::unique_ptr<IOADAPTOR_T>& writer,
 const type& buffer) {
 grape::InArchive arc;
 arc << buffer;
 CHECK(writer->WriteArchive(arc));
 }
 
 template <typename IOADAPTOR_T>
 static void deserialize(std::unique_ptr<IOADAPTOR_T>& reader, type& buffer) {
 grape::OutArchive arc;
 CHECK(reader->ReadArchive(arc));
 arc >> buffer;
 }
};
 
template <>
struct KeyBuffer<std::string_view> {
 using type = StringViewVector;
 
 template <typename IOADAPTOR_T>
 static void serialize(std::unique_ptr<IOADAPTOR_T>& writer,
 const type& buffer) {
 size_t content_buffer_size = buffer.content_buffer().size();
 CHECK(writer->Write(&content_buffer_size, sizeof(size_t)));
 if (content_buffer_size > 0) {
 CHECK(writer->Write(buffer.content_buffer().data(),
 content_buffer_size * sizeof(char)));
 }
 size_t offset_buffer_size = buffer.offset_buffer().size();
 CHECK(writer->Write(&offset_buffer_size, sizeof(size_t)));
 if (offset_buffer_size > 0) {
 CHECK(writer->Write(buffer.offset_buffer().data(),
 offset_buffer_size * sizeof(size_t)));
 }
 }
 
 template <typename IOADAPTOR_T>
 static void deserialize(std::unique_ptr<IOADAPTOR_T>& reader, type& buffer) {
 size_t content_buffer_size;
 CHECK(reader->Read(&content_buffer_size, sizeof(size_t)));
 if (content_buffer_size > 0) {
 buffer.content_buffer().resize(content_buffer_size);
 CHECK(reader->Read(buffer.content_buffer().data(),
 content_buffer_size * sizeof(char)));
 }
 size_t offset_buffer_size;
 CHECK(reader->Read(&offset_buffer_size, sizeof(size_t)));
 if (offset_buffer_size > 0) {
 buffer.offset_buffer().resize(offset_buffer_size);
 CHECK(reader->Read(buffer.offset_buffer().data(),
 offset_buffer_size * sizeof(size_t)));
 }
 }
};
 
} // namespace id_indexer_impl
 
template <typename T>
struct GHash {
 size_t operator()(const T& val) const { return std::hash<T>()(val); }
};
 
template <>
struct GHash<int64_t> {
 size_t operator()(const int64_t& val) const {
 uint64_t x = static_cast<uint64_t>(val);
 x = (x ^ (x >> 30)) * UINT64_C(0xbf58476d1ce4e5b9);
 x = (x ^ (x >> 27)) * UINT64_C(0x94d049bb133111eb);
 x = x ^ (x >> 31);
 return x;
 }
};
 
template <>
struct GHash<Any> {
 size_t operator()(const Any& val) const {
 if (val.type == PropertyType::kInt64) {
 return GHash<int64_t>()(val.AsInt64());
 } else if (val.type == PropertyType::kInt32) {
 return GHash<int32_t>()(val.AsInt32());
 } else if (val.type == PropertyType::kUInt64) {
 return GHash<uint64_t>()(val.AsUInt64());
 } else if (val.type == PropertyType::kUInt32) {
 return GHash<uint32_t>()(val.AsUInt32());
 } else {
 return GHash<std::string_view>()(val.AsStringView());
 }
 }
};
 
template <typename KEY_T, typename INDEX_T>
class IdIndexer;
 
template <typename INDEX_T>
class LFIndexer;
 
template <typename KEY_T, typename INDEX_T>
void build_lf_indexer(const IdIndexer<KEY_T, INDEX_T>& input,
 const std::string& filename, LFIndexer<INDEX_T>& lf,
 const std::string& snapshot_dir,
 const std::string& work_dir, PropertyType type);
 
template <typename INDEX_T>
class LFIndexer {
 public:
 LFIndexer()
 : indices_(),
 indices_size_(0),
 num_elements_(0),
 num_slots_minus_one_(0),
 keys_(nullptr),
 hasher_() {}
 LFIndexer(LFIndexer&& rhs)
 : indices_(std::move(rhs.indices_)),
 indices_size_(rhs.indices_size_),
 num_elements_(rhs.num_elements_.load()),
 num_slots_minus_one_(rhs.num_slots_minus_one_),
 hasher_(rhs.hasher_) {
 if (keys_ != rhs.keys_) {
 if (keys_ != nullptr) {
 delete keys_;
 }
 keys_ = rhs.keys_;
 }
 hash_policy_.set_mod_function_by_index(
 rhs.hash_policy_.get_mod_function_index());
 }
 
 ~LFIndexer() {
 if (keys_ != nullptr) {
 delete keys_;
 }
 }
 
 static std::string prefix() { return "indexer"; }
 
 void init(const PropertyType& type) {
 if (keys_ != nullptr) {
 delete keys_;
 }
 keys_ = nullptr;
 if (type == PropertyType::kInt64) {
 keys_ = new TypedColumn<int64_t>(StorageStrategy::kMem);
 } else if (type == PropertyType::kInt32) {
 keys_ = new TypedColumn<int32_t>(StorageStrategy::kMem);
 } else if (type == PropertyType::kUInt64) {
 keys_ = new TypedColumn<uint64_t>(StorageStrategy::kMem);
 } else if (type == PropertyType::kUInt32) {
 keys_ = new TypedColumn<uint32_t>(StorageStrategy::kMem);
 } else if (type.type_enum == impl::PropertyTypeImpl::kVarChar) {
 keys_ = new StringColumn(StorageStrategy::kMem,
 type.additional_type_info.max_length);
 } else if (type.type_enum == impl::PropertyTypeImpl::kStringView) {
 LOG(WARNING) << "String type is a deprecated type, use varchar instead.";
 LOG(WARNING) << "Use default max length"
 << PropertyType::STRING_DEFAULT_MAX_LENGTH
 << " for varchar type.";
 keys_ = new StringColumn(StorageStrategy::kMem,
 PropertyType::STRING_DEFAULT_MAX_LENGTH);
 } else {
 LOG(FATAL) << "Not support type [" << type << "] as pk type ..";
 }
 }
 
 void build_empty_LFIndexer(const std::string& filename,
 const std::string& snapshot_dir,
 const std::string& work_dir) {
 keys_->open(filename + ".keys", "", work_dir);
 indices_.open(work_dir + "/" + filename + ".indices", true);
 
 num_elements_.store(0);
 indices_size_ = 0;
 dump_meta(work_dir + "/" + filename + ".meta");
 indices_.reset();
 keys_->close();
 }
 
 void reserve(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 capacity() const { return keys_->size(); }
 
 size_t size() const { return num_elements_.load(); }
 PropertyType get_type() const { return keys_->type(); }
 
 INDEX_T insert(const Any& oid) {
 assert(oid.type == get_type());
 INDEX_T ind = static_cast<INDEX_T>(num_elements_.fetch_add(1));
 keys_->set_any(ind, oid);
 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 (__sync_bool_compare_and_swap(&indices_.data()[index], sentinel,
 ind)) {
 break;
 }
 index = (index + 1) % (num_slots_minus_one_ + 1);
 }
 return ind;
 }
 
 INDEX_T get_index(const Any& oid) const {
 assert(oid.type == get_type());
 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) {
 INDEX_T ind = indices_.get(index);
 if (ind == sentinel) {
 VLOG(10) << "cannot find " << oid.to_string() << " in lf_indexer";
 return ind;
 } else if (keys_->get(ind) == oid) {
 return ind;
 } else {
 index = (index + 1) % (num_slots_minus_one_ + 1);
 }
 }
 }
 
 bool get_index(const Any& oid, INDEX_T& ret) const {
 if (oid.type != get_type()) {
 return false;
 }
 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) {
 INDEX_T ind = indices_.get(index);
 if (ind == sentinel) {
 return false;
 } else if (keys_->get(ind) == oid) {
 ret = ind;
 return true;
 } else {
 index = (index + 1) % (num_slots_minus_one_ + 1);
 }
 }
 return false;
 }
 
 Any get_key(const INDEX_T& index) const { return keys_->get(index); }
 
 void copy_to_tmp(const std::string& cur_path, const std::string& tmp_path) {
 copy_file(cur_path + ".meta", tmp_path + ".meta");
 load_meta(tmp_path + ".meta");
 keys_->copy_to_tmp(cur_path + ".keys", tmp_path + ".keys");
 copy_file(cur_path + ".indices", tmp_path + ".indices");
 }
 
 void open(const std::string& name, const std::string& snapshot_dir,
 const std::string& work_dir) {
 if (std::filesystem::exists(snapshot_dir + "/" + name + ".meta")) {
 copy_to_tmp(snapshot_dir + "/" + name, work_dir + "/" + name);
 } else {
 build_empty_LFIndexer(name, "", work_dir);
 }
 
 load_meta(work_dir + "/" + name + ".meta");
 keys_->open(name + ".keys", "", work_dir);
 indices_.open(work_dir + "/" + name + ".indices", true);
 size_t num_elements = num_elements_.load();
 
 keys_->resize(num_elements + (num_elements >> 2));
 
 indices_size_ = indices_.size();
 }
 
 void open_in_memory(const std::string& name) {
 if (std::filesystem::exists(name + ".meta")) {
 load_meta(name + ".meta");
 } else {
 num_elements_.store(0);
 }
 keys_->open_in_memory(name + ".keys");
 indices_.open(name + ".indices", false);
 indices_size_ = indices_.size();
 size_t num_elements = num_elements_.load();
 keys_->resize(num_elements + (num_elements >> 2));
 }
 
 void open_with_hugepages(const std::string& name, bool hugepage_table) {
 if (std::filesystem::exists(name + ".meta")) {
 load_meta(name + ".meta");
 } else {
 num_elements_.store(0);
 }
 keys_->open_with_hugepages(name + ".keys", true);
 if (hugepage_table) {
 indices_.open_with_hugepages(name + ".indices");
 } else {
 indices_.open(name + ".indices", false);
 }
 indices_size_ = indices_.size();
 size_t num_elements = num_elements_.load();
 keys_->resize(num_elements + (num_elements >> 2));
 }
 
 void dump(const std::string& name, const std::string& snapshot_dir) {
 keys_->resize(num_elements_.load());
 keys_->dump(snapshot_dir + "/" + name + ".keys");
 indices_.dump(snapshot_dir + "/" + name + ".indices");
 dump_meta(snapshot_dir + "/" + name + ".meta");
 close();
 }
 
 void close() {
 keys_->close();
 indices_.reset();
 }
 
 void dump_meta(const std::string& filename) const {
 grape::InArchive arc;
 arc << get_type() << num_elements_.load() << num_slots_minus_one_
 << hash_policy_.get_mod_function_index();
 FILE* fout = fopen(filename.c_str(), "wb");
 fwrite(arc.GetBuffer(), sizeof(char), arc.GetSize(), fout);
 fflush(fout);
 fclose(fout);
 }
 
 void load_meta(const std::string& filename) {
 grape::OutArchive arc;
 FILE* fin = fopen(filename.c_str(), "r");
 size_t meta_file_size = std::filesystem::file_size(filename);
 std::vector<char> buf(meta_file_size);
 CHECK_EQ(fread(buf.data(), sizeof(char), meta_file_size, fin),
 meta_file_size);
 arc.SetSlice(buf.data(), meta_file_size);
 size_t mod_function_index;
 PropertyType type;
 arc >> type;
 size_t num_elements;
 arc >> num_elements;
 
 num_elements_.store(num_elements);
 arc >> num_slots_minus_one_ >> mod_function_index;
 init(type);
 hash_policy_.set_mod_function_by_index(mod_function_index);
 fclose(fin);
 }
 
 // get keys
 const ColumnBase& get_keys() const { return *keys_; }
 void warmup(int thread_num) const {
 size_t keys_size = num_elements_.load();
 size_t indices_size = indices_.size();
 std::atomic<size_t> k_i(0), i_i(0);
 std::atomic<size_t> output(0);
 size_t chunk = 4096;
 std::vector<std::thread> threads;
 for (int i = 0; i < thread_num; ++i) {
 threads.emplace_back([&]() {
 size_t ret = 0;
 while (true) {
 size_t begin = std::min(k_i.fetch_add(chunk), keys_size);
 size_t end = std::min(begin + chunk, keys_size);
 if (begin == end) {
 break;
 }
 while (begin < end) {
 keys_->get(begin);
 ++begin;
 }
 }
 while (true) {
 size_t begin = std::min(i_i.fetch_add(chunk), indices_size);
 size_t end = std::min(begin + chunk, indices_size);
 if (begin >= end) {
 break;
 }
 while (begin < end) {
 ret += indices_.get(begin);
 ++begin;
 }
 }
 output.fetch_add(ret);
 });
 }
 for (auto& thrd : threads) {
 thrd.join();
 }
 (void) output.load();
 }
 
 private:
 mmap_array<INDEX_T>
 indices_; // size() == indices_size_ == num_slots_minus_one_ +
 // log(num_slots_minus_one_)
 size_t indices_size_;
 std::atomic<size_t> num_elements_;
 size_t num_slots_minus_one_;
 ColumnBase* keys_;
 
 ska::ska::prime_number_hash_policy hash_policy_;
 GHash<Any> hasher_;
 
 template <typename _KEY_T, typename _INDEX_T>
 friend void build_lf_indexer(const IdIndexer<_KEY_T, _INDEX_T>& input,
 const std::string& filename,
 LFIndexer<_INDEX_T>& output,
 const std::string& snapshot_dir,
 const std::string& work_dir, PropertyType type);
};
 
template <typename INDEX_T>
class IdIndexerBase {
 public:
 IdIndexerBase() = default;
 virtual ~IdIndexerBase() = default;
 virtual PropertyType get_type() const = 0;
 virtual void _add(const Any& oid) = 0;
 virtual bool add(const Any& oid, INDEX_T& lid) = 0;
 virtual bool get_key(const INDEX_T& lid, Any& oid) const = 0;
 virtual bool get_index(const Any& oid, INDEX_T& lid) const = 0;
 virtual size_t size() const = 0;
};
 
template <typename KEY_T, typename INDEX_T>
class IdIndexer : public IdIndexerBase<INDEX_T> {
 public:
 using key_buffer_t = typename id_indexer_impl::KeyBuffer<KEY_T>::type;
 using ind_buffer_t = std::vector<INDEX_T>;
 using dist_buffer_t = std::vector<int8_t>;
 
 IdIndexer() : hasher_() { reset_to_empty_state(); }
 ~IdIndexer() {}
 
 PropertyType get_type() const override { return AnyConverter<KEY_T>::type(); }
 
 void _add(const Any& oid) override {
 assert(get_type() == oid.type);
 KEY_T oid_;
 ConvertAny<KEY_T>::to(oid, oid_);
 _add(oid_);
 }
 
 bool add(const Any& oid, INDEX_T& lid) override {
 assert(get_type() == oid.type);
 KEY_T oid_;
 ConvertAny<KEY_T>::to(oid, oid_);
 return add(oid_, lid);
 }
 
 bool get_key(const INDEX_T& lid, Any& oid) const override {
 KEY_T oid_;
 bool flag = get_key(lid, oid_);
 if (flag) {
 oid = Any::From(oid_);
 }
 return flag;
 }
 
 bool get_index(const Any& oid, INDEX_T& lid) const override {
 assert(get_type() == oid.type);
 KEY_T oid_;
 ConvertAny<KEY_T>::to(oid, oid_);
 return get_index(oid_, lid);
 }
 void Clear() {
 keys_.clear();
 indices_.clear();
 distances_.clear();
 num_elements_ = 0;
 num_slots_minus_one_ = 0;
 hash_policy_.reset();
 }
 
 size_t entry_num() const { return distances_.size(); }
 
 bool add(const KEY_T& oid, INDEX_T& lid) {
 size_t index =
 hash_policy_.index_for_hash(hasher_(oid), num_slots_minus_one_);
 
 int8_t distance_from_desired = 0;
 for (; distances_[index] >= distance_from_desired;
 ++index, ++distance_from_desired) {
 INDEX_T cur_lid = indices_[index];
 if (keys_[cur_lid] == oid) {
 lid = cur_lid;
 return false;
 }
 }
 
 lid = static_cast<INDEX_T>(keys_.size());
 keys_.push_back(oid);
 assert(keys_.size() == num_elements_ + 1);
 emplace_new_value(distance_from_desired, index, lid);
 assert(keys_.size() == num_elements_);
 return true;
 }
 
 bool add(KEY_T&& oid, INDEX_T& lid) {
 size_t index =
 hash_policy_.index_for_hash(hasher_(oid), num_slots_minus_one_);
 
 int8_t distance_from_desired = 0;
 for (; distances_[index] >= distance_from_desired;
 ++index, ++distance_from_desired) {
 INDEX_T cur_lid = indices_[index];
 if (keys_[cur_lid] == oid) {
 lid = cur_lid;
 return false;
 }
 }
 
 lid = static_cast<INDEX_T>(keys_.size());
 keys_.push_back(std::move(oid));
 assert(keys_.size() == num_elements_ + 1);
 emplace_new_value(distance_from_desired, index, lid);
 assert(keys_.size() == num_elements_);
 return true;
 }
 
 bool _add(const KEY_T& oid, size_t hash_value, INDEX_T& lid) {
 size_t index =
 hash_policy_.index_for_hash(hash_value, num_slots_minus_one_);
 
 int8_t distance_from_desired = 0;
 for (; distances_[index] >= distance_from_desired;
 ++index, ++distance_from_desired) {
 INDEX_T cur_lid = indices_[index];
 if (keys_[cur_lid] == oid) {
 lid = cur_lid;
 return false;
 }
 }
 
 lid = static_cast<INDEX_T>(keys_.size());
 keys_.push_back(oid);
 assert(keys_.size() == num_elements_ + 1);
 emplace_new_value(distance_from_desired, index, lid);
 assert(keys_.size() == num_elements_);
 return true;
 }
 
 bool _add(KEY_T&& oid, size_t hash_value, INDEX_T& lid) {
 size_t index =
 hash_policy_.index_for_hash(hash_value, num_slots_minus_one_);
 
 int8_t distance_from_desired = 0;
 for (; distances_[index] >= distance_from_desired;
 ++index, ++distance_from_desired) {
 INDEX_T cur_lid = indices_[index];
 if (keys_[cur_lid] == oid) {
 lid = cur_lid;
 return false;
 }
 }
 
 lid = static_cast<INDEX_T>(keys_.size());
 keys_.push_back(std::move(oid));
 assert(keys_.size() == num_elements_ + 1);
 emplace_new_value(distance_from_desired, index, lid);
 assert(keys_.size() == num_elements_);
 return true;
 }
 
 void _add(const KEY_T& oid) {
 size_t index =
 hash_policy_.index_for_hash(hasher_(oid), num_slots_minus_one_);
 
 int8_t distance_from_desired = 0;
 for (; distances_[index] >= distance_from_desired;
 ++index, ++distance_from_desired) {
 if (keys_[indices_[index]] == oid) {
 return;
 }
 }
 
 INDEX_T lid = static_cast<INDEX_T>(keys_.size());
 keys_.push_back(oid);
 assert(keys_.size() == num_elements_ + 1);
 emplace_new_value(distance_from_desired, index, lid);
 assert(keys_.size() == num_elements_);
 }
 
 void _add(KEY_T&& oid) {
 size_t index =
 hash_policy_.index_for_hash(hasher_(oid), num_slots_minus_one_);
 
 int8_t distance_from_desired = 0;
 for (; distances_[index] >= distance_from_desired;
 ++index, ++distance_from_desired) {
 if (keys_[indices_[index]] == oid) {
 return;
 }
 }
 
 INDEX_T lid = static_cast<INDEX_T>(keys_.size());
 keys_.push_back(std::move(oid));
 assert(keys_.size() == num_elements_ + 1);
 emplace_new_value(distance_from_desired, index, lid);
 assert(keys_.size() == num_elements_);
 }
 
 size_t bucket_count() const {
 return num_slots_minus_one_ ? num_slots_minus_one_ + 1 : 0;
 }
 
 bool empty() const { return (num_elements_ == 0); }
 
 size_t size() const override { return num_elements_; }
 
 bool get_key(INDEX_T lid, KEY_T& oid) const {
 if (static_cast<size_t>(lid) >= num_elements_) {
 return false;
 }
 oid = keys_[lid];
 return true;
 }
 
 bool get_index(const KEY_T& oid, INDEX_T& lid) const {
 size_t index =
 hash_policy_.index_for_hash(hasher_(oid), num_slots_minus_one_);
 for (int8_t distance = 0; distances_[index] >= distance;
 ++distance, ++index) {
 INDEX_T ret = indices_[index];
 if (keys_[ret] == oid) {
 lid = ret;
 return true;
 }
 }
 return false;
 }
 
 bool _get_index(const KEY_T& oid, size_t hash, INDEX_T& lid) const {
 size_t index = hash_policy_.index_for_hash(hash, num_slots_minus_one_);
 for (int8_t distance = 0; distances_[index] >= distance;
 ++distance, ++index) {
 INDEX_T ret = indices_[index];
 if (keys_[ret] == oid) {
 lid = ret;
 return true;
 }
 }
 return false;
 }
 
 void swap(IdIndexer<KEY_T, INDEX_T>& rhs) {
 keys_.swap(rhs.keys_);
 indices_.swap(rhs.indices_);
 distances_.swap(rhs.distances_);
 
 hash_policy_.swap(rhs.hash_policy_);
 std::swap(max_lookups_, rhs.max_lookups_);
 std::swap(num_elements_, rhs.num_elements_);
 std::swap(num_slots_minus_one_, rhs.num_slots_minus_one_);
 
 std::swap(hasher_, rhs.hasher_);
 }
 
 const key_buffer_t& keys() const { return keys_; }
 
 key_buffer_t& keys() { return keys_; }
 
 void Serialize(std::unique_ptr<grape::LocalIOAdaptor>& writer) const {
 id_indexer_impl::KeyBuffer<KEY_T>::serialize(writer, keys_);
 grape::InArchive arc;
 arc << hash_policy_.get_mod_function_index() << max_lookups_
 << num_elements_ << num_slots_minus_one_ << indices_.size()
 << distances_.size();
 CHECK(writer->WriteArchive(arc));
 arc.Clear();
 
 if (indices_.size() > 0) {
 CHECK(writer->Write(const_cast<uint8_t*>(indices_.data()),
 indices_.size() * sizeof(INDEX_T)));
 }
 if (distances_.size() > 0) {
 CHECK(writer->Write(const_cast<int8_t*>(distances_.data()),
 distances_.size() * sizeof(int8_t)));
 }
 }
 
 void Deserialize(std::unique_ptr<grape::LocalIOAdaptor>& reader) {
 id_indexer_impl::KeyBuffer<KEY_T>::deserialize(reader, keys_);
 grape::OutArchive arc;
 CHECK(reader->ReadArchive(arc));
 size_t mod_function_index;
 size_t indices_size, distances_size;
 arc >> mod_function_index >> max_lookups_ >> num_elements_ >>
 num_slots_minus_one_ >> indices_size >> distances_size;
 arc.Clear();
 
 hash_policy_.set_mod_function_by_index(mod_function_index);
 indices_.resize(indices_size);
 distances_.resize(distances_size);
 if (indices_size > 0) {
 CHECK(reader->Read(indices_.data(), indices_.size() * sizeof(INDEX_T)));
 }
 if (distances_size > 0) {
 CHECK(
 reader->Read(distances_.data(), distances_.size() * sizeof(int8_t)));
 }
 }
 
 void _rehash(size_t num) { rehash(num); }
 
 private:
 void emplace(INDEX_T lid) {
 KEY_T key = keys_[lid];
 size_t index =
 hash_policy_.index_for_hash(hasher_(key), num_slots_minus_one_);
 int8_t distance_from_desired = 0;
 for (; distances_[index] >= distance_from_desired;
 ++index, ++distance_from_desired) {
 if (indices_[index] == lid) {
 return;
 }
 }
 
 emplace_new_value(distance_from_desired, index, lid);
 }
 
 void emplace_new_value(int8_t distance_from_desired, size_t index,
 INDEX_T lid) {
 if (num_slots_minus_one_ == 0 || distance_from_desired == max_lookups_ ||
 num_elements_ + 1 >
 (num_slots_minus_one_ + 1) * id_indexer_impl::max_load_factor) {
 grow();
 return;
 } else if (distances_[index] < 0) {
 indices_[index] = lid;
 distances_[index] = distance_from_desired;
 ++num_elements_;
 return;
 }
 INDEX_T to_insert = lid;
 std::swap(distance_from_desired, distances_[index]);
 std::swap(to_insert, indices_[index]);
 for (++distance_from_desired, ++index;; ++index) {
 if (distances_[index] < 0) {
 indices_[index] = to_insert;
 distances_[index] = distance_from_desired;
 ++num_elements_;
 return;
 } else if (distances_[index] < distance_from_desired) {
 std::swap(distance_from_desired, distances_[index]);
 std::swap(to_insert, indices_[index]);
 ++distance_from_desired;
 } else {
 ++distance_from_desired;
 if (distance_from_desired == max_lookups_) {
 grow();
 return;
 }
 }
 }
 }
 
 void grow() { rehash(std::max(size_t(4), 2 * bucket_count())); }
 
 void rehash(size_t num_buckets) {
 num_buckets = std::max(
 num_buckets, static_cast<size_t>(std::ceil(
 num_elements_ / id_indexer_impl::max_load_factor)));
 
 if (num_buckets == 0) {
 reset_to_empty_state();
 return;
 }
 
 auto new_prime_index = hash_policy_.next_size_over(num_buckets);
 if (num_buckets == bucket_count()) {
 return;
 }
 
 int8_t new_max_lookups = compute_max_lookups(num_buckets);
 
 dist_buffer_t new_distances(num_buckets + new_max_lookups);
 ind_buffer_t new_indices(num_buckets + new_max_lookups,
 std::numeric_limits<INDEX_T>::max());
 
 size_t special_end_index = num_buckets + new_max_lookups - 1;
 for (size_t i = 0; i != special_end_index; ++i) {
 new_distances[i] = -1;
 }
 new_distances[special_end_index] = 0;
 
 new_indices.swap(indices_);
 new_distances.swap(distances_);
 
 std::swap(num_slots_minus_one_, num_buckets);
 --num_slots_minus_one_;
 hash_policy_.commit(new_prime_index);
 
 max_lookups_ = new_max_lookups;
 
 num_elements_ = 0;
 INDEX_T elem_num = static_cast<INDEX_T>(keys_.size());
 for (INDEX_T lid = 0; lid < elem_num; ++lid) {
 emplace(lid);
 }
 }
 
 void reset_to_empty_state() {
 keys_.clear();
 
 indices_.clear();
 distances_.clear();
 indices_.resize(id_indexer_impl::min_lookups,
 std::numeric_limits<INDEX_T>::max());
 distances_.resize(id_indexer_impl::min_lookups, -1);
 distances_[id_indexer_impl::min_lookups - 1] = 0;
 
 num_slots_minus_one_ = 0;
 hash_policy_.reset();
 max_lookups_ = id_indexer_impl::min_lookups - 1;
 num_elements_ = 0;
 }
 
 static int8_t compute_max_lookups(size_t num_buckets) {
 int8_t desired = id_indexer_impl::log2(num_buckets);
 return std::max(id_indexer_impl::min_lookups, desired);
 }
 
 key_buffer_t keys_;
 ind_buffer_t indices_;
 dist_buffer_t distances_;
 
 ska::ska::prime_number_hash_policy hash_policy_;
 int8_t max_lookups_ = id_indexer_impl::min_lookups - 1;
 size_t num_elements_ = 0;
 size_t num_slots_minus_one_ = 0;
 
 GHash<KEY_T> hasher_;
 
 template <typename _KEY_T, typename _INDEX_T>
 friend void build_lf_indexer(const IdIndexer<_KEY_T, _INDEX_T>& input,
 const std::string& filename,
 LFIndexer<_INDEX_T>& output,
 const std::string& snapshot_dir,
 const std::string& work_dir, PropertyType type);
};
 
template <typename KEY_T, typename INDEX_T>
struct _move_data {
 using key_buffer_t = typename id_indexer_impl::KeyBuffer<KEY_T>::type;
 void operator()(const key_buffer_t& input, ColumnBase& col, size_t size) {
 auto& keys = dynamic_cast<TypedColumn<KEY_T>&>(col);
 for (size_t idx = 0; idx < size; ++idx) {
 keys.set_value(idx, input[idx]);
 }
 }
};
 
template <typename INDEX_T>
struct _move_data<std::string_view, INDEX_T> {
 using key_buffer_t =
 typename id_indexer_impl::KeyBuffer<std::string_view>::type;
 void operator()(const key_buffer_t& input, ColumnBase& col, size_t size) {
 auto& keys = dynamic_cast<StringColumn&>(col);
 for (size_t idx = 0; idx < size; ++idx) {
 keys.set_value(idx, input[idx]);
 }
 }
};
 
template <typename KEY_T, typename INDEX_T>
void build_lf_indexer(const IdIndexer<KEY_T, INDEX_T>& input,
 const std::string& filename, LFIndexer<INDEX_T>& lf,
 const std::string& snapshot_dir,
 const std::string& work_dir, PropertyType type) {
 size_t size = input.keys_.size();
 lf.init(type);
 lf.keys_->open(filename + ".keys", "", work_dir);
 lf.keys_->resize(size);
 _move_data<KEY_T, INDEX_T>()(input.keys_, *lf.keys_, size);
 lf.num_elements_.store(size);
 
 lf.indices_.open(snapshot_dir + "/" + filename + ".indices", true);
 lf.indices_.resize(input.num_slots_minus_one_ + 1);
 
 lf.indices_size_ = input.indices_.size();
 
 lf.hash_policy_.set_mod_function_by_index(
 input.hash_policy_.get_mod_function_index());
 lf.num_slots_minus_one_ = input.num_slots_minus_one_;
 memcpy(lf.indices_.data(), input.indices_.data(),
 lf.indices_.size() * sizeof(INDEX_T));
 static constexpr INDEX_T sentinel = std::numeric_limits<INDEX_T>::max();
 
 std::vector<INDEX_T> residuals;
 for (size_t idx = lf.indices_.size(); idx < lf.indices_size_; ++idx) {
 if (input.indices_[idx] != sentinel) {
 residuals.push_back(input.indices_[idx]);
 }
 }
 for (const auto& lid : residuals) {
 auto oid = input.keys_[lid];
 size_t index = input.hash_policy_.index_for_hash(
 input.hasher_(oid), input.num_slots_minus_one_);
 while (true) {
 if (lf.indices_[index] == lid) {
 break;
 } else if (lf.indices_[index] == sentinel) {
 lf.indices_[index] = lid;
 break;
 }
 index = (index + 1) % (input.num_slots_minus_one_ + 1);
 }
 }
 lf.dump_meta(snapshot_dir + "/" + filename + ".meta");
 
 lf.keys_->dump(snapshot_dir + "/" + filename + ".keys");
 std::filesystem::remove(work_dir + "/" + filename + ".meta");
 lf.keys_->close();
 lf.keys_->open(filename + ".keys", snapshot_dir, "");
}
 
} // namespace gs
 
#endif // GRAPHSCOPE_GRAPH_ID_INDEXER_H_