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

enhance: add cache shard disk usage metric (#50733)

Browse Source 
## Summary
- Thread the insert-channel shard through field, index, JSON stats, text
index, storage v1, and storage v2 load metadata into cachinglayer
translator metadata.
- Add cgo shard setters so Go-side load requests can pass shard into C++
load info.
- Keep the actual cachinglayer metric definition and CacheSlot
accounting in companion `milvus-common` PR:
https://github.com/zilliztech/milvus-common/pull/98

issue: #50941

---------

Signed-off-by: sunby <sunbingyi1992@gmail.com>
This commit is contained in:
Bingyi Sun
2026-06-30 18:22:28 +08:00
committed by GitHub
parent 1065ab0e27
commit 6316edd334
32 changed files with 194 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/core/conanfile.py
+1 -1
View File
@@ -28,7 +28,7 @@ class MilvusConan(ConanFile):
"prometheus-cpp/1.2.4#0918d66c13f97acb7809759f9de49b3f",
"re2/20230301#f8efaf45f98d0193cd0b2ea08b6b4060",
"folly/2026.04.20.00@milvus/dev#06852bea5b6449f0c4eb0df002b5779c",
"milvus-common/1.0.0-4f41e32@milvus/dev#1ce0b1d539952e383c64a51881384d31",
"milvus-common/1.0.0-835fcd0@milvus/dev#f908c64d5f16981c6b9f4d4e02358562",
"google-cloud-cpp/2.28.0@milvus/dev#468918b43cec43624531a0340398cf43",
"opentelemetry-cpp/1.23.0@milvus/dev#11bc565ec6e82910ae8f7471da756720",
"librdkafka/1.9.1#ec1a00d5414f618555799be9566adfb7",
internal/core/src/common/Consts.h
+1
View File
@@ -136,6 +136,7 @@ const std::string SEGMENT_MANIFEST_KEY = "segment_manifest";
const std::string EXTERNAL_SPEC_KEY = "external_spec";
const std::string LOON_FFI_PROPERTIES_KEY = "loon_ffi_properties";
const std::string STATS_BASE_PATH_KEY = "stats_base_path";
const std::string JSON_STATS_CACHE_SHARD_KEY = "json_stats_cache_shard";
// storage version
const int64_t STORAGE_V1 = 1;
internal/core/src/common/LoadInfo.h
+1
View File
@@ -43,6 +43,7 @@ struct LoadFieldDataInfo {
milvus::proto::common::LoadPriority load_priority =
milvus::proto::common::LoadPriority::HIGH;
std::vector<int64_t> child_field_ids;
std::string shard;
};
struct LoadDeletedRecordInfo {
internal/core/src/index/json_stats/JsonKeyStats.cpp
+5 -2
View File
@@ -1019,8 +1019,8 @@ JsonKeyStats::LoadColumnGroup(int64_t column_group_id,
segment_id_);
auto enable_mmap = !mmap_filepath_.empty();
auto column_group_info =
FieldDataInfo(column_group_id, field_id_, num_rows, mmap_filepath_);
auto column_group_info = FieldDataInfo(
column_group_id, field_id_, num_rows, mmap_filepath_, false, shard_);
LOG_INFO(
"loads column group {} with num_rows {} for segment "
"{}",
@@ -1126,6 +1126,7 @@ JsonKeyStats::LoadSharedKeyIndex(
load_info.index_size = index_size;
load_info.load_priority = load_priority_;
load_info.warmup_policy = warmup_policy;
load_info.shard = shard_;
std::unique_ptr<cachinglayer::Translator<index::BsonInvertedIndex>>
translator = std::make_unique<
segcore::storagev1translator::BsonInvertedIndexTranslator>(
@@ -1160,6 +1161,8 @@ JsonKeyStats::Load(milvus::tracer::TraceContext ctx, const Config& config) {
LOG_INFO("load json stats for segment {} with load priority: {}",
segment_id_,
static_cast<int>(load_priority_));
shard_ = GetValueFromConfig<std::string>(config, JSON_STATS_CACHE_SHARD_KEY)
.value_or("");
auto index_files =
GetValueFromConfig<std::vector<std::string>>(config, "index_files");
internal/core/src/index/json_stats/JsonKeyStats.h
+1
View File
@@ -686,6 +686,7 @@ class JsonKeyStats : public ScalarIndex<std::string> {
std::set<JsonKey> column_keys_;
std::shared_ptr<JsonStatsParquetWriter> parquet_writer_;
std::shared_ptr<BsonInvertedIndex> bson_inverted_index_;
std::string shard_;
// cache slot for bson inverted index when using translator
std::shared_ptr<milvus::cachinglayer::CacheSlot<BsonInvertedIndex>>
bson_index_cache_slot_;
internal/core/src/mmap/Types.h
+9 -4
View File
@@ -32,11 +32,13 @@ struct FieldDataInfo {
FieldDataInfo(int64_t field_id,
size_t row_count,
std::string mmap_dir_path = "",
bool in_load_list = false)
bool in_load_list = false,
std::string shard = "")
: field_id(field_id),
row_count(row_count),
mmap_dir_path(std::move(mmap_dir_path)),
in_load_list(in_load_list) {
in_load_list(in_load_list),
shard(std::move(shard)) {
arrow_reader_channel = std::make_shared<ArrowReaderChannel>();
}
@@ -44,12 +46,14 @@ struct FieldDataInfo {
int64_t main_field_id,
size_t row_count,
std::string mmap_dir_path = "",
bool in_load_list = false)
bool in_load_list = false,
std::string shard = "")
: field_id(field_id),
row_count(row_count),
main_field_id(main_field_id),
mmap_dir_path(std::move(mmap_dir_path)),
in_load_list(in_load_list) {
in_load_list(in_load_list),
shard(std::move(shard)) {
arrow_reader_channel = std::make_shared<ArrowReaderChannel>();
}
@@ -59,5 +63,6 @@ struct FieldDataInfo {
std::string mmap_dir_path{};
std::shared_ptr<ArrowReaderChannel> arrow_reader_channel;
bool in_load_list{false};
std::string shard{};
};
} // namespace milvus
internal/core/src/monitor/monitor_c.cpp
+3 -1
View File
@@ -13,15 +13,17 @@
#include <string.h>
#include <string>
#include "cachinglayer/Metrics.h"
#include "common/FastMem.h"
#include "common/init_c.h"
#include "monitor_c.h"
#include "common/PrometheusClient.h"
#include "monitor_c.h"
char*
GetCoreMetrics() {
UpdateArrowIOThreadPoolMetrics();
static_cast<void>(
milvus::cachinglayer::monitor::collect_cache_shard_disk_usage_stats());
auto str = milvus::monitor::getPrometheusClient().GetMetrics();
auto len = str.length();
char* res = static_cast<char*>(malloc(len + 1));
internal/core/src/segcore/CacheMetricAttribution.h
+30
View File
@@ -0,0 +1,30 @@
// Copyright (C) 2019-2026 Zilliz. All rights reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance
// with the License. You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software distributed under the License
// is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
// or implied. See the License for the specific language governing permissions and limitations under the License
#pragma once
#include <optional>
#include <string>
#include <utility>
#include "cachinglayer/Translator.h"
namespace milvus::segcore {
inline std::optional<milvus::cachinglayer::MetricAttribution>
MetricAttributionFromShard(std::string shard) {
if (shard.empty()) {
return std::nullopt;
}
return milvus::cachinglayer::MetricAttribution{std::move(shard)};
}
} // namespace milvus::segcore
internal/core/src/segcore/ChunkedSegmentSealedImpl.cpp
+20 -9
View File
@@ -1074,7 +1074,8 @@ ChunkedSegmentSealedImpl::load_column_group_data_internal(
auto column_group_info = FieldDataInfo(column_group_id.get(),
num_rows,
mmap_dir_path,
merged_in_load_list);
merged_in_load_list,
load_info.shard);
LOG_INFO(
"[StorageV2] segment {} loads column group {} with field ids "
"{} "
@@ -1185,11 +1186,11 @@ ChunkedSegmentSealedImpl::load_field_data_internal(
milvus::storage::LocalChunkManagerSingleton::GetInstance()
.GetChunkManager()
->GetRootPath();
auto field_data_info =
FieldDataInfo(field_id.get(),
num_rows,
mmap_dir_path,
schema_->ShouldLoadField(field_id));
auto field_data_info = FieldDataInfo(field_id.get(),
num_rows,
mmap_dir_path,
schema_->ShouldLoadField(field_id),
load_info.shard);
LOG_INFO("segment {} loads field {} with num_rows {}, sorted by pk {}",
this->get_segment_id(),
field_id.get(),
@@ -3447,7 +3448,8 @@ ChunkedSegmentSealedImpl::LoadTextIndex(
info_proto->fieldid(),
field_meta.get_analyzer_params(),
info_proto->index_size(),
info_proto->warmup_policy()};
info_proto->warmup_policy(),
std::atomic_load(&segment_load_info_)->GetInsertChannel()};
std::unique_ptr<
milvus::cachinglayer::Translator<milvus::index::TextMatchIndex>>
@@ -3530,6 +3532,8 @@ ChunkedSegmentSealedImpl::LoadJsonKeyIndex(
if (!info_proto->base_path().empty()) {
config[STATS_BASE_PATH_KEY] = info_proto->base_path();
}
config[JSON_STATS_CACHE_SHARD_KEY] =
std::atomic_load(&segment_load_info_)->GetInsertChannel();
milvus::storage::FileManagerContext file_ctx(
field_data_meta, index_meta, remote_chunk_manager, fs);
@@ -5086,7 +5090,12 @@ 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 field_data_info = FieldDataInfo(field_id.get(), size, mmap_dir_path);
auto field_data_info = FieldDataInfo(
field_id.get(),
size,
mmap_dir_path,
false,
std::atomic_load(&segment_load_info_)->GetInsertChannel());
auto [field_has_warmup, field_warmup_policy] = schema_->WarmupPolicy(
field_id, IsVectorDataType(data_type), /*is_index=*/false);
@@ -5474,7 +5483,8 @@ ChunkedSegmentSealedImpl::LoadColumnGroup(
eager_load,
warmup_policy,
cache_key_suffix,
load_info->GetEstimatedBytesPerRow());
load_info->GetEstimatedBytesPerRow(),
load_info->GetInsertChannel());
auto chunked_column_group =
std::make_shared<ChunkedColumnGroup>(std::move(translator));
@@ -5624,6 +5634,7 @@ ChunkedSegmentSealedImpl::LoadBatchFieldData(
LoadFieldDataInfo load_field_data_info;
load_field_data_info.storage_version =
load_info_snapshot->GetStorageVersion();
load_field_data_info.shard = load_info_snapshot->GetInsertChannel();
auto fields_to_load = field_ids;
AssertInfo(!fields_to_load.empty(),
"load field data with empty field list");
internal/core/src/segcore/SegmentGrowingImpl.cpp
+3 -2
View File
@@ -995,8 +995,8 @@ SegmentGrowingImpl::load_column_group_data_internal(
auto insert_files = info.insert_files;
storage::SortByPath(insert_files);
auto fs = milvus::segcore::GetDefaultArrowFileSystem();
auto column_group_info =
FieldDataInfo(column_group_id.get(), num_rows, "");
auto column_group_info = FieldDataInfo(
column_group_id.get(), num_rows, "", false, infos.shard);
column_group_info.arrow_reader_channel->set_capacity(parallel_degree);
LOG_INFO(
@@ -2509,6 +2509,7 @@ SegmentGrowingImpl::Load(milvus::tracer::TraceContext& trace_ctx,
// Set load priority
field_data_info.load_priority = load_info_.priority();
field_data_info.shard = load_info_.insert_channel();
auto manifest_path = load_info_.manifest_path();
if (manifest_path != "") {
internal/core/src/segcore/SegmentLoadInfo.cpp
+1
View File
@@ -67,6 +67,7 @@ SegmentLoadInfo::ConvertFieldIndexInfoToLoadIndexInfo(
load_index_info.field_id = field_id.get();
load_index_info.collection_id = GetCollectionID();
load_index_info.partition_id = GetPartitionID();
load_index_info.shard = GetInsertChannel();
// Get field type from schema
const auto& field_meta = schema_->operator[](field_id);
internal/core/src/segcore/Types.h
+3
View File
@@ -63,6 +63,7 @@ struct LoadIndexInfo {
int64_t dim;
std::string
warmup_policy; // "disable", "sync", or "async"; empty means use global config
std::string shard;
std::optional<LoadResourceRequest> load_resource_request;
// Default constructor
@@ -99,6 +100,7 @@ struct LoadIndexInfo {
num_rows(other.num_rows),
dim(other.dim),
warmup_policy(other.warmup_policy),
shard(other.shard),
load_resource_request(other.load_resource_request) {
}
@@ -129,6 +131,7 @@ struct LoadIndexInfo {
num_rows = other.num_rows;
dim = other.dim;
warmup_policy = other.warmup_policy;
shard = other.shard;
load_resource_request = other.load_resource_request;
}
return *this;
internal/core/src/segcore/load_field_data_c.cpp
+10
View File
@@ -75,6 +75,16 @@ AppendLoadFieldInfo(CLoadFieldDataInfo c_load_field_data_info,
}
}
void
SetLoadFieldDataInfoShard(CLoadFieldDataInfo c_load_field_data_info,
const char* shard) {
SCOPE_CGO_CALL_METRIC();
auto load_field_data_info =
static_cast<LoadFieldDataInfo*>(c_load_field_data_info);
load_field_data_info->shard = shard == nullptr ? "" : shard;
}
CStatus
SetLoadFieldInfoChildFields(CLoadFieldDataInfo c_load_field_data_info,
int64_t field_id,
internal/core/src/segcore/load_field_data_c.h
+4
View File
@@ -37,6 +37,10 @@ AppendLoadFieldInfo(CLoadFieldDataInfo c_load_field_data_info,
int64_t field_id,
int64_t row_count);
void
SetLoadFieldDataInfoShard(CLoadFieldDataInfo c_load_field_data_info,
const char* shard);
CStatus
SetLoadFieldInfoChildFields(CLoadFieldDataInfo c_load_field_data_info,
int64_t field_id,
internal/core/src/segcore/load_index_c.cpp
+9
View File
@@ -391,3 +391,12 @@ FinishLoadIndexInfo(CLoadIndexInfo c_load_index_info,
return status;
}
}
void
SetLoadIndexInfoShard(CLoadIndexInfo c_load_index_info, const char* shard) {
SCOPE_CGO_CALL_METRIC();
auto load_index_info =
static_cast<milvus::segcore::LoadIndexInfo*>(c_load_index_info);
load_index_info->shard = shard == nullptr ? "" : shard;
}
internal/core/src/segcore/load_index_c.h
+3
View File
@@ -62,6 +62,9 @@ CStatus
FinishLoadIndexInfo(CLoadIndexInfo c_load_index_info,
const uint8_t* serialized_load_index_info,
const uint64_t len);
void
SetLoadIndexInfoShard(CLoadIndexInfo c_load_index_info, const char* shard);
#ifdef __cplusplus
}
#endif
internal/core/src/segcore/storagev1translator/BsonInvertedIndexTranslator.cpp
+5 -1
View File
@@ -18,6 +18,7 @@
#include <algorithm>
#include <functional>
#include <optional>
#include <string_view>
#include <utility>
@@ -27,6 +28,7 @@
#include "index/json_stats/bson_inverted.h"
#include "log/Log.h"
#include "pb/common.pb.h"
#include "segcore/CacheMetricAttribution.h"
#include "segcore/Utils.h"
#include "storage/DiskFileManagerImpl.h"
@@ -48,7 +50,9 @@ BsonInvertedIndexTranslator::BsonInvertedIndexTranslator(
milvus::segcore::getCacheWarmupPolicy(load_info_.warmup_policy,
/* is_vector */ false,
/* is_index */ true),
/* support_eviction */ true) {
/* support_eviction */ true,
std::nullopt,
milvus::segcore::MetricAttributionFromShard(load_info_.shard)) {
}
size_t
internal/core/src/segcore/storagev1translator/BsonInvertedIndexTranslator.h
+1
View File
@@ -32,6 +32,7 @@ struct BsonInvertedIndexLoadInfo {
uint32_t load_priority;
std::string
warmup_policy; // "disable", "sync", or "async"; empty means use global config
std::string shard;
};
// Translator for BsonInvertedIndex in json stats. It loads a single-cell
internal/core/src/segcore/storagev1translator/ChunkTranslator.cpp
+2 -1
View File
@@ -97,7 +97,8 @@ ChunkTranslator::ChunkTranslator(
IsVectorDataType(field_meta.get_data_type()),
/* is_index */ false,
/* in_load_list*/ field_data_info.in_load_list),
/* support_eviction */ true),
/* support_eviction */ true,
field_data_info.shard),
load_priority_(load_priority) {
AssertInfo(!SystemProperty::Instance().IsSystem(FieldId(field_id_)),
"ChunkTranslator not supported for system field");
internal/core/src/segcore/storagev1translator/ChunkTranslator.h
+12 -6
View File
@@ -12,6 +12,7 @@
#pragma once
#include <string>
#include <utility>
#include <vector>
#include "cachinglayer/Translator.h"
@@ -19,6 +20,7 @@
#include "common/Chunk.h"
#include "common/type_c.h"
#include "mmap/Types.h"
#include "segcore/CacheMetricAttribution.h"
namespace milvus::segcore::storagev1translator {
@@ -34,12 +36,16 @@ struct CTMeta : public milvus::cachinglayer::Meta {
milvus::cachinglayer::CellIdMappingMode cell_id_mapping_mode,
milvus::cachinglayer::CellDataType cell_data_type,
CacheWarmupPolicy cache_warmup_policy,
bool support_eviction)
: milvus::cachinglayer::Meta(storage_type,
cell_id_mapping_mode,
cell_data_type,
cache_warmup_policy,
support_eviction) {
bool support_eviction,
std::string shard = "")
: milvus::cachinglayer::Meta(
storage_type,
cell_id_mapping_mode,
cell_data_type,
cache_warmup_policy,
support_eviction,
std::nullopt,
milvus::segcore::MetricAttributionFromShard(std::move(shard))) {
}
};
internal/core/src/segcore/storagev1translator/DefaultValueChunkTranslator.cpp
+2 -1
View File
@@ -62,7 +62,8 @@ DefaultValueChunkTranslator::DefaultValueChunkTranslator(
IsVectorDataType(field_meta.get_data_type()),
/* is_index */ false,
/* in_load_list, set to false to reduce memory usage */ false),
/* support_eviction */ false) {
/* support_eviction */ false,
field_data_info.shard) {
// Split rows into ~64KB cells according to value_size().
// Fallback to single-cell if value_size() is not well-defined.
auto vsize = this->value_size();
internal/core/src/segcore/storagev1translator/SealedIndexTranslator.cpp
+5 -1
View File
@@ -1,6 +1,7 @@
#include "segcore/storagev1translator/SealedIndexTranslator.h"
#include <filesystem>
#include <optional>
#include <utility>
#include "cachinglayer/LoadingOverheadTracker.h"
@@ -15,6 +16,7 @@
#include "index/Utils.h"
#include "log/Log.h"
#include "nlohmann/json.hpp"
#include "segcore/CacheMetricAttribution.h"
#include "segcore/Types.h"
#include "segcore/Utils.h"
#include "segcore/memory_planner.h"
@@ -75,7 +77,9 @@ SealedIndexTranslator::SealedIndexTranslator(
// currently only vector index is possible to support lazy load
!(IsVectorDataType(load_index_info->field_type) &&
knowhere::IndexFactory::Instance().FeatureCheck(
index_info_.index_type, knowhere::feature::LAZY_LOAD))) {
index_info_.index_type, knowhere::feature::LAZY_LOAD)),
std::nullopt,
milvus::segcore::MetricAttributionFromShard(load_index_info->shard)) {
load_resource_request_ = EstimateLoadResource();
auto scalar_version =
internal/core/src/segcore/storagev1translator/TextMatchIndexTranslator.cpp
+5 -1
View File
@@ -18,6 +18,7 @@
#include <algorithm>
#include <functional>
#include <optional>
#include <string_view>
#include <type_traits>
#include <utility>
@@ -27,6 +28,7 @@
#include "glog/logging.h"
#include "index/TextMatchIndex.h"
#include "log/Log.h"
#include "segcore/CacheMetricAttribution.h"
#include "segcore/Utils.h"
namespace milvus::segcore::storagev1translator {
@@ -48,7 +50,9 @@ TextMatchIndexTranslator::TextMatchIndexTranslator(
milvus::segcore::getCacheWarmupPolicy(load_info_.warmup_policy,
/* is_vector */ false,
/* is_index */ true),
/* support_eviction */ true) {
/* support_eviction */ true,
std::nullopt,
milvus::segcore::MetricAttributionFromShard(load_info_.shard)) {
}
size_t
internal/core/src/segcore/storagev1translator/TextMatchIndexTranslator.h
+1
View File
@@ -31,6 +31,7 @@ struct TextMatchIndexLoadInfo {
int64_t index_size;
std::string
warmup_policy; // "disable", "sync", or "async"; empty means use global config
std::string shard;
};
// Translator for TextMatchIndex (non-knowhere index). It loads a single-cell
internal/core/src/segcore/storagev1translator/V1SealedIndexTranslator.cpp
+16 -12
View File
@@ -14,6 +14,7 @@
#include "index/Utils.h"
#include "milvus-storage/filesystem/fs.h"
#include "nlohmann/json.hpp"
#include "segcore/CacheMetricAttribution.h"
#include "segcore/Types.h"
#include "segcore/Utils.h"
#include "storage/FileManager.h"
@@ -45,18 +46,21 @@ V1SealedIndexTranslator::V1SealedIndexTranslator(
key_(fmt::format("seg_{}_si_{}",
load_index_info->segment_id,
load_index_info->field_id)),
meta_(load_index_info->enable_mmap
? milvus::cachinglayer::StorageType::DISK
: milvus::cachinglayer::StorageType::MEMORY,
milvus::cachinglayer::CellIdMappingMode::ALWAYS_ZERO,
milvus::segcore::getCellDataType(
/* is_vector */ IsVectorDataType(load_index_info->field_type),
/* is_index */ true),
milvus::segcore::getCacheWarmupPolicy(
load_index_info->warmup_policy,
/* is_vector */ IsVectorDataType(load_index_info->field_type),
/* is_index */ true),
/* support_eviction */ false) {
meta_(
load_index_info->enable_mmap
? milvus::cachinglayer::StorageType::DISK
: milvus::cachinglayer::StorageType::MEMORY,
milvus::cachinglayer::CellIdMappingMode::ALWAYS_ZERO,
milvus::segcore::getCellDataType(
/* is_vector */ IsVectorDataType(load_index_info->field_type),
/* is_index */ true),
milvus::segcore::getCacheWarmupPolicy(
load_index_info->warmup_policy,
/* is_vector */ IsVectorDataType(load_index_info->field_type),
/* is_index */ true),
/* support_eviction */ false,
std::nullopt,
milvus::segcore::MetricAttributionFromShard(load_index_info->shard)) {
}
size_t
internal/core/src/segcore/storagev2translator/GroupCTMeta.h
+13 -7
View File
@@ -18,11 +18,13 @@
#include <algorithm>
#include <atomic>
#include <cstdint>
#include <string>
#include <type_traits>
#include <utility>
#include <vector>
#include "cachinglayer/Translator.h"
#include "segcore/CacheMetricAttribution.h"
namespace milvus::segcore::storagev2translator {
@@ -94,12 +96,16 @@ struct GroupCTMeta : public milvus::cachinglayer::Meta {
milvus::cachinglayer::CellIdMappingMode cell_id_mapping_mode,
milvus::cachinglayer::CellDataType cell_data_type,
CacheWarmupPolicy cache_warmup_policy,
bool support_eviction)
: milvus::cachinglayer::Meta(storage_type,
cell_id_mapping_mode,
cell_data_type,
cache_warmup_policy,
support_eviction),
bool support_eviction,
std::string shard = "")
: milvus::cachinglayer::Meta(
storage_type,
cell_id_mapping_mode,
cell_data_type,
cache_warmup_policy,
support_eviction,
std::nullopt,
milvus::segcore::MetricAttributionFromShard(std::move(shard))),
num_fields_(num_fields),
total_row_groups_(0) {
}
@@ -111,4 +117,4 @@ struct GroupCTMeta : public milvus::cachinglayer::Meta {
}
};
} // namespace milvus::segcore::storagev2translator
} // namespace milvus::segcore::storagev2translator
internal/core/src/segcore/storagev2translator/GroupChunkTranslator.cpp
+2 -1
View File
@@ -126,7 +126,8 @@ GroupChunkTranslator::GroupChunkTranslator(
return false;
}(),
/* is_index */ false),
/* support_eviction */ true),
/* support_eviction */ true,
column_group_info.shard),
use_mmap_(use_mmap),
mmap_populate_(mmap_populate),
has_array_field_(std::any_of(field_metas_.begin(),
internal/core/src/segcore/storagev2translator/ManifestGroupTranslator.cpp
+5 -2
View File
@@ -27,6 +27,7 @@
#include <string>
#include <type_traits>
#include <unordered_map>
#include <utility>
#include <vector>
#include "NamedType/named_type_impl.hpp"
@@ -75,7 +76,8 @@ ManifestGroupTranslator::ManifestGroupTranslator(
bool eager_load,
const std::string& warmup_policy,
const std::string& cache_key_suffix,
int64_t fallback_bytes_per_row)
int64_t fallback_bytes_per_row,
std::string shard)
: segment_id_(segment_id),
group_chunk_type_(group_chunk_type),
column_group_index_(column_group_index),
@@ -117,7 +119,8 @@ ManifestGroupTranslator::ManifestGroupTranslator(
}(),
/* is_index */ false,
/* in_load_list*/ eager_load),
/* support_eviction */ true),
/* support_eviction */ true,
std::move(shard)),
use_mmap_(use_mmap),
mmap_populate_(mmap_populate),
has_array_field_(std::any_of(field_metas_.begin(),
internal/core/src/segcore/storagev2translator/ManifestGroupTranslator.h
+2 -1
View File
@@ -79,7 +79,8 @@ class ManifestGroupTranslator
bool eager_load,
const std::string& warmup_policy,
const std::string& cache_key_suffix = "",
int64_t fallback_bytes_per_row = 0);
int64_t fallback_bytes_per_row = 0,
std::string shard = "");
~ManifestGroupTranslator() = default;
/**
internal/querynodev2/segments/load_index_info.go
+10
View File
@@ -19,6 +19,7 @@ package segments
/*
#cgo pkg-config: milvus_core
#include <stdlib.h>
#include "segcore/load_index_c.h"
#include "common/binary_set_c.h"
*/
@@ -89,6 +90,15 @@ func (li *LoadIndexInfo) appendLoadIndexInfo(ctx context.Context, info *cgopb.Lo
return HandleCStatus(ctx, &status, "FinishLoadIndexInfo failed")
}
func (li *LoadIndexInfo) setShard(shard string) {
if shard == "" {
return
}
cShard := C.CString(shard)
defer C.free(unsafe.Pointer(cShard))
C.SetLoadIndexInfoShard(li.cLoadIndexInfo, cShard)
}
func (li *LoadIndexInfo) loadIndex(ctx context.Context) error {
var status C.CStatus
_, _ = GetLoadPool().Submit(func() (any, error) {
internal/querynodev2/segments/segment.go
+3
View File
@@ -1001,6 +1001,8 @@ func (s *LocalSegment) LoadFieldData(ctx context.Context, fieldID int64, rowCoun
}},
RowCount: rowCount,
StorageVersion: s.LoadInfo().GetStorageVersion(),
LoadPriority: s.LoadInfo().GetPriority(),
Shard: s.LoadInfo().GetInsertChannel(),
}
GetLoadPool().Submit(func() (any, error) {
@@ -1177,6 +1179,7 @@ func GetCLoadInfoWithFunc(ctx context.Context,
mlog.Warn(ctx, "fail to append load index info", mlog.Err(err))
return err
}
loadIndexInfo.setShard(loadInfo.GetInsertChannel())
return f(loadIndexInfo)
}
internal/util/segcore/requests.go
+6
View File
@@ -45,6 +45,7 @@ type LoadFieldDataRequest struct {
RowCount int64
StorageVersion int64
LoadPriority commonpb.LoadPriority
Shard string
}
type LoadFieldDataInfo struct {
@@ -64,6 +65,11 @@ func (req *LoadFieldDataRequest) getCLoadFieldDataRequest() (result *cLoadFieldD
C.DeleteLoadFieldDataInfo(cLoadFieldDataInfo)
}
}()
if req.Shard != "" {
cShard := C.CString(req.Shard)
defer C.free(unsafe.Pointer(cShard))
C.SetLoadFieldDataInfoShard(cLoadFieldDataInfo, cShard)
}
rowCount := C.int64_t(req.RowCount)
for _, field := range req.Fields {