furyhawk/milvus
1
0
Fork 0
mirror of https://github.com/milvus-io/milvus.git synced 2026-07-21 10:15:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: cap recovered growing manifest rows (#50952)

Browse Source 
## Summary

issue: #50910
issue: #50945

Growing StorageV3 recovery could load all rows from the manifest even
when `SegmentLoadInfo.num_of_rows` represented an earlier checkpoint.
That allowed the growing text-match index to advance beyond the
checkpoint row count; the next WAL replay insert then reserved the
checkpoint offset and Tantivy rejected the lower doc id.

This PR:
- Caps recovered column-group rows to the segment checkpoint row count
before raw fields, PKs, and text indexes are populated.
- Adds `Growing.LoadStorageV3ManifestCapsRowsAtCheckpoint`, which writes
a 6-row StorageV3 manifest, loads it with a 4-row checkpoint, then
inserts the next row at offset 4.

## Verification

- `clang-format-12` on changed C++ files
- `git diff --check`
- Fresh C++ configure in `/tmp/milvus-50910-build` was blocked before
compilation because generated `internal/core/src/pb` sources were
missing: `No SOURCES given to target: milvus_pb`

---------

Signed-off-by: Buqian Zheng <zhengbuqian@gmail.com>
This commit is contained in:
Buqian Zheng
2026-07-03 11:02:29 +08:00
committed by GitHub
parent 331cf54bef
commit cd8a943b9c
4 changed files with 296 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/core/src/segcore/SegmentGrowingImpl.cpp
+141 -11
View File
@@ -28,6 +28,7 @@
#include <type_traits>
#include <unordered_map>
#include <variant>
#include <vector>
#include "segcore/default_fs.h"
#include "NamedType/named_type_impl.hpp"
@@ -94,6 +95,75 @@ using namespace milvus::cachinglayer;
namespace {
int64_t
GetLoadedFieldRows(
const std::vector<std::unordered_map<FieldId, std::vector<FieldDataPtr>>>&
column_group_results,
FieldId field_id) {
int64_t rows = 0;
bool found = false;
for (const auto& column_group_result : column_group_results) {
auto it = column_group_result.find(field_id);
if (it == column_group_result.end()) {
continue;
}
found = true;
for (const auto& field_data : it->second) {
rows += field_data->get_num_rows();
}
}
return found ? rows : -1;
}
void
AssertLoadedFieldRows(
const std::vector<std::unordered_map<FieldId, std::vector<FieldDataPtr>>>&
column_group_results,
FieldId field_id,
int64_t expected_rows,
const char* field_name) {
if (expected_rows == 0) {
return;
}
auto rows = GetLoadedFieldRows(column_group_results, field_id);
AssertInfo(rows == expected_rows,
"growing segment StorageV3 manifest loads {} rows for {} "
"field {}, but SegmentLoadInfo expects {} rows",
rows,
field_name,
field_id.get(),
expected_rows);
}
void
AssertAllLoadedFieldRows(
const std::vector<std::unordered_map<FieldId, std::vector<FieldDataPtr>>>&
column_group_results,
int64_t expected_rows) {
if (expected_rows == 0) {
return;
}
std::unordered_map<FieldId, int64_t> loaded_rows;
for (const auto& column_group_result : column_group_results) {
for (const auto& [field_id, field_data] : column_group_result) {
auto& rows = loaded_rows[field_id];
for (const auto& data : field_data) {
rows += data->get_num_rows();
}
}
}
for (const auto& [field_id, rows] : loaded_rows) {
AssertInfo(rows == expected_rows,
"growing segment StorageV3 manifest loads {} rows for "
"field {}, but SegmentLoadInfo expects {} rows",
rows,
field_id.get(),
expected_rows);
}
}
int32_t
GetVectorArrayLength(const proto::schema::VectorField& vec_field,
DataType element_type,
@@ -2620,9 +2690,10 @@ SegmentGrowingImpl::LoadColumnsGroups(std::string manifest_path) {
std::future<std::unordered_map<FieldId, std::vector<FieldDataPtr>>>>
load_group_futures;
for (int64_t i = 0; i < column_groups->size(); ++i) {
auto future = pool.Submit([this, column_groups, properties, i] {
return LoadColumnGroup(column_groups, properties, i);
});
auto future =
pool.Submit([this, column_groups, properties, i, num_rows] {
return LoadColumnGroup(column_groups, properties, i, num_rows);
});
load_group_futures.emplace_back(std::move(future));
}
@@ -2648,9 +2719,31 @@ SegmentGrowingImpl::LoadColumnsGroups(std::string manifest_path) {
std::rethrow_exception(load_exceptions[0]);
}
AssertInfo(primary_field_id.get() != INVALID_FIELD_ID, "Primary key is -1");
AssertLoadedFieldRows(
column_group_results, primary_field_id, num_rows, "primary");
AssertAllLoadedFieldRows(column_group_results, num_rows);
auto timestamp_rows =
GetLoadedFieldRows(column_group_results, TimestampFieldID);
auto row_id_rows = GetLoadedFieldRows(column_group_results, RowFieldID);
auto reserved_offset = PreInsert(num_rows);
text_loaded_row_count_ = num_rows;
if (timestamp_rows == -1 && num_rows > 0) {
std::vector<Timestamp> timestamps(num_rows, 0);
insert_record_.timestamps_.set_data_raw(
reserved_offset, timestamps.data(), num_rows);
stats_.mem_size += num_rows * sizeof(Timestamp);
}
if (row_id_rows == -1 && num_rows > 0) {
std::vector<int64_t> row_ids(num_rows);
std::iota(row_ids.begin(), row_ids.end(), reserved_offset);
insert_record_.row_ids_.set_data_raw(
reserved_offset, row_ids.data(), num_rows);
stats_.mem_size += num_rows * sizeof(int64_t);
}
for (auto& column_group_result : column_group_results) {
for (auto& [field_id, field_data] : column_group_result) {
load_field_data_common(field_id,
@@ -2675,7 +2768,8 @@ std::unordered_map<FieldId, std::vector<FieldDataPtr>>
SegmentGrowingImpl::LoadColumnGroup(
const std::shared_ptr<milvus_storage::api::ColumnGroups>& column_groups,
const std::shared_ptr<milvus_storage::api::Properties>& properties,
int64_t index) {
int64_t index,
int64_t row_limit) {
AssertInfo(index < column_groups->size(),
"load column group index out of range");
auto column_group = column_groups->at(index);
@@ -2694,14 +2788,38 @@ SegmentGrowingImpl::LoadColumnGroup(
auto parallel_degree =
static_cast<uint64_t>(DEFAULT_FIELD_MAX_MEMORY_LIMIT / FILE_SLICE_SIZE);
std::vector<int64_t> all_row_groups(chunk_reader->total_number_of_chunks());
AssertInfo(row_limit >= 0,
"load column group row limit should be non-negative, got {}",
row_limit);
auto chunk_rows_result = chunk_reader->get_chunk_rows();
AssertInfo(chunk_rows_result.ok(),
"get chunk rows failed, segment {}, column group index {}, "
"error: {}",
get_segment_id(),
index,
chunk_rows_result.status().ToString());
std::iota(all_row_groups.begin(), all_row_groups.end(), 0);
auto chunk_rows = std::move(chunk_rows_result).ValueOrDie();
std::vector<int64_t> row_groups_to_load;
row_groups_to_load.reserve(chunk_rows.size());
auto rows_to_read = int64_t{0};
for (auto chunk_index = size_t{0};
chunk_index < chunk_rows.size() && rows_to_read < row_limit;
++chunk_index) {
row_groups_to_load.push_back(static_cast<int64_t>(chunk_index));
rows_to_read += static_cast<int64_t>(chunk_rows[chunk_index]);
}
AssertInfo(rows_to_read >= row_limit,
"column group {} has fewer rows than checkpoint row count, "
"loaded rows {}, expected rows {}",
index,
rows_to_read,
row_limit);
// create parallel degree split strategy
auto strategy =
std::make_unique<ParallelDegreeSplitStrategy>(parallel_degree);
auto split_result = strategy->split(all_row_groups);
auto split_result = strategy->split(row_groups_to_load);
auto& thread_pool =
ThreadPools::GetThreadPool(milvus::ThreadPoolPriority::HIGH);
@@ -2721,10 +2839,18 @@ SegmentGrowingImpl::LoadColumnGroup(
}
std::unordered_map<FieldId, std::vector<FieldDataPtr>> field_data_map;
for (auto& future : part_futures) {
auto part_result = future.get();
// Growing recovery may reuse a manifest that already contains rows past the
// segment checkpoint. Only load rows covered by SegmentLoadInfo so WAL
// replay can append the remaining rows at the expected offsets.
int64_t loaded_rows = 0;
storage::ProcessFuturesInOrder(part_futures, [&](auto part_result) {
for (auto& record_batch : part_result) {
if (loaded_rows >= row_limit) {
break;
}
auto batch_num_rows = record_batch->num_rows();
auto rows_to_load =
std::min<int64_t>(batch_num_rows, row_limit - loaded_rows);
for (auto i = 0; i < record_batch->num_columns(); ++i) {
auto column = record_batch->column_name(i);
@@ -2741,13 +2867,17 @@ SegmentGrowingImpl::LoadColumnGroup(
!IsSparseFloatVectorDataType(data_type)
? field.get_dim()
: 1,
batch_num_rows);
rows_to_load);
auto array = record_batch->column(i);
if (rows_to_load < batch_num_rows) {
array = array->Slice(0, rows_to_load);
}
field_data->FillFieldData(array);
field_data_map[field_id].push_back(field_data);
}
loaded_rows += rows_to_load;
}
}
});
LOG_INFO("Finished loading segment {} column group {}", id_, index);
return field_data_map;
internal/core/src/segcore/SegmentGrowingImpl.h
+2 -1
View File
@@ -828,7 +828,8 @@ class SegmentGrowingImpl : public SegmentGrowing {
LoadColumnGroup(
const std::shared_ptr<milvus_storage::api::ColumnGroups>& column_groups,
const std::shared_ptr<milvus_storage::api::Properties>& properties,
int64_t index);
int64_t index,
int64_t row_limit);
void
InitializeArrayOffsets();
internal/core/src/segcore/SegmentGrowingTest.cpp
+139
View File
@@ -16,6 +16,7 @@
#include <algorithm>
#include <array>
#include <cstdint>
#include <filesystem>
#include <iostream>
#include <map>
#include <memory>
@@ -61,11 +62,42 @@
#include "segcore/SegmentGrowingImpl.h"
#include "segcore/Utils.h"
#include "test_utils/DataGen.h"
#include "test_utils/ManifestTestUtil.h"
#include "test_utils/storage_test_utils.h"
using namespace milvus::segcore;
using namespace milvus;
namespace {
void
AddStorageV3SystemFields(const SchemaPtr& schema) {
schema->AddField(
FieldName("RowID"), RowFieldID, DataType::INT64, false, std::nullopt);
schema->AddField(FieldName("Timestamp"),
TimestampFieldID,
DataType::INT64,
false,
std::nullopt);
}
bool
ManifestHasField(const milvus::test::V3SegmentTestData& test_data,
FieldId field_id) {
auto field_column = std::to_string(field_id.get());
auto column_groups = test_data.GetColumnGroups();
for (size_t i = 0; i < column_groups->size(); ++i) {
const auto& columns = column_groups->at(i)->columns;
if (std::find(columns.begin(), columns.end(), field_column) !=
columns.end()) {
return true;
}
}
return false;
}
} // namespace
TEST(Growing, DeleteCount) {
auto schema = std::make_shared<Schema>();
auto pk = schema->AddDebugField("pk", DataType::INT64);
@@ -138,6 +170,113 @@ TEST(Growing, RealCount) {
ASSERT_EQ(0, segment->get_real_count());
}
TEST(Growing, LoadStorageV3ManifestCapsRowsAtCheckpoint) {
auto schema = std::make_shared<Schema>();
AddStorageV3SystemFields(schema);
auto pk = schema->AddDebugField("pk", DataType::INT64);
constexpr int64_t dim = 4;
auto vec = schema->AddDebugField(
"vec", DataType::VECTOR_FLOAT, dim, knowhere::metric::L2);
std::map<std::string, std::string> analyzer_params;
auto text = schema->AddDebugVarcharField(FieldName("text"),
DataType::VARCHAR,
65535,
false,
true,
true,
analyzer_params,
std::nullopt);
schema->set_primary_field_id(pk);
auto base_path = (std::filesystem::path(TestLocalPath) /
"growing_recovery_checkpoint_row_cap")
.string();
std::filesystem::remove_all(base_path);
milvus::test::V3SegmentTestData test_data(
schema, 2, 3, dim, TestLocalPath, base_path);
ASSERT_EQ(test_data.NumColumnGroups(), 2);
ASSERT_TRUE(ManifestHasField(test_data, RowFieldID));
ASSERT_TRUE(ManifestHasField(test_data, TimestampFieldID));
constexpr int64_t checkpoint_rows = 4;
milvus::proto::segcore::SegmentLoadInfo load_info;
load_info.set_collectionid(1);
load_info.set_partitionid(2);
load_info.set_segmentid(3);
load_info.set_storageversion(STORAGE_V3);
load_info.set_num_of_rows(checkpoint_rows);
load_info.set_manifest_path(test_data.ManifestPathJson());
load_info.set_insert_channel("by-dev-rootcoord-dml_0_1v0");
auto segment =
CreateGrowingSegment(schema, empty_index_meta, load_info.segmentid());
segment->SetLoadInfo(load_info);
milvus::tracer::TraceContext trace_ctx;
segment->Load(trace_ctx, nullptr);
ASSERT_EQ(segment->get_row_count(), checkpoint_rows);
EXPECT_EQ(segment->get_real_count(), checkpoint_rows);
std::vector<int64_t> row_ids = {checkpoint_rows};
std::vector<Timestamp> timestamps = {100};
std::vector<int64_t> pks = {100000};
std::vector<std::string> texts = {"text after recovery checkpoint"};
std::vector<float> vectors(dim, 1.0F);
auto insert_data = std::make_unique<InsertRecordProto>();
insert_data->set_num_rows(1);
insert_data->mutable_fields_data()->AddAllocated(
CreateDataArrayFrom(pks.data(), nullptr, 1, (*schema)[pk]).release());
insert_data->mutable_fields_data()->AddAllocated(
CreateDataArrayFrom(texts.data(), nullptr, 1, (*schema)[text])
.release());
insert_data->mutable_fields_data()->AddAllocated(
CreateDataArrayFrom(vectors.data(), nullptr, 1, (*schema)[vec])
.release());
auto offset = segment->PreInsert(1);
ASSERT_EQ(offset, checkpoint_rows);
ASSERT_NO_THROW(segment->Insert(
offset, 1, row_ids.data(), timestamps.data(), insert_data.get()));
EXPECT_EQ(segment->get_row_count(), checkpoint_rows + 1);
std::filesystem::remove_all(base_path);
}
TEST(Growing, LoadStorageV3ManifestRejectsShortRequiredRows) {
auto schema = std::make_shared<Schema>();
AddStorageV3SystemFields(schema);
auto pk = schema->AddDebugField("pk", DataType::INT64);
schema->set_primary_field_id(pk);
auto base_path = (std::filesystem::path(TestLocalPath) /
"growing_recovery_short_required_rows")
.string();
std::filesystem::remove_all(base_path);
milvus::test::V3SegmentTestData test_data(
schema, 1, 2, 1, TestLocalPath, base_path);
ASSERT_TRUE(ManifestHasField(test_data, RowFieldID));
ASSERT_TRUE(ManifestHasField(test_data, TimestampFieldID));
milvus::proto::segcore::SegmentLoadInfo load_info;
load_info.set_collectionid(1);
load_info.set_partitionid(2);
load_info.set_segmentid(4);
load_info.set_storageversion(STORAGE_V3);
load_info.set_num_of_rows(test_data.TotalRows() + 1);
load_info.set_manifest_path(test_data.ManifestPathJson());
load_info.set_insert_channel("by-dev-rootcoord-dml_0_1v0");
auto segment =
CreateGrowingSegment(schema, empty_index_meta, load_info.segmentid());
segment->SetLoadInfo(load_info);
milvus::tracer::TraceContext trace_ctx;
ASSERT_ANY_THROW(segment->Load(trace_ctx, nullptr));
EXPECT_EQ(segment->get_row_count(), 0);
std::filesystem::remove_all(base_path);
}
TEST(Growing, InsertSkipsMissingFunctionOutputField) {
auto schema = std::make_shared<Schema>();
schema->set_schema_version(2);
internal/core/unittest/test_utils/ManifestTestUtil.h
+14 -2
View File
@@ -160,12 +160,12 @@ class V3SegmentTestData {
AssertInfo(commit_result.ok(),
"Failed to commit transaction: {}",
commit_result.status().ToString());
auto version = commit_result.ValueOrDie();
committed_version_ = commit_result.ValueOrDie();
// Read back from manifest to get the committed column groups
auto read_txn_result =
milvus_storage::api::transaction::Transaction::Open(
fs, base_path_, version);
fs, base_path_, committed_version_);
AssertInfo(read_txn_result.ok(),
"Failed to open read transaction: {}",
read_txn_result.status().ToString());
@@ -251,6 +251,17 @@ class V3SegmentTestData {
return base_path_;
}
int64_t
Version() const {
return committed_version_;
}
std::string
ManifestPathJson() const {
return "{\"base_path\":\"" + base_path_ +
"\",\"ver\":" + std::to_string(committed_version_) + "}";
}
private:
SchemaPtr schema_;
std::shared_ptr<arrow::Schema> loon_schema_;
@@ -258,6 +269,7 @@ class V3SegmentTestData {
std::string base_path_;
std::string root_path_;
int64_t total_rows_ = 0;
int64_t committed_version_ = 0;
};
} // namespace milvus::test