Skip to content

Commit

Permalink
fix: fix directory confusion between indexnode and querynode
Browse files Browse the repository at this point in the history
Signed-off-by: luzhang <luzhang@zilliz.com>
  • Loading branch information
luzhang committed Jan 8, 2025
1 parent 84f8047 commit 36e62d4
Show file tree
Hide file tree
Showing 33 changed files with 217 additions and 102 deletions.
26 changes: 26 additions & 0 deletions internal/core/src/common/Common.h
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,32 @@ extern int64_t LOW_PRIORITY_THREAD_CORE_COEFFICIENT;
extern int CPU_NUM;
extern int64_t EXEC_EVAL_EXPR_BATCH_SIZE;

enum class Role { QueryNode, IndexNode };

const std::unordered_map<Role, std::string> RoleToStringMap = {
{Role::QueryNode, "querynode"}, {Role::IndexNode, "indexnode"}};

// convert role to string
inline std::string
ToString(Role role) {
auto it = RoleToStringMap.find(role);
if (it != RoleToStringMap.end()) {
return it->second;
}
PanicInfo(UnexpectedError, "role {} not found", int(role));
}

// convert string to role
inline Role
FromString(const std::string& role_str) {
for (const auto& pair : RoleToStringMap) {
if (pair.second == role_str) {
return pair.first;
}
}
PanicInfo(UnexpectedError, "role {} not found", role_str);
}

void
SetIndexSliceSize(const int64_t size);

Expand Down
9 changes: 9 additions & 0 deletions internal/core/src/index/Index.h
Original file line number Diff line number Diff line change
Expand Up @@ -71,12 +71,21 @@ class IndexBase {
return index_type_;
}

bool
IsLoadingIndex() const {
return is_loading_index_;
}

protected:
explicit IndexBase(IndexType index_type)
: index_type_(std::move(index_type)) {
}

IndexType index_type_ = "";

// current index obj is used to build and load index in one class,
// this flag is used to indicate whether building index or loading index
bool is_loading_index_{false};
};

