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: loading overhead estimation and LoadingOverheadConfig for caching layer (#48773)

Browse Source 
## Summary

- Change \`estimated_byte_size_of_cell\` return semantics from
\`{loaded, total_loading}\` to \`{loaded, loading_overhead}\` across
**all** translators
- Integrate with milvus-common \`LoadingOverheadConfig\` API to cap
total loading overhead per group via \`LoadingOverheadTracker\`
- Add cell-batch loading constants (\`kChannelCapacityMultiplier\`,
\`kLoadingOverheadInflationRatio\`) to \`memory_planner.h\`
- Update milvus-common to \`sparknack/milvus-common\` loading-opt branch
(\`b2efad3\`)

issue: #48772

## Changes

### 1. Overhead semantics change (all translators)
All \`estimated_byte_size_of_cell\` implementations updated so the
second element is pure overhead (temporary buffers during loading), not
total loading cost:

| Translator | mmap overhead | non-mmap overhead |
|-----------|--------------|-------------------|
| ChunkTranslator | \`{2x mem, 1x disk}\` | \`{1x mem, 0}\` |
| DefaultValueChunkTranslator | \`{0, 0}\` | \`{0, 0}\` |
| InterimSealedIndexTranslator | build buffer only | build buffer only |
| SealedIndexTranslator | \`max - final\` | \`max - final\` |
| TextMatchIndexTranslator | temp mem during load | temp disk during
load |
| BsonInvertedIndexTranslator | temp mem (download) | temp disk (local
file) |
| GroupChunkTranslator | via LoadingOverheadConfig | via
LoadingOverheadConfig |
| ManifestGroupTranslator | via LoadingOverheadConfig | via
LoadingOverheadConfig |

### 2. LoadingOverheadConfig for cell-batch translators
\`GroupChunkTranslator\` and \`ManifestGroupTranslator\` set
\`Meta::loading_overhead\` with:
- \`upper_bound\`: \`max_inflight * max_cell_size * inflation_ratio\`
- \`group\`: per-translator-type + CellDataType

This prevents N concurrent column group loads from each reserving full
overhead independently.

### 3. Cell-batch loading constants
Added to \`memory_planner.h\`:
- \`kChannelCapacityMultiplier = 1.5\` — bounded channel capacity
relative to pool size
- \`kLoadingOverheadInflationRatio = 1.2\` — safety margin for overhead
estimation

### 4. milvus-common dependency
Points to \`sparknack/milvus-common\` loading-opt branch (\`b2efad3\`)
which adds:
- \`LoadingOverheadConfig\` struct in \`Translator.h\`
- \`LoadingOverheadTracker\` for per-group overhead capping
- Updated \`CacheSlot\` and \`DList\` to integrate tracker

## Test plan

- [ ] Existing unit tests pass
- [ ] Load collection with multiple column groups, verify reduced
overhead reservation in logs
- [ ] Verify LoadingOverheadTracker metrics show capped overhead


🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Signed-off-by: Shawn Wang <shawn.wang@zilliz.com>
This commit is contained in:
sparknack
2026-04-08 14:17:38 +08:00
committed by GitHub
parent 151dc63ef2
commit d8bc52d470
10 changed files with 86 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/core/src/segcore/memory_planner.h
+9
View File
@@ -100,6 +100,15 @@ LoadWithStrategy(const std::vector<std::string>& remote_files,
// ---- Cell-batch loading ----
// The channel capacity multiplier relative to pool size.
// The bounded channel holds at most (pool_size * kChannelCapacityMultiplier) items,
// providing backpressure to producer threads while keeping them busy.
constexpr double kChannelCapacityMultiplier = 1.5;
// Safety margin for loading overhead upper bound estimation.
// Covers Arrow Table alignment overhead, metadata, and other estimation errors.
constexpr double kLoadingOverheadInflationRatio = 1.2;
// A cell specification: identifies a cell's location within a specific file.
struct CellSpec {
int64_t cid; // cell id
internal/core/src/segcore/storagev1translator/BsonInvertedIndexTranslator.cpp
+4 -4
View File
@@ -66,11 +66,11 @@ BsonInvertedIndexTranslator::estimated_byte_size_of_cell(
milvus::cachinglayer::cid_t) const {
// ignore the cid checking, because there is only one cell
if (load_info_.enable_mmap) {
return {{0, load_info_.index_size},
{load_info_.index_size, load_info_.index_size}};
// loaded: on disk; overhead: temp memory for download buffer
return {{0, load_info_.index_size}, {load_info_.index_size, 0}};
} else {
return {{load_info_.index_size, 0},
{load_info_.index_size, load_info_.index_size}};
// loaded: in memory; overhead: temp disk for local file before loading
return {{load_info_.index_size, 0}, {0, load_info_.index_size}};
}
}
internal/core/src/segcore/storagev1translator/ChunkTranslator.cpp
+2 -2
View File
@@ -140,10 +140,10 @@ ChunkTranslator::estimated_byte_size_of_cell(
int64_t memory_size = file_infos_[cid].memory_size;
if (use_mmap_) {
// For mmap, the memory is counted as disk usage
return {{0, memory_size}, {memory_size * 2, memory_size * 2}};
return {{0, memory_size}, {memory_size * 2, memory_size}};
} else {
// For non-mmap, the memory is counted as memory usage
return {{memory_size, 0}, {memory_size * 2, 0}};
return {{memory_size, 0}, {memory_size, 0}};
}
}
internal/core/src/segcore/storagev1translator/DefaultValueChunkTranslator.cpp
+2 -2
View File
@@ -214,9 +214,9 @@ DefaultValueChunkTranslator::estimated_byte_size_of_cell(
auto rows = rows_end - rows_begin;
auto cell_bytes = value_size * rows;
if (use_mmap_) {
return {{0, cell_bytes}, {0, cell_bytes}};
return {{0, cell_bytes}, {0, 0}};
} else {
return {{cell_bytes, 0}, {cell_bytes, 0}};
return {{cell_bytes, 0}, {0, 0}};
}
}
internal/core/src/segcore/storagev1translator/InterimSealedIndexTranslator.cpp
+1 -2
View File
@@ -96,8 +96,7 @@ InterimSealedIndexTranslator::estimated_byte_size_of_cell(
} else if (index_type_ == knowhere::IndexEnum::INDEX_FAISS_IVFFLAT_CC) {
// fp16/bf16 all use float32 to build index
int64_t fp32_size = row_count * sizeof(float) * dim_;
return {{fp32_size, 0},
{static_cast<int64_t>(fp32_size + fp32_size * 0.5), 0}};
return {{fp32_size, 0}, {static_cast<int64_t>(fp32_size * 0.5), 0}};
} else {
// SPARSE_WAND_CC and SPARSE_INVERTED_INDEX_CC basically has the same size as the
// raw data.
internal/core/src/segcore/storagev1translator/SealedIndexTranslator.cpp
+3 -2
View File
@@ -100,8 +100,9 @@ SealedIndexTranslator::estimated_byte_size_of_cell(
// this is an estimation, error could be up to 20%.
return {milvus::cachinglayer::ResourceUsage(request.final_memory_cost,
request.final_disk_cost),
milvus::cachinglayer::ResourceUsage(request.max_memory_cost,
request.max_disk_cost * 2)};
milvus::cachinglayer::ResourceUsage(
request.max_memory_cost - request.final_memory_cost,
request.max_disk_cost * 2 - request.final_disk_cost)};
}
const std::string&
internal/core/src/segcore/storagev1translator/TextMatchIndexTranslator.cpp
+2 -4
View File
@@ -67,14 +67,12 @@ TextMatchIndexTranslator::estimated_byte_size_of_cell(
milvus::cachinglayer::cid_t) const {
// ignore the cid checking, because there is only one cell
if (load_info_.enable_mmap) {
return {{0, load_info_.index_size},
{load_info_.index_size, load_info_.index_size}};
return {{0, load_info_.index_size}, {load_info_.index_size, 0}};
} else {
// The reason the maximum disk usage is not zero is that the text match index
// is first written to the disk, then loaded into memory. Only after that are
// the disk files deleted.
return {{load_info_.index_size, 0},
{load_info_.index_size, load_info_.index_size}};
return {{load_info_.index_size, 0}, {0, load_info_.index_size}};
}
}
internal/core/src/segcore/storagev2translator/GroupChunkTranslator.cpp
+33 -1
View File
@@ -231,6 +231,37 @@ GroupChunkTranslator::GroupChunkTranslator(
total_row_groups,
num_cells,
kRowGroupsPerCell);
// Set loading overhead config to cap total overhead reservation.
// During get_cells, decoded Arrow Tables exist simultaneously in:
// - pool threads (pool_size): reading batches, pending push
// - bounded channel (pool_size * kChannelCapacityMultiplier): pushed, awaiting pop
// - main thread (1): being converted to GroupChunk
if (!meta_.chunk_memory_size_.empty()) {
// Use THREAD_POOL_MAX_THREADS_SIZE as the upper bound for pool size.
// This is the global cap applied to all priority pools.
int pool_size = milvus::THREAD_POOL_MAX_THREADS_SIZE.load();
if (pool_size <= 0) {
pool_size = static_cast<int>(std::round(
milvus::CPU_NUM *
milvus::HIGH_PRIORITY_THREAD_CORE_COEFFICIENT.load()));
}
auto max_inflight = static_cast<int64_t>(
pool_size * (1.0 + kChannelCapacityMultiplier) + 1);
int64_t max_cell_sz = *std::max_element(
meta_.chunk_memory_size_.begin(), meta_.chunk_memory_size_.end());
auto ub = static_cast<int64_t>(max_inflight * max_cell_sz *
kLoadingOverheadInflationRatio);
auto upper_bound = use_mmap_
? milvus::cachinglayer::ResourceUsage{ub, ub}
: milvus::cachinglayer::ResourceUsage{ub, 0};
// Group by CellDataType name so all CacheSlots of the same type
// share one overhead upper bound via LoadingOverheadTracker.
auto group = fmt::format("GroupChunkTranslator_{}",
static_cast<int>(meta_.cell_data_type));
meta_.loading_overhead =
milvus::cachinglayer::LoadingOverheadConfig{upper_bound, group};
}
}
GroupChunkTranslator::~GroupChunkTranslator() {
@@ -352,7 +383,8 @@ GroupChunkTranslator::get_cells(milvus::OpContext* ctx,
auto& pool = milvus::ThreadPools::GetThreadPool(
milvus::PriorityForLoad(load_priority_));
auto channel = std::make_shared<milvus::segcore::CellReaderChannel>(
static_cast<size_t>(pool.GetMaxThreadNum() * 1.5));
static_cast<size_t>(pool.GetMaxThreadNum() *
milvus::segcore::kChannelCapacityMultiplier));
auto fs = milvus_storage::ArrowFileSystemSingleton::GetInstance()
.GetArrowFileSystem();
internal/core/src/segcore/storagev2translator/ManifestGroupTranslator.cpp
+29 -1
View File
@@ -170,6 +170,33 @@ ManifestGroupTranslator::ManifestGroupTranslator(
total_row_groups,
num_cells,
kRowGroupsPerCell);
// Set loading overhead config to cap total overhead reservation.
if (!meta_.chunk_memory_size_.empty()) {
// Use THREAD_POOL_MAX_THREADS_SIZE as the upper bound for pool size.
// This is the global cap applied to all priority pools.
int pool_size = milvus::THREAD_POOL_MAX_THREADS_SIZE.load();
if (pool_size <= 0) {
pool_size = static_cast<int>(std::round(
milvus::CPU_NUM *
milvus::HIGH_PRIORITY_THREAD_CORE_COEFFICIENT.load()));
}
auto max_inflight = static_cast<int64_t>(
pool_size * (1.0 + kChannelCapacityMultiplier) + 1);
int64_t max_cell_sz = *std::max_element(
meta_.chunk_memory_size_.begin(), meta_.chunk_memory_size_.end());
auto ub = static_cast<int64_t>(max_inflight * max_cell_sz *
kLoadingOverheadInflationRatio);
auto upper_bound = use_mmap_
? milvus::cachinglayer::ResourceUsage{ub, ub}
: milvus::cachinglayer::ResourceUsage{ub, 0};
// Group by CellDataType name so all CacheSlots of the same type
// share one overhead upper bound via LoadingOverheadTracker.
auto group = fmt::format("ManifestGroupTranslator_{}",
static_cast<int>(meta_.cell_data_type));
meta_.loading_overhead =
milvus::cachinglayer::LoadingOverheadConfig{upper_bound, group};
}
}
size_t
@@ -246,7 +273,8 @@ ManifestGroupTranslator::get_cells(
auto& pool = milvus::ThreadPools::GetThreadPool(
milvus::PriorityForLoad(load_priority_));
auto channel = std::make_shared<milvus::segcore::CellReaderChannel>(
static_cast<size_t>(pool.GetMaxThreadNum() * 1.5));
static_cast<size_t>(pool.GetMaxThreadNum() *
milvus::segcore::kChannelCapacityMultiplier));
auto load_futures =
milvus::segcore::LoadCellBatchAsync(ctx,
internal/core/thirdparty/milvus-common/CMakeLists.txt
Vendored
+1 -1
View File
@@ -13,7 +13,7 @@
milvus_add_pkg_config("milvus-common")
set_property(DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR} PROPERTY INCLUDE_DIRECTORIES "")
set( MILVUS-COMMON-VERSION 5983a18 )
set( MILVUS-COMMON-VERSION 2eb54ae )
set( GIT_REPOSITORY "https://github.com/zilliztech/milvus-common.git")
message(STATUS "milvus-common repo: ${GIT_REPOSITORY}")