mirror of
https://github.com/milvus-io/milvus.git
synced 2026-07-21 10:15:43 +00:00
enhance: configurable byte-size threshold for storage v2 cell packing (#49205)
## Summary This PR ships two related commits: 1. **`enhance: fold calculate_size and write_to_target passes for variable-length chunk writers`** — refactor String/JSON/Geometry chunk writers from two Arrow passes (count + cache) to one Arrow pass that fills a shared `offsets_` member. Eliminates per-row heap allocation (no more `simdjson::padded_string` copies). `AssertInfo` guards `cursor <= UINT32_MAX` before each narrowing cast so oversize chunks fail loudly instead of silently wrapping. 2. **`enhance: configurable byte-size threshold for storage v2 cell packing`** — replace the hardcoded `4 row groups per cell` constant with a runtime-configurable byte-size target driven by `queryNode.segcore.storageV2.cellTargetSizeBytes` (default `4 MiB`, refreshable). - At cell-build time, `rgs_per_cell = max(1, target / avg_row_group_size)`, with cells never crossing file boundaries. - Paramtable `Formatter` rejects non-positive values and falls back to 4 MiB with a warn log. - Value pushed into segcore via a new CGO setter (`SetStorageV2CellTargetSizeBytes`) on QueryNode startup and on paramtable change, backed by an inline atomic in `GroupCTMeta.h`. ## Why The fixed `4` constant does not scale across workloads: large row groups make each cell too fat (coarse eviction, unpredictable memory), small row groups waste cache slots and increase bookkeeping overhead. A byte-target keeps cache cell footprint predictable regardless of parquet row group layout, and gives operators a runtime tuning knob. Default 4 MiB preserves the prior `4 rgs × ~1 MiB/rg` footprint. ## Test plan - [x] Dedicated `ComputeRowGroupsPerCell` unit tests with hardcoded expected outputs pin helper behavior independently of the translator integration tests. - [x] `GroupChunkTranslatorTest` / `ManifestGroupTranslatorTest` read `GetCellTargetSizeBytes()` so assertions track production. - [ ] CI: `ci-v2/build-ut-cov`, `ci-v2/e2e-amd` - [ ] Manual: toggle `queryNode.segcore.storageV2.cellTargetSizeBytes` at runtime and confirm newly loaded segments observe the new pack size. issue: #49204 --------- Signed-off-by: Shawn Wang <shawn.wang@zilliz.com> Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
sparknack
co-authored by
Claude Opus 4.7
Claude Opus 4.7
parent
9cd614f10d
commit
794710a7d5
@@ -518,6 +518,8 @@ queryNode:
|
||||
cacheTtl: 0
|
||||
storageUsageTrackingEnabled: false # Enable storage usage tracking for Tiered Storage. Defaults to false.
|
||||
knowhereScoreConsistency: false # Enable knowhere strong consistency score computation logic
|
||||
storageV2:
|
||||
cellTargetSizeBytes: 4194304 # Target average byte size per storage v2 cache cell. Parquet row groups are greedily packed so that rgs_per_cell * avg_row_group_size ≈ this target. Each cell always contains at least one row group and cells never cross file boundaries. Tune larger for bigger batch IO (fewer cells) or smaller to get finer cache granularity. Default 4 MiB.
|
||||
deleteDumpBatchSize: 10000 # Batch size for delete snapshot dump in segcore.
|
||||
loadMemoryUsageFactor: 1 # The multiply factor of calculating the memory usage while loading segments
|
||||
enableDisk: false # enable querynode load disk index, and search on disk index
|
||||
|
||||
@@ -12,6 +12,7 @@
|
||||
#include "common/ChunkWriter.h"
|
||||
|
||||
#include <cstdint>
|
||||
#include <limits>
|
||||
#include <memory>
|
||||
#include <tuple>
|
||||
#include <utility>
|
||||
@@ -38,164 +39,162 @@ namespace milvus {
|
||||
|
||||
std::pair<size_t, size_t>
|
||||
StringChunkWriter::calculate_size(const arrow::ArrayVector& array_vec) {
|
||||
// Single pass over Arrow: compute row count, absolute offsets, total
|
||||
// size. write_to_target reuses offsets_ and walks Arrow just once more
|
||||
// to emit the string bytes (2 arrow passes total instead of 4).
|
||||
row_nums_ = 0;
|
||||
size_t size = 0;
|
||||
// tuple <data, size, offset>
|
||||
std::vector<std::tuple<const uint8_t*, int64_t, int64_t>> null_bitmaps;
|
||||
for (const auto& data : array_vec) {
|
||||
// for bson, we use binary array to store the string
|
||||
row_nums_ += data->length();
|
||||
}
|
||||
|
||||
const int offset_num = row_nums_ + 1;
|
||||
const size_t null_bitmap_bytes = nullable_ ? (row_nums_ + 7) / 8 : 0;
|
||||
size_t cursor = null_bitmap_bytes + sizeof(uint32_t) * offset_num;
|
||||
|
||||
offsets_.clear();
|
||||
offsets_.reserve(offset_num);
|
||||
for (const auto& data : array_vec) {
|
||||
auto array = std::dynamic_pointer_cast<arrow::BinaryArray>(data);
|
||||
for (int i = 0; i < array->length(); i++) {
|
||||
auto str = array->GetView(i);
|
||||
size += str.size();
|
||||
offsets_.push_back(static_cast<uint32_t>(cursor));
|
||||
cursor += array->GetView(i).size();
|
||||
}
|
||||
row_nums_ += array->length();
|
||||
}
|
||||
if (nullable_) {
|
||||
size += (row_nums_ + 7) / 8;
|
||||
}
|
||||
size += sizeof(uint32_t) * (row_nums_ + 1) + MMAP_STRING_PADDING;
|
||||
// String chunk uses uint32 offsets on disk; reject oversize chunks loudly
|
||||
// rather than silently wrapping.
|
||||
AssertInfo(cursor <= std::numeric_limits<uint32_t>::max(),
|
||||
"string chunk size {} exceeds uint32 offset limit",
|
||||
cursor);
|
||||
offsets_.push_back(static_cast<uint32_t>(cursor));
|
||||
|
||||
size_t size = cursor + MMAP_STRING_PADDING;
|
||||
return {size, row_nums_};
|
||||
}
|
||||
|
||||
void
|
||||
StringChunkWriter::write_to_target(const arrow::ArrayVector& array_vec,
|
||||
const std::shared_ptr<ChunkTarget>& target) {
|
||||
strs_.clear();
|
||||
strs_.reserve(row_nums_);
|
||||
// tuple <data, size, offset>
|
||||
std::vector<std::tuple<const uint8_t*, int64_t, int64_t>> null_bitmaps;
|
||||
for (const auto& data : array_vec) {
|
||||
// for bson, we use binary array to store the string
|
||||
auto array = std::dynamic_pointer_cast<arrow::BinaryArray>(data);
|
||||
for (int i = 0; i < array->length(); i++) {
|
||||
auto str = array->GetView(i);
|
||||
strs_.emplace_back(str);
|
||||
}
|
||||
if (nullable_) {
|
||||
// chunk layout: null bitmap, offsets[row_nums_+1], str1..strN, padding
|
||||
if (nullable_) {
|
||||
std::vector<std::tuple<const uint8_t*, int64_t, int64_t>> null_bitmaps;
|
||||
null_bitmaps.reserve(array_vec.size());
|
||||
for (const auto& data : array_vec) {
|
||||
null_bitmaps.emplace_back(
|
||||
data->null_bitmap_data(), data->length(), data->offset());
|
||||
}
|
||||
write_null_bit_maps(null_bitmaps, target);
|
||||
}
|
||||
|
||||
// chunk layout: null bitmap, offset1, offset2, ..., offsetn, str1, str2, ..., strn, padding
|
||||
// write null bitmaps
|
||||
write_null_bit_maps(null_bitmaps, target);
|
||||
target->write(offsets_.data(), offsets_.size() * sizeof(uint32_t));
|
||||
|
||||
// write data
|
||||
const int offset_num = row_nums_ + 1;
|
||||
const uint32_t null_bitmap_bytes =
|
||||
nullable_ ? static_cast<uint32_t>((row_nums_ + 7) / 8) : 0;
|
||||
uint32_t offset_start_pos =
|
||||
null_bitmap_bytes + sizeof(uint32_t) * offset_num;
|
||||
std::vector<uint32_t> offsets;
|
||||
offsets.reserve(offset_num);
|
||||
for (const auto& str : strs_) {
|
||||
offsets.push_back(offset_start_pos);
|
||||
offset_start_pos += str.size();
|
||||
}
|
||||
offsets.push_back(offset_start_pos);
|
||||
|
||||
target->write(offsets.data(), offsets.size() * sizeof(uint32_t));
|
||||
for (const auto& str : strs_) {
|
||||
target->write(str.data(), str.size());
|
||||
for (const auto& data : array_vec) {
|
||||
auto array = std::dynamic_pointer_cast<arrow::BinaryArray>(data);
|
||||
for (int i = 0; i < array->length(); i++) {
|
||||
auto str = array->GetView(i);
|
||||
target->write(str.data(), str.size());
|
||||
}
|
||||
}
|
||||
|
||||
// write padding, maybe not needed anymore
|
||||
// FIXME
|
||||
char padding[MMAP_STRING_PADDING];
|
||||
char padding[MMAP_STRING_PADDING] = {};
|
||||
target->write(padding, MMAP_STRING_PADDING);
|
||||
|
||||
offsets_.clear();
|
||||
offsets_.shrink_to_fit();
|
||||
}
|
||||
|
||||
std::pair<size_t, size_t>
|
||||
JSONChunkWriter::calculate_size(const arrow::ArrayVector& array_vec) {
|
||||
// Single pass over Arrow: compute row count, absolute offsets, total
|
||||
// size. No per-row simdjson::padded_string allocation — write_to_target
|
||||
// copies bytes directly from the Arrow buffer and emits a single
|
||||
// SIMDJSON_PADDING region at the tail.
|
||||
row_nums_ = 0;
|
||||
size_t size = 0;
|
||||
cached_jsons_.clear();
|
||||
|
||||
// First pass: count total rows for reserve
|
||||
for (const auto& data : array_vec) {
|
||||
row_nums_ += data->length();
|
||||
}
|
||||
cached_jsons_.reserve(row_nums_);
|
||||
|
||||
// Second pass: parse and cache Json objects
|
||||
const int offset_num = row_nums_ + 1;
|
||||
const size_t null_bitmap_bytes = nullable_ ? (row_nums_ + 7) / 8 : 0;
|
||||
size_t cursor = null_bitmap_bytes + sizeof(uint32_t) * offset_num;
|
||||
|
||||
offsets_.clear();
|
||||
offsets_.reserve(offset_num);
|
||||
for (const auto& data : array_vec) {
|
||||
auto array = std::dynamic_pointer_cast<arrow::BinaryArray>(data);
|
||||
for (int i = 0; i < array->length(); i++) {
|
||||
auto str = array->GetView(i);
|
||||
cached_jsons_.emplace_back(simdjson::padded_string(str));
|
||||
size += cached_jsons_.back().data().size();
|
||||
offsets_.push_back(static_cast<uint32_t>(cursor));
|
||||
cursor += array->GetView(i).size();
|
||||
}
|
||||
}
|
||||
if (nullable_) {
|
||||
size += (row_nums_ + 7) / 8;
|
||||
}
|
||||
size += sizeof(uint32_t) * (row_nums_ + 1) + simdjson::SIMDJSON_PADDING;
|
||||
AssertInfo(cursor <= std::numeric_limits<uint32_t>::max(),
|
||||
"json chunk size {} exceeds uint32 offset limit",
|
||||
cursor);
|
||||
offsets_.push_back(static_cast<uint32_t>(cursor));
|
||||
|
||||
size_t size = cursor + simdjson::SIMDJSON_PADDING;
|
||||
return {size, row_nums_};
|
||||
}
|
||||
|
||||
void
|
||||
JSONChunkWriter::write_to_target(const arrow::ArrayVector& array_vec,
|
||||
const std::shared_ptr<ChunkTarget>& target) {
|
||||
// Collect null bitmaps
|
||||
std::vector<std::tuple<const uint8_t*, int64_t, int64_t>> null_bitmaps;
|
||||
// chunk layout: null bitmap, offsets[row_nums_+1], json1..jsonN, padding
|
||||
if (nullable_) {
|
||||
std::vector<std::tuple<const uint8_t*, int64_t, int64_t>> null_bitmaps;
|
||||
null_bitmaps.reserve(array_vec.size());
|
||||
for (const auto& data : array_vec) {
|
||||
null_bitmaps.emplace_back(
|
||||
data->null_bitmap_data(), data->length(), data->offset());
|
||||
}
|
||||
write_null_bit_maps(null_bitmaps, target);
|
||||
}
|
||||
|
||||
// chunk layout: null bitmaps, offset1, offset2, ... ,json1, json2, ..., jsonn
|
||||
// write null bitmaps
|
||||
write_null_bit_maps(null_bitmaps, target);
|
||||
target->write(offsets_.data(), offsets_.size() * sizeof(uint32_t));
|
||||
|
||||
const int offset_num = row_nums_ + 1;
|
||||
const uint32_t null_bitmap_bytes =
|
||||
nullable_ ? static_cast<uint32_t>((row_nums_ + 7) / 8) : 0;
|
||||
uint32_t offset_start_pos =
|
||||
null_bitmap_bytes + sizeof(uint32_t) * offset_num;
|
||||
std::vector<uint32_t> offsets;
|
||||
offsets.reserve(offset_num);
|
||||
for (const auto& json : cached_jsons_) {
|
||||
offsets.push_back(offset_start_pos);
|
||||
offset_start_pos += json.data().size();
|
||||
}
|
||||
offsets.push_back(offset_start_pos);
|
||||
|
||||
target->write(offsets.data(), offset_num * sizeof(uint32_t));
|
||||
|
||||
// write data
|
||||
for (const auto& json : cached_jsons_) {
|
||||
target->write(json.data().data(), json.data().size());
|
||||
for (const auto& data : array_vec) {
|
||||
auto array = std::dynamic_pointer_cast<arrow::BinaryArray>(data);
|
||||
for (int i = 0; i < array->length(); i++) {
|
||||
auto str = array->GetView(i);
|
||||
target->write(str.data(), str.size());
|
||||
}
|
||||
}
|
||||
|
||||
char padding[simdjson::SIMDJSON_PADDING];
|
||||
char padding[simdjson::SIMDJSON_PADDING] = {};
|
||||
target->write(padding, simdjson::SIMDJSON_PADDING);
|
||||
|
||||
// Release cached data
|
||||
cached_jsons_.clear();
|
||||
cached_jsons_.shrink_to_fit();
|
||||
offsets_.clear();
|
||||
offsets_.shrink_to_fit();
|
||||
}
|
||||
|
||||
std::pair<size_t, size_t>
|
||||
GeometryChunkWriter::calculate_size(const arrow::ArrayVector& array_vec) {
|
||||
// Same pattern as String/JSON: single Arrow pass produces offsets_ and
|
||||
// total size; write_to_target reuses offsets_ + walks Arrow once for
|
||||
// the WKB bytes.
|
||||
row_nums_ = 0;
|
||||
size_t size = 0;
|
||||
for (const auto& data : array_vec) {
|
||||
row_nums_ += data->length();
|
||||
}
|
||||
|
||||
const int offset_num = row_nums_ + 1;
|
||||
const size_t null_bitmap_bytes = nullable_ ? (row_nums_ + 7) / 8 : 0;
|
||||
size_t cursor = null_bitmap_bytes + sizeof(uint32_t) * offset_num;
|
||||
|
||||
offsets_.clear();
|
||||
offsets_.reserve(offset_num);
|
||||
for (const auto& data : array_vec) {
|
||||
auto array = std::dynamic_pointer_cast<arrow::BinaryArray>(data);
|
||||
for (int64_t i = 0; i < array->length(); ++i) {
|
||||
auto str = array->GetView(i);
|
||||
size += str.size();
|
||||
offsets_.push_back(static_cast<uint32_t>(cursor));
|
||||
cursor += array->GetView(i).size();
|
||||
}
|
||||
row_nums_ += array->length();
|
||||
}
|
||||
if (nullable_) {
|
||||
size += (row_nums_ + 7) / 8;
|
||||
}
|
||||
size += sizeof(uint32_t) * (row_nums_ + 1) + MMAP_GEOMETRY_PADDING;
|
||||
AssertInfo(cursor <= std::numeric_limits<uint32_t>::max(),
|
||||
"geometry chunk size {} exceeds uint32 offset limit",
|
||||
cursor);
|
||||
offsets_.push_back(static_cast<uint32_t>(cursor));
|
||||
|
||||
size_t size = cursor + MMAP_GEOMETRY_PADDING;
|
||||
return {size, row_nums_};
|
||||
}
|
||||
|
||||
@@ -203,74 +202,75 @@ void
|
||||
GeometryChunkWriter::write_to_target(
|
||||
const arrow::ArrayVector& array_vec,
|
||||
const std::shared_ptr<ChunkTarget>& target) {
|
||||
std::vector<std::string_view> wkb_strs;
|
||||
std::vector<std::tuple<const uint8_t*, int64_t, int64_t>> null_bitmaps;
|
||||
wkb_strs.reserve(row_nums_);
|
||||
// chunk layout: null bitmap, offsets, wkb strings, padding
|
||||
if (nullable_) {
|
||||
std::vector<std::tuple<const uint8_t*, int64_t, int64_t>> null_bitmaps;
|
||||
null_bitmaps.reserve(array_vec.size());
|
||||
for (const auto& data : array_vec) {
|
||||
null_bitmaps.emplace_back(
|
||||
data->null_bitmap_data(), data->length(), data->offset());
|
||||
}
|
||||
write_null_bit_maps(null_bitmaps, target);
|
||||
}
|
||||
|
||||
target->write(offsets_.data(), offsets_.size() * sizeof(uint32_t));
|
||||
|
||||
for (const auto& data : array_vec) {
|
||||
auto array = std::dynamic_pointer_cast<arrow::BinaryArray>(data);
|
||||
for (int64_t i = 0; i < array->length(); ++i) {
|
||||
auto str = array->GetView(i);
|
||||
wkb_strs.emplace_back(str);
|
||||
}
|
||||
if (nullable_) {
|
||||
null_bitmaps.emplace_back(
|
||||
data->null_bitmap_data(), data->length(), data->offset());
|
||||
target->write(str.data(), str.size());
|
||||
}
|
||||
}
|
||||
|
||||
// chunk layout: null bitmap, offsets, wkb strings, padding
|
||||
write_null_bit_maps(null_bitmaps, target);
|
||||
|
||||
const int offset_num = row_nums_ + 1;
|
||||
const uint32_t null_bitmap_bytes =
|
||||
nullable_ ? static_cast<uint32_t>((row_nums_ + 7) / 8) : 0;
|
||||
uint32_t offset_start_pos =
|
||||
null_bitmap_bytes +
|
||||
static_cast<uint32_t>(sizeof(uint32_t) * offset_num);
|
||||
std::vector<uint32_t> offsets;
|
||||
offsets.reserve(offset_num);
|
||||
for (const auto& str : wkb_strs) {
|
||||
offsets.push_back(offset_start_pos);
|
||||
offset_start_pos += str.size();
|
||||
}
|
||||
offsets.push_back(offset_start_pos);
|
||||
|
||||
target->write(offsets.data(), offsets.size() * sizeof(uint32_t));
|
||||
|
||||
for (const auto& str : wkb_strs) {
|
||||
target->write(str.data(), str.size());
|
||||
}
|
||||
|
||||
char padding[MMAP_GEOMETRY_PADDING];
|
||||
char padding[MMAP_GEOMETRY_PADDING] = {};
|
||||
target->write(padding, MMAP_GEOMETRY_PADDING);
|
||||
|
||||
offsets_.clear();
|
||||
offsets_.shrink_to_fit();
|
||||
}
|
||||
|
||||
std::pair<size_t, size_t>
|
||||
ArrayChunkWriter::calculate_size(const arrow::ArrayVector& array_vec) {
|
||||
row_nums_ = 0;
|
||||
size_t size = 0;
|
||||
// Parse ScalarFieldProto once per row here and cache the resulting
|
||||
// Array objects so write_to_target does not re-parse. Also produce the
|
||||
// interleaved [off0, len0, off1, len1, ..., offN-1, lenN-1, offN] header
|
||||
// so write_to_target can emit it in a single target->write call.
|
||||
const bool is_string = IsStringDataType(element_type_);
|
||||
|
||||
row_nums_ = 0;
|
||||
for (const auto& data : array_vec) {
|
||||
row_nums_ += data->length();
|
||||
}
|
||||
|
||||
cached_arrays_.clear();
|
||||
cached_arrays_.reserve(row_nums_);
|
||||
header_.clear();
|
||||
header_.reserve(row_nums_ * 2 + 1);
|
||||
|
||||
const int header_entries = row_nums_ * 2 + 1;
|
||||
const size_t null_bitmap_bytes = nullable_ ? (row_nums_ + 7) / 8 : 0;
|
||||
size_t cursor = null_bitmap_bytes + sizeof(uint32_t) * header_entries;
|
||||
|
||||
for (const auto& data : array_vec) {
|
||||
auto array = std::dynamic_pointer_cast<arrow::BinaryArray>(data);
|
||||
for (int64_t i = 0; i < array->length(); ++i) {
|
||||
auto str = array->GetView(i);
|
||||
ScalarFieldProto scalar_array;
|
||||
scalar_array.ParseFromArray(str.data(), str.size());
|
||||
Array arr(scalar_array);
|
||||
size += arr.byte_size();
|
||||
cached_arrays_.emplace_back(scalar_array);
|
||||
const auto& arr = cached_arrays_.back();
|
||||
header_.push_back(static_cast<uint32_t>(cursor)); // off_i
|
||||
header_.push_back(static_cast<uint32_t>(arr.length())); // len_i
|
||||
if (is_string) {
|
||||
size += sizeof(uint32_t) * arr.length();
|
||||
cursor += sizeof(uint32_t) * arr.length();
|
||||
}
|
||||
cursor += arr.byte_size();
|
||||
}
|
||||
}
|
||||
header_.push_back(static_cast<uint32_t>(cursor)); // off_N (sentinel)
|
||||
|
||||
row_nums_ += array->length();
|
||||
}
|
||||
if (nullable_) {
|
||||
size += (row_nums_ + 7) / 8;
|
||||
}
|
||||
size += sizeof(uint32_t) * (row_nums_ * 2 + 1) + MMAP_ARRAY_PADDING;
|
||||
size_t size = cursor + MMAP_ARRAY_PADDING;
|
||||
return {size, row_nums_};
|
||||
}
|
||||
|
||||
@@ -278,57 +278,21 @@ void
|
||||
ArrayChunkWriter::write_to_target(const arrow::ArrayVector& array_vec,
|
||||
const std::shared_ptr<ChunkTarget>& target) {
|
||||
const bool is_string = IsStringDataType(element_type_);
|
||||
std::vector<Array> arrays;
|
||||
arrays.reserve(row_nums_);
|
||||
std::vector<std::tuple<const uint8_t*, int64_t, int64_t>> null_bitmaps;
|
||||
|
||||
for (const auto& data : array_vec) {
|
||||
auto array = std::dynamic_pointer_cast<arrow::BinaryArray>(data);
|
||||
for (int64_t i = 0; i < array->length(); ++i) {
|
||||
auto str = array->GetView(i);
|
||||
ScalarFieldProto scalar_array;
|
||||
scalar_array.ParseFromArray(str.data(), str.size());
|
||||
arrays.emplace_back(scalar_array);
|
||||
}
|
||||
if (nullable_) {
|
||||
if (nullable_) {
|
||||
std::vector<std::tuple<const uint8_t*, int64_t, int64_t>> null_bitmaps;
|
||||
null_bitmaps.reserve(array_vec.size());
|
||||
for (const auto& data : array_vec) {
|
||||
null_bitmaps.emplace_back(
|
||||
data->null_bitmap_data(), data->length(), data->offset());
|
||||
}
|
||||
write_null_bit_maps(null_bitmaps, target);
|
||||
}
|
||||
|
||||
write_null_bit_maps(null_bitmaps, target);
|
||||
// Header: interleaved [off0, len0, off1, len1, ..., offN-1, lenN-1, offN]
|
||||
target->write(header_.data(), header_.size() * sizeof(uint32_t));
|
||||
|
||||
const int offsets_num = row_nums_ + 1;
|
||||
const int len_num = row_nums_;
|
||||
const uint32_t null_bitmap_bytes =
|
||||
nullable_ ? static_cast<uint32_t>((row_nums_ + 7) / 8) : 0;
|
||||
uint32_t offset_start_pos =
|
||||
null_bitmap_bytes + sizeof(uint32_t) * (offsets_num + len_num);
|
||||
|
||||
std::vector<uint32_t> offsets(offsets_num);
|
||||
std::vector<uint32_t> lens(len_num);
|
||||
|
||||
for (size_t i = 0; i < arrays.size(); ++i) {
|
||||
auto& arr = arrays[i];
|
||||
offsets[i] = offset_start_pos;
|
||||
lens[i] = arr.length();
|
||||
if (is_string) {
|
||||
offset_start_pos += sizeof(uint32_t) * lens[i];
|
||||
}
|
||||
offset_start_pos += arr.byte_size();
|
||||
}
|
||||
|
||||
if (!offsets.empty()) {
|
||||
offsets.back() = offset_start_pos;
|
||||
}
|
||||
|
||||
for (int i = 0; i < row_nums_; ++i) {
|
||||
target->write(&offsets[i], sizeof(uint32_t));
|
||||
target->write(&lens[i], sizeof(uint32_t));
|
||||
}
|
||||
target->write(&offsets.back(), sizeof(uint32_t));
|
||||
|
||||
for (auto& arr : arrays) {
|
||||
for (auto& arr : cached_arrays_) {
|
||||
if (is_string) {
|
||||
target->write(arr.get_offsets_data(),
|
||||
arr.length() * sizeof(uint32_t));
|
||||
@@ -336,8 +300,13 @@ ArrayChunkWriter::write_to_target(const arrow::ArrayVector& array_vec,
|
||||
target->write(arr.data(), arr.byte_size());
|
||||
}
|
||||
|
||||
char padding[MMAP_ARRAY_PADDING];
|
||||
char padding[MMAP_ARRAY_PADDING] = {};
|
||||
target->write(padding, MMAP_ARRAY_PADDING);
|
||||
|
||||
cached_arrays_.clear();
|
||||
cached_arrays_.shrink_to_fit();
|
||||
header_.clear();
|
||||
header_.shrink_to_fit();
|
||||
}
|
||||
|
||||
std::pair<size_t, size_t>
|
||||
|
||||
@@ -230,7 +230,11 @@ class StringChunkWriter : public ChunkWriterBase {
|
||||
const std::shared_ptr<ChunkTarget>& target) override;
|
||||
|
||||
private:
|
||||
std::vector<std::string_view> strs_;
|
||||
// Pre-computed absolute offsets (offsets_[i] = byte offset of row i from
|
||||
// chunk start, offsets_[row_nums_] = end offset). Populated in
|
||||
// calculate_size, consumed in write_to_target to avoid a second pass over
|
||||
// Arrow for sizing.
|
||||
std::vector<uint32_t> offsets_;
|
||||
};
|
||||
|
||||
class JSONChunkWriter : public ChunkWriterBase {
|
||||
@@ -245,7 +249,10 @@ class JSONChunkWriter : public ChunkWriterBase {
|
||||
const std::shared_ptr<ChunkTarget>& target) override;
|
||||
|
||||
private:
|
||||
std::vector<Json> cached_jsons_;
|
||||
// Same shape as StringChunkWriter::offsets_. No per-row simdjson padding
|
||||
// buffer is kept — we write Arrow bytes directly and emit a single
|
||||
// SIMDJSON_PADDING region at the tail.
|
||||
std::vector<uint32_t> offsets_;
|
||||
};
|
||||
|
||||
class GeometryChunkWriter : public ChunkWriterBase {
|
||||
@@ -258,6 +265,9 @@ class GeometryChunkWriter : public ChunkWriterBase {
|
||||
void
|
||||
write_to_target(const arrow::ArrayVector& array_vec,
|
||||
const std::shared_ptr<ChunkTarget>& target) override;
|
||||
|
||||
private:
|
||||
std::vector<uint32_t> offsets_;
|
||||
};
|
||||
|
||||
class ArrayChunkWriter : public ChunkWriterBase {
|
||||
@@ -275,6 +285,13 @@ class ArrayChunkWriter : public ChunkWriterBase {
|
||||
|
||||
private:
|
||||
const milvus::DataType element_type_;
|
||||
// Parsed protobufs cached by calculate_size so write_to_target does not
|
||||
// pay a second ScalarFieldProto parse per row.
|
||||
std::vector<Array> cached_arrays_;
|
||||
// Interleaved [off0, len0, off1, len1, ..., offN-1, lenN-1, offN] header.
|
||||
// Populated by calculate_size so write_to_target can emit the whole
|
||||
// header in a single target->write call.
|
||||
std::vector<uint32_t> header_;
|
||||
};
|
||||
|
||||
class VectorArrayChunkWriter : public ChunkWriterBase {
|
||||
|
||||
@@ -25,6 +25,7 @@
|
||||
#include "common/init_c.h"
|
||||
#include "exec/expression/ExprCache.h"
|
||||
#include "log/Log.h"
|
||||
#include "segcore/storagev2translator/GroupCTMeta.h"
|
||||
#include "storage/ThreadPool.h"
|
||||
|
||||
std::once_flag traceFlag;
|
||||
@@ -127,6 +128,11 @@ SetArrowIOThreadPoolCapacity(int threads) {
|
||||
LOG_INFO("arrow io thread pool capacity set to {}", threads);
|
||||
}
|
||||
|
||||
void
|
||||
SetStorageV2CellTargetSizeBytes(int64_t bytes) {
|
||||
milvus::segcore::storagev2translator::SetCellTargetSizeBytes(bytes);
|
||||
}
|
||||
|
||||
void
|
||||
LogOpenSSLFIPSStatus() {
|
||||
std::call_once(fipsFlag, []() {
|
||||
|
||||
@@ -98,6 +98,11 @@ SetExprResCacheCapacityBytes(int64_t bytes);
|
||||
void
|
||||
SetArrowIOThreadPoolCapacity(int threads);
|
||||
|
||||
// Target average byte size of one storage v2 cache cell. Row groups are
|
||||
// packed into cells so that rgs_per_cell * avg_row_group_size ≈ this value.
|
||||
void
|
||||
SetStorageV2CellTargetSizeBytes(int64_t bytes);
|
||||
|
||||
#ifdef __cplusplus
|
||||
};
|
||||
#endif
|
||||
|
||||
@@ -16,7 +16,9 @@
|
||||
#pragma once
|
||||
|
||||
#include <algorithm>
|
||||
#include <atomic>
|
||||
#include <cstdint>
|
||||
#include <type_traits>
|
||||
#include <utility>
|
||||
#include <vector>
|
||||
|
||||
@@ -24,12 +26,55 @@
|
||||
|
||||
namespace milvus::segcore::storagev2translator {
|
||||
|
||||
// Number of row groups (parquet row groups) merged into one cache cell,
|
||||
// for now it is a constant.
|
||||
// hierarchy: 1 group chunk <-> 1 cache cell <-> kRowGroupsPerCell row groups
|
||||
constexpr size_t kRowGroupsPerCell = 4;
|
||||
static_assert(kRowGroupsPerCell > 0,
|
||||
"kRowGroupsPerCell must be greater than 0");
|
||||
// Target average byte size per storage-v2 cache cell. Parquet row groups
|
||||
// are packed into cells so that `rgs_per_cell * avg_row_group_size ≈ target`.
|
||||
// Runtime-configurable via SetCellTargetSizeBytes (wired from paramtable
|
||||
// `queryNode.segcore.storageV2.cellTargetSizeBytes`).
|
||||
inline std::atomic<int64_t>&
|
||||
cell_target_size_bytes_atomic() {
|
||||
static std::atomic<int64_t> instance{4LL * 1024 * 1024}; // init: 4 MiB
|
||||
return instance;
|
||||
}
|
||||
|
||||
inline int64_t
|
||||
GetCellTargetSizeBytes() {
|
||||
return cell_target_size_bytes_atomic().load(std::memory_order_acquire);
|
||||
}
|
||||
|
||||
inline void
|
||||
SetCellTargetSizeBytes(int64_t v) {
|
||||
if (v <= 0) {
|
||||
return; // ignore invalid
|
||||
}
|
||||
cell_target_size_bytes_atomic().store(v, std::memory_order_release);
|
||||
}
|
||||
|
||||
// Derive the number of row groups per cell from a byte-size target so
|
||||
// that each cell holds >= 1 row group and the average cell size is
|
||||
// close to cell_target_size_bytes.
|
||||
// Templated so it accepts both std::vector<int64_t> (built locally by
|
||||
// GroupChunkTranslator) and std::vector<uint64_t>/size_t (returned by
|
||||
// milvus-storage's chunk_reader->get_chunk_size()).
|
||||
template <typename T>
|
||||
inline size_t
|
||||
ComputeRowGroupsPerCell(const std::vector<T>& row_group_sizes,
|
||||
int64_t cell_target_size_bytes) {
|
||||
static_assert(std::is_arithmetic_v<T>,
|
||||
"ComputeRowGroupsPerCell expects a vector of numeric sizes");
|
||||
if (row_group_sizes.empty()) {
|
||||
return 1;
|
||||
}
|
||||
int64_t total = 0;
|
||||
for (auto s : row_group_sizes) {
|
||||
total += static_cast<int64_t>(s);
|
||||
}
|
||||
int64_t avg = total / static_cast<int64_t>(row_group_sizes.size());
|
||||
if (avg <= 0) {
|
||||
return 1;
|
||||
}
|
||||
size_t n = static_cast<size_t>(cell_target_size_bytes / avg);
|
||||
return std::max<size_t>(n, 1);
|
||||
}
|
||||
|
||||
struct GroupCTMeta : public milvus::cachinglayer::Meta {
|
||||
// num_rows_until_chunk_[i] = total rows(prefix sum) in cells [0, i-1]
|
||||
|
||||
@@ -150,15 +150,19 @@ GroupChunkTranslator::GroupChunkTranslator(
|
||||
}
|
||||
|
||||
// Build cell mapping: cells DO NOT span files — each cell's row groups
|
||||
// come entirely from one file.
|
||||
// come entirely from one file. Derive row-groups-per-cell from the
|
||||
// runtime-configurable target cell byte size so avg cell size ≈ target.
|
||||
const int64_t cell_target_size_bytes = GetCellTargetSizeBytes();
|
||||
meta_.total_row_groups_ = total_row_groups;
|
||||
const size_t rgs_per_cell =
|
||||
ComputeRowGroupsPerCell(row_group_sizes, cell_target_size_bytes);
|
||||
size_t global_rg_offset = 0;
|
||||
for (const auto& rg_meta : row_group_meta_list_) {
|
||||
size_t file_rg_count = rg_meta.size();
|
||||
for (size_t local_start = 0; local_start < file_rg_count;
|
||||
local_start += kRowGroupsPerCell) {
|
||||
local_start += rgs_per_cell) {
|
||||
size_t local_end =
|
||||
std::min(local_start + kRowGroupsPerCell, file_rg_count);
|
||||
std::min(local_start + rgs_per_cell, file_rg_count);
|
||||
meta_.cell_row_group_ranges_.push_back(
|
||||
{global_rg_offset + local_start, global_rg_offset + local_end});
|
||||
}
|
||||
@@ -194,12 +198,12 @@ GroupChunkTranslator::GroupChunkTranslator(
|
||||
column_group_info_.row_count));
|
||||
|
||||
LOG_INFO(
|
||||
"[StorageV2] translator {} merged {} row groups into {} cells ({} "
|
||||
"row groups per cell)",
|
||||
"[StorageV2] translator {} merged {} row groups into {} cells "
|
||||
"(cell_target_size_bytes={})",
|
||||
key_,
|
||||
total_row_groups,
|
||||
num_cells,
|
||||
kRowGroupsPerCell);
|
||||
cell_target_size_bytes);
|
||||
|
||||
// Set loading overhead config to cap total overhead reservation.
|
||||
// During get_cells, decoded Arrow Tables exist simultaneously in:
|
||||
|
||||
@@ -142,12 +142,18 @@ TEST_P(GroupChunkTranslatorTest, TestWithMmap) {
|
||||
"[StorageV2] Failed to create file row group reader: " +
|
||||
reader_result.status().ToString());
|
||||
auto fr = reader_result.ValueOrDie();
|
||||
auto expected_num_cells =
|
||||
(fr->file_metadata()->GetRowGroupMetadataVector().size() +
|
||||
kRowGroupsPerCell - 1) /
|
||||
kRowGroupsPerCell;
|
||||
auto row_group_metadata_vector =
|
||||
fr->file_metadata()->GetRowGroupMetadataVector();
|
||||
std::vector<int64_t> row_group_sizes;
|
||||
row_group_sizes.reserve(row_group_metadata_vector.size());
|
||||
for (int i = 0; i < row_group_metadata_vector.size(); ++i) {
|
||||
row_group_sizes.push_back(static_cast<int64_t>(
|
||||
row_group_metadata_vector.Get(i).memory_size()));
|
||||
}
|
||||
auto rgs_per_cell =
|
||||
ComputeRowGroupsPerCell(row_group_sizes, GetCellTargetSizeBytes());
|
||||
auto expected_num_cells =
|
||||
(row_group_metadata_vector.size() + rgs_per_cell - 1) / rgs_per_cell;
|
||||
auto status = fr->Close();
|
||||
AssertInfo(status.ok(), "failed to close file reader");
|
||||
EXPECT_EQ(translator->num_cells(), expected_num_cells);
|
||||
@@ -297,11 +303,27 @@ TEST_P(GroupChunkTranslatorTest, TestMultipleFiles) {
|
||||
/* warmup_policy */ "");
|
||||
|
||||
// Test total number of cells across all files
|
||||
// Cells never span files, so count per-file ceil
|
||||
// Cells never span files, so count per-file ceil. The cell-per-count is
|
||||
// derived from the average row-group size (kDefaultCellTargetSizeBytes)
|
||||
// across the same aggregated sizes the translator sees.
|
||||
std::vector<int64_t> all_row_group_sizes;
|
||||
for (const auto& file_path : multi_file_paths) {
|
||||
auto fr_result =
|
||||
milvus_storage::FileRowGroupReader::Make(fs_, file_path);
|
||||
ASSERT_TRUE(fr_result.ok());
|
||||
auto fr = fr_result.ValueOrDie();
|
||||
auto rgmv = fr->file_metadata()->GetRowGroupMetadataVector();
|
||||
for (int i = 0; i < rgmv.size(); ++i) {
|
||||
all_row_group_sizes.push_back(
|
||||
static_cast<int64_t>(rgmv.Get(i).memory_size()));
|
||||
}
|
||||
ASSERT_TRUE(fr->Close().ok());
|
||||
}
|
||||
auto rgs_per_cell =
|
||||
ComputeRowGroupsPerCell(all_row_group_sizes, GetCellTargetSizeBytes());
|
||||
int64_t expected_total_cells = 0;
|
||||
for (auto row_groups : expected_row_groups_per_file) {
|
||||
expected_total_cells +=
|
||||
(row_groups + kRowGroupsPerCell - 1) / kRowGroupsPerCell;
|
||||
expected_total_cells += (row_groups + rgs_per_cell - 1) / rgs_per_cell;
|
||||
}
|
||||
EXPECT_EQ(translator->num_cells(), expected_total_cells);
|
||||
|
||||
@@ -402,3 +424,59 @@ TEST_P(GroupChunkTranslatorTest, TestMultipleFiles) {
|
||||
INSTANTIATE_TEST_SUITE_P(GroupChunkTranslatorTest,
|
||||
GroupChunkTranslatorTest,
|
||||
testing::Bool());
|
||||
|
||||
// Pins ComputeRowGroupsPerCell behavior with hardcoded inputs so a regression
|
||||
// in the helper fails here independently of the integration tests above,
|
||||
// which feed the helper's own output into their expectations.
|
||||
TEST(ComputeRowGroupsPerCellTest, EmptyReturnsOne) {
|
||||
std::vector<int64_t> sizes;
|
||||
EXPECT_EQ(ComputeRowGroupsPerCell(sizes, 2 * 1024 * 1024), 1u);
|
||||
}
|
||||
|
||||
TEST(ComputeRowGroupsPerCellTest, SingleRowGroup) {
|
||||
std::vector<int64_t> sizes{512 * 1024};
|
||||
// target much larger than rg, but n is floored by rg count semantics -
|
||||
// helper itself returns target/avg; caller clamps by rg count.
|
||||
EXPECT_EQ(ComputeRowGroupsPerCell(sizes, 4 * 1024 * 1024), 8u);
|
||||
EXPECT_EQ(ComputeRowGroupsPerCell(sizes, 512 * 1024), 1u);
|
||||
}
|
||||
|
||||
TEST(ComputeRowGroupsPerCellTest, TargetEqualToAverage) {
|
||||
std::vector<int64_t> sizes{1024 * 1024, 1024 * 1024, 1024 * 1024};
|
||||
EXPECT_EQ(ComputeRowGroupsPerCell(sizes, 1024 * 1024), 1u);
|
||||
}
|
||||
|
||||
TEST(ComputeRowGroupsPerCellTest, TargetMultipleOfAverage) {
|
||||
std::vector<int64_t> sizes{1024 * 1024, 1024 * 1024, 1024 * 1024};
|
||||
EXPECT_EQ(ComputeRowGroupsPerCell(sizes, 4 * 1024 * 1024), 4u);
|
||||
EXPECT_EQ(ComputeRowGroupsPerCell(sizes, 8 * 1024 * 1024), 8u);
|
||||
}
|
||||
|
||||
TEST(ComputeRowGroupsPerCellTest, TargetSmallerThanAverageClampsToOne) {
|
||||
std::vector<int64_t> sizes{4 * 1024 * 1024, 4 * 1024 * 1024};
|
||||
EXPECT_EQ(ComputeRowGroupsPerCell(sizes, 1024 * 1024), 1u);
|
||||
EXPECT_EQ(ComputeRowGroupsPerCell(sizes, 0), 1u);
|
||||
}
|
||||
|
||||
TEST(ComputeRowGroupsPerCellTest, ZeroSizeRowGroups) {
|
||||
std::vector<int64_t> sizes{0, 0, 0};
|
||||
EXPECT_EQ(ComputeRowGroupsPerCell(sizes, 4 * 1024 * 1024), 1u);
|
||||
}
|
||||
|
||||
TEST(ComputeRowGroupsPerCellTest, UsesGlobalAverage) {
|
||||
// Documents current behavior: helper uses a single average across the
|
||||
// input vector. Mixing small (128 KiB) and large (4 MiB) row groups
|
||||
// yields avg ~2 MiB and therefore 2 rgs/cell at the 4 MiB target -
|
||||
// callers that want per-file sizing must split the vector themselves.
|
||||
std::vector<int64_t> sizes{
|
||||
128 * 1024, 128 * 1024, 4 * 1024 * 1024, 4 * 1024 * 1024};
|
||||
int64_t total = 128 * 1024 * 2 + 4 * 1024 * 1024 * 2;
|
||||
int64_t avg = total / 4;
|
||||
size_t expected = static_cast<size_t>((4 * 1024 * 1024) / avg);
|
||||
EXPECT_EQ(ComputeRowGroupsPerCell(sizes, 4 * 1024 * 1024), expected);
|
||||
}
|
||||
|
||||
TEST(ComputeRowGroupsPerCellTest, AcceptsUnsignedSizes) {
|
||||
std::vector<uint64_t> sizes{1024 * 1024, 1024 * 1024};
|
||||
EXPECT_EQ(ComputeRowGroupsPerCell(sizes, 4 * 1024 * 1024), 4u);
|
||||
}
|
||||
|
||||
@@ -133,17 +133,21 @@ ManifestGroupTranslator::ManifestGroupTranslator(
|
||||
}
|
||||
const auto& row_group_rows = rows_result.ValueOrDie();
|
||||
|
||||
// Merge row groups into group chunks(cache cells)
|
||||
// Merge row groups into group chunks(cache cells). Derive row-groups-
|
||||
// per-cell from the runtime-configurable target byte size so avg cell
|
||||
// byte size ≈ target.
|
||||
const int64_t cell_target_size_bytes = GetCellTargetSizeBytes();
|
||||
size_t total_row_groups = row_group_sizes.size();
|
||||
meta_.total_row_groups_ = total_row_groups;
|
||||
size_t num_cells =
|
||||
(total_row_groups + kRowGroupsPerCell - 1) / kRowGroupsPerCell;
|
||||
const size_t rgs_per_cell =
|
||||
ComputeRowGroupsPerCell(row_group_sizes, cell_target_size_bytes);
|
||||
size_t num_cells = (total_row_groups + rgs_per_cell - 1) / rgs_per_cell;
|
||||
|
||||
// Populate cell_row_group_ranges_ (single data source, no multi-file)
|
||||
meta_.cell_row_group_ranges_.reserve(num_cells);
|
||||
for (size_t cid = 0; cid < num_cells; ++cid) {
|
||||
size_t start = cid * kRowGroupsPerCell;
|
||||
size_t end = std::min(start + kRowGroupsPerCell, total_row_groups);
|
||||
size_t start = cid * rgs_per_cell;
|
||||
size_t end = std::min(start + rgs_per_cell, total_row_groups);
|
||||
meta_.cell_row_group_ranges_.push_back({start, end});
|
||||
}
|
||||
|
||||
@@ -193,12 +197,12 @@ ManifestGroupTranslator::ManifestGroupTranslator(
|
||||
}
|
||||
|
||||
LOG_INFO(
|
||||
"[StorageV2] translator {} merged {} row groups into {} cells ({} "
|
||||
"row groups per cell)",
|
||||
"[StorageV2] translator {} merged {} row groups into {} cells "
|
||||
"(cell_target_size_bytes={})",
|
||||
key_,
|
||||
total_row_groups,
|
||||
num_cells,
|
||||
kRowGroupsPerCell);
|
||||
cell_target_size_bytes);
|
||||
|
||||
// Set loading overhead config to cap total overhead reservation.
|
||||
if (!meta_.chunk_memory_size_.empty()) {
|
||||
|
||||
@@ -112,8 +112,11 @@ TEST_P(ManifestGroupTranslatorTest, TestScalarColumnGroup) {
|
||||
auto num_cells = translator->num_cells();
|
||||
auto chunk_reader = test_data_->CreateChunkReader(0);
|
||||
auto expected_num_chunks = chunk_reader->total_number_of_chunks();
|
||||
auto row_group_sizes = chunk_reader->get_chunk_size().ValueOrDie();
|
||||
auto rgs_per_cell =
|
||||
ComputeRowGroupsPerCell(row_group_sizes, GetCellTargetSizeBytes());
|
||||
auto expected_num_cells =
|
||||
(expected_num_chunks + kRowGroupsPerCell - 1) / kRowGroupsPerCell;
|
||||
(expected_num_chunks + rgs_per_cell - 1) / rgs_per_cell;
|
||||
EXPECT_EQ(num_cells, expected_num_cells);
|
||||
|
||||
// cell_id_of — identity mapping
|
||||
@@ -346,13 +349,18 @@ TEST_P(ManifestGroupTranslatorTest, TestRowGroupRangesCoverage) {
|
||||
|
||||
EXPECT_EQ(meta->cell_row_group_ranges_.size(), num_cells);
|
||||
|
||||
auto chunk_reader = test_data_->CreateChunkReader(0);
|
||||
auto row_group_sizes = chunk_reader->get_chunk_size().ValueOrDie();
|
||||
auto rgs_per_cell =
|
||||
ComputeRowGroupsPerCell(row_group_sizes, GetCellTargetSizeBytes());
|
||||
|
||||
// Ranges should be contiguous and cover [0, total_row_groups_)
|
||||
size_t expected_start = 0;
|
||||
for (size_t cid = 0; cid < num_cells; ++cid) {
|
||||
auto [start, end] = meta->get_row_group_range(cid);
|
||||
EXPECT_EQ(start, expected_start) << "gap at cid " << cid;
|
||||
EXPECT_GT(end, start) << "empty range at cid " << cid;
|
||||
EXPECT_LE(end - start, kRowGroupsPerCell)
|
||||
EXPECT_LE(end - start, rgs_per_cell)
|
||||
<< "range too large at cid " << cid;
|
||||
expected_start = end;
|
||||
}
|
||||
|
||||
@@ -294,6 +294,16 @@ func (node *QueryNode) RegisterSegcoreConfigWatcher() {
|
||||
pt.Watch(pt.CommonCfg.ArrowIOThreadPoolMaxCapacity.Key,
|
||||
config.NewHandler(pt.CommonCfg.ArrowIOThreadPoolMaxCapacity.Key,
|
||||
arrowIOThreadHandler(pt.CommonCfg.ArrowIOThreadPoolMaxCapacity.Key)))
|
||||
pt.Watch(pt.QueryNodeCfg.StorageV2CellTargetSizeBytes.Key,
|
||||
config.NewHandler("queryNode.segcore.storageV2.cellTargetSizeBytes", func(evt *config.Event) {
|
||||
if !evt.HasUpdated {
|
||||
return
|
||||
}
|
||||
newBytes := paramtable.Get().QueryNodeCfg.StorageV2CellTargetSizeBytes.GetAsInt64()
|
||||
initcore.UpdateStorageV2CellTargetSizeBytes(newBytes)
|
||||
log.Info("queryNode.segcore.storageV2.cellTargetSizeBytes updated",
|
||||
zap.Int64("bytes", newBytes))
|
||||
}))
|
||||
}
|
||||
|
||||
func getIndexEngineVersion() (minimal, current, maximum int32) {
|
||||
|
||||
@@ -157,6 +157,9 @@ func doInitQueryNodeOnce(ctx context.Context) error {
|
||||
|
||||
C.SetArrowIOThreadPoolCapacity(C.int(ResolveArrowIOThreadPoolCapacity()))
|
||||
|
||||
cStorageV2CellTargetSizeBytes := C.int64_t(paramtable.Get().QueryNodeCfg.StorageV2CellTargetSizeBytes.GetAsInt64())
|
||||
C.SetStorageV2CellTargetSizeBytes(cStorageV2CellTargetSizeBytes)
|
||||
|
||||
enableParquetStatsSkipIndex := paramtable.Get().CommonCfg.ParquetStatsSkipIndex.GetAsBool()
|
||||
C.SetDefaultEnableParquetStatsSkipIndex(C.bool(enableParquetStatsSkipIndex))
|
||||
|
||||
|
||||
@@ -107,6 +107,10 @@ func ResolveArrowIOThreadPoolCapacity() int {
|
||||
return threads
|
||||
}
|
||||
|
||||
func UpdateStorageV2CellTargetSizeBytes(bytes int64) {
|
||||
C.SetStorageV2CellTargetSizeBytes(C.int64_t(bytes))
|
||||
}
|
||||
|
||||
func UpdateDefaultGrowingJSONKeyStatsEnable(enable bool) {
|
||||
C.SetDefaultGrowingJSONKeyStatsEnable(C.bool(enable))
|
||||
}
|
||||
|
||||
@@ -3508,6 +3508,10 @@ type queryNodeConfig struct {
|
||||
// CGOPoolSize ratio to MaxReadConcurrency
|
||||
CGOPoolSizeRatio ParamItem `refreshable:"true"`
|
||||
|
||||
// Target average byte size per storage v2 cache cell. Parquet row groups
|
||||
// are packed into cells so rgs_per_cell * avg_rg_size ≈ this value.
|
||||
StorageV2CellTargetSizeBytes ParamItem `refreshable:"true"`
|
||||
|
||||
EnableWorkerSQCostMetrics ParamItem `refreshable:"true"`
|
||||
|
||||
ExprEvalBatchSize ParamItem `refreshable:"false"`
|
||||
@@ -4604,6 +4608,27 @@ user-task-polling:
|
||||
}
|
||||
p.CGOPoolSizeRatio.Init(base.mgr)
|
||||
|
||||
p.StorageV2CellTargetSizeBytes = ParamItem{
|
||||
Key: "queryNode.segcore.storageV2.cellTargetSizeBytes",
|
||||
Version: "3.0.0",
|
||||
DefaultValue: "4194304", // 4 MiB
|
||||
Doc: `Target average byte size per storage v2 cache cell. Parquet row groups are ` +
|
||||
`greedily packed so that rgs_per_cell * avg_row_group_size ≈ this target. ` +
|
||||
`Each cell always contains at least one row group and cells never cross file ` +
|
||||
`boundaries. Tune larger for bigger batch IO (fewer cells) or smaller to get ` +
|
||||
`finer cache granularity. Default 4 MiB.`,
|
||||
Export: true,
|
||||
Formatter: func(v string) string {
|
||||
if getAsInt64(v) <= 0 {
|
||||
log.Warn("queryNode.segcore.storageV2.cellTargetSizeBytes must be positive, using default 4 MiB",
|
||||
zap.String("configured", v))
|
||||
return "4194304"
|
||||
}
|
||||
return v
|
||||
},
|
||||
}
|
||||
p.StorageV2CellTargetSizeBytes.Init(base.mgr)
|
||||
|
||||
p.EnableWorkerSQCostMetrics = ParamItem{
|
||||
Key: "queryNode.enableWorkerSQCostMetrics",
|
||||
Version: "2.3.0",
|
||||
|
||||
Reference in New Issue
Block a user