Skip to content

milvus
milvus

BaiXuePrincess / milvus
与 Fork 源项目一致

Fork自 milvus / milvus

通知 7
Star 4
Fork 0
milvus
milvus

体验新版 GitCode，发现更多精彩内容 >>

提交 57dfdae3 编写于 作者: X Xu Peng
浏览文件
操作

feat(db): fix bug and add exception catch in meta operation 


Former-commit-id: 6d33945997e7a528bc729efdad6076d6532f79b3
上级 486c175b
隐藏空白更改
内联 并排
Showing with 241 addition and 187 deletion
cpp/src/db/DBImpl.cpp
浏览文件 @ 57dfdae3
...@@ -184,7 +184,8 @@ void DBImpl::background_call() { ...@@ -184,7 +184,8 @@ void DBImpl::background_call() {
std::lock_guard<std::mutex> lock(_mutex); std::lock_guard<std::mutex> lock(_mutex);
assert(_bg_compaction_scheduled); assert(_bg_compaction_scheduled);
if (!_bg_error.ok()) return; if (!_bg_error.ok() || _shutting_down.load(std::memory_order_acquire))
return ;
background_compaction(); background_compaction();
......
cpp/src/db/DBMetaImpl.cpp
浏览文件 @ 57dfdae3
...@@ -129,31 +129,41 @@ Status DBMetaImpl::get_group(GroupSchema& group_info) { ...@@ -129,31 +129,41 @@ Status DBMetaImpl::get_group(GroupSchema& group_info) {
} }
Status DBMetaImpl::get_group_no_lock(GroupSchema& group_info) { Status DBMetaImpl::get_group_no_lock(GroupSchema& group_info) {
auto groups = ConnectorPtr->select(columns(&GroupSchema::id, try {
&GroupSchema::group_id, auto groups = ConnectorPtr->select(columns(&GroupSchema::id,
&GroupSchema::files_cnt, &GroupSchema::group_id,
&GroupSchema::dimension), &GroupSchema::files_cnt,
where(c(&GroupSchema::group_id) == group_info.group_id)); &GroupSchema::dimension),
assert(groups.size() <= 1); where(c(&GroupSchema::group_id) == group_info.group_id));
if (groups.size() == 1) { assert(groups.size() <= 1);
group_info.id = std::get<0>(groups[0]); if (groups.size() == 1) {
group_info.files_cnt = std::get<2>(groups[0]); group_info.id = std::get<0>(groups[0]);
group_info.dimension = std::get<3>(groups[0]); group_info.files_cnt = std::get<2>(groups[0]);
} else { group_info.dimension = std::get<3>(groups[0]);
return Status::NotFound("Group " + group_info.group_id + " not found"); } else {
return Status::NotFound("Group " + group_info.group_id + " not found");
}
} catch (std::exception &e) {
LOG(DEBUG) << e.what();
throw e;
} }
return Status::OK(); return Status::OK();
} }
Status DBMetaImpl::has_group(const std::string& group_id, bool& has_or_not) { Status DBMetaImpl::has_group(const std::string& group_id, bool& has_or_not) {
auto groups = ConnectorPtr->select(columns(&GroupSchema::id), try {
where(c(&GroupSchema::group_id) == group_id)); auto groups = ConnectorPtr->select(columns(&GroupSchema::id),
assert(groups.size() <= 1); where(c(&GroupSchema::group_id) == group_id));
if (groups.size() == 1) { assert(groups.size() <= 1);
has_or_not = true; if (groups.size() == 1) {
} else { has_or_not = true;
has_or_not = false; } else {
has_or_not = false;
}
} catch (std::exception &e) {
LOG(DEBUG) << e.what();
throw e;
} }
return Status::OK(); return Status::OK();
} }
...@@ -183,6 +193,7 @@ Status DBMetaImpl::add_group_file(GroupFileSchema& group_file) { ...@@ -183,6 +193,7 @@ Status DBMetaImpl::add_group_file(GroupFileSchema& group_file) {
try { try {
auto id = ConnectorPtr->insert(group_file); auto id = ConnectorPtr->insert(group_file);
group_file.id = id; group_file.id = id;
LOG(DEBUG) << "Add group_file of file_id=" << group_file.file_id;
} catch (...) { } catch (...) {
return Status::DBTransactionError("Add file Error"); return Status::DBTransactionError("Add file Error");
} }
...@@ -200,37 +211,42 @@ Status DBMetaImpl::add_group_file(GroupFileSchema& group_file) { ...@@ -200,37 +211,42 @@ Status DBMetaImpl::add_group_file(GroupFileSchema& group_file) {
Status DBMetaImpl::files_to_index(GroupFilesSchema& files) { Status DBMetaImpl::files_to_index(GroupFilesSchema& files) {
files.clear(); files.clear();
auto selected = ConnectorPtr->select(columns(&GroupFileSchema::id, try {
&GroupFileSchema::group_id, auto selected = ConnectorPtr->select(columns(&GroupFileSchema::id,
&GroupFileSchema::file_id, &GroupFileSchema::group_id,
&GroupFileSchema::file_type, &GroupFileSchema::file_id,
&GroupFileSchema::rows, &GroupFileSchema::file_type,
&GroupFileSchema::date), &GroupFileSchema::rows,
where(c(&GroupFileSchema::file_type) == (int)GroupFileSchema::TO_INDEX)); &GroupFileSchema::date),
where(c(&GroupFileSchema::file_type) == (int)GroupFileSchema::TO_INDEX));
std::map<std::string, GroupSchema> groups;
std::map<std::string, GroupSchema> groups;
for (auto& file : selected) {
GroupFileSchema group_file; for (auto& file : selected) {
group_file.id = std::get<0>(file); GroupFileSchema group_file;
group_file.group_id = std::get<1>(file); group_file.id = std::get<0>(file);
group_file.file_id = std::get<2>(file); group_file.group_id = std::get<1>(file);
group_file.file_type = std::get<3>(file); group_file.file_id = std::get<2>(file);
group_file.rows = std::get<4>(file); group_file.file_type = std::get<3>(file);
group_file.date = std::get<5>(file); group_file.rows = std::get<4>(file);
GetGroupFilePath(group_file); group_file.date = std::get<5>(file);
auto groupItr = groups.find(group_file.group_id); GetGroupFilePath(group_file);
if (groupItr == groups.end()) { auto groupItr = groups.find(group_file.group_id);
GroupSchema group_info; if (groupItr == groups.end()) {
group_info.group_id = group_file.group_id; GroupSchema group_info;
auto status = get_group_no_lock(group_info); group_info.group_id = group_file.group_id;
if (!status.ok()) { auto status = get_group_no_lock(group_info);
return status; if (!status.ok()) {
return status;
}
groups[group_file.group_id] = group_info;
} }
groups[group_file.group_id] = group_info; group_file.dimension = groups[group_file.group_id].dimension;
files.push_back(group_file);
} }
group_file.dimension = groups[group_file.group_id].dimension; } catch (std::exception & e) {
files.push_back(group_file); LOG(DEBUG) << e.what();
throw e;
} }
return Status::OK(); return Status::OK();
...@@ -241,38 +257,43 @@ Status DBMetaImpl::files_to_search(const std::string &group_id, ...@@ -241,38 +257,43 @@ Status DBMetaImpl::files_to_search(const std::string &group_id,
DatePartionedGroupFilesSchema &files) { DatePartionedGroupFilesSchema &files) {
// TODO: support data partition // TODO: support data partition
files.clear(); files.clear();
auto selected = ConnectorPtr->select(columns(&GroupFileSchema::id, try {
&GroupFileSchema::group_id, auto selected = ConnectorPtr->select(columns(&GroupFileSchema::id,
&GroupFileSchema::file_id, &GroupFileSchema::group_id,
&GroupFileSchema::file_type, &GroupFileSchema::file_id,
&GroupFileSchema::rows, &GroupFileSchema::file_type,
&GroupFileSchema::date), &GroupFileSchema::rows,
where(c(&GroupFileSchema::group_id) == group_id and &GroupFileSchema::date),
(c(&GroupFileSchema::file_type) == (int) GroupFileSchema::RAW or where(c(&GroupFileSchema::group_id) == group_id and
c(&GroupFileSchema::file_type) == (int) GroupFileSchema::INDEX))); (c(&GroupFileSchema::file_type) == (int) GroupFileSchema::RAW or
c(&GroupFileSchema::file_type) == (int) GroupFileSchema::INDEX)));
GroupSchema group_info;
group_info.group_id = group_id; GroupSchema group_info;
auto status = get_group_no_lock(group_info); group_info.group_id = group_id;
if (!status.ok()) { auto status = get_group_no_lock(group_info);
return status; if (!status.ok()) {
} return status;
}
for (auto& file : selected) { for (auto& file : selected) {
GroupFileSchema group_file; GroupFileSchema group_file;
group_file.id = std::get<0>(file); group_file.id = std::get<0>(file);
group_file.group_id = std::get<1>(file); group_file.group_id = std::get<1>(file);
group_file.file_id = std::get<2>(file); group_file.file_id = std::get<2>(file);
group_file.file_type = std::get<3>(file); group_file.file_type = std::get<3>(file);
group_file.rows = std::get<4>(file); group_file.rows = std::get<4>(file);
group_file.date = std::get<5>(file); group_file.date = std::get<5>(file);
group_file.dimension = group_info.dimension; group_file.dimension = group_info.dimension;
GetGroupFilePath(group_file); GetGroupFilePath(group_file);
auto dateItr = files.find(group_file.date); auto dateItr = files.find(group_file.date);
if (dateItr == files.end()) { if (dateItr == files.end()) {
files[group_file.date] = GroupFilesSchema(); files[group_file.date] = GroupFilesSchema();
}
files[group_file.date].push_back(group_file);
} }
files[group_file.date].push_back(group_file); } catch (std::exception & e) {
LOG(DEBUG) << e.what();
throw e;
} }
return Status::OK(); return Status::OK();
...@@ -282,37 +303,42 @@ Status DBMetaImpl::files_to_merge(const std::string& group_id, ...@@ -282,37 +303,42 @@ Status DBMetaImpl::files_to_merge(const std::string& group_id,
DatePartionedGroupFilesSchema& files) { DatePartionedGroupFilesSchema& files) {
files.clear(); files.clear();
auto selected = ConnectorPtr->select(columns(&GroupFileSchema::id, try {
&GroupFileSchema::group_id, auto selected = ConnectorPtr->select(columns(&GroupFileSchema::id,
&GroupFileSchema::file_id, &GroupFileSchema::group_id,
&GroupFileSchema::file_type, &GroupFileSchema::file_id,
&GroupFileSchema::rows, &GroupFileSchema::file_type,
&GroupFileSchema::date), &GroupFileSchema::rows,
where(c(&GroupFileSchema::file_type) == (int)GroupFileSchema::RAW and &GroupFileSchema::date),
c(&GroupFileSchema::group_id) == group_id)); where(c(&GroupFileSchema::file_type) == (int)GroupFileSchema::RAW and
c(&GroupFileSchema::group_id) == group_id));
GroupSchema group_info;
group_info.group_id = group_id; GroupSchema group_info;
auto status = get_group_no_lock(group_info); group_info.group_id = group_id;
if (!status.ok()) { auto status = get_group_no_lock(group_info);
return status; if (!status.ok()) {
} return status;
}
for (auto& file : selected) { for (auto& file : selected) {
GroupFileSchema group_file; GroupFileSchema group_file;
group_file.id = std::get<0>(file); group_file.id = std::get<0>(file);
group_file.group_id = std::get<1>(file); group_file.group_id = std::get<1>(file);
group_file.file_id = std::get<2>(file); group_file.file_id = std::get<2>(file);
group_file.file_type = std::get<3>(file); group_file.file_type = std::get<3>(file);
group_file.rows = std::get<4>(file); group_file.rows = std::get<4>(file);
group_file.date = std::get<5>(file); group_file.date = std::get<5>(file);
group_file.dimension = group_info.dimension; group_file.dimension = group_info.dimension;
GetGroupFilePath(group_file); GetGroupFilePath(group_file);
auto dateItr = files.find(group_file.date); auto dateItr = files.find(group_file.date);
if (dateItr == files.end()) { if (dateItr == files.end()) {
files[group_file.date] = GroupFilesSchema(); files[group_file.date] = GroupFilesSchema();
}
files[group_file.date].push_back(group_file);
} }
files[group_file.date].push_back(group_file); } catch (std::exception & e) {
LOG(DEBUG) << e.what();
throw e;
} }
return Status::OK(); return Status::OK();
...@@ -341,88 +367,109 @@ Status DBMetaImpl::get_group_files(const std::string& group_id_, ...@@ -341,88 +367,109 @@ Status DBMetaImpl::get_group_files(const std::string& group_id_,
Status DBMetaImpl::update_group_file(GroupFileSchema& group_file) { Status DBMetaImpl::update_group_file(GroupFileSchema& group_file) {
group_file.updated_time = GetMicroSecTimeStamp(); group_file.updated_time = GetMicroSecTimeStamp();
auto commited = ConnectorPtr->transaction([&] () mutable { try {
ConnectorPtr->update(group_file); auto commited = ConnectorPtr->transaction([&] () mutable {
return true; ConnectorPtr->update(group_file);
}); return true;
if (!commited) { });
return Status::DBTransactionError("Update file Error"); if (!commited) {
return Status::DBTransactionError("Update file Error");
}
} catch (std::exception & e) {
LOG(DEBUG) << e.what();
LOG(DEBUG) << "id= " << group_file.id << " file_id=" << group_file.file_id;
throw e;
} }
return Status::OK(); return Status::OK();
} }
Status DBMetaImpl::update_files(GroupFilesSchema& files) { Status DBMetaImpl::update_files(GroupFilesSchema& files) {
auto commited = ConnectorPtr->transaction([&] () mutable { try {
for (auto& file : files) { auto commited = ConnectorPtr->transaction([&] () mutable {
file.updated_time = GetMicroSecTimeStamp(); for (auto& file : files) {
ConnectorPtr->update(file); file.updated_time = GetMicroSecTimeStamp();
ConnectorPtr->update(file);
}
return true;
});
if (!commited) {
return Status::DBTransactionError("Update files Error");
} }
return true; } catch (std::exception & e) {
}); LOG(DEBUG) << e.what();
if (!commited) { throw e;
return Status::DBTransactionError("Update files Error");
} }
return Status::OK(); return Status::OK();
} }
Status DBMetaImpl::cleanup_ttl_files(uint16_t seconds) { Status DBMetaImpl::cleanup_ttl_files(uint16_t seconds) {
auto now = GetMicroSecTimeStamp(); auto now = GetMicroSecTimeStamp();
auto selected = ConnectorPtr->select(columns(&GroupFileSchema::id, try {
&GroupFileSchema::group_id, auto selected = ConnectorPtr->select(columns(&GroupFileSchema::id,
&GroupFileSchema::file_id, &GroupFileSchema::group_id,
&GroupFileSchema::file_type, &GroupFileSchema::file_id,
&GroupFileSchema::rows, &GroupFileSchema::file_type,
&GroupFileSchema::date), &GroupFileSchema::rows,
where(c(&GroupFileSchema::file_type) == (int)GroupFileSchema::TO_DELETE and &GroupFileSchema::date),
c(&GroupFileSchema::updated_time) > now - 1000000*seconds)); where(c(&GroupFileSchema::file_type) == (int)GroupFileSchema::TO_DELETE and
c(&GroupFileSchema::updated_time) > now - 1000000*seconds));
GroupFilesSchema updated;
GroupFilesSchema updated;
for (auto& file : selected) {
GroupFileSchema group_file; for (auto& file : selected) {
group_file.id = std::get<0>(file); GroupFileSchema group_file;
group_file.group_id = std::get<1>(file); group_file.id = std::get<0>(file);
group_file.file_id = std::get<2>(file); group_file.group_id = std::get<1>(file);
group_file.file_type = std::get<3>(file); group_file.file_id = std::get<2>(file);
group_file.rows = std::get<4>(file); group_file.file_type = std::get<3>(file);
group_file.date = std::get<5>(file); group_file.rows = std::get<4>(file);
GetGroupFilePath(group_file); group_file.date = std::get<5>(file);
if (group_file.file_type == GroupFileSchema::TO_DELETE) { GetGroupFilePath(group_file);
boost::filesystem::remove(group_file.location); if (group_file.file_type == GroupFileSchema::TO_DELETE) {
boost::filesystem::remove(group_file.location);
}
ConnectorPtr->remove<GroupFileSchema>(group_file.id);
LOG(DEBUG) << "Removing deleted id=" << group_file.id << " location=" << group_file.location << std::endl;
} }
ConnectorPtr->remove<GroupFileSchema>(group_file.id); } catch (std::exception & e) {
LOG(DEBUG) << "Removing deleted id=" << group_file.id << " location=" << group_file.location << std::endl; LOG(DEBUG) << e.what();
throw e;
} }
return Status::OK(); return Status::OK();
} }
Status DBMetaImpl::cleanup() { Status DBMetaImpl::cleanup() {
auto selected = ConnectorPtr->select(columns(&GroupFileSchema::id, try {
&GroupFileSchema::group_id, auto selected = ConnectorPtr->select(columns(&GroupFileSchema::id,
&GroupFileSchema::file_id, &GroupFileSchema::group_id,
&GroupFileSchema::file_type, &GroupFileSchema::file_id,
&GroupFileSchema::rows, &GroupFileSchema::file_type,
&GroupFileSchema::date), &GroupFileSchema::rows,
where(c(&GroupFileSchema::file_type) == (int)GroupFileSchema::TO_DELETE or &GroupFileSchema::date),
c(&GroupFileSchema::file_type) == (int)GroupFileSchema::NEW)); where(c(&GroupFileSchema::file_type) == (int)GroupFileSchema::TO_DELETE or
c(&GroupFileSchema::file_type) == (int)GroupFileSchema::NEW));
GroupFilesSchema updated;
GroupFilesSchema updated;
for (auto& file : selected) {
GroupFileSchema group_file; for (auto& file : selected) {
group_file.id = std::get<0>(file); GroupFileSchema group_file;
group_file.group_id = std::get<1>(file); group_file.id = std::get<0>(file);
group_file.file_id = std::get<2>(file); group_file.group_id = std::get<1>(file);
group_file.file_type = std::get<3>(file); group_file.file_id = std::get<2>(file);
group_file.rows = std::get<4>(file); group_file.file_type = std::get<3>(file);
group_file.date = std::get<5>(file); group_file.rows = std::get<4>(file);
GetGroupFilePath(group_file); group_file.date = std::get<5>(file);
if (group_file.file_type == GroupFileSchema::TO_DELETE) { GetGroupFilePath(group_file);
boost::filesystem::remove(group_file.location); if (group_file.file_type == GroupFileSchema::TO_DELETE) {
boost::filesystem::remove(group_file.location);
}
ConnectorPtr->remove<GroupFileSchema>(group_file.id);
LOG(DEBUG) << "Removing id=" << group_file.id << " location=" << group_file.location << std::endl;
} }
ConnectorPtr->remove<GroupFileSchema>(group_file.id); } catch (std::exception & e) {
LOG(DEBUG) << "Removing id=" << group_file.id << " location=" << group_file.location << std::endl; LOG(DEBUG) << e.what();
throw e;
} }
return Status::OK(); return Status::OK();
...@@ -430,26 +477,32 @@ Status DBMetaImpl::cleanup() { ...@@ -430,26 +477,32 @@ Status DBMetaImpl::cleanup() {
Status DBMetaImpl::count(const std::string& group_id, long& result) { Status DBMetaImpl::count(const std::string& group_id, long& result) {
auto selected = ConnectorPtr->select(columns(&GroupFileSchema::rows, try {
&GroupFileSchema::date), auto selected = ConnectorPtr->select(columns(&GroupFileSchema::rows,
where((c(&GroupFileSchema::file_type) == (int)GroupFileSchema::RAW or &GroupFileSchema::date),
c(&GroupFileSchema::file_type) == (int)GroupFileSchema::TO_INDEX or where((c(&GroupFileSchema::file_type) == (int)GroupFileSchema::RAW or
c(&GroupFileSchema::file_type) == (int)GroupFileSchema::INDEX) and c(&GroupFileSchema::file_type) == (int)GroupFileSchema::TO_INDEX or
c(&GroupFileSchema::group_id) == group_id)); c(&GroupFileSchema::file_type) == (int)GroupFileSchema::INDEX) and
c(&GroupFileSchema::group_id) == group_id));
GroupSchema group_info;
group_info.group_id = group_id;
auto status = get_group_no_lock(group_info);
if (!status.ok()) {
return status;
}
result = 0;
for (auto& file : selected) {
result += std::get<0>(file);
}
GroupSchema group_info; result /= group_info.dimension;
group_info.group_id = group_id;
auto status = get_group_no_lock(group_info);
if (!status.ok()) {
return status;
}
result = 0; } catch (std::exception & e) {
for (auto& file : selected) { LOG(DEBUG) << e.what();
result += std::get<0>(file); throw e;
} }
result /= group_info.dimension;
return Status::OK(); return Status::OK();
} }
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或