using IndexBasePtr = std::unique_ptr<IndexBase>;
Expand Down
5 changes: 3 additions & 2 deletions internal/core/src/index/InvertedIndexTantivy.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -92,10 +92,11 @@ InvertedIndexTantivy<T>::InvertedIndexTantivy(
const storage::FileManagerContext& ctx)
: ScalarIndex<T>(INVERTED_INDEX_TYPE),
schema_(ctx.fieldDataMeta.field_schema) {
this->is_loading_index_ = ctx.for_loading_index;
mem_file_manager_ = std::make_shared<MemFileManager>(ctx);
disk_file_manager_ = std::make_shared<DiskFileManager>(ctx);
// push init wrapper to load process
if (ctx.for_loading_index) {
if (this->is_loading_index_) {
return;
}
InitForBuildIndex();
Expand All @@ -107,7 +108,7 @@ InvertedIndexTantivy<T>::~InvertedIndexTantivy() {
wrapper_->free();
}
auto local_chunk_manager =
storage::LocalChunkManagerSingleton::GetInstance().GetChunkManager();
storage::LocalChunkManagerFactory::GetInstance().GetChunkManager();
auto prefix = path_;
LOG_INFO("inverted index remove path:{}", path_);
local_chunk_manager->RemoveDir(prefix);
Expand Down
44 changes: 28 additions & 16 deletions internal/core/src/index/VectorDiskIndex.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -46,18 +46,24 @@ VectorDiskAnnIndex<T>::VectorDiskAnnIndex(
file_manager_ =
std::make_shared<storage::DiskFileManagerImpl>(file_manager_context);
AssertInfo(file_manager_ != nullptr, "create file manager failed!");
auto local_chunk_manager =
storage::LocalChunkManagerSingleton::GetInstance().GetChunkManager();
auto local_index_path_prefix = file_manager_->GetLocalIndexObjectPrefix();

// As we have guarded dup-load in QueryNode,
// this assertion failed only if the Milvus rebooted in the same pod,
// need to remove these files then re-load the segment
if (local_chunk_manager->Exist(local_index_path_prefix)) {
local_chunk_manager->RemoveDir(local_index_path_prefix);
is_loading_index_ = file_manager_context.for_loading_index;
if (is_loading_index_) {
// As we have guarded dup-load in QueryNode,
// this assertion failed only if the Milvus rebooted in the same pod,
// need to remove these files then re-load the segment
auto local_chunk_manager =
storage::LocalChunkManagerFactory::GetInstance().GetChunkManager();
auto local_index_path_prefix =
file_manager_->GetLocalIndexObjectPrefix();

if (local_chunk_manager->Exist(local_index_path_prefix)) {
local_chunk_manager->RemoveDir(local_index_path_prefix);
}
CheckCompatible(version);
local_chunk_manager->CreateDir(local_index_path_prefix);
}
CheckCompatible(version);
local_chunk_manager->CreateDir(local_index_path_prefix);

auto diskann_index_pack =
knowhere::Pack(std::shared_ptr<knowhere::FileManager>(file_manager_));
auto get_index_obj = knowhere::IndexFactory::Instance().Create<T>(
Expand Down Expand Up @@ -135,7 +141,8 @@ template <typename T>
void
VectorDiskAnnIndex<T>::Build(const Config& config) {
auto local_chunk_manager =
storage::LocalChunkManagerSingleton::GetInstance().GetChunkManager();
storage::LocalChunkManagerFactory::GetInstance().GetChunkManager(
milvus::Role::IndexNode);
knowhere::Json build_config;
build_config.update(config);

Expand Down Expand Up @@ -185,7 +192,8 @@ void
VectorDiskAnnIndex<T>::BuildWithDataset(const DatasetPtr& dataset,
const Config& config) {
auto local_chunk_manager =
storage::LocalChunkManagerSingleton::GetInstance().GetChunkManager();
storage::LocalChunkManagerFactory::GetInstance().GetChunkManager(
milvus::Role::IndexNode);
knowhere::Json build_config;
build_config.update(config);
// set data path
Expand Down Expand Up @@ -353,10 +361,14 @@ template <typename T>
void
VectorDiskAnnIndex<T>::CleanLocalData() {
auto local_chunk_manager =
storage::LocalChunkManagerSingleton::GetInstance().GetChunkManager();
local_chunk_manager->RemoveDir(file_manager_->GetLocalIndexObjectPrefix());
local_chunk_manager->RemoveDir(
file_manager_->GetLocalRawDataObjectPrefix());
storage::LocalChunkManagerFactory::GetInstance().GetChunkManager();
if (is_loading_index_) {
local_chunk_manager->RemoveDir(
file_manager_->GetLocalIndexObjectPrefix());
} else {
local_chunk_manager->RemoveDir(
file_manager_->GetLocalRawDataObjectPrefix());
}
}

template <typename T>
Expand Down
7 changes: 4 additions & 3 deletions internal/core/src/segcore/load_index_c.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@
#include "common/EasyAssert.h"
#include "common/Types.h"
#include "common/type_c.h"
#include "common/Common.h"
#include "index/Index.h"
#include "index/IndexFactory.h"
#include "index/Meta.h"
Expand Down Expand Up @@ -156,7 +157,7 @@ appendVecIndex(CLoadIndexInfo c_load_index_info, CBinarySet c_binary_set) {
config["index_files"] = load_index_info->index_files;

milvus::storage::FileManagerContext fileManagerContext(
field_meta, index_meta, remote_chunk_manager);
field_meta, index_meta, remote_chunk_manager, true);
fileManagerContext.set_for_loading_index(true);

load_index_info->index =
Expand Down Expand Up @@ -424,8 +425,8 @@ CleanLoadedIndex(CLoadIndexInfo c_load_index_info) {
auto load_index_info =
(milvus::segcore::LoadIndexInfo*)c_load_index_info;
auto local_chunk_manager =
milvus::storage::LocalChunkManagerSingleton::GetInstance()
.GetChunkManager();
milvus::storage::LocalChunkManagerFactory::GetInstance()
.GetChunkManager(milvus::Role::QueryNode);
auto index_file_path_prefix =
milvus::storage::GenIndexPathPrefix(local_chunk_manager,
load_index_info->index_build_id,
Expand Down
1 change: 1 addition & 0 deletions internal/core/src/segcore/segment_c.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -466,6 +466,7 @@ LoadTextIndex(CSegmentInterface c_segment,

milvus::storage::FileManagerContext ctx(
field_meta, index_meta, remote_chunk_manager);
ctx.set_for_loading_index(true);

auto index = std::make_unique<milvus::index::TextMatchIndex>(ctx);
index->Load(config);
Expand Down
36 changes: 23 additions & 13 deletions internal/core/src/storage/DiskFileManagerImpl.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -48,13 +48,18 @@ namespace milvus::storage {
DiskFileManagerImpl::DiskFileManagerImpl(
const FileManagerContext& fileManagerContext)
: FileManagerImpl(fileManagerContext.fieldDataMeta,
fileManagerContext.indexMeta) {
fileManagerContext.indexMeta,
fileManagerContext.for_loading_index) {
rcm_ = fileManagerContext.chunkManagerPtr;
}

DiskFileManagerImpl::~DiskFileManagerImpl() {
auto local_chunk_manager =
LocalChunkManagerSingleton::GetInstance().GetChunkManager();
for_loading_index_
? LocalChunkManagerFactory::GetInstance().GetChunkManager(
milvus::Role::QueryNode)
: LocalChunkManagerFactory::GetInstance().GetChunkManager(
milvus::Role::IndexNode);
local_chunk_manager->RemoveDir(GetIndexPathPrefixWithBuildID(
local_chunk_manager, index_meta_.build_id));
}
Expand Down Expand Up @@ -82,7 +87,7 @@ DiskFileManagerImpl::GetRemoteTextLogPath(const std::string& file_name,
bool
DiskFileManagerImpl::AddFile(const std::string& file) noexcept {
auto local_chunk_manager =
LocalChunkManagerSingleton::GetInstance().GetChunkManager();
LocalChunkManagerFactory::GetInstance().GetChunkManager();
FILEMANAGER_TRY
if (!local_chunk_manager->Exist(file)) {
LOG_ERROR("local file {} not exists", file);
Expand Down Expand Up @@ -134,7 +139,7 @@ DiskFileManagerImpl::AddFile(const std::string& file) noexcept {
bool
DiskFileManagerImpl::AddTextLog(const std::string& file) noexcept {
auto local_chunk_manager =
LocalChunkManagerSingleton::GetInstance().GetChunkManager();
LocalChunkManagerFactory::GetInstance().GetChunkManager();
FILEMANAGER_TRY
if (!local_chunk_manager->Exist(file)) {
LOG_ERROR("local file {} not exists", file);
Expand Down Expand Up @@ -190,7 +195,7 @@ DiskFileManagerImpl::AddBatchIndexFiles(
const std::vector<std::string>& remote_files,
const std::vector<int64_t>& remote_file_sizes) {
auto local_chunk_manager =
LocalChunkManagerSingleton::GetInstance().GetChunkManager();
LocalChunkManagerFactory::GetInstance().GetChunkManager();
auto& pool = ThreadPools::GetThreadPool(milvus::ThreadPoolPriority::HIGH);

std::vector<std::future<std::shared_ptr<uint8_t[]>>> futures;
Expand Down Expand Up @@ -239,7 +244,7 @@ void
DiskFileManagerImpl::CacheIndexToDisk(
const std::vector<std::string>& remote_files) {
auto local_chunk_manager =
LocalChunkManagerSingleton::GetInstance().GetChunkManager();
LocalChunkManagerFactory::GetInstance().GetChunkManager();

std::map<std::string, std::vector<int>> index_slices;
for (auto& file_path : remote_files) {
Expand Down Expand Up @@ -307,7 +312,7 @@ void
DiskFileManagerImpl::CacheTextLogToDisk(
const std::vector<std::string>& remote_files) {
auto local_chunk_manager =
LocalChunkManagerSingleton::GetInstance().GetChunkManager();
LocalChunkManagerFactory::GetInstance().GetChunkManager();

std::map<std::string, std::vector<int>> index_slices;
for (auto& file_path : remote_files) {
Expand Down Expand Up @@ -376,7 +381,8 @@ DiskFileManagerImpl::CacheRawDataToDisk(std::vector<std::string> remote_files) {
auto field_id = GetFieldDataMeta().field_id;

auto local_chunk_manager =
LocalChunkManagerSingleton::GetInstance().GetChunkManager();
LocalChunkManagerFactory::GetInstance().GetChunkManager(
milvus::Role::IndexNode);
std::string local_data_path;
bool file_created = false;

Expand Down Expand Up @@ -619,7 +625,8 @@ DiskFileManagerImpl::CacheOptFieldToDisk(OptFieldT& fields_map) {
auto segment_id = GetFieldDataMeta().segment_id;
auto vec_field_id = GetFieldDataMeta().field_id;
auto local_chunk_manager =
LocalChunkManagerSingleton::GetInstance().GetChunkManager();
LocalChunkManagerFactory::GetInstance().GetChunkManager(
milvus::Role::IndexNode);
auto local_data_path = storage::GenFieldRawDataPathPrefix(
local_chunk_manager, segment_id, vec_field_id) +
std::string(VEC_OPT_FIELDS);
Expand Down Expand Up @@ -696,15 +703,17 @@ DiskFileManagerImpl::GetFileName(const std::string& localfile) {
std::string
DiskFileManagerImpl::GetLocalIndexObjectPrefix() {
auto local_chunk_manager =
LocalChunkManagerSingleton::GetInstance().GetChunkManager();
LocalChunkManagerFactory::GetInstance().GetChunkManager(
milvus::Role::QueryNode);
return GenIndexPathPrefix(
local_chunk_manager, index_meta_.build_id, index_meta_.index_version);
}

std::string
DiskFileManagerImpl::GetLocalTextIndexPrefix() {
auto local_chunk_manager =
LocalChunkManagerSingleton::GetInstance().GetChunkManager();
LocalChunkManagerFactory::GetInstance().GetChunkManager(
milvus::Role::QueryNode);
return GenTextIndexPathPrefix(local_chunk_manager,
index_meta_.build_id,
index_meta_.index_version,
Expand All @@ -729,7 +738,8 @@ DiskFileManagerImpl::GetTextIndexIdentifier() {
std::string
DiskFileManagerImpl::GetLocalRawDataObjectPrefix() {
auto local_chunk_manager =
LocalChunkManagerSingleton::GetInstance().GetChunkManager();
LocalChunkManagerFactory::GetInstance().GetChunkManager(
milvus::Role::IndexNode);
return GenFieldRawDataPathPrefix(
local_chunk_manager, field_meta_.segment_id, field_meta_.field_id);
}
Expand All @@ -744,7 +754,7 @@ std::optional<bool>
DiskFileManagerImpl::IsExisted(const std::string& file) noexcept {
bool isExist = false;
auto local_chunk_manager =
LocalChunkManagerSingleton::GetInstance().GetChunkManager();
LocalChunkManagerFactory::GetInstance().GetChunkManager();
try {
isExist = local_chunk_manager->Exist(file);
} catch (std::exception& e) {
Expand Down
20 changes: 18 additions & 2 deletions internal/core/src/storage/FileManager.h
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,16 @@ struct FileManagerContext {
chunkManagerPtr(chunkManagerPtr) {
}

explicit FileManagerContext(const FieldDataMeta& fieldDataMeta,
const IndexMeta& indexMeta,
const ChunkManagerPtr& chunkManagerPtr,
bool for_loading)
: fieldDataMeta(fieldDataMeta),
indexMeta(indexMeta),
chunkManagerPtr(chunkManagerPtr),
for_loading_index(for_loading) {
}

bool
Valid() const {
return chunkManagerPtr != nullptr;
Expand Down Expand Up @@ -74,8 +84,11 @@ struct FileManagerContext {
class FileManagerImpl : public knowhere::FileManager {
public:
explicit FileManagerImpl(const FieldDataMeta& field_mata,
IndexMeta index_meta)
: field_meta_(field_mata), index_meta_(std::move(index_meta)) {
IndexMeta index_meta,
bool for_loading_index)
: field_meta_(field_mata),
index_meta_(std::move(index_meta)),
for_loading_index_(for_loading_index) {
}

public:
Expand Down Expand Up @@ -176,6 +189,9 @@ class FileManagerImpl : public knowhere::FileManager {
// index meta
IndexMeta index_meta_;
ChunkManagerPtr rcm_;

// indicate whether file manager is used for building index or load index
bool for_loading_index_{false};
};

using FileManagerImplPtr = std::shared_ptr<FileManagerImpl>;
Expand Down
Loading

0 comments on commit 36e62d4

Please sign in to comment.