Status IndexManager::Load()

in src/search/index_manager.cc [40:131]
71 lines of code
23 McCabe index (conditional complexity)

Status IndexManager::Load(engine::Context &ctx, const std::string &ns) {
  // currently index cannot work in cluster mode
  if (storage->GetConfig()->cluster_enabled) {
    return Status::OK();
  }
  util::UniqueIterator iter(ctx, ctx.DefaultScanOptions(), ColumnFamilyID::Search);
  auto begin = SearchKey{ns, ""}.ConstructIndexMeta();

  for (iter->Seek(begin); iter->Valid(); iter->Next()) {
    auto key = iter->key();

    uint8_t ns_size = 0;
    if (!GetFixed8(&key, &ns_size)) break;
    if (ns_size != ns.size()) break;
    if (!key.starts_with(ns)) break;
    key.remove_prefix(ns_size);

    uint8_t subkey_type = 0;
    if (!GetFixed8(&key, &subkey_type)) break;
    if (subkey_type != (uint8_t)SearchSubkeyType::INDEX_META) break;

    Slice index_name;
    if (!GetSizedString(&key, &index_name)) break;

    IndexMetadata metadata;
    auto index_meta_value = iter->value();
    if (auto s = metadata.Decode(&index_meta_value); !s.ok()) {
      return {Status::NotOK, fmt::format("fail to decode index metadata for index {}: {}", index_name, s.ToString())};
    }

    auto index_key = SearchKey(ns, index_name.ToStringView());
    std::string prefix_value;
    if (auto s = storage->Get(ctx, ctx.DefaultMultiGetOptions(), storage->GetCFHandle(ColumnFamilyID::Search),
                              index_key.ConstructIndexPrefixes(), &prefix_value);
        !s.ok()) {
      return {Status::NotOK, fmt::format("fail to find index prefixes for index {}: {}", index_name, s.ToString())};
    }

    IndexPrefixes prefixes;
    Slice prefix_slice = prefix_value;
    if (auto s = prefixes.Decode(&prefix_slice); !s.ok()) {
      return {Status::NotOK, fmt::format("fail to decode index prefixes for index {}: {}", index_name, s.ToString())};
    }

    auto info = std::make_unique<kqir::IndexInfo>(index_name.ToString(), metadata, ns);
    info->prefixes = prefixes;

    util::UniqueIterator field_iter(ctx, ctx.DefaultScanOptions(), ColumnFamilyID::Search);
    auto field_begin = index_key.ConstructFieldMeta();

    for (field_iter->Seek(field_begin); field_iter->Valid(); field_iter->Next()) {
      auto key = field_iter->key();

      uint8_t ns_size = 0;
      if (!GetFixed8(&key, &ns_size)) break;
      if (ns_size != ns.size()) break;
      if (!key.starts_with(ns)) break;
      key.remove_prefix(ns_size);

      uint8_t subkey_type = 0;
      if (!GetFixed8(&key, &subkey_type)) break;
      if (subkey_type != (uint8_t)SearchSubkeyType::FIELD_META) break;

      Slice value;
      if (!GetSizedString(&key, &value)) break;
      if (value != index_name) break;

      if (!GetSizedString(&key, &value)) break;

      auto field_name = value;
      auto field_value = field_iter->value();

      std::unique_ptr<IndexFieldMetadata> field_meta;
      if (auto s = IndexFieldMetadata::Decode(&field_value, field_meta); !s.ok()) {
        return {Status::NotOK, fmt::format("fail to decode index field metadata for index {}, field {}: {}", index_name,
                                           field_name, s.ToString())};
      }

      info->Add(kqir::FieldInfo(field_name.ToString(), std::move(field_meta)));
    }

    auto updater = std::make_unique<IndexUpdater>(info.get());
    indexer->Add(std::move(updater));
    index_map.Insert(std::move(info));
  }

  if (auto s = iter->status(); !s.ok()) {
    return {Status::NotOK, fmt::format("fail to load index metadata: {}", s.ToString())};
  }

  return Status::OK();
}