Skip to content

Commit

Permalink
fix: the result of select on the deleted index col is empty (4paradig…
Browse files Browse the repository at this point in the history 
…m#3426)
  • Loading branch information
@dl239
dl239 authored Aug 31, 2023
1 parent bc00d67 commit 6eff021
Show file tree
Hide file tree
Showing 4 changed files with 59 additions and 14 deletions.
48 changes: 40 additions & 8 deletions src/catalog/tablet_catalog.cc
View file
Original file line number Diff line number Diff line change
Expand Up @@ -92,6 +92,9 @@ bool TabletTableHandler::UpdateIndex(
index_hint_vec_[pos].clear();
for (int32_t i = 0; i < indexs.size(); i++) {
const auto& column_key = indexs.Get(i);
if (column_key.flag() != 0) {
continue;
}
::hybridse::vm::IndexSt index_st;
index_st.index = i;
index_st.ts_pos = ::hybridse::vm::INVALID_POS;
Expand Down Expand Up @@ -229,7 +232,9 @@ int TabletTableHandler::DeleteTable(uint32_t pid) {
return new_tables->size();
}

void TabletTableHandler::Update(const ::openmldb::nameserver::TableInfo& meta, const ClientManager& client_manager) {
bool TabletTableHandler::Update(const ::openmldb::nameserver::TableInfo& meta, const ClientManager& client_manager,
bool* index_updated) {
*index_updated = false;
::openmldb::storage::TableSt new_table_st(meta);
for (const auto& partition_st : *(new_table_st.GetPartitions())) {
uint32_t pid = partition_st.GetPid();
Expand All @@ -238,10 +243,28 @@ void TabletTableHandler::Update(const ::openmldb::nameserver::TableInfo& meta, c
}
table_client_manager_->UpdatePartitionClientManager(partition_st, client_manager);
}
if (meta.column_key_size() != static_cast<int>(GetIndex().size())) {
LOG(INFO) << "index size changed, update index" << meta.column_key_size() << " " << GetIndex().size();
UpdateIndex(meta.column_key());
const auto& index_hint = GetIndex();
size_t index_cnt = 0;
bool has_new_index = false;
for (const auto& column_key : meta.column_key()) {
if (column_key.flag() != 0) {
continue;
}
index_cnt++;
if (index_hint.find(column_key.index_name()) == index_hint.end()) {
has_new_index = true;
break;
}
}
if (index_cnt != index_hint.size() || has_new_index) {
LOG(INFO) << "index size changed. current index size " << index_hint.size()
<< " , new index size " << index_cnt;
if (!UpdateIndex(meta.column_key())) {
return false;
}
*index_updated = true;
}
return true;
}

std::shared_ptr<::hybridse::vm::Tablet> TabletTableHandler::GetTablet(const std::string& index_name,
Expand Down Expand Up @@ -433,7 +456,8 @@ bool TabletCatalog::UpdateTableMeta(const ::openmldb::api::TableMeta& meta) {
return handler->UpdateIndex(meta.column_key());
}

bool TabletCatalog::UpdateTableInfo(const ::openmldb::nameserver::TableInfo& table_info) {
bool TabletCatalog::UpdateTableInfo(const ::openmldb::nameserver::TableInfo& table_info, bool* index_updated) {
*index_updated = false;
const std::string& db_name = table_info.db();
const std::string& table_name = table_info.name();
std::shared_ptr<TabletTableHandler> handler;
Expand All @@ -456,22 +480,29 @@ bool TabletCatalog::UpdateTableInfo(const ::openmldb::nameserver::TableInfo& tab
} else {
handler = it->second;
}
handler->Update(table_info, client_manager_);
if (bool updated = false; !handler->Update(table_info, client_manager_, &updated)) {
return false;
} else if (updated) {
*index_updated = true;
}
}
return true;
}

void TabletCatalog::Refresh(const std::vector<::openmldb::nameserver::TableInfo>& table_info_vec, uint64_t version,
const Procedures& db_sp_map) {
const Procedures& db_sp_map, bool* updated) {
*updated = false;
std::map<std::string, std::set<std::string>> table_map;
for (const auto& table_info : table_info_vec) {
const std::string& db_name = table_info.db();
const std::string& table_name = table_info.name();
if (db_name.empty()) {
continue;
}
if (!UpdateTableInfo(table_info)) {
if (bool index_updated = false; !UpdateTableInfo(table_info, &index_updated)) {
continue;
} else if (index_updated) {
*updated = true;
}
auto cur_db_it = table_map.find(db_name);
if (cur_db_it == table_map.end()) {
Expand All @@ -494,6 +525,7 @@ void TabletCatalog::Refresh(const std::vector<::openmldb::nameserver::TableInfo>
!table_it->second->HasLocalTable()) {
LOG(INFO) << "delete table from catalog. db: " << db_it->first << ", table: " << table_it->first;
table_it = db_it->second.erase(table_it);
*updated = true;
continue;
}
++table_it;
Expand Down
7 changes: 4 additions & 3 deletions src/catalog/tablet_catalog.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -182,7 +182,8 @@ class TabletTableHandler : public ::hybridse::vm::TableHandler,

int DeleteTable(uint32_t pid);

void Update(const ::openmldb::nameserver::TableInfo &meta, const ClientManager &client_manager);
bool Update(const ::openmldb::nameserver::TableInfo &meta, const ClientManager &client_manager,
bool* index_updated);

private:
inline int32_t GetColumnIndex(const std::string &column) {
Expand Down Expand Up @@ -223,7 +224,7 @@ class TabletCatalog : public ::hybridse::vm::Catalog {

bool UpdateTableMeta(const ::openmldb::api::TableMeta &meta);

bool UpdateTableInfo(const ::openmldb::nameserver::TableInfo& table_info);
bool UpdateTableInfo(const ::openmldb::nameserver::TableInfo& table_info, bool* index_updated);

std::shared_ptr<::hybridse::type::Database> GetDatabase(const std::string &db) override;

Expand All @@ -237,7 +238,7 @@ class TabletCatalog : public ::hybridse::vm::Catalog {
bool DeleteDB(const std::string &db);

void Refresh(const std::vector<::openmldb::nameserver::TableInfo> &table_info_vec, uint64_t version,
const Procedures &db_sp_map);
const Procedures &db_sp_map, bool* updated);

bool AddProcedure(const std::string &db, const std::string &sp_name,
const std::shared_ptr<hybridse::sdk::ProcedureInfo> &sp_info);
Expand Down
5 changes: 4 additions & 1 deletion src/cmd/sql_cmd_test.cc
View file
Original file line number Diff line number Diff line change
Expand Up @@ -3681,10 +3681,13 @@ TEST_F(SqlCmdTest, SelectWithAddNewIndex) {
ASSERT_EQ(res->Size(), 3);
res = sr->ExecuteSQL(absl::StrCat("select id,c1,c2,c3 from ", tb1_name, " where c2=1;"), &status);
ASSERT_EQ(res->Size(), 3);
ProcessSQLs(sr, {absl::StrCat("drop index ", db1_name, ".", tb1_name, ".index1")});
absl::SleepFor(absl::Seconds(2));
res = sr->ExecuteSQL(absl::StrCat("select id,c1,c2,c3 from ", tb1_name, " where c2=1;"), &status);
ASSERT_EQ(res->Size(), 3);

ProcessSQLs(sr, {
absl::StrCat("use ", db1_name, ";"),
absl::StrCat("drop index ", db1_name, ".", tb1_name, ".index1"),
absl::StrCat("drop table ", tb1_name),
absl::StrCat("drop database ", db1_name),
});
Expand Down
13 changes: 11 additions & 2 deletions src/tablet/tablet_impl.cc
View file
Original file line number Diff line number Diff line change
Expand Up @@ -4247,7 +4247,12 @@ bool TabletImpl::RefreshSingleTable(uint32_t tid) {
LOG(WARNING) << "fail to parse table proto. tid: " << tid << " value: " << value;
return false;
}
return catalog_->UpdateTableInfo(table_info);
if (bool index_updated = false; !catalog_->UpdateTableInfo(table_info, &index_updated)) {
return false;
} else if (index_updated) {
engine_->ClearCacheLocked("");
}
return true;
}

void TabletImpl::UpdateGlobalVarTable() {
Expand Down Expand Up @@ -4369,7 +4374,11 @@ void TabletImpl::RefreshTableInfo() {
}
}
auto old_db_sp_map = catalog_->GetProcedures();
catalog_->Refresh(table_info_vec, version, db_sp_map);
bool updated = false;
catalog_->Refresh(table_info_vec, version, db_sp_map, &updated);
if (updated) {
engine_->ClearCacheLocked("");
}
// skip exist procedure, don`t need recompile
for (const auto& db_sp_map_kv : db_sp_map) {
const auto& db = db_sp_map_kv.first;
Expand Down

0 comments on commit 6eff021

Please sign in to comment.