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

feat: impl StructArray --- get raw data from index when index has raw data (#48889)

Browse Source 
issue: https://github.com/milvus-io/milvus/issues/42148
design doc:
https://github.com/milvus-io/milvus-design-docs/blob/main/design_docs/20260306-struct.md

---------

Signed-off-by: SpadeA <tangchenjie1210@gmail.com>
This commit is contained in:
Spade A
2026-04-22 19:13:44 +08:00
committed by GitHub
parent b50538eede
commit f531b405d2
12 changed files with 727 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/core/src/index/IndexFactory.cpp
+15 -3
View File
@@ -264,6 +264,9 @@ IndexFactory::VecIndexLoadResource(
num_rows,
dim,
config);
has_raw_data =
knowhere::IndexStaticFaced<knowhere::fp32>::HasRawData(
index_type, index_version, config);
break;
case milvus::DataType::VECTOR_FLOAT16:
resource = knowhere::IndexStaticFaced<knowhere::fp16>::
@@ -273,6 +276,9 @@ IndexFactory::VecIndexLoadResource(
num_rows,
dim,
config);
has_raw_data =
knowhere::IndexStaticFaced<knowhere::fp16>::HasRawData(
index_type, index_version, config);
break;
case milvus::DataType::VECTOR_BFLOAT16:
resource = knowhere::IndexStaticFaced<knowhere::bf16>::
@@ -282,6 +288,9 @@ IndexFactory::VecIndexLoadResource(
num_rows,
dim,
config);
has_raw_data =
knowhere::IndexStaticFaced<knowhere::bf16>::HasRawData(
index_type, index_version, config);
break;
case milvus::DataType::VECTOR_BINARY:
resource = knowhere::IndexStaticFaced<knowhere::bin1>::
@@ -291,6 +300,9 @@ IndexFactory::VecIndexLoadResource(
num_rows,
dim,
config);
has_raw_data =
knowhere::IndexStaticFaced<knowhere::bin1>::HasRawData(
index_type, index_version, config);
break;
case milvus::DataType::VECTOR_INT8:
resource = knowhere::IndexStaticFaced<knowhere::int8>::
@@ -300,6 +312,9 @@ IndexFactory::VecIndexLoadResource(
num_rows,
dim,
config);
has_raw_data =
knowhere::IndexStaticFaced<knowhere::int8>::HasRawData(
index_type, index_version, config);
break;
default:
@@ -310,9 +325,6 @@ IndexFactory::VecIndexLoadResource(
element_type);
return LoadResourceRequest{0, 0, 0, 0, true};
}
// For VectorArray, has_raw_data is always false as get_vector of index does not provide offsets which
// is required for reconstructing the raw data
has_raw_data = false;
break;
}
default:
internal/core/src/index/VectorDiskIndex.cpp
+19 -8
View File
@@ -521,14 +521,25 @@ VectorDiskAnnIndex<T>::GetVector(const DatasetPtr dataset) const {
KnowhereStatusString(res.error()),
res.what()));
}
auto tensor = res.value()->GetTensor();
auto row_num = res.value()->GetRows();
auto dim = res.value()->GetDim();
int64_t data_size = milvus::GetVecRowSize<T>(dim) * row_num;
std::vector<uint8_t> raw_data;
raw_data.resize(data_size);
memcpy(raw_data.data(), tensor, data_size);
return raw_data;
return this->template DecodeVectorByIdsResult<T>(res.value());
}
template <typename T>
std::pair<std::vector<uint8_t>, std::vector<size_t>>
VectorDiskAnnIndex<T>::GetEmbListByIds(const DatasetPtr dataset,
const std::string& metric_type) const {
if (dataset->GetRows() == 0) {
return {{}, {0}};
}
auto res = index_.GetEmbListByIds(dataset, metric_type);
if (!res.has_value()) {
ThrowInfo(ErrorCode::UnexpectedError,
fmt::format("failed to get emb list: {}: {}",
KnowhereStatusString(res.error()),
res.what()));
}
return this->template DecodeEmbListByIdsResult<T>(res.value());
}
template <typename T>
internal/core/src/index/VectorDiskIndex.h
+4
View File
@@ -85,6 +85,10 @@ class VectorDiskAnnIndex : public VectorIndex {
std::vector<uint8_t>
GetVector(const DatasetPtr dataset) const override;
std::pair<std::vector<uint8_t>, std::vector<size_t>>
GetEmbListByIds(const DatasetPtr dataset,
const std::string& metric_type) const override;
std::unique_ptr<const knowhere::sparse::SparseRow<SparseValueType>[]>
GetSparseVector(const DatasetPtr dataset) const override {
ThrowInfo(ErrorCode::Unsupported,
internal/core/src/index/VectorIndex.h
+51
View File
@@ -16,6 +16,7 @@
#pragma once
#include <cstring>
#include <map>
#include <memory>
#include <string>
@@ -32,6 +33,7 @@
#include "common/QueryResult.h"
#include "common/QueryInfo.h"
#include "common/OpContext.h"
#include "knowhere/comp/index_param.h"
#include "knowhere/version.h"
namespace milvus::index {
@@ -99,6 +101,21 @@ class VectorIndex : public IndexBase {
virtual std::vector<uint8_t>
GetVector(const DatasetPtr dataset) const = 0;
/**
* @brief Retrieve embedding lists by their IDs from the index.
*
* @param dataset Contains the embedding list IDs (rows = count, ids = el_ids)
* @param metric_type The metric type (e.g., MAX_SIM, MAX_SIM_IP)
* @return A pair of (raw_vector_data, offsets) where offsets has size count+1
* and raw_vector_data contains all vectors concatenated.
*/
virtual std::pair<std::vector<uint8_t>, std::vector<size_t>>
GetEmbListByIds(const DatasetPtr dataset,
const std::string& metric_type) const {
ThrowInfo(NotImplemented,
"GetEmbListByIds not supported for current index type");
}
virtual std::unique_ptr<
const knowhere::sparse::SparseRow<SparseValueType>[]>
GetSparseVector(const DatasetPtr dataset) const = 0;
@@ -213,6 +230,40 @@ class VectorIndex : public IndexBase {
}
protected:
template <typename T>
static std::vector<uint8_t>
DecodeVectorByIdsResult(const knowhere::DataSetPtr& result) {
auto tensor = result->GetTensor();
auto row_num = result->GetRows();
auto dim = result->GetDim();
size_t data_size =
static_cast<size_t>(milvus::GetVecRowSize<T>(dim)) * row_num;
std::vector<uint8_t> raw_data(data_size);
std::memcpy(raw_data.data(), tensor, data_size);
return raw_data;
}
template <typename T>
static std::pair<std::vector<uint8_t>, std::vector<size_t>>
DecodeEmbListByIdsResult(const knowhere::DataSetPtr& result) {
auto tensor = result->GetTensor();
auto dim = result->GetDim();
auto num_el_ids = result->GetRows();
const size_t* offsets_ptr =
result->Get<const size_t*>(knowhere::meta::EMB_LIST_OFFSET);
AssertInfo(offsets_ptr != nullptr,
"EMB_LIST_OFFSET not found in result");
size_t total_vecs = offsets_ptr[num_el_ids];
size_t data_size =
static_cast<size_t>(milvus::GetVecRowSize<T>(dim)) * total_vecs;
std::vector<uint8_t> raw_data(data_size);
std::memcpy(raw_data.data(), tensor, data_size);
std::vector<size_t> offsets(offsets_ptr, offsets_ptr + num_el_ids + 1);
return {std::move(raw_data), std::move(offsets)};
}
milvus::OffsetMapping offset_mapping_;
private:
internal/core/src/index/VectorMemIndex.cpp
+18 -8
View File
@@ -671,14 +671,24 @@ VectorMemIndex<T>::GetVector(const DatasetPtr dataset) const {
"failed to get vector, {}",
KnowhereStatusString(res.error()));
}
auto tensor = res.value()->GetTensor();
auto row_num = res.value()->GetRows();
auto dim = res.value()->GetDim();
int64_t data_size = milvus::GetVecRowSize<T>(dim) * row_num;
std::vector<uint8_t> raw_data;
raw_data.resize(data_size);
memcpy(raw_data.data(), tensor, data_size);
return raw_data;
return this->template DecodeVectorByIdsResult<T>(res.value());
}
template <typename T>
std::pair<std::vector<uint8_t>, std::vector<size_t>>
VectorMemIndex<T>::GetEmbListByIds(const DatasetPtr dataset,
const std::string& metric_type) const {
if (dataset->GetRows() == 0) {
return {{}, {0}};
}
auto res = index_.GetEmbListByIds(dataset, metric_type);
if (!res.has_value()) {
ThrowInfo(
ErrorCode::UnexpectedError,
"failed to get emb list, " + KnowhereStatusString(res.error()));
}
return this->template DecodeEmbListByIdsResult<T>(res.value());
}
template <typename T>
internal/core/src/index/VectorMemIndex.h
+4
View File
@@ -103,6 +103,10 @@ class VectorMemIndex : public VectorIndex {
std::vector<uint8_t>
GetVector(const DatasetPtr dataset) const override;
std::pair<std::vector<uint8_t>, std::vector<size_t>>
GetEmbListByIds(const DatasetPtr dataset,
const std::string& metric_type) const override;
std::unique_ptr<const knowhere::sparse::SparseRow<SparseValueType>[]>
GetSparseVector(const DatasetPtr dataset) const override;
internal/core/src/segcore/ChunkedSegmentSealedImpl.cpp
+109 -3
View File
@@ -1640,8 +1640,6 @@ ChunkedSegmentSealedImpl::get_vector(milvus::OpContext* op_ctx,
auto vec_index = dynamic_cast<index::VectorIndex*>(ca->get_cell_of(0));
AssertInfo(vec_index, "invalid vector indexing");
auto index_type = vec_index->GetIndexType();
auto metric_type = vec_index->GetMetricType();
auto has_raw_data = vec_index->HasRawData();
if (has_raw_data) {
@@ -1676,6 +1674,109 @@ ChunkedSegmentSealedImpl::get_vector(milvus::OpContext* op_ctx,
return nullptr;
}
std::unique_ptr<DataArray>
ChunkedSegmentSealedImpl::get_emb_list(milvus::OpContext* op_ctx,
FieldId field_id,
const FieldMeta& field_meta,
const int64_t* seg_offsets,
int64_t count) const {
AssertInfo(field_meta.get_data_type() == DataType::VECTOR_ARRAY,
"get_emb_list only supports VECTOR_ARRAY");
if (!get_bit(index_ready_bitset_, field_id) &&
!get_bit(binlog_index_bitset_, field_id)) {
return fill_with_empty(field_id, count);
}
AssertInfo(vector_indexings_.is_ready(field_id),
"vector index is not ready");
auto field_indexing = vector_indexings_.get_field_indexing(field_id);
auto cache_index = field_indexing->indexing_;
auto ca = SemiInlineGet(cache_index->PinCells(op_ctx, {0}));
auto vec_index = dynamic_cast<index::VectorIndex*>(ca->get_cell_of(0));
AssertInfo(vec_index, "invalid vector indexing");
auto has_raw_data = vec_index->HasRawData();
AssertInfo(has_raw_data,
"get_emb_list called on vector index without raw data");
auto metric_type = vec_index->GetMetricType();
// Build el_ids dataset from seg_offsets (seg_offsets are el_ids for VECTOR_ARRAY)
auto ids_ds = GenIdsDataset(count, seg_offsets);
auto [raw_data, offsets] = vec_index->GetEmbListByIds(ids_ds, metric_type);
AssertInfo(offsets.size() == static_cast<size_t>(count + 1),
"GetEmbListByIds returned invalid offsets size {}, expected {}",
offsets.size(),
count + 1);
auto dim = field_meta.get_dim();
auto element_type = field_meta.get_element_type();
const size_t vec_size_per_element =
milvus::vector_bytes_per_element(element_type, dim);
auto data_array = std::make_unique<DataArray>();
data_array->set_field_id(field_meta.get_id().get());
data_array->set_type(static_cast<milvus::proto::schema::DataType>(
field_meta.get_data_type()));
auto vector_array = data_array->mutable_vectors();
auto obj = vector_array->mutable_vector_array();
obj->set_dim(dim);
obj->set_element_type(milvus::ToProtoDataType(element_type));
// Build a VectorFieldProto for each embedding list
for (int64_t i = 0; i < count; i++) {
auto* entry = obj->mutable_data()->Add();
entry->set_dim(dim);
size_t vec_start = offsets[i];
size_t vec_count = offsets[i + 1] - offsets[i];
if (vec_count == 0) {
continue;
}
size_t byte_offset = vec_start * vec_size_per_element;
size_t byte_len = vec_count * vec_size_per_element;
switch (element_type) {
case DataType::VECTOR_FLOAT: {
auto* src = reinterpret_cast<const float*>(raw_data.data() +
byte_offset);
entry->mutable_float_vector()->mutable_data()->Add(
src, src + vec_count * dim);
break;
}
case DataType::VECTOR_BINARY: {
auto* src = reinterpret_cast<const char*>(raw_data.data() +
byte_offset);
entry->mutable_binary_vector()->assign(src, byte_len);
break;
}
case DataType::VECTOR_FLOAT16: {
auto* src = reinterpret_cast<const char*>(raw_data.data() +
byte_offset);
entry->mutable_float16_vector()->assign(src, byte_len);
break;
}
case DataType::VECTOR_BFLOAT16: {
auto* src = reinterpret_cast<const char*>(raw_data.data() +
byte_offset);
entry->mutable_bfloat16_vector()->assign(src, byte_len);
break;
}
case DataType::VECTOR_INT8: {
auto* src = reinterpret_cast<const char*>(raw_data.data() +
byte_offset);
entry->mutable_int8_vector()->assign(src, byte_len);
break;
}
default:
break;
}
}
return data_array;
}
void
ChunkedSegmentSealedImpl::DropFieldData(const FieldId field_id) {
AssertInfo(!SystemProperty::Instance().IsSystem(field_id),
@@ -3104,7 +3205,12 @@ ChunkedSegmentSealedImpl::bulk_subscript(milvus::OpContext* op_ctx,
std::unique_ptr<DataArray> vector{nullptr};
// Try index first: if vector index exists and has raw data, read from index
if (IndexHasRawData(field_id)) {
vector = get_vector(op_ctx, field_id, seg_offsets, count);
if (IsVectorArrayDataType(field_meta.get_data_type())) {
vector =
get_emb_list(op_ctx, field_id, field_meta, seg_offsets, count);
} else {
vector = get_vector(op_ctx, field_id, seg_offsets, count);
}
} else {
// Fallback to field data
auto [field, exist] = GetFieldDataIfExist(field_id);
internal/core/src/segcore/ChunkedSegmentSealedImpl.h
+7
View File
@@ -316,6 +316,13 @@ class ChunkedSegmentSealedImpl : public SegmentSealed {
const int64_t* ids,
int64_t count) const override;
std::unique_ptr<DataArray>
get_emb_list(milvus::OpContext* op_ctx,
FieldId field_id,
const FieldMeta& field_meta,
const int64_t* seg_offsets,
int64_t count) const;
bool
is_nullable(FieldId field_id) const override {
auto& field_meta = schema_->operator[](field_id);
internal/core/src/segcore/Utils.cpp
+2 -1
View File
@@ -504,7 +504,8 @@ CreateEmptyVectorDataArray(int64_t count, const FieldMeta& field_meta) {
field_meta.get_element_type()));
obj->mutable_data()->Reserve(count);
for (int i = 0; i < count; i++) {
*(obj->mutable_data()->Add()) = proto::schema::VectorField();
auto* row = obj->mutable_data()->Add();
row->set_dim(dim);
}
break;
}
internal/core/thirdparty/knowhere/CMakeLists.txt
Vendored
+1 -1
View File
@@ -14,7 +14,7 @@
# Update KNOWHERE_VERSION for the first occurrence
milvus_add_pkg_config("knowhere")
set_property(DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR} PROPERTY INCLUDE_DIRECTORIES "")
set( KNOWHERE_VERSION 216ddba )
set( KNOWHERE_VERSION 2b560877 )
set( GIT_REPOSITORY "https://github.com/zilliztech/knowhere.git")
message(STATUS "Knowhere repo: ${GIT_REPOSITORY}")
internal/core/unittest/test_loading.cpp
+50 -1
View File
@@ -51,12 +51,26 @@ class IndexLoadTest : public ::testing::TestWithParam<Param> {
data_type = milvus::DataType::VECTOR_SPARSE_U32_F32;
} else if (field_type == "vector_int8") {
data_type = milvus::DataType::VECTOR_INT8;
} else if (field_type == "vector_array") {
data_type = milvus::DataType::VECTOR_ARRAY;
} else if (field_type == "array") {
data_type = milvus::DataType::ARRAY;
} else {
data_type = milvus::DataType::STRING;
}
element_data_type = milvus::DataType::NONE;
if (index_params.find("element_type") != index_params.end()) {
std::string et = index_params["element_type"];
if (et == "vector_float") {
element_data_type = milvus::DataType::VECTOR_FLOAT;
} else if (et == "vector_fp16") {
element_data_type = milvus::DataType::VECTOR_FLOAT16;
} else if (et == "vector_bf16") {
element_data_type = milvus::DataType::VECTOR_BFLOAT16;
}
}
expected = param.second;
}
@@ -69,6 +83,7 @@ class IndexLoadTest : public ::testing::TestWithParam<Param> {
std::map<std::string, std::string> index_params;
bool enable_mmap;
milvus::DataType data_type;
milvus::DataType element_data_type;
LoadResourceRequest expected;
};
@@ -238,7 +253,40 @@ static const auto kIndexLoadTestValues = ::testing::Values(
1UL * 1024 * 1024 * 1024,
1UL * 1024 * 1024 * 1024,
1UL * 1024 * 1024 * 1024,
false}));
false}),
// VECTOR_ARRAY + HNSW (FLAT): has_raw_data = true
std::pair<std::map<std::string, std::string>, LoadResourceRequest>(
{{"index_type", "HNSW"},
{"metric_type", "MAX_SIM"},
{"efConstrcution", "300"},
{"M", "30"},
{"mmap", "false"},
{"field_type", "vector_array"},
{"element_type", "vector_float"}},
{2UL * 1024 * 1024 * 1024, 0UL, 1UL * 1024 * 1024 * 1024, 0UL, true}),
// VECTOR_ARRAY + HNSW_SQ (SQ8, fp32): has_raw_data = false (lossy)
std::pair<std::map<std::string, std::string>, LoadResourceRequest>(
{{"index_type", "HNSW_SQ"},
{"metric_type", "MAX_SIM"},
{"efConstrcution", "300"},
{"M", "30"},
{"sq_type", "SQ8"},
{"mmap", "false"},
{"field_type", "vector_array"},
{"element_type", "vector_float"}},
{2UL * 1024 * 1024 * 1024, 0UL, 1UL * 1024 * 1024 * 1024, 0UL, false}),
// VECTOR_ARRAY + HNSW_PQ (no refine): has_raw_data = false
std::pair<std::map<std::string, std::string>, LoadResourceRequest>(
{{"index_type", "HNSW_PQ"},
{"metric_type", "MAX_SIM"},
{"efConstrcution", "300"},
{"M", "30"},
{"pq_m", "8"},
{"nbits", "8"},
{"mmap", "false"},
{"field_type", "vector_array"},
{"element_type", "vector_float"}},
{2UL * 1024 * 1024 * 1024, 0UL, 1UL * 1024 * 1024 * 1024, 0UL, false}));
INSTANTIATE_TEST_SUITE_P(IndexTypeLoadInfo,
IndexLoadTest,
@@ -252,6 +300,7 @@ TEST_P(IndexLoadTest, ResourceEstimate) {
loadIndexInfo.segment_id = 3;
loadIndexInfo.field_id = 4;
loadIndexInfo.field_type = data_type;
loadIndexInfo.element_type = element_data_type;
loadIndexInfo.enable_mmap = enable_mmap;
loadIndexInfo.mmap_dir_path = "/tmp/mmap";
loadIndexInfo.index_id = 5;
internal/core/unittest/test_sealed.cpp
+447 -43
View File
@@ -76,12 +76,14 @@
#include "storage/FileManager.h"
#include "storage/InsertData.h"
#include "storage/PayloadReader.h"
#include "storage/ThreadPools.h"
#include "storage/RemoteChunkManagerSingleton.h"
#include "storage/Types.h"
#include "storage/Util.h"
#include "test_utils/Constants.h"
#include "test_utils/DataGen.h"
#include "test_utils/cachinglayer_test_utils.h"
#include "test_utils/indexbuilder_test_utils.h"
#include "test_utils/storage_test_utils.h"
using namespace milvus;
@@ -2250,6 +2252,49 @@ TEST(Sealed, QueryAllNullableFields) {
using VectorArrayTestParam =
std::tuple<DataType, std::string, int, std::string>;
void
VerifyVectorResults(const VectorFieldProto& result_vec,
const VectorFieldProto& expected_vec,
DataType element_type) {
switch (element_type) {
case DataType::VECTOR_FLOAT: {
auto result_data = result_vec.float_vector().data();
auto expected_data = expected_vec.float_vector().data();
EXPECT_EQ(result_data.size(), expected_data.size());
for (int64_t i = 0; i < result_data.size(); ++i) {
EXPECT_NEAR(result_data[i], expected_data[i], 1e-6f);
}
break;
}
case DataType::VECTOR_BINARY: {
auto result_data = result_vec.binary_vector();
auto expected_data = expected_vec.binary_vector();
EXPECT_EQ(result_data, expected_data);
break;
}
case DataType::VECTOR_FLOAT16: {
auto result_data = result_vec.float16_vector();
auto expected_data = expected_vec.float16_vector();
EXPECT_EQ(result_data, expected_data);
break;
}
case DataType::VECTOR_BFLOAT16: {
auto result_data = result_vec.bfloat16_vector();
auto expected_data = expected_vec.bfloat16_vector();
EXPECT_EQ(result_data, expected_data);
break;
}
case DataType::VECTOR_INT8: {
auto result_data = result_vec.int8_vector();
auto expected_data = expected_vec.int8_vector();
EXPECT_EQ(result_data, expected_data);
break;
}
default:
break;
}
}
class SealedVectorArrayTest
: public ::testing::TestWithParam<VectorArrayTestParam> {
protected:
@@ -2271,49 +2316,6 @@ class SealedVectorArrayTest
ASSERT_EQ(dim % 8, 0) << "Binary vector dim must be multiple of 8";
}
}
void
VerifyVectorResults(const VectorFieldProto& result_vec,
const VectorFieldProto& expected_vec,
DataType element_type) {
switch (element_type) {
case DataType::VECTOR_FLOAT: {
auto result_data = result_vec.float_vector().data();
auto expected_data = expected_vec.float_vector().data();
EXPECT_EQ(result_data.size(), expected_data.size());
for (int64_t i = 0; i < result_data.size(); ++i) {
EXPECT_NEAR(result_data[i], expected_data[i], 1e-6f);
}
break;
}
case DataType::VECTOR_BINARY: {
auto result_data = result_vec.binary_vector();
auto expected_data = expected_vec.binary_vector();
EXPECT_EQ(result_data, expected_data);
break;
}
case DataType::VECTOR_FLOAT16: {
auto result_data = result_vec.float16_vector();
auto expected_data = expected_vec.float16_vector();
EXPECT_EQ(result_data, expected_data);
break;
}
case DataType::VECTOR_BFLOAT16: {
auto result_data = result_vec.bfloat16_vector();
auto expected_data = expected_vec.bfloat16_vector();
EXPECT_EQ(result_data, expected_data);
break;
}
case DataType::VECTOR_INT8: {
auto result_data = result_vec.int8_vector();
auto expected_data = expected_vec.int8_vector();
EXPECT_EQ(result_data, expected_data);
break;
}
default:
break;
}
}
};
TEST_P(SealedVectorArrayTest, QueryVectorArrayAllFields) {
@@ -2642,6 +2644,408 @@ TEST_P(SealedVectorArrayTest, SearchVectorArray) {
}
}
TEST_P(SealedVectorArrayTest, BulkSubscriptVectorArrayFromIndex) {
int64_t collection_id = 1;
int64_t partition_id = 2;
int64_t segment_id = 3;
int64_t index_build_id = 5000;
int64_t index_version = 5000;
auto schema = std::make_shared<Schema>();
auto int64_field = schema->AddDebugField("int64", DataType::INT64);
auto array_vec = schema->AddDebugVectorArrayField(
"array_vec", element_type, dim, metric_type);
schema->set_primary_field_id(int64_field);
auto field_meta = milvus::segcore::gen_field_meta(collection_id,
partition_id,
segment_id,
array_vec.get(),
DataType::VECTOR_ARRAY,
element_type,
false);
auto index_meta = gen_index_meta(
segment_id, array_vec.get(), index_build_id, index_version);
int64_t dataset_size = 100;
auto emb_list_len = 2;
auto dataset = DataGen(schema, dataset_size, 42, 0, 1, emb_list_len);
// prepare field data and binlog
std::string root_path = TestLocalPath;
auto storage_config = gen_local_storage_config(root_path);
auto cm = CreateChunkManager(storage_config);
auto fs = milvus::storage::InitArrowFileSystem(storage_config);
auto vec_array_col = dataset.get_col<VectorFieldProto>(array_vec);
std::vector<milvus::VectorArray> vector_arrays;
for (auto& v : vec_array_col) {
vector_arrays.push_back(milvus::VectorArray(v));
}
auto field_data = storage::CreateFieldData(
DataType::VECTOR_ARRAY, element_type, false, dim);
field_data->FillFieldData(vector_arrays.data(), vector_arrays.size());
auto segment = CreateSealedSegment(schema);
auto field_data_info = PrepareSingleFieldInsertBinlog(collection_id,
partition_id,
segment_id,
array_vec.get(),
{field_data},
cm);
segment->LoadFieldData(field_data_info);
auto sealed = dynamic_cast<ChunkedSegmentSealedImpl*>(segment.get());
// serialize binlog for index building
auto payload_reader =
std::make_shared<milvus::storage::PayloadReader>(field_data);
storage::InsertData insert_data(payload_reader);
insert_data.SetFieldDataMeta(field_meta);
insert_data.SetTimestamps(0, 100);
auto serialized_bytes = insert_data.Serialize(storage::Remote);
auto log_path = fmt::format("{}{}/{}/{}/{}/{}",
TestLocalPath,
collection_id,
partition_id,
segment_id,
array_vec.get(),
0);
auto cm_w = ChunkManagerWrapper(cm);
cm_w.Write(log_path, serialized_bytes.data(), serialized_bytes.size());
// build HNSW index (FLAT base = has_raw_data = true)
milvus::index::CreateIndexInfo create_index_info;
create_index_info.field_type = DataType::VECTOR_ARRAY;
create_index_info.metric_type = metric_type;
create_index_info.index_type = knowhere::IndexEnum::INDEX_HNSW;
create_index_info.index_engine_version =
knowhere::Version::GetCurrentVersion().VersionNumber();
auto emb_list_hnsw_index =
milvus::index::IndexFactory::GetInstance().CreateIndex(
create_index_info,
storage::FileManagerContext(field_meta, index_meta, cm, fs));
Config config;
config[milvus::index::INDEX_TYPE] = knowhere::IndexEnum::INDEX_HNSW;
config[INSERT_FILES_KEY] = std::vector<std::string>{log_path};
config[knowhere::meta::METRIC_TYPE] = create_index_info.metric_type;
config[knowhere::indexparam::M] = "16";
config[knowhere::indexparam::EF] = "10";
config[DIM_KEY] = dim;
emb_list_hnsw_index->Build(config);
// drop field data and load index
LoadIndexInfo load_info;
load_info.field_id = array_vec.get();
load_info.field_type = DataType::VECTOR_ARRAY;
load_info.element_type = element_type;
load_info.index_params = GenIndexParams(emb_list_hnsw_index.get());
load_info.cache_index =
CreateTestCacheIndex("test", std::move(emb_list_hnsw_index));
load_info.index_params["metric_type"] = metric_type;
segment->DropFieldData(array_vec);
segment->LoadIndex(load_info);
// bulk_subscript should now go through get_emb_list -> GetEmbListByIds
auto ids_ds = GenRandomIds(dataset_size);
auto result = sealed->bulk_subscript(
nullptr, array_vec, ids_ds->GetIds(), dataset_size);
ASSERT_NE(result, nullptr);
EXPECT_EQ(result->vectors().vector_array().data_size(), dataset_size);
// verify each row's embedding list matches original data
for (int64_t i = 0; i < dataset_size; ++i) {
auto result_vec = result->vectors().vector_array().data()[i];
auto expected_vec = vec_array_col[ids_ds->GetIds()[i]];
VerifyVectorResults(result_vec, expected_vec, element_type);
}
}
TEST(SealedVectorArrayFallback,
BulkSubscriptVectorArrayFallsBackToFieldDataWhenIndexHasNoRawData) {
constexpr int64_t collection_id = 1;
constexpr int64_t partition_id = 2;
constexpr int64_t segment_id = 3;
constexpr int64_t index_build_id = 7000;
constexpr int64_t index_version = 7000;
constexpr int64_t dataset_size = 100;
constexpr int64_t emb_list_len = 2;
constexpr int64_t dim = 4;
const auto element_type = DataType::VECTOR_FLOAT;
const auto metric_type = knowhere::metric::MAX_SIM;
auto schema = std::make_shared<Schema>();
auto int64_field = schema->AddDebugField("int64", DataType::INT64);
auto array_vec = schema->AddDebugVectorArrayField(
"array_vec", element_type, dim, metric_type);
schema->set_primary_field_id(int64_field);
auto field_meta = milvus::segcore::gen_field_meta(collection_id,
partition_id,
segment_id,
array_vec.get(),
DataType::VECTOR_ARRAY,
element_type,
false);
auto index_meta = gen_index_meta(
segment_id, array_vec.get(), index_build_id, index_version);
auto dataset = DataGen(schema, dataset_size, 42, 0, 1, emb_list_len);
std::string root_path = TestLocalPath;
auto storage_config = gen_local_storage_config(root_path);
auto cm = CreateChunkManager(storage_config);
auto fs = milvus::storage::InitArrowFileSystem(storage_config);
auto vec_array_col = dataset.get_col<VectorFieldProto>(array_vec);
std::vector<milvus::VectorArray> vector_arrays;
for (auto& v : vec_array_col) {
vector_arrays.push_back(milvus::VectorArray(v));
}
auto field_data = storage::CreateFieldData(
DataType::VECTOR_ARRAY, element_type, false, dim);
field_data->FillFieldData(vector_arrays.data(), vector_arrays.size());
auto segment = CreateSealedSegment(schema);
auto field_data_info = PrepareSingleFieldInsertBinlog(collection_id,
partition_id,
segment_id,
array_vec.get(),
{field_data},
cm);
segment->LoadFieldData(field_data_info);
auto sealed = dynamic_cast<ChunkedSegmentSealedImpl*>(segment.get());
auto payload_reader =
std::make_shared<milvus::storage::PayloadReader>(field_data);
storage::InsertData insert_data(payload_reader);
insert_data.SetFieldDataMeta(field_meta);
insert_data.SetTimestamps(0, 100);
auto serialized_bytes = insert_data.Serialize(storage::Remote);
auto log_path = fmt::format("{}{}/{}/{}/{}/{}",
TestLocalPath,
collection_id,
partition_id,
segment_id,
array_vec.get(),
0);
auto cm_w = ChunkManagerWrapper(cm);
cm_w.Write(log_path, serialized_bytes.data(), serialized_bytes.size());
milvus::index::CreateIndexInfo create_index_info;
create_index_info.field_type = DataType::VECTOR_ARRAY;
create_index_info.metric_type = metric_type;
create_index_info.index_type = knowhere::IndexEnum::INDEX_HNSW_SQ;
create_index_info.index_engine_version =
knowhere::Version::GetCurrentVersion().VersionNumber();
auto emb_list_hnsw_sq_index =
milvus::index::IndexFactory::GetInstance().CreateIndex(
create_index_info,
storage::FileManagerContext(field_meta, index_meta, cm, fs));
Config config;
config[milvus::index::INDEX_TYPE] = knowhere::IndexEnum::INDEX_HNSW_SQ;
config[INSERT_FILES_KEY] = std::vector<std::string>{log_path};
config[knowhere::meta::METRIC_TYPE] = create_index_info.metric_type;
config[knowhere::indexparam::M] = "16";
config[knowhere::indexparam::EF] = "10";
config[knowhere::indexparam::SQ_TYPE] = "SQ8";
config[DIM_KEY] = dim;
emb_list_hnsw_sq_index->Build(config);
auto index_params = GenIndexParams(emb_list_hnsw_sq_index.get());
index_params["metric_type"] = metric_type;
auto request = milvus::index::IndexFactory::GetInstance().IndexLoadResource(
DataType::VECTOR_ARRAY,
element_type,
create_index_info.index_engine_version,
0,
index_params,
false,
dataset_size * emb_list_len,
dim);
ASSERT_FALSE(request.has_raw_data);
LoadIndexInfo load_info;
load_info.field_id = array_vec.get();
load_info.field_type = DataType::VECTOR_ARRAY;
load_info.element_type = element_type;
load_info.index_params = index_params;
load_info.cache_index =
CreateTestCacheIndex("test", std::move(emb_list_hnsw_sq_index));
ASSERT_TRUE(segment->HasFieldData(array_vec));
segment->LoadIndex(load_info);
ASSERT_TRUE(segment->HasFieldData(array_vec));
auto ids_ds = GenRandomIds(dataset_size);
auto result = sealed->bulk_subscript(
nullptr, array_vec, ids_ds->GetIds(), dataset_size);
ASSERT_NE(result, nullptr);
EXPECT_EQ(result->vectors().vector_array().data_size(), dataset_size);
for (int64_t i = 0; i < dataset_size; ++i) {
auto result_vec = result->vectors().vector_array().data()[i];
auto expected_vec = vec_array_col[ids_ds->GetIds()[i]];
VerifyVectorResults(result_vec, expected_vec, element_type);
}
}
#ifdef BUILD_DISK_ANN
TEST_P(SealedVectorArrayTest, BulkSubscriptVectorArrayFromDiskIndex) {
// DiskANN only registers fp32, fp16, bf16
if (element_type == DataType::VECTOR_BINARY ||
element_type == DataType::VECTOR_INT8) {
GTEST_SKIP() << "DiskANN does not support this element type";
}
int64_t collection_id = 1;
int64_t partition_id = 2;
int64_t segment_id = 3;
int64_t index_build_id = 6000;
int64_t index_version = 6000;
auto schema = std::make_shared<Schema>();
auto int64_field = schema->AddDebugField("int64", DataType::INT64);
auto array_vec = schema->AddDebugVectorArrayField(
"array_vec", element_type, dim, metric_type);
schema->set_primary_field_id(int64_field);
auto field_meta = milvus::segcore::gen_field_meta(collection_id,
partition_id,
segment_id,
array_vec.get(),
DataType::VECTOR_ARRAY,
element_type,
false);
auto index_meta = gen_index_meta(
segment_id, array_vec.get(), index_build_id, index_version);
int64_t dataset_size = 100;
auto emb_list_len = 2;
auto dataset = DataGen(schema, dataset_size, 42, 0, 1, emb_list_len);
// prepare field data and binlog
std::string root_path = TestLocalPath;
auto storage_config = gen_local_storage_config(root_path);
auto cm = CreateChunkManager(storage_config);
auto fs = milvus::storage::InitArrowFileSystem(storage_config);
auto vec_array_col = dataset.get_col<VectorFieldProto>(array_vec);
std::vector<milvus::VectorArray> vector_arrays;
for (auto& v : vec_array_col) {
vector_arrays.push_back(milvus::VectorArray(v));
}
auto field_data = storage::CreateFieldData(
DataType::VECTOR_ARRAY, element_type, false, dim);
field_data->FillFieldData(vector_arrays.data(), vector_arrays.size());
auto segment = CreateSealedSegment(schema);
auto field_data_info = PrepareSingleFieldInsertBinlog(collection_id,
partition_id,
segment_id,
array_vec.get(),
{field_data},
cm);
segment->LoadFieldData(field_data_info);
auto sealed = dynamic_cast<ChunkedSegmentSealedImpl*>(segment.get());
// serialize binlog for index building
auto payload_reader =
std::make_shared<milvus::storage::PayloadReader>(field_data);
storage::InsertData insert_data(payload_reader);
insert_data.SetFieldDataMeta(field_meta);
insert_data.SetTimestamps(0, 100);
auto serialized_bytes = insert_data.Serialize(storage::Remote);
auto log_path = fmt::format("{}{}/{}/{}/{}/{}",
TestLocalPath,
collection_id,
partition_id,
segment_id,
array_vec.get(),
0);
auto cm_w = ChunkManagerWrapper(cm);
cm_w.Write(log_path, serialized_bytes.data(), serialized_bytes.size());
storage::FileManagerContext file_mgr_ctx(field_meta, index_meta, cm, fs);
// build DiskANN index
milvus::index::CreateIndexInfo create_index_info;
create_index_info.field_type = DataType::VECTOR_ARRAY;
create_index_info.metric_type = metric_type;
create_index_info.index_type = knowhere::IndexEnum::INDEX_DISKANN;
create_index_info.index_engine_version =
knowhere::Version::GetCurrentVersion().VersionNumber();
auto disk_index = milvus::index::IndexFactory::GetInstance().CreateIndex(
create_index_info, file_mgr_ctx);
Config config;
config[milvus::index::INDEX_TYPE] = knowhere::IndexEnum::INDEX_DISKANN;
config[INSERT_FILES_KEY] = std::vector<std::string>{log_path};
config[knowhere::meta::METRIC_TYPE] = metric_type;
config[DIM_KEY] = dim;
config[milvus::index::DISK_ANN_MAX_DEGREE] = std::to_string(24);
config[milvus::index::DISK_ANN_SEARCH_LIST_SIZE] = std::to_string(56);
config[milvus::index::DISK_ANN_PQ_CODE_BUDGET] = std::to_string(0.001);
config[milvus::index::DISK_ANN_BUILD_DRAM_BUDGET] = std::to_string(2);
config[milvus::index::DISK_ANN_BUILD_THREAD_NUM] = std::to_string(2);
disk_index->Build(config);
// Upload to serialize, then reload
auto upload_result = disk_index->Upload();
auto index_files = upload_result->GetIndexFiles();
disk_index.reset();
auto loaded_index = milvus::index::IndexFactory::GetInstance().CreateIndex(
create_index_info, file_mgr_ctx);
auto vec_index =
dynamic_cast<milvus::index::VectorIndex*>(loaded_index.get());
auto load_conf = generate_load_conf(knowhere::IndexEnum::INDEX_DISKANN,
metric_type,
dataset_size * emb_list_len);
load_conf["index_files"] = index_files;
load_conf[milvus::LOAD_PRIORITY] =
milvus::proto::common::LoadPriority::HIGH;
vec_index->Load(milvus::tracer::TraceContext{}, load_conf);
// drop field data and load index into segment
LoadIndexInfo load_info;
load_info.field_id = array_vec.get();
load_info.field_type = DataType::VECTOR_ARRAY;
load_info.element_type = element_type;
load_info.index_params = GenIndexParams(vec_index);
load_info.cache_index =
CreateTestCacheIndex("test", std::move(loaded_index));
load_info.index_params["metric_type"] = metric_type;
segment->DropFieldData(array_vec);
segment->LoadIndex(load_info);
// bulk_subscript should go through get_emb_list -> GetEmbListByIds on disk index
auto ids_ds = GenRandomIds(dataset_size);
auto result = sealed->bulk_subscript(
nullptr, array_vec, ids_ds->GetIds(), dataset_size);
ASSERT_NE(result, nullptr);
EXPECT_EQ(result->vectors().vector_array().data_size(), dataset_size);
for (int64_t i = 0; i < dataset_size; ++i) {
auto result_vec = result->vectors().vector_array().data()[i];
auto expected_vec = vec_array_col[ids_ds->GetIds()[i]];
VerifyVectorResults(result_vec, expected_vec, element_type);
}
}
#endif // BUILD_DISK_ANN
INSTANTIATE_TEST_SUITE_P(
VectorArrayTypes,
SealedVectorArrayTest,