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

fix: warm no-index vector fields as indexes (#50836)

Browse Source 
issue: #50834

Fix no-index sealed vector raw data warmup selection. When a sealed
segment has no usable built or interim vector index, segcore now treats
raw vector field data as the vector search representation and resolves
its warmup from vector-index policy. Once a built or interim index
exists, raw vector data keeps normal vector-field warmup semantics.

Go load paths continue to pass field warmup unchanged; segcore resolves
the effective policy from current segment state.

---------

Signed-off-by: Shawn Wang <shawn.wang@zilliz.com>
This commit is contained in:
sparknack
2026-07-18 15:22:41 +08:00
committed by GitHub
parent ddf3e4d558
commit 28a7661a2e
10 changed files with 681 additions and 105 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/core/src/common/Schema.cpp
+5 -1
View File
@@ -550,7 +550,11 @@ Schema::WarmupPolicy(const FieldId& field_id,
return {true, it->second};
}
// Fallback to appropriate collection-level config based on field type
return CollectionWarmupPolicy(is_vector, is_index);
}
std::pair<bool, std::string>
Schema::CollectionWarmupPolicy(bool is_vector, bool is_index) const {
if (is_vector) {
if (is_index) {
return {warmup_vector_index_.has_value(),
internal/core/src/common/Schema.h
+3
View File
@@ -637,6 +637,9 @@ class Schema {
std::pair<bool, std::string>
WarmupPolicy(const FieldId& field, bool is_vector, bool is_index) const;
std::pair<bool, std::string>
CollectionWarmupPolicy(bool is_vector, bool is_index) const;
// True if the field carries FieldSchema::is_function_output.
bool
is_function_output(const FieldId& field_id) const {
internal/core/src/mmap/ChunkedColumnGroup.h
+14
View File
@@ -192,6 +192,13 @@ class ChunkedColumnGroup {
return memory_size;
}
#ifdef MILVUS_UNIT_TEST
CacheWarmupPolicy
TestCacheWarmupPolicy() const {
return slot_->meta()->cache_warmup_policy;
}
#endif
protected:
mutable std::shared_ptr<CacheSlot<GroupChunk>> slot_;
size_t num_chunks_{0};
@@ -297,6 +304,13 @@ class ProxyChunkColumn : public ChunkedColumnInterface {
return group_->memory_size();
}
#ifdef MILVUS_UNIT_TEST
CacheWarmupPolicy
TestCacheWarmupPolicy() const {
return group_->TestCacheWarmupPolicy();
}
#endif
int64_t
chunk_row_nums(int64_t chunk_id) const override {
return group_->GetNumRowsUntilChunk(chunk_id + 1) -
internal/core/src/segcore/ChunkedSegmentSealedImpl.cpp
+167 -102
View File
@@ -2102,9 +2102,9 @@ ChunkedSegmentSealedImpl::LoadColumnGroups(
eager_fields.push_back(field_id);
continue;
}
auto [has_warmup, warmup_str] = schema_snapshot->WarmupPolicy(
field_id, field_is_vector, /*is_index=*/false);
auto resolved = getCacheWarmupPolicy(has_warmup ? warmup_str : "",
auto warmup_str = resolve_field_data_warmup_policy(
field_id, segment_load_info, schema_snapshot);
auto resolved = getCacheWarmupPolicy(warmup_str,
field_is_vector,
/*is_index=*/false,
/*in_load_list=*/true);
@@ -2281,6 +2281,27 @@ ChunkedSegmentSealedImpl::SynthesizeExternalSystemFields(
namespace {
// A column group or field-binlog group may contain multiple fields but has only
// one translator warmup policy. Accumulate per-field policies into the most
// aggressive group policy: sync > async > disable. Empty means no field has
// provided an explicit warmup setting yet, so downstream logic may still fall
// back to global config.
void
AccumulateWarmupPolicyForGroup(const std::string& policy,
std::string& aggregated_warmup_policy) {
if (policy.empty()) {
return;
}
if (policy == "sync") {
aggregated_warmup_policy = "sync";
} else if (policy == "async" && aggregated_warmup_policy != "sync") {
aggregated_warmup_policy = "async";
} else if (policy == "disable" && aggregated_warmup_policy.empty()) {
aggregated_warmup_policy = "disable";
}
}
struct FileMetadataLoadResult {
milvus_storage::RowGroupMetadataVector row_group_meta;
// per field_id → per-row-group statistics; nullptr entry means the row
@@ -2477,6 +2498,11 @@ ChunkedSegmentSealedImpl::load_column_group_data_internal(
mmap_dir_path);
auto field_metas = schema_snapshot->get_field_metas(milvus_field_ids);
auto warmup_policy =
resolve_field_data_group_warmup_policy(field_metas,
segment_load_info,
schema_snapshot,
info.warmup_policy);
std::vector<FieldId> fields_for_stats;
if (ENABLE_PARQUET_STATS_SKIP_INDEX) {
@@ -2509,7 +2535,7 @@ ChunkedSegmentSealedImpl::load_column_group_data_internal(
mmap_config.GetMmapPopulate(),
milvus_field_ids.size(),
load_info.load_priority,
info.warmup_policy);
warmup_policy);
auto chunked_column_group =
std::make_shared<ChunkedColumnGroup>(std::move(translator));
@@ -2544,7 +2570,7 @@ ChunkedSegmentSealedImpl::load_column_group_data_internal(
std::move(ts), num_rows, runtime);
} else {
init_storage_v2_timestamp_index(
column, num_rows, info.warmup_policy, runtime);
column, num_rows, warmup_policy, runtime);
}
if (runtime == nullptr) {
PublishSystemFieldStateLocked();
@@ -2625,6 +2651,11 @@ ChunkedSegmentSealedImpl::load_column_group_data_internal(
mmap_dir_path);
auto field_metas = schema_snapshot->get_field_metas(milvus_field_ids);
auto warmup_policy =
resolve_field_data_group_warmup_policy(field_metas,
segment_load_info,
schema_snapshot,
info.warmup_policy);
std::vector<FieldId> fields_for_stats;
if (ENABLE_PARQUET_STATS_SKIP_INDEX) {
@@ -2657,7 +2688,7 @@ ChunkedSegmentSealedImpl::load_column_group_data_internal(
mmap_config.GetMmapPopulate(),
milvus_field_ids.size(),
load_info.load_priority,
info.warmup_policy);
warmup_policy);
auto chunked_column_group =
std::make_shared<ChunkedColumnGroup>(std::move(translator));
@@ -2820,6 +2851,11 @@ ChunkedSegmentSealedImpl::load_field_data_internal(
storage::SortByPath(file_infos);
auto field_meta = schema_snapshot->operator[](field_id);
auto warmup_policy =
resolve_field_data_warmup_policy(field_id,
segment_load_info,
schema_snapshot,
info.warmup_policy);
std::unique_ptr<Translator<milvus::Chunk>> translator =
std::make_unique<storagev1translator::ChunkTranslator>(
this->get_segment_id(),
@@ -2829,7 +2865,7 @@ ChunkedSegmentSealedImpl::load_field_data_internal(
info.enable_mmap,
mmap_config.GetMmapPopulate(),
load_info.load_priority,
info.warmup_policy);
warmup_policy);
auto data_type = field_meta.get_data_type();
auto slot = cachinglayer::Manager::GetInstance().CreateCacheSlot(
@@ -6390,6 +6426,91 @@ ChunkedSegmentSealedImpl::mask_with_timestamps(BitsetTypeView& bitset_chunk,
bitset_chunk |= mask;
}
std::string
ChunkedSegmentSealedImpl::resolve_field_data_warmup_policy(
FieldId field_id,
const SegmentLoadInfo& segment_load_info,
const SchemaPtr& schema_snapshot,
const std::string& explicit_warmup_policy) const {
// System fields do not carry user-field warmup settings and are not
// represented in user-field bitsets. They should not affect group warmup
// aggregation.
if (SystemProperty::Instance().IsSystem(field_id)) {
return explicit_warmup_policy;
}
// "Has index" here means an index is usable now:
// - SegmentLoadInfo covers DataCoord-built index files loaded with segment
// metadata.
// - published_state_.binlog_index_bitset covers interim indexes already
// generated from raw binlog data.
// Do not predict whether an interim index may be generated later. Before an
// index exists, raw vector data is the search-critical representation and
// follows vector-index warmup.
bool has_index = segment_load_info.HasIndexInfo(field_id);
if (!has_index) {
auto snapshot = CapturePublishedState();
has_index = snapshot != nullptr &&
get_bit_if_present(snapshot->binlog_index_bitset, field_id);
}
const auto& field_meta = schema_snapshot->operator[](field_id);
auto is_vector = IsVectorDataType(field_meta.get_data_type());
// explicit_warmup_policy is the field-data override carried by the current
// load request. It is normally authoritative. The only exception is vector
// raw data without a usable index: QueryCoord may have propagated
// warmup.vectorField=disable into the field TypeParams, but in this state
// the raw vector column is effectively the index/search path and must use
// vector-index warmup instead.
if (!explicit_warmup_policy.empty() && (!is_vector || has_index)) {
return explicit_warmup_policy;
}
if (is_vector && !has_index) {
auto [has_index_warmup, index_warmup_policy] =
schema_snapshot->CollectionWarmupPolicy(/*is_vector=*/true,
/*is_index=*/true);
if (has_index_warmup) {
return index_warmup_policy;
}
switch (milvus::cachinglayer::Manager::GetInstance()
.getVectorIndexCacheWarmupPolicy()) {
case CacheWarmupPolicy::CacheWarmupPolicy_Sync:
return "sync";
case CacheWarmupPolicy::CacheWarmupPolicy_Async:
return "async";
case CacheWarmupPolicy::CacheWarmupPolicy_Disable:
return "disable";
default:
return "";
}
}
auto [has_field_warmup, field_warmup_policy] =
schema_snapshot->WarmupPolicy(field_id, is_vector, /*is_index=*/false);
return has_field_warmup ? field_warmup_policy : "";
}
std::string
ChunkedSegmentSealedImpl::resolve_field_data_group_warmup_policy(
const std::unordered_map<FieldId, FieldMeta>& field_metas,
const SegmentLoadInfo& segment_load_info,
const SchemaPtr& schema_snapshot,
const std::string& explicit_warmup_policy) const {
std::string aggregated_warmup_policy;
for (const auto& field_meta_pair : field_metas) {
AccumulateWarmupPolicyForGroup(
resolve_field_data_warmup_policy(field_meta_pair.first,
segment_load_info,
schema_snapshot,
explicit_warmup_policy),
aggregated_warmup_policy);
}
return aggregated_warmup_policy;
}
bool
ChunkedSegmentSealedImpl::generate_interim_index(
const FieldId field_id,
@@ -6974,8 +7095,10 @@ ChunkedSegmentSealedImpl::PrepareLoadDiffForReopen(
CheckCancellation(op_ctx, id_, "ChunkedSegmentSealedImpl::ApplyLoadDiff()");
if (!diff.fields_to_fill_default.empty()) {
FillDefaultValueFields(
diff.fields_to_fill_default, schema_snapshot, committer);
FillDefaultValueFields(diff.fields_to_fill_default,
segment_load_info,
schema_snapshot,
committer);
}
CheckCancellation(op_ctx, id_, "ChunkedSegmentSealedImpl::ApplyLoadDiff()");
@@ -7266,9 +7389,11 @@ ChunkedSegmentSealedImpl::ApplyLoadDiff(milvus::OpContext* op_ctx,
}
void
ChunkedSegmentSealedImpl::fill_empty_field(const FieldMeta& field_meta,
const SchemaPtr& schema_snapshot,
RuntimeResourceState& runtime) {
ChunkedSegmentSealedImpl::fill_empty_field(
const FieldMeta& field_meta,
const SchemaPtr& schema_snapshot,
const SegmentLoadInfo& segment_load_info,
RuntimeResourceState& runtime) {
auto field_id = field_meta.get_id();
auto data_type = field_meta.get_data_type();
LOG_INFO(
@@ -7290,18 +7415,14 @@ ChunkedSegmentSealedImpl::fill_empty_field(const FieldMeta& field_meta,
->GetRootPath();
int64_t size = num_rows_.value();
AssertInfo(size > 0, "Chunked Sealed segment must have more than 0 row");
auto load_info_snapshot = CaptureLoadInfoSnapshot();
auto field_data_info =
FieldDataInfo(field_id.get(),
size,
mmap_dir_path,
false,
load_info_snapshot->GetInsertChannel());
auto field_data_info = FieldDataInfo(field_id.get(),
size,
mmap_dir_path,
false,
segment_load_info.GetInsertChannel());
auto [field_has_warmup, field_warmup_policy] =
schema_snapshot->WarmupPolicy(
field_id, IsVectorDataType(data_type), /*is_index=*/false);
std::string warmup_policy = field_has_warmup ? field_warmup_policy : "";
auto warmup_policy = resolve_field_data_warmup_policy(
field_id, segment_load_info, schema_snapshot);
std::unique_ptr<Translator<milvus::Chunk>> translator =
std::make_unique<storagev1translator::DefaultValueChunkTranslator>(
get_segment_id(),
@@ -7323,16 +7444,6 @@ ChunkedSegmentSealedImpl::fill_empty_field(const FieldMeta& field_meta,
id_);
}
void
ChunkedSegmentSealedImpl::fill_empty_field(const FieldMeta& field_meta) {
auto runtime = CloneMutableRuntimeResourceState();
fill_empty_field(field_meta, CaptureSchemaSnapshot(), *runtime);
EnsureArrayOffsetsForStructField(
field_meta, num_rows_.value_or(0), *runtime);
PublishFieldDataReadyLocked(field_meta.get_id(),
ToConstRuntimeState(std::move(runtime)));
}
void
ChunkedSegmentSealedImpl::EnsureArrayOffsetsForStructField(
const FieldMeta& field_meta,
@@ -7357,6 +7468,7 @@ ChunkedSegmentSealedImpl::EnsureArrayOffsetsForStructField(
void
ChunkedSegmentSealedImpl::FillDefaultValueFields(
const std::vector<FieldId>& field_ids,
const SegmentLoadInfo& segment_load_info,
const SchemaPtr& schema_snapshot,
RuntimeResourceState* runtime,
PublishedSegmentState* staged_state) {
@@ -7384,7 +7496,8 @@ ChunkedSegmentSealedImpl::FillDefaultValueFields(
continue;
}
const auto& field_meta = schema_snapshot->operator[](field_id);
fill_empty_field(field_meta, schema_snapshot, *target_runtime);
fill_empty_field(
field_meta, schema_snapshot, segment_load_info, *target_runtime);
EnsureArrayOffsetsForStructField(
field_meta, num_rows_.value_or(0), *target_runtime);
filled_fields.push_back(field_id);
@@ -7405,15 +7518,17 @@ ChunkedSegmentSealedImpl::FillDefaultValueFields(
void
ChunkedSegmentSealedImpl::FillDefaultValueFields(
const std::vector<FieldId>& field_ids) {
FillDefaultValueFields(field_ids, CaptureSchemaSnapshot(), nullptr);
auto snapshot = CapturePublishedState();
FillDefaultValueFields(
field_ids, *snapshot->load_info, snapshot->schema, nullptr);
}
void
ChunkedSegmentSealedImpl::FillDefaultValueFields(
const std::vector<FieldId>& field_ids,
const SegmentLoadInfo& segment_load_info,
const SchemaPtr& schema_snapshot,
StagedStateCommitter& committer) {
auto snapshot = CapturePublishedState();
const auto* staged_state = committer.staged_state();
std::vector<std::pair<FieldMeta, std::shared_ptr<ChunkedColumnInterface>>>
@@ -7455,20 +7570,15 @@ ChunkedSegmentSealedImpl::FillDefaultValueFields(
int64_t size = num_rows_.value();
AssertInfo(size > 0,
"Chunked Sealed segment must have more than 0 row");
auto load_info_snapshot = staged_state->load_info != nullptr
? staged_state->load_info
: snapshot->load_info;
auto field_data_info =
FieldDataInfo(field_id.get(),
size,
mmap_dir_path,
false,
load_info_snapshot->GetInsertChannel());
segment_load_info.GetInsertChannel());
auto [field_has_warmup, field_warmup_policy] =
schema_snapshot->WarmupPolicy(
field_id, IsVectorDataType(data_type), /*is_index=*/false);
std::string warmup_policy = field_has_warmup ? field_warmup_policy : "";
auto warmup_policy = resolve_field_data_warmup_policy(
field_id, segment_load_info, schema_snapshot);
std::unique_ptr<Translator<milvus::Chunk>> translator =
std::make_unique<storagev1translator::DefaultValueChunkTranslator>(
get_segment_id(),
@@ -7793,17 +7903,18 @@ ChunkedSegmentSealedImpl::LoadColumnGroup(
}
auto field_metas = schema_snapshot->get_field_metas(milvus_field_ids);
auto aggregated_warmup_policy = resolve_field_data_group_warmup_policy(
field_metas, segment_load_info, schema_snapshot);
// assumption: vector field occupies whole column group
bool is_vector = false;
bool has_mmap_setting = false;
bool mmap_enabled = false;
bool has_warmup_setting = false;
std::string aggregated_warmup_policy = "disable";
for (auto& [field_id, field_meta] : field_metas) {
if (IsVectorDataType(field_meta.get_data_type())) {
is_vector = true;
}
// if field has mmap setting, use it
// - mmap setting at collection level, then all field are the same
// - mmap setting at field level, we define that as long as one field shall be mmap, then whole group shall be mmaped
@@ -7811,24 +7922,6 @@ ChunkedSegmentSealedImpl::LoadColumnGroup(
schema_snapshot->MmapEnabled(field_id);
has_mmap_setting = has_mmap_setting || field_has_setting;
mmap_enabled = mmap_enabled || field_mmap_enabled;
// if field has warmup setting, use it
// - warmup setting at collection level, uses appropriate key based on field type
// - warmup setting at field level, use the most aggressive policy (sync > async > disable)
// Note: this is for field data loading, not index (is_index = false)
bool field_is_vector = IsVectorDataType(field_meta.get_data_type());
auto [field_has_warmup, field_warmup_policy] =
schema_snapshot->WarmupPolicy(
field_id, field_is_vector, /*is_index=*/false);
if (field_has_warmup) {
has_warmup_setting = true;
if (field_warmup_policy == "sync") {
aggregated_warmup_policy = "sync";
} else if (field_warmup_policy == "async" &&
aggregated_warmup_policy != "sync") {
aggregated_warmup_policy = "async";
}
}
}
auto& mmap_config = storage::MmapManager::GetInstance().GetMmapConfig();
@@ -7873,8 +7966,7 @@ ChunkedSegmentSealedImpl::LoadColumnGroup(
// Determine warmup policy: use per-field settings if any,
// otherwise pass empty string to fall back to global config
std::string warmup_policy =
has_warmup_setting ? aggregated_warmup_policy : "";
std::string warmup_policy = aggregated_warmup_policy;
// Multiple lazy entries can share the same column-group index (one per
// field), so the translator cache key must be disambiguated by the
@@ -7965,12 +8057,12 @@ ChunkedSegmentSealedImpl::LoadColumnGroup(
}
auto field_metas = schema_snapshot->get_field_metas(milvus_field_ids);
auto aggregated_warmup_policy = resolve_field_data_group_warmup_policy(
field_metas, segment_load_info, schema_snapshot);
bool is_vector = false;
bool has_mmap_setting = false;
bool mmap_enabled = false;
bool has_warmup_setting = false;
std::string aggregated_warmup_policy = "disable";
for (auto& [field_id, field_meta] : field_metas) {
if (IsVectorDataType(field_meta.get_data_type())) {
is_vector = true;
@@ -7979,20 +8071,6 @@ ChunkedSegmentSealedImpl::LoadColumnGroup(
schema_snapshot->MmapEnabled(field_id);
has_mmap_setting = has_mmap_setting || field_has_setting;
mmap_enabled = mmap_enabled || field_mmap_enabled;
bool field_is_vector = IsVectorDataType(field_meta.get_data_type());
auto [field_has_warmup, field_warmup_policy] =
schema_snapshot->WarmupPolicy(
field_id, field_is_vector, /*is_index=*/false);
if (field_has_warmup) {
has_warmup_setting = true;
if (field_warmup_policy == "sync") {
aggregated_warmup_policy = "sync";
} else if (field_warmup_policy == "async" &&
aggregated_warmup_policy != "sync") {
aggregated_warmup_policy = "async";
}
}
}
auto& mmap_config = storage::MmapManager::GetInstance().GetMmapConfig();
@@ -8030,8 +8108,7 @@ ChunkedSegmentSealedImpl::LoadColumnGroup(
.GetChunkManager()
->GetRootPath();
std::string warmup_policy =
has_warmup_setting ? aggregated_warmup_policy : "";
std::string warmup_policy = aggregated_warmup_policy;
std::string cache_key_suffix;
if (!eager_load) {
@@ -8318,8 +8395,7 @@ ChunkedSegmentSealedImpl::LoadBatchFieldData(
bool mmap_enabled = false;
bool is_vector = false;
bool has_warmup_setting = false;
std::string aggregated_warmup_policy = "disable";
std::string aggregated_warmup_policy;
for (const auto& child_field_id : fields_to_load) {
auto& field_meta = schema_snapshot->operator[](child_field_id);
if (IsVectorDataType(field_meta.get_data_type())) {
@@ -8346,20 +8422,10 @@ ChunkedSegmentSealedImpl::LoadBatchFieldData(
index_has_raw_data = false;
}
auto [field_has_warmup, field_warmup_policy] =
schema_snapshot->WarmupPolicy(
child_field_id,
IsVectorDataType(field_meta.get_data_type()),
/*is_index=*/false);
if (field_has_warmup) {
has_warmup_setting = true;
if (field_warmup_policy == "sync") {
aggregated_warmup_policy = "sync";
} else if (field_warmup_policy == "async" &&
aggregated_warmup_policy != "sync") {
aggregated_warmup_policy = "async";
}
}
AccumulateWarmupPolicyForGroup(
resolve_field_data_warmup_policy(
child_field_id, segment_load_info, schema_snapshot),
aggregated_warmup_policy);
}
auto group_id = field_binlog.fieldid();
@@ -8408,8 +8474,7 @@ ChunkedSegmentSealedImpl::LoadBatchFieldData(
// Determine group warmup policy: use per-field settings if any,
// otherwise fall back to global warmup policy
field_binlog_info.warmup_policy =
has_warmup_setting ? aggregated_warmup_policy : "";
field_binlog_info.warmup_policy = aggregated_warmup_policy;
// Store in map
load_field_data_info.field_infos[group_id] = field_binlog_info;
internal/core/src/segcore/ChunkedSegmentSealedImpl.h
+77 -2
View File
@@ -1371,10 +1371,22 @@ class ChunkedSegmentSealedImpl : public SegmentSealed {
void
fill_empty_field(const FieldMeta& field_meta,
const SchemaPtr& schema_snapshot,
const SegmentLoadInfo& segment_load_info,
RuntimeResourceState& runtime);
void
fill_empty_field(const FieldMeta& field_meta);
std::string
resolve_field_data_warmup_policy(
FieldId field_id,
const SegmentLoadInfo& segment_load_info,
const SchemaPtr& schema_snapshot,
const std::string& explicit_warmup_policy = "") const;
std::string
resolve_field_data_group_warmup_policy(
const std::unordered_map<FieldId, FieldMeta>& field_metas,
const SegmentLoadInfo& segment_load_info,
const SchemaPtr& schema_snapshot,
const std::string& explicit_warmup_policy = "") const;
SchemaPtr
CaptureSchemaSnapshot() const;
@@ -1811,12 +1823,14 @@ class ChunkedSegmentSealedImpl : public SegmentSealed {
void
FillDefaultValueFields(const std::vector<FieldId>& field_ids,
const SegmentLoadInfo& segment_load_info,
const SchemaPtr& schema_snapshot,
RuntimeResourceState* runtime = nullptr,
PublishedSegmentState* staged_state = nullptr);
void
FillDefaultValueFields(const std::vector<FieldId>& field_ids,
const SegmentLoadInfo& segment_load_info,
const SchemaPtr& schema_snapshot,
StagedStateCommitter& committer);
@@ -2279,6 +2293,67 @@ class ChunkedSegmentSealedImpl : public SegmentSealed {
return CapturePublishedState();
}
std::string
TestResolveFieldDataWarmupPolicy(
FieldId field_id,
const SegmentLoadInfo& segment_load_info,
const SchemaPtr& schema_snapshot,
const std::string& explicit_warmup_policy = "") const {
return resolve_field_data_warmup_policy(field_id,
segment_load_info,
schema_snapshot,
explicit_warmup_policy);
}
std::string
TestResolveFieldDataGroupWarmupPolicy(
const std::vector<FieldId>& field_ids,
const SegmentLoadInfo& segment_load_info,
const SchemaPtr& schema_snapshot) const {
return resolve_field_data_group_warmup_policy(
schema_snapshot->get_field_metas(field_ids),
segment_load_info,
schema_snapshot);
}
std::shared_ptr<ChunkedColumnInterface>
TestStageLoadColumnGroupWithReader(
const std::shared_ptr<milvus_storage::api::ColumnGroups>& column_groups,
const std::shared_ptr<milvus_storage::api::Properties>& properties,
int64_t index,
const std::vector<FieldId>& field_ids,
const SegmentLoadInfo& segment_load_info,
const SchemaPtr& schema_snapshot,
std::shared_ptr<milvus_storage::api::Reader> reader,
bool eager_load = true) {
auto current = CapturePublishedState();
auto runtime = CloneMutableRuntimeResourceState();
runtime->reader = std::move(reader);
auto staged = ClonePublishedState(current);
staged->schema = schema_snapshot;
staged->load_info =
std::make_shared<const SegmentLoadInfo>(segment_load_info);
staged->runtime = ToConstRuntimeState(runtime);
staged->commit_ts = current->commit_ts;
NormalizePublishedState(*staged);
StagedStateCommitter committer(*this, runtime.get(), staged.get());
LoadColumnGroup(column_groups,
properties,
index,
field_ids,
segment_load_info,
schema_snapshot,
eager_load,
nullptr,
false,
committer);
auto it = runtime->fields.find(field_ids.front());
AssertInfo(it != runtime->fields.end(), "test field was not loaded");
return it->second;
}
void
TestPublishSystemFieldState() {
PublishSystemFieldStateLocked();
internal/core/src/segcore/ChunkedSegmentSealedStorageV2Test.cpp
+109
View File
@@ -58,6 +58,7 @@
#include "milvus-storage/common/config.h"
#include "milvus-storage/filesystem/fs.h"
#include "milvus-storage/packed/writer.h"
#include "mmap/ChunkedColumnGroup.h"
#include "pb/plan.pb.h"
#include "pb/schema.pb.h"
#include "plan/PlanNode.h"
@@ -185,8 +186,116 @@ class StorageV2CellTargetGuard {
private:
int64_t old_bytes_;
};
class StorageV2TempDirGuard {
public:
StorageV2TempDirGuard(milvus_storage::ArrowFileSystemPtr fs,
std::string path)
: fs_(std::move(fs)), path_(std::move(path)) {
static_cast<void>(fs_->DeleteDir(path_));
}
~StorageV2TempDirGuard() {
static_cast<void>(fs_->DeleteDir(path_));
}
private:
milvus_storage::ArrowFileSystemPtr fs_;
std::string path_;
};
void
AddWarmupProperty(milvus::proto::schema::CollectionSchema& schema_proto,
const std::string& key,
const std::string& value) {
auto* prop = schema_proto.add_properties();
prop->set_key(key);
prop->set_value(value);
}
} // namespace
TEST(ChunkedSegmentSealedStorageV2,
DirectLoadFieldDataUsesVectorIndexWarmupForNoIndexVector) {
constexpr int64_t kPkFieldId = START_USER_FIELDID;
constexpr int64_t kVectorFieldId = START_USER_FIELDID + 1;
constexpr int64_t kDim = 4;
constexpr int64_t kRowCount = 4;
milvus::proto::schema::CollectionSchema schema_proto;
auto* pk_field = schema_proto.add_fields();
pk_field->set_fieldid(kPkFieldId);
pk_field->set_name("pk");
pk_field->set_data_type(milvus::proto::schema::DataType::Int64);
pk_field->set_is_primary_key(true);
auto* vector_field = schema_proto.add_fields();
vector_field->set_fieldid(kVectorFieldId);
vector_field->set_name("vec");
vector_field->set_data_type(milvus::proto::schema::DataType::FloatVector);
auto* dim = vector_field->add_type_params();
dim->set_key("dim");
dim->set_value(std::to_string(kDim));
AddWarmupProperty(schema_proto, "warmup.vectorField", "disable");
AddWarmupProperty(schema_proto, "warmup.vectorIndex", "sync");
auto schema = Schema::ParseFrom(schema_proto);
auto fs = milvus::segcore::GetDefaultArrowFileSystem();
const std::string dir = "test_data/storage_v2_direct_warmup";
StorageV2TempDirGuard dir_guard(fs, dir);
const std::string path = dir + "/vec.parquet";
ASSERT_TRUE(fs->CreateDir(dir).ok());
auto arrow_schema = schema->ConvertToArrowSchema();
std::vector<std::string> paths{path};
auto storage_config = milvus_storage::StorageConfig();
std::vector<std::vector<int>> column_groups{{1}};
auto writer_result = milvus_storage::PackedRecordBatchWriter::Make(
fs,
paths,
arrow_schema,
storage_config,
column_groups,
16 * 1024 * 1024,
::parquet::default_writer_properties());
ASSERT_TRUE(writer_result.ok()) << writer_result.status().ToString();
auto writer = writer_result.ValueOrDie();
auto dataset = DataGen(schema, kRowCount);
auto record_batch = ConvertToArrowRecordBatch(dataset, kDim, arrow_schema);
ASSERT_NE(record_batch, nullptr);
ASSERT_TRUE(writer->Write(record_batch).ok());
ASSERT_TRUE(writer->Close().ok());
LoadFieldDataInfo load_info;
load_info.storage_version = 2;
FieldBinlogInfo field_info{
kVectorFieldId,
kRowCount,
std::vector<int64_t>{kRowCount},
std::vector<int64_t>{kRowCount * kDim *
static_cast<int64_t>(sizeof(float))},
false,
"disable",
std::vector<std::string>{path},
std::vector<int64_t>{kVectorFieldId}};
load_info.field_infos.emplace(kVectorFieldId, std::move(field_info));
auto segment = segcore::CreateSealedSegment(
schema, nullptr, -1, segcore::SegcoreConfig::default_config(), true);
segment->LoadFieldData(load_info);
auto* sealed = dynamic_cast<ChunkedSegmentSealedImpl*>(segment.get());
ASSERT_NE(sealed, nullptr);
auto runtime = sealed->TestCloneMutableRuntimeResourceState();
auto field = runtime->fields.find(FieldId(kVectorFieldId));
ASSERT_NE(field, runtime->fields.end());
auto proxy_column =
std::dynamic_pointer_cast<ProxyChunkColumn>(field->second);
ASSERT_NE(proxy_column, nullptr);
EXPECT_EQ(proxy_column->TestCacheWarmupPolicy(),
CacheWarmupPolicy::CacheWarmupPolicy_Sync);
}
class TestChunkSegmentStorageV2 : public testing::TestWithParam<bool> {
protected:
segcore::SegmentSealedUPtr
internal/core/unittest/test_sealed.cpp
+186
View File
@@ -21,6 +21,7 @@
#include <chrono>
#include <cstdint>
#include <exception>
#include <functional>
#include <future>
#include <iostream>
#include <limits>
@@ -63,6 +64,7 @@
#include "knowhere/config.h"
#include "knowhere/dataset.h"
#include "knowhere/version.h"
#include "mmap/ChunkedColumnGroup.h"
#include "pb/common.pb.h"
#include "pb/schema.pb.h"
#include "query/Plan.h"
@@ -104,6 +106,126 @@ GetFieldBit(const BitsetType& bitset, FieldId field_id) {
return bitset[pos];
}
constexpr int64_t kWarmupPkFieldId = START_USER_FIELDID;
constexpr int64_t kWarmupVectorFieldId = START_USER_FIELDID + 1;
void
AddWarmupProperty(milvus::proto::schema::CollectionSchema& schema_proto,
const std::string& key,
const std::string& value) {
auto* prop = schema_proto.add_properties();
prop->set_key(key);
prop->set_value(value);
}
SchemaPtr
CreateWarmupPolicySchema(bool include_vector) {
milvus::proto::schema::CollectionSchema schema_proto;
auto* pk_field = schema_proto.add_fields();
pk_field->set_fieldid(kWarmupPkFieldId);
pk_field->set_name("pk");
pk_field->set_data_type(milvus::proto::schema::DataType::Int64);
pk_field->set_is_primary_key(true);
if (include_vector) {
auto* vector_field = schema_proto.add_fields();
vector_field->set_fieldid(kWarmupVectorFieldId);
vector_field->set_name("vec");
vector_field->set_data_type(
milvus::proto::schema::DataType::FloatVector);
auto* dim = vector_field->add_type_params();
dim->set_key("dim");
dim->set_value("4");
}
AddWarmupProperty(schema_proto, "warmup.vectorField", "sync");
AddWarmupProperty(schema_proto, "warmup.vectorIndex", "disable");
auto schema = Schema::ParseFrom(schema_proto);
schema->set_schema_version(include_vector ? 200 : 100);
return schema;
}
std::shared_ptr<milvus_storage::api::ColumnGroups>
MakeWarmupTestColumnGroups() {
auto column_groups = std::make_shared<milvus_storage::api::ColumnGroups>();
auto column_group = std::make_shared<milvus_storage::api::ColumnGroup>();
column_group->columns.emplace_back(std::to_string(kWarmupVectorFieldId));
column_groups->emplace_back(std::move(column_group));
return column_groups;
}
class WarmupTestChunkReader : public milvus_storage::api::ChunkReader {
public:
size_t
total_number_of_chunks() const override {
return 1;
}
arrow::Result<std::vector<int64_t>>
get_chunk_indices(const std::vector<int64_t>& row_indices) override {
return std::vector<int64_t>(row_indices.size(), 0);
}
arrow::Result<std::shared_ptr<arrow::RecordBatch>>
get_chunk(int64_t) override {
return arrow::Status::NotImplemented("warmup test chunk reader");
}
arrow::Result<std::vector<std::shared_ptr<arrow::RecordBatch>>>
get_chunks(const std::vector<int64_t>&, size_t) override {
return arrow::Status::NotImplemented("warmup test chunk reader");
}
arrow::Result<std::vector<uint64_t>>
get_chunk_size() override {
return std::vector<uint64_t>{1};
}
arrow::Result<std::vector<uint64_t>>
get_chunk_rows() override {
return std::vector<uint64_t>{1};
}
};
class WarmupTestReader : public milvus_storage::api::Reader {
public:
explicit WarmupTestReader(
std::shared_ptr<milvus_storage::api::ColumnGroups> column_groups)
: column_groups_(std::move(column_groups)) {
}
std::shared_ptr<milvus_storage::api::ColumnGroups>
get_column_groups() const override {
return column_groups_;
}
arrow::Result<std::shared_ptr<arrow::RecordBatchReader>>
get_record_batch_reader(const std::string&) const override {
return arrow::Status::NotImplemented("warmup test reader");
}
arrow::Result<std::unique_ptr<milvus_storage::api::ChunkReader>>
get_chunk_reader(int64_t, const std::shared_ptr<std::vector<std::string>>&)
const override {
return std::make_unique<WarmupTestChunkReader>();
}
arrow::Result<std::shared_ptr<arrow::Table>>
take(const std::vector<int64_t>&,
size_t,
const std::shared_ptr<std::vector<std::string>>&) override {
return arrow::Status::NotImplemented("warmup test reader");
}
void
set_keyretriever(
const std::function<std::string(const std::string&)>&) override {
}
private:
std::shared_ptr<milvus_storage::api::ColumnGroups> column_groups_;
};
class CancellationObservingIndexTranslator
: public cachinglayer::Translator<index::IndexBase> {
public:
@@ -4367,6 +4489,70 @@ TEST(SealedSegmentCowState, StagedVectorIndexLoadUsesResizedNewSchemaBitset) {
EXPECT_TRUE(sealed->TestVectorIndexReady(new_vec));
}
TEST(SealedSegmentCowState,
WarmupResolverHandlesAddedVectorAbsentFromPublishedBitset) {
auto old_schema = CreateWarmupPolicySchema(/*include_vector=*/false);
auto new_schema = CreateWarmupPolicySchema(/*include_vector=*/true);
const FieldId vec(kWarmupVectorFieldId);
auto segment = CreateSealedSegment(old_schema);
auto* sealed = dynamic_cast<ChunkedSegmentSealedImpl*>(segment.get());
ASSERT_NE(sealed, nullptr);
auto current = sealed->TestGetPublishedStateSnapshot();
ASSERT_NE(current, nullptr);
ASSERT_LE(current->binlog_index_bitset.size(), 1);
ASSERT_FALSE(current->schema->get_fields().count(vec));
SegmentLoadInfo staged_load_info(current->load_info->GetProto(),
new_schema);
auto [has_field_warmup, field_warmup_policy] =
new_schema->WarmupPolicy(vec, /*is_vector=*/true, /*is_index=*/false);
ASSERT_TRUE(has_field_warmup);
ASSERT_EQ(field_warmup_policy, "sync");
std::string resolved;
ASSERT_NO_THROW(resolved = sealed->TestResolveFieldDataWarmupPolicy(
vec, staged_load_info, new_schema));
EXPECT_EQ(resolved, "disable");
}
TEST(SealedSegmentCowState,
StagedStorageV2ColumnGroupUsesVectorIndexWarmupForNoIndexVector) {
auto schema = CreateWarmupPolicySchema(/*include_vector=*/true);
const FieldId vec(kWarmupVectorFieldId);
auto segment = CreateSealedSegment(schema);
auto* sealed = dynamic_cast<ChunkedSegmentSealedImpl*>(segment.get());
ASSERT_NE(sealed, nullptr);
proto::segcore::SegmentLoadInfo load_proto;
load_proto.set_segmentid(1006);
load_proto.set_num_of_rows(1);
SegmentLoadInfo segment_load_info(load_proto, schema);
ASSERT_EQ(sealed->TestResolveFieldDataGroupWarmupPolicy(
{vec}, segment_load_info, schema),
"disable");
auto column_groups = MakeWarmupTestColumnGroups();
auto reader = std::make_shared<WarmupTestReader>(column_groups);
auto column = sealed->TestStageLoadColumnGroupWithReader(
column_groups,
std::make_shared<milvus_storage::api::Properties>(),
/*index=*/0,
{vec},
segment_load_info,
schema,
std::move(reader),
/*eager_load=*/true);
auto proxy_column = std::dynamic_pointer_cast<ProxyChunkColumn>(column);
ASSERT_NE(proxy_column, nullptr);
EXPECT_EQ(proxy_column->TestCacheWarmupPolicy(),
CacheWarmupPolicy::CacheWarmupPolicy_Disable);
}
TEST(SealedSegmentCowState, StagedVectorIndexSkipsInterimIndexGeneration) {
auto schema = std::make_shared<Schema>();
auto pk = schema->AddDebugField("pk", DataType::INT64);
internal/querycoordv2/task/utils.go
+13
View File
@@ -283,6 +283,19 @@ func applyCollectionSettings(schema *schemapb.CollectionSchema,
schemaCloned = applyCollectionMmapSetting(schemaCloned, collectionProperties)
schemaCloned = applyCollectionWarmupSetting(schemaCloned, collectionProperties)
// Index warmup is normally materialized into each IndexInfo by
// applyIndexWarmupSetting. No-index vector fields have no IndexInfo carrier,
// but segcore treats their raw data as the vector-index/search path until an
// index exists. Carry QueryCoord's auto-warmup vector-index fallback through
// the effective load schema without changing warmup.vectorField semantics.
if _, exist := common.GetWarmupPolicyByKey(common.WarmupVectorIndexKey, schemaCloned.GetProperties()...); !exist &&
autoWarmupForNonPKIsolationCollection(collectionProperties) {
schemaCloned.Properties = append(schemaCloned.Properties, &commonpb.KeyValuePair{
Key: common.WarmupVectorIndexKey,
Value: common.WarmupSync,
})
}
return schemaCloned
}
internal/querycoordv2/task/utils_test.go
+62
View File
@@ -722,6 +722,68 @@ func TestApplyCollectionWarmupSettingAutoWarmup(t *testing.T) {
})
}
func TestApplyCollectionSettingsAutoWarmupVectorIndexCarrier(t *testing.T) {
paramtable.Init()
t.Run("autoWarmupForNonPKIsolationCollection carries vector index warmup on effective schema", func(t *testing.T) {
paramtable.Get().Save(paramtable.Get().QueryCoordCfg.AutoWarmupForNonPKIsolationCollection.Key, "true")
defer paramtable.Get().Reset(paramtable.Get().QueryCoordCfg.AutoWarmupForNonPKIsolationCollection.Key)
schema := &schemapb.CollectionSchema{
Fields: []*schemapb.FieldSchema{
{FieldID: 1, DataType: schemapb.DataType_FloatVector},
},
}
result := applyCollectionSettings(schema, nil)
warmup, exist := common.GetWarmupPolicyByKey(common.WarmupVectorIndexKey, result.GetProperties()...)
assert.True(t, exist)
assert.Equal(t, common.WarmupSync, warmup)
_, fieldWarmupExist := common.GetWarmupPolicy(result.GetFields()[0].GetTypeParams()...)
assert.False(t, fieldWarmupExist, "vector field warmup should not be changed by autoWarmupForNonPKIsolationCollection")
})
t.Run("explicit vector index warmup keeps priority over autoWarmupForNonPKIsolationCollection", func(t *testing.T) {
paramtable.Get().Save(paramtable.Get().QueryCoordCfg.AutoWarmupForNonPKIsolationCollection.Key, "true")
defer paramtable.Get().Reset(paramtable.Get().QueryCoordCfg.AutoWarmupForNonPKIsolationCollection.Key)
schema := &schemapb.CollectionSchema{
Fields: []*schemapb.FieldSchema{
{FieldID: 1, DataType: schemapb.DataType_FloatVector},
},
}
collectionProps := []*commonpb.KeyValuePair{
{Key: common.WarmupVectorIndexKey, Value: common.WarmupDisable},
}
result := applyCollectionSettings(schema, collectionProps)
warmup, exist := common.GetWarmupPolicyByKey(common.WarmupVectorIndexKey, result.GetProperties()...)
assert.True(t, exist)
assert.Equal(t, common.WarmupDisable, warmup)
})
t.Run("PKI collection does not carry vector index auto warmup", func(t *testing.T) {
paramtable.Get().Save(paramtable.Get().QueryCoordCfg.AutoWarmupForNonPKIsolationCollection.Key, "true")
defer paramtable.Get().Reset(paramtable.Get().QueryCoordCfg.AutoWarmupForNonPKIsolationCollection.Key)
schema := &schemapb.CollectionSchema{
Fields: []*schemapb.FieldSchema{
{FieldID: 1, DataType: schemapb.DataType_FloatVector},
},
}
collectionProps := []*commonpb.KeyValuePair{
{Key: common.PartitionKeyIsolationKey, Value: "true"},
}
result := applyCollectionSettings(schema, collectionProps)
_, exist := common.GetWarmupPolicyByKey(common.WarmupVectorIndexKey, result.GetProperties()...)
assert.False(t, exist)
})
}
func TestApplyIndexWarmupSettingAutoWarmup(t *testing.T) {
paramtable.Init()
internal/querynodev2/segments/utils_test.go
+45
View File
@@ -287,6 +287,16 @@ func TestGetFieldWarmupPolicy(t *testing.T) {
assert.Equal(t, common.WarmupAsync, policy)
})
t.Run("vector field TypeParams warmup is returned", func(t *testing.T) {
policy := getFieldWarmupPolicy(&schemapb.FieldSchema{
DataType: schemapb.DataType_FloatVector,
TypeParams: []*commonpb.KeyValuePair{
{Key: common.WarmupKey, Value: common.WarmupDisable},
},
})
assert.Equal(t, common.WarmupDisable, policy)
})
t.Run("fallback to global config for scalar field", func(t *testing.T) {
paramtable.Get().Save(paramtable.Get().QueryNodeCfg.TieredWarmupScalarField.Key, common.WarmupSync)
defer paramtable.Get().Reset(paramtable.Get().QueryNodeCfg.TieredWarmupScalarField.Key)
@@ -722,6 +732,24 @@ func TestIsExternalCollectionLazyLoad(t *testing.T) {
assert.True(t, isExternalCollectionLazyLoad(schema))
})
t.Run("external_vector_without_raw_warmup_follows_vector_field", func(t *testing.T) {
paramtable.Get().Save(paramtable.Get().QueryNodeCfg.TieredWarmupVectorField.Key, common.WarmupDisable)
defer paramtable.Get().Reset(paramtable.Get().QueryNodeCfg.TieredWarmupVectorField.Key)
paramtable.Get().Save(paramtable.Get().QueryNodeCfg.TieredWarmupVectorIndex.Key, common.WarmupSync)
defer paramtable.Get().Reset(paramtable.Get().QueryNodeCfg.TieredWarmupVectorIndex.Key)
schema := &schemapb.CollectionSchema{
Fields: []*schemapb.FieldSchema{
{
Name: "ext_vector",
DataType: schemapb.DataType_FloatVector,
ExternalField: "source_vector",
},
},
}
assert.True(t, isExternalCollectionLazyLoad(schema))
})
t.Run("milvus_table_real_pk_forces_eager_load", func(t *testing.T) {
schema := &schemapb.CollectionSchema{
ExternalSpec: `{"format":"milvus-table"}`,
@@ -739,4 +767,21 @@ func TestIsExternalCollectionLazyLoad(t *testing.T) {
}
assert.False(t, isExternalCollectionLazyLoad(schema))
})
t.Run("external_vector_follows_raw_field_warmup", func(t *testing.T) {
paramtable.Get().Save(paramtable.Get().QueryNodeCfg.TieredWarmupVectorIndex.Key, common.WarmupSync)
defer paramtable.Get().Reset(paramtable.Get().QueryNodeCfg.TieredWarmupVectorIndex.Key)
schema := &schemapb.CollectionSchema{
Fields: []*schemapb.FieldSchema{
{
Name: "ext_vector",
DataType: schemapb.DataType_FloatVector,
ExternalField: "source_vector",
TypeParams: []*commonpb.KeyValuePair{{Key: common.WarmupKey, Value: common.WarmupDisable}},
},
},
}
assert.True(t, isExternalCollectionLazyLoad(schema))
})
}