xiaoxue_admin
/
zvec
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix(index): snapshot read-only indexes without flush (#569)

This commit is contained in:
egolearner 2026-07-10 10:55:50 +08:00 committed by GitHub
parent 7449d7193b
commit 821a9333fc
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 154 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
src/db/common/rocksdb_context.h
View File

@ -120,6 +120,11 @@ struct RocksdbContext {
size_t count();
bool read_only() const {
return read_only_;
}
// Create a Rocksdb instance from Args
Status create(Args args);
@ -142,4 +147,4 @@ struct RocksdbContext {
};
} // namespace zvec
} // namespace zvec

12
src/db/index/column/fts_column/fts_indexer.cc
View File

@ -143,12 +143,14 @@ Status FtsIndexer::close() {
}
Status FtsIndexer::create_snapshot(const std::string &snapshot_path) {
auto s = flush();
if (!s.ok()) {
LOG_ERROR("FtsIndexer: flush failed during snapshot");
return s;
if (fts_ctx_ && !fts_ctx_->read_only()) {
auto s = flush();
if (!s.ok()) {
LOG_ERROR("FtsIndexer: flush failed during snapshot");
return s;
}
}
s = fts_ctx_->create_checkpoint(snapshot_path);
auto s = fts_ctx_->create_checkpoint(snapshot_path);
if (!s.ok()) {
LOG_ERROR("FtsIndexer: create_checkpoint to [%s] failed: %s",
snapshot_path.c_str(), s.message().c_str());

10
src/db/index/column/inverted_column/inverted_indexer.cc
View File

@ -107,9 +107,11 @@ Status InvertedIndexer::flush() {
Status InvertedIndexer::create_snapshot(const std::string &snapshot_dir) {
Status s;
if (s = flush(); !s.ok()) {
LOG_ERROR("Failed to flush %s during creating a snapshot", ID().c_str());
return s;
if (!rocksdb_context_.read_only()) {
if (s = flush(); !s.ok()) {
LOG_ERROR("Failed to flush %s during creating a snapshot", ID().c_str());
return s;
}
}
if (s = rocksdb_context_.create_checkpoint(snapshot_dir); s.ok()) {
@ -264,4 +266,4 @@ Status InvertedIndexer::remove_column_indexer(const std::string &field_name) {
}
} // namespace zvec
} // namespace zvec

3
src/db/index/segment/segment.cc
View File

@ -2133,9 +2133,10 @@ Status SegmentImpl::drop_scalar_index(const std::vector<std::string> &columns,
auto s = invert_indexers_->create_snapshot(new_invert_index_path);
CHECK_RETURN_STATUS(s);
// The snapshot copy is mutated below to remove dropped columns and seal.
auto new_scalar_indexer = InvertedIndexer::CreateAndOpen(
collection_schema_->name(), new_invert_index_path, false, invert_fields,
options_.read_only_);
false);
if (!new_scalar_indexer) {
LOG_ERROR("Failed to create scalar indexer");
return Status::InternalError("Failed to create scalar indexer");

139
tests/db/collection_test.cc
View File

@ -2544,6 +2544,60 @@ TEST_F(CollectionTest, Feature_DropIndex_Scalar) {
func("int32", true);
func("int32", false);
{
FileHelper::RemoveDirectory(col_path);
int doc_count = 100;
auto schema = TestHelper::CreateSchemaWithScalarIndex(false, true);
auto options = CollectionOptions{false, true, 64 * 1024 * 1024};
auto collection = TestHelper::CreateCollectionWithDoc(
col_path, *schema, options, 0, doc_count, false);
ASSERT_TRUE(collection->Optimize().ok());
collection.reset();
auto reopen_result = Collection::Open(col_path, options);
ASSERT_TRUE(reopen_result.has_value()) << reopen_result.error().message();
collection = std::move(reopen_result.value());
auto s = collection->DropIndex("int32");
ASSERT_TRUE(s.ok()) << s.message();
auto expected_schema = std::make_shared<CollectionSchema>(*schema);
s = expected_schema->drop_index("int32");
ASSERT_TRUE(s.ok()) << s.message();
auto schema_after_drop = collection->Schema();
ASSERT_TRUE(schema_after_drop.has_value())
<< schema_after_drop.error().message();
ASSERT_EQ(*expected_schema, schema_after_drop.value());
collection.reset();
reopen_result = Collection::Open(col_path, options);
ASSERT_TRUE(reopen_result.has_value()) << reopen_result.error().message();
collection = std::move(reopen_result.value());
schema_after_drop = collection->Schema();
ASSERT_TRUE(schema_after_drop.has_value())
<< schema_after_drop.error().message();
ASSERT_EQ(*expected_schema, schema_after_drop.value());
ASSERT_EQ(collection->Stats().value().doc_count, doc_count);
for (int i = 0; i < doc_count; i++) {
auto expect_doc = TestHelper::CreateDoc(i, *schema);
auto result = collection->Fetch({expect_doc.pk()});
ASSERT_TRUE(result.has_value());
ASSERT_EQ(result.value().size(), 1);
ASSERT_EQ(result.value().count(expect_doc.pk()), 1);
auto doc = result.value()[expect_doc.pk()];
ASSERT_NE(doc, nullptr);
ASSERT_EQ(*doc, expect_doc);
}
collection.reset();
FileHelper::RemoveDirectory(col_path);
}
}
TEST_F(CollectionTest, Feature_DropIndex_AfterCreate) {
@ -6078,7 +6132,7 @@ TEST_F(CollectionTest, Feature_CreateOrDropFtsIndex) {
col.reset();
auto reopen_res = Collection::Open(col_path, options);
ASSERT_TRUE(reopen_res.has_value()) << reopen_res.error().message();
col = reopen_res.value();
col = std::move(reopen_res.value());
auto q_reopen = fts_search(col, "hello");
ASSERT_TRUE(q_reopen.has_value()) << q_reopen.error().message();
@ -6120,7 +6174,7 @@ TEST_F(CollectionTest, Feature_CreateOrDropFtsIndex) {
col.reset();
auto reopen_res = Collection::Open(col_path, options);
ASSERT_TRUE(reopen_res.has_value()) << reopen_res.error().message();
col = reopen_res.value();
col = std::move(reopen_res.value());
auto q_reopen = fts_search(col, "hello");
ASSERT_FALSE(q_reopen.has_value());
@ -6129,7 +6183,80 @@ TEST_F(CollectionTest, Feature_CreateOrDropFtsIndex) {
FileHelper::RemoveDirectory(col_path);
}
// Case 3: Create → Drop → Create → Drop cycle on the same column.
// Case 3: Drop one FTS index from a reopened optimized collection while
// another FTS index remains.
{
FileHelper::RemoveDirectory(col_path);
auto schema = std::make_shared<CollectionSchema>("fts_drop_reopen");
schema->add_field(std::make_shared<FieldSchema>("title", DataType::STRING));
schema->add_field(
std::make_shared<FieldSchema>("content", DataType::STRING, false,
std::make_shared<FtsIndexParams>()));
schema->add_field(
std::make_shared<FieldSchema>("other_content", DataType::STRING, false,
std::make_shared<FtsIndexParams>()));
schema->add_field(std::make_shared<FieldSchema>(
"vec", DataType::VECTOR_FP32, 4, false,
std::make_shared<FlatIndexParams>(MetricType::IP)));
CollectionOptions options{false, true};
auto col_res = Collection::CreateAndOpen(col_path, *schema, options);
ASSERT_TRUE(col_res.has_value()) << col_res.error().message();
auto col = std::move(col_res.value());
std::vector<Doc> docs;
for (uint64_t i = 0; i < 20; i++) {
Doc d;
d.set_pk("pk_" + std::to_string(i));
d.set<std::string>("title", "title_" + std::to_string(i));
d.set<std::string>("content", "hello content " + std::to_string(i));
d.set<std::string>("other_content", "hello other " + std::to_string(i));
d.set<std::vector<float>>("vec", std::vector<float>(4, float(i) + 0.1f));
docs.push_back(d);
}
ASSERT_TRUE(col->Insert(docs).has_value());
ASSERT_TRUE(col->Optimize().ok());
col.reset();
auto reopen_res = Collection::Open(col_path, options);
ASSERT_TRUE(reopen_res.has_value()) << reopen_res.error().message();
col = std::move(reopen_res.value());
auto s = col->DropIndex("content");
ASSERT_TRUE(s.ok()) << s.message();
auto schema_after_drop = col->Schema();
ASSERT_TRUE(schema_after_drop.has_value())
<< schema_after_drop.error().message();
ASSERT_EQ(schema_after_drop.value().get_field("content")->index_params(),
nullptr);
ASSERT_NE(
schema_after_drop.value().get_field("other_content")->index_params(),
nullptr);
col.reset();
reopen_res = Collection::Open(col_path, options);
ASSERT_TRUE(reopen_res.has_value()) << reopen_res.error().message();
col = std::move(reopen_res.value());
schema_after_drop = col->Schema();
ASSERT_TRUE(schema_after_drop.has_value())
<< schema_after_drop.error().message();
ASSERT_EQ(schema_after_drop.value().get_field("content")->index_params(),
nullptr);
ASSERT_NE(
schema_after_drop.value().get_field("other_content")->index_params(),
nullptr);
ASSERT_EQ(col->Stats().value().doc_count, 20u);
auto fetched = col->Fetch({"pk_0", "pk_19"});
ASSERT_TRUE(fetched.has_value()) << fetched.error().message();
ASSERT_EQ(fetched.value().size(), 2u);
col.reset();
FileHelper::RemoveDirectory(col_path);
}
// Case 4: Create → Drop → Create → Drop cycle on the same column.
{
FileHelper::RemoveDirectory(col_path);
auto schema = build_schema(false);
@ -6175,7 +6302,7 @@ TEST_F(CollectionTest, Feature_CreateOrDropFtsIndex) {
col.reset();
auto reopen_res = Collection::Open(col_path, options);
ASSERT_TRUE(reopen_res.has_value()) << reopen_res.error().message();
col = reopen_res.value();
col = std::move(reopen_res.value());
q = fts_search(col, "hello");
ASSERT_FALSE(q.has_value());
@ -6184,7 +6311,7 @@ TEST_F(CollectionTest, Feature_CreateOrDropFtsIndex) {
FileHelper::RemoveDirectory(col_path);
}
// Case 4: CreateIndex with different FtsIndexParams on a column that already
// Case 5: CreateIndex with different FtsIndexParams on a column that already
// has an FTS index — should remove the old index and rebuild with new params.
{
FileHelper::RemoveDirectory(col_path);
@ -6231,7 +6358,7 @@ TEST_F(CollectionTest, Feature_CreateOrDropFtsIndex) {
col.reset();
auto reopen_res = Collection::Open(col_path, options);
ASSERT_TRUE(reopen_res.has_value()) << reopen_res.error().message();
col = reopen_res.value();
col = std::move(reopen_res.value());
q = fts_search(col, "hello");
ASSERT_TRUE(q.has_value()) << q.error().message();