furyhawk/milvus
1
0
Fork 0
mirror of https://github.com/milvus-io/milvus.git synced 2026-07-20 09:45:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: fix three-values nullbility and race in growing segmentArrayOffets (#51415)

Browse Source 
issue: https://github.com/milvus-io/milvus/issues/51381

---------

Signed-off-by: SpadeA <tangchenjie1210@gmail.com>
Signed-off-by: SpadeA-Tang <tangchenjie1210@gmail.com>
This commit is contained in:
Spade A
2026-07-20 00:22:41 +08:00
committed by GitHub
parent 42f6bcb69a
commit 149903e0fd
19 changed files with 494 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/core/src/exec/expression/Expr.h
+34 -1
View File
@@ -1769,6 +1769,39 @@ class SegmentExpr : public Expr {
return ProcessIndexChunksImpl<T>(func, true, validity_mode, values...);
}
TargetBitmap
GetFieldRowValidity(int64_t row_count) const {
TargetBitmap valid_result(row_count, true);
if (row_count == 0) {
return valid_result;
}
int64_t processed_size = 0;
for (int64_t chunk_id = 0;
chunk_id < num_data_chunk_ && processed_size < row_count;
++chunk_id) {
auto chunk_size = segment_->is_chunked()
? segment_->chunk_size(field_id_, chunk_id)
: size_per_chunk_;
auto size = std::min(chunk_size, row_count - processed_size);
if (size == 0) {
continue;
}
segment_->ApplyFieldValidData(op_ctx_,
field_id_,
chunk_id,
0,
size,
valid_result.view() + processed_size);
processed_size += size;
}
AssertInfo(processed_size == row_count,
"field validity row count mismatch: expected {}, got {}",
row_count,
processed_size);
return valid_result;
}
template <typename T, typename FUNC, typename... ValTypes>
VectorPtr
ProcessIndexChunksImpl(FUNC func,
@@ -1874,7 +1907,7 @@ class SegmentExpr : public Expr {
}
} else if (cached_is_nested_index_ &&
func_returns_row_level) {
valid_res = TargetBitmap(active_count_, true);
valid_res = GetFieldRowValidity(active_count_);
} else {
valid_res = index_ptr->IsNotNull();
}
internal/core/src/exec/expression/JsonContainsExpr.cpp
+19 -3
View File
@@ -136,13 +136,29 @@ PhyJsonContainsFilterExpr::Eval(EvalCtx& context, VectorPtr& result) {
auto input = context.get_offset_input();
SetHasOffsetInput((input != nullptr));
if (expr_->vals_.empty()) {
auto real_batch_size = has_offset_input_
? context.get_offset_input()->size()
: GetNextBatchSize();
auto real_batch_size =
has_offset_input_ ? input->size() : GetNextBatchSize();
if (real_batch_size == 0) {
result = nullptr;
return;
}
if (expr_->column_.data_type_ == DataType::ARRAY &&
expr_->column_.nullable_) {
auto valid_result =
has_offset_input_
? ProcessChunksForValidByOffsets<ArrayView>(false, *input)
: ProcessDataChunksForValid<ArrayView>();
const bool empty_matches =
expr_->op_ == proto::plan::JSONContainsExpr_JSONOp_ContainsAll;
TargetBitmap value_result(real_batch_size, empty_matches);
value_result &= valid_result;
result = std::make_shared<ColumnVector>(std::move(value_result),
std::move(valid_result));
return;
}
if (expr_->op_ == proto::plan::JSONContainsExpr_JSONOp_ContainsAll) {
result = std::make_shared<ColumnVector>(
TargetBitmap(real_batch_size, true),
internal/core/src/exec/expression/JsonContainsExpr.h
+5
View File
@@ -495,6 +495,11 @@ class PhyJsonContainsFilterExpr : public SegmentExpr {
exec_path_ = ExprExecPath::JsonStats;
return;
}
if (expr_->column_.data_type_ == DataType::ARRAY &&
expr_->vals_.empty()) {
exec_path_ = ExprExecPath::RawData;
return;
}
SegmentExpr::DetermineExecPath();
}
internal/core/src/exec/expression/MatchExpr.cpp
+54
View File
@@ -288,6 +288,8 @@ PhyMatchFilterExpr::Eval(EvalCtx& context, VectorPtr& result) {
if (MatchEmptyElements(match_type, threshold)) {
bitset_view.set();
}
ApplyStructRowValidity(
col_vec.get(), field_meta.get_id(), input, batch_rows);
if (!has_offset_input_) {
current_pos_ += batch_rows;
}
@@ -386,10 +388,62 @@ PhyMatchFilterExpr::Eval(EvalCtx& context, VectorPtr& result) {
dispatch.template operator()<false>();
}
ApplyStructRowValidity(
col_vec.get(), field_meta.get_id(), input, batch_rows);
if (!has_offset_input_) {
current_pos_ += batch_rows;
}
}
void
PhyMatchFilterExpr::ApplyStructRowValidity(ColumnVector* col_vec,
FieldId field_id,
const OffsetVector* input,
int64_t batch_rows) {
TargetBitmapView value_view(col_vec->GetRawData(), col_vec->size());
TargetBitmapView valid_view(col_vec->GetValidRawData(), col_vec->size());
if (input != nullptr) {
AssertInfo(static_cast<int64_t>(input->size()) == batch_rows,
"offset input size {} does not match batch row count {}",
input->size(),
batch_rows);
std::vector<int64_t> row_offsets(batch_rows);
for (int64_t i = 0; i < batch_rows; ++i) {
row_offsets[i] = static_cast<int64_t>((*input)[i]);
}
segment_->ApplyFieldValidDataByOffsets(
op_ctx_, field_id, row_offsets.data(), batch_rows, valid_view);
} else {
int64_t processed = 0;
int64_t row_offset = current_pos_;
while (processed < batch_rows) {
auto [chunk_id, offset_in_chunk] =
segment_->get_chunk_by_offset(field_id, row_offset);
auto count = std::min(
batch_rows - processed,
segment_->chunk_size(field_id, chunk_id) - offset_in_chunk);
AssertInfo(count > 0,
"invalid field validity range at row offset {} for "
"field {}",
row_offset,
field_id.get());
segment_->ApplyFieldValidData(op_ctx_,
field_id,
chunk_id,
offset_in_chunk,
count,
valid_view + processed);
processed += count;
row_offset += count;
}
}
for (int64_t i = 0; i < batch_rows; ++i) {
if (!valid_view[i]) {
value_view[i] = false;
}
}
}
} // namespace exec
} // namespace milvus
internal/core/src/exec/expression/MatchExpr.h
+7
View File
@@ -83,6 +83,13 @@ class PhyMatchFilterExpr : public Expr {
return false;
}
private:
void
ApplyStructRowValidity(ColumnVector* col_vec,
FieldId field_id,
const OffsetVector* input,
int64_t batch_rows);
private:
std::shared_ptr<const milvus::expr::MatchExpr> expr_;
const segcore::SegmentInternalInterface* segment_;
internal/core/src/exec/expression/MatchExprTest.cpp
+224
View File
@@ -24,10 +24,12 @@
#include <iostream>
#include <map>
#include <memory>
#include <numeric>
#include <optional>
#include <random>
#include <set>
#include <string>
#include <thread>
#include <utility>
#include <vector>
@@ -487,6 +489,228 @@ TEST(MatchExprZeroElementBatch, MatchAnyTreatsEmptyRowsAsNoMatch) {
EXPECT_EQ(result->offset(0), 2);
}
namespace {
std::unique_ptr<InsertRecordProto>
BuildNullableStructInsertData(const std::shared_ptr<Schema>& schema,
FieldId int64_fid,
FieldId sub_int_fid) {
constexpr int64_t row_count = 5;
auto insert_data = std::make_unique<InsertRecordProto>();
std::vector<int64_t> ids(row_count);
std::iota(ids.begin(), ids.end(), 0);
auto id_array = CreateDataArrayFrom(
ids.data(), nullptr, row_count, schema->operator[](int64_fid));
insert_data->mutable_fields_data()->AddAllocated(id_array.release());
auto* sub_field = insert_data->add_fields_data();
sub_field->set_field_id(sub_int_fid.get());
sub_field->set_field_name("struct_array[sub_int]");
sub_field->set_type(proto::schema::DataType::Array);
auto* array_data = sub_field->mutable_scalars()->mutable_array_data();
array_data->set_element_type(proto::schema::DataType::Int32);
const std::vector<bool> valid = {true, false, true, true, false};
for (auto is_valid : valid) {
sub_field->add_valid_data(is_valid);
}
auto append_row = [array_data](std::initializer_list<int32_t> values) {
auto* row = array_data->add_data();
for (auto value : values) {
row->mutable_int_data()->add_data(value);
}
};
append_row({1, 2});
append_row({});
append_row({});
append_row({9001});
append_row({});
insert_data->set_num_rows(row_count);
return insert_data;
}
std::set<int64_t>
RetrieveOffsets(SegmentInternalInterface* segment,
const std::shared_ptr<Schema>& schema,
const std::string& expr) {
ScopedSchemaHandle schema_handle(*schema);
auto plan_str = schema_handle.Parse(expr);
auto plan =
CreateRetrievePlanByExpr(schema, plan_str.data(), plan_str.size());
EXPECT_NE(plan, nullptr);
auto result = segment->Retrieve(
nullptr, plan.get(), 1L << 63, DEFAULT_MAX_OUTPUT_SIZE, false);
EXPECT_NE(result, nullptr);
std::set<int64_t> offsets;
if (result != nullptr) {
offsets.insert(result->offset().begin(), result->offset().end());
}
return offsets;
}
void
CheckNullableStructExpressions(SegmentInternalInterface* segment,
const std::shared_ptr<Schema>& schema) {
EXPECT_EQ(
RetrieveOffsets(
segment, schema, "match_any(struct_array, $[sub_int] >= 9000)"),
(std::set<int64_t>{3}));
EXPECT_EQ(RetrieveOffsets(
segment, schema, "match_all(struct_array, $[sub_int] >= 0)"),
(std::set<int64_t>{0, 2, 3}));
EXPECT_EQ(
RetrieveOffsets(
segment, schema, "not match_any(struct_array, $[sub_int] >= 9000)"),
(std::set<int64_t>{0, 2}));
EXPECT_EQ(
RetrieveOffsets(
segment, schema, "json_contains_all(struct_array[sub_int], [])"),
(std::set<int64_t>{0, 2, 3}));
EXPECT_TRUE(RetrieveOffsets(segment,
schema,
"json_contains_any(struct_array[sub_int], [])")
.empty());
EXPECT_EQ(
RetrieveOffsets(segment,
schema,
"not json_contains_any(struct_array[sub_int], [])"),
(std::set<int64_t>{0, 2, 3}));
}
std::shared_ptr<Schema>
BuildNullableStructSchema(FieldId& int64_fid, FieldId& sub_int_fid) {
auto schema = std::make_shared<Schema>();
int64_fid = schema->AddDebugField("id", DataType::INT64);
schema->set_primary_field_id(int64_fid);
sub_int_fid = schema->AddDebugArrayField(
"struct_array[sub_int]", DataType::INT32, true);
return schema;
}
} // namespace
TEST(MatchExprNullableStruct, SealedPropagatesStructRowValidity) {
FieldId int64_fid;
FieldId sub_int_fid;
auto schema = BuildNullableStructSchema(int64_fid, sub_int_fid);
auto insert_data =
BuildNullableStructInsertData(schema, int64_fid, sub_int_fid);
GeneratedData generated_data;
generated_data.schema_ = schema;
generated_data.raw_ = insert_data.release();
for (int64_t i = 0; i < 5; ++i) {
generated_data.row_ids_.push_back(i);
generated_data.timestamps_.push_back(i);
}
auto segment = CreateSealedWithFieldDataLoaded(schema, generated_data);
CheckNullableStructExpressions(segment.get(), schema);
}
TEST(MatchExprNullableStruct, GrowingPropagatesStructRowValidity) {
FieldId int64_fid;
FieldId sub_int_fid;
auto schema = BuildNullableStructSchema(int64_fid, sub_int_fid);
auto insert_data =
BuildNullableStructInsertData(schema, int64_fid, sub_int_fid);
std::vector<idx_t> row_ids = {0, 1, 2, 3, 4};
std::vector<Timestamp> timestamps = {0, 1, 2, 3, 4};
auto segment = CreateGrowingSegment(schema, empty_index_meta);
segment->PreInsert(row_ids.size());
segment->Insert(0,
row_ids.size(),
row_ids.data(),
timestamps.data(),
insert_data.get());
CheckNullableStructExpressions(segment.get(), schema);
}
TEST(MatchExprNullableStruct, NestedIndexUsesPhysicalRowValidity) {
FieldId int64_fid;
FieldId sub_int_fid;
auto schema = BuildNullableStructSchema(int64_fid, sub_int_fid);
auto insert_data =
BuildNullableStructInsertData(schema, int64_fid, sub_int_fid);
GeneratedData generated_data;
generated_data.schema_ = schema;
generated_data.raw_ = insert_data.release();
for (int64_t i = 0; i < 5; ++i) {
generated_data.row_ids_.push_back(i);
generated_data.timestamps_.push_back(i);
}
auto segment = CreateSealedWithFieldDataLoaded(schema, generated_data);
std::vector<boost::container::vector<int32_t>> arrays = {
{1, 2}, {}, {}, {9001}, {}};
auto index = std::make_unique<index::InvertedIndexTantivy<int32_t>>();
Config cfg;
cfg["is_array"] = true;
cfg["is_nested_index"] = true;
index->BuildWithRawDataForUT(arrays.size(), arrays.data(), cfg);
LoadIndexInfo info{};
info.field_id = sub_int_fid.get();
info.index_params = GenIndexParams(index.get());
info.cache_index =
CreateTestCacheIndex("nullable_sub_int", std::move(index));
segment->LoadIndex(info);
EXPECT_EQ(
RetrieveOffsets(segment.get(),
schema,
"not array_contains(struct_array[sub_int], 9001)"),
(std::set<int64_t>{0, 2}));
}
TEST(StructArrayOffsetsReopen, ConcurrentReadersAreSynchronized) {
auto schema = std::make_shared<Schema>();
auto int64_fid = schema->AddDebugField("id", DataType::INT64);
schema->set_primary_field_id(int64_fid);
auto base_sub_fid = schema->AddDebugArrayField(
"base_struct[sub_int]", DataType::INT32, true);
schema->set_schema_version(1);
auto segment = CreateGrowingSegment(schema, empty_index_meta);
std::atomic<bool> stop{false};
std::atomic<int64_t> missing{0};
std::vector<std::thread> readers;
for (int i = 0; i < 4; ++i) {
readers.emplace_back([&]() {
while (!stop.load(std::memory_order_relaxed)) {
if (segment->GetArrayOffsets(base_sub_fid) == nullptr) {
missing.fetch_add(1, std::memory_order_relaxed);
}
}
});
}
auto latest_schema = schema;
for (int version = 2; version <= 32; ++version) {
auto next_schema = std::make_shared<Schema>(*latest_schema);
next_schema->AddDebugArrayField(
"struct_" + std::to_string(version) + "[sub_int]",
DataType::INT32,
true);
next_schema->set_schema_version(version);
segment->LazyCheckSchema(next_schema, nullptr);
latest_schema = std::move(next_schema);
}
stop.store(true, std::memory_order_relaxed);
for (auto& reader : readers) {
reader.join();
}
EXPECT_EQ(missing.load(), 0);
}
class SealedMatchExprTest : public ::testing::Test {
protected:
void
internal/core/src/index/BitmapIndex.h
+1 -2
View File
@@ -188,8 +188,7 @@ class BitmapIndex : public ScalarIndex<T> {
const bool
HasRawData() const override {
if (schema_.data_type() == proto::schema::DataType::Array &&
!is_nested_index_) {
if (schema_.data_type() == proto::schema::DataType::Array) {
return false;
}
return true;
internal/core/src/index/BitmapIndexArrayTest.cpp
+2 -1
View File
@@ -578,7 +578,7 @@ TEST(BitmapIndexArrayNestedTest, BuildAndLoadElementLevelBitmap) {
auto index = std::make_unique<index::BitmapIndex<int32_t>>(ctx, true);
index->BuildWithFieldData(std::vector<FieldDataPtr>{field_data});
ASSERT_TRUE(index->IsNestedIndex());
ASSERT_TRUE(index->HasRawData());
ASSERT_FALSE(index->HasRawData());
ASSERT_EQ(index->Count(), 4);
auto binary_set = index->Serialize({});
@@ -586,6 +586,7 @@ TEST(BitmapIndexArrayNestedTest, BuildAndLoadElementLevelBitmap) {
std::make_unique<index::BitmapIndex<int32_t>>(ctx, false);
loaded_index->Load(binary_set, {});
ASSERT_TRUE(loaded_index->IsNestedIndex());
ASSERT_FALSE(loaded_index->HasRawData());
ASSERT_EQ(loaded_index->Count(), 4);
int32_t value = 2;
internal/core/src/index/IndexFactory.cpp
+3 -2
View File
@@ -215,7 +215,8 @@ ResolveHybridInternalIndexType(
bool
IndexFactory::CanUseIndexRawDataForField(DataType field_type,
bool has_raw_data) {
return has_raw_data && field_type != DataType::ARRAY;
return has_raw_data && field_type != DataType::ARRAY &&
field_type != DataType::VECTOR_ARRAY;
}
template <typename T>
@@ -535,7 +536,7 @@ IndexFactory::VecIndexLoadResource(
LoadResourceRequest request{};
const auto& res = resource.value();
request.has_raw_data = has_raw_data;
request.has_raw_data = CanUseIndexRawDataForField(field_type, has_raw_data);
request.final_disk_cost = res.diskCost;
request.final_memory_cost = res.memoryCost;
if (knowhere::UseDiskLoad(index_type, index_version) || mmaped) {
internal/core/src/mmap/ChunkedColumnInterface.h
+2 -2
View File
@@ -164,7 +164,7 @@ class ChunkedColumnInterface {
offset,
size,
chunk->RowNums());
auto& valid_data = chunk->Valid();
const auto& valid_data = chunk->Valid();
AssertInfo(
offset + size <= static_cast<int64_t>(valid_data.size()),
"Valid-data range out of valid-data bounds, offset: {}, size: {}, "
@@ -173,7 +173,7 @@ class ChunkedColumnInterface {
size,
valid_data.size());
for (int64_t i = 0; i < size; ++i) {
if (!chunk->isValid(offset + i)) {
if (!valid_data[offset + i]) {
valid_result[i] = false;
}
}
internal/core/src/segcore/ChunkedSegmentSealedImpl.cpp
+3
View File
@@ -698,6 +698,9 @@ ChunkedSegmentSealedImpl::LoadVecIndex(LoadIndexInfo& info,
info.num_rows,
info.dim);
}
request.has_raw_data =
milvus::index::IndexFactory::CanUseIndexRawDataForField(
field_meta.get_data_type(), request.has_raw_data);
// Note: raw data lifecycle (eviction/drop) is handled by LoadDiff + ApplyLoadDiff,
// not here. This avoids unsafe ManualEvictCache on column groups.
internal/core/src/segcore/SegmentGrowingImpl.cpp
+47 -18
View File
@@ -344,6 +344,8 @@ SchemaHasTextField(const Schema& schema) {
void
SegmentGrowingImpl::InitializeArrayOffsets() {
std::unique_lock lock(array_offsets_map_mutex_);
// Group fields by struct_name
std::unordered_map<std::string, std::vector<FieldId>> struct_fields;
@@ -751,8 +753,17 @@ SegmentGrowingImpl::Insert(int64_t reserved_offset,
field_meta);
}
// update ArrayOffsetsGrowing for struct fields
if (struct_representative_fields_.count(field_id) > 0) {
std::shared_ptr<ArrayOffsetsGrowing> array_offsets;
{
std::shared_lock lock(array_offsets_map_mutex_);
if (struct_representative_fields_.count(field_id) > 0) {
auto offsets_it = array_offsets_map_.find(field_id);
if (offsets_it != array_offsets_map_.end()) {
array_offsets = offsets_it->second;
}
}
}
if (array_offsets != nullptr) {
const auto& field_data =
insert_record_proto->fields_data(data_offset);
@@ -760,11 +771,8 @@ SegmentGrowingImpl::Insert(int64_t reserved_offset,
ExtractArrayLengths(
field_data, field_meta, num_rows, array_lengths.data());
auto offsets_it = array_offsets_map_.find(field_id);
if (offsets_it != array_offsets_map_.end()) {
offsets_it->second->Insert(
reserved_offset, array_lengths.data(), num_rows);
}
array_offsets->Insert(
reserved_offset, array_lengths.data(), num_rows);
}
// index text.
@@ -1008,17 +1016,22 @@ SegmentGrowingImpl::load_field_data_common(
}
}
// update ArrayOffsetsGrowing for struct fields
if (struct_representative_fields_.count(field_id) > 0) {
std::shared_ptr<ArrayOffsetsGrowing> array_offsets;
{
std::shared_lock lock(array_offsets_map_mutex_);
if (struct_representative_fields_.count(field_id) > 0) {
auto offsets_it = array_offsets_map_.find(field_id);
if (offsets_it != array_offsets_map_.end()) {
array_offsets = offsets_it->second;
}
}
}
if (array_offsets != nullptr) {
std::vector<int32_t> array_lengths(num_rows);
ExtractArrayLengthsFromFieldData(
field_data, field_meta, array_lengths.data());
auto offsets_it = array_offsets_map_.find(field_id);
if (offsets_it != array_offsets_map_.end()) {
offsets_it->second->Insert(
reserved_offset, array_lengths.data(), num_rows);
}
array_offsets->Insert(reserved_offset, array_lengths.data(), num_rows);
LOG_INFO("Updated ArrayOffsetsGrowing for field {} with {} rows",
field_id.get(),
@@ -2707,6 +2720,8 @@ SegmentGrowingImpl::FillAbsentFields() {
insert_record_.is_valid_data_exist(field_id) &&
insert_record_.get_valid_data(field_id)->get_data().empty()) {
fill_empty_field(field_meta);
EnsureArrayOffsetsForStructField(field_meta,
insert_record_.row_count());
}
continue;
}
@@ -2714,6 +2729,8 @@ SegmentGrowingImpl::FillAbsentFields() {
// so we must check data empty here
if (insert_record_.get_data_base(field_id)->empty()) {
fill_empty_field(field_meta);
EnsureArrayOffsetsForStructField(field_meta,
insert_record_.row_count());
}
}
}
@@ -3003,6 +3020,8 @@ SegmentGrowingImpl::EnsureArrayOffsetsForStructField(
return;
}
std::unique_lock lock(array_offsets_map_mutex_);
std::shared_ptr<ArrayOffsetsGrowing> array_offsets;
for (const auto& [field_id, offsets] : array_offsets_map_) {
auto field_it = schema_->get_fields().find(field_id);
@@ -3020,13 +3039,23 @@ SegmentGrowingImpl::EnsureArrayOffsetsForStructField(
if (!array_offsets) {
array_offsets = std::make_shared<ArrayOffsetsGrowing>();
if (row_count > 0) {
std::vector<int32_t> empty_lengths(row_count, 0);
array_offsets->Insert(0, empty_lengths.data(), row_count);
}
struct_representative_fields_.insert(field_meta.get_id());
}
auto current_row_count = array_offsets->GetRowCount();
AssertInfo(current_row_count <= row_count,
"struct array offsets row count {} exceeds segment row count "
"{} for field {}",
current_row_count,
row_count,
field_meta.get_id().get());
if (current_row_count < row_count) {
auto missing_row_count = row_count - current_row_count;
std::vector<int32_t> empty_lengths(missing_row_count, 0);
array_offsets->Insert(
current_row_count, empty_lengths.data(), missing_row_count);
}
array_offsets_map_[field_meta.get_id()] = array_offsets;
}
internal/core/src/segcore/SegmentGrowingImpl.h
+3
View File
@@ -662,6 +662,7 @@ class SegmentGrowingImpl : public SegmentGrowing {
std::shared_ptr<const IArrayOffsets>
GetArrayOffsets(FieldId field_id) const override {
std::shared_lock lock(array_offsets_map_mutex_);
auto it = array_offsets_map_.find(field_id);
if (it != array_offsets_map_.end()) {
return it->second;
@@ -893,6 +894,8 @@ class SegmentGrowingImpl : public SegmentGrowing {
// One field_id per struct, since all fields in the same struct have identical array lengths
std::unordered_set<FieldId> struct_representative_fields_;
mutable std::shared_mutex array_offsets_map_mutex_;
// Tracked resource usage for refund-then-charge pattern
// This stores the last estimated resource usage that was charged to the cache manager
ResourceUsage tracked_resource_{};
internal/core/src/segcore/SegmentGrowingTest.cpp
+39
View File
@@ -349,6 +349,45 @@ TEST(Growing, MissingStructArrayOffsetsReturnsEmptyForOldRows) {
}
}
TEST(Growing, LoadMissingStructArrayOffsetsReturnsEmptyForOldRows) {
auto schema = std::make_shared<Schema>();
auto pk = schema->AddDebugField("pk", DataType::INT64);
schema->set_primary_field_id(pk);
auto label =
schema->AddDebugArrayField("chunks[label]", DataType::VARCHAR, true);
auto score =
schema->AddDebugArrayField("chunks[score]", DataType::INT32, true);
constexpr int64_t row_count = 5;
auto dataset = DataGen(schema, row_count);
auto config = SegcoreConfig::default_config();
auto segment = CreateGrowingWithFieldDataLoaded(
schema,
empty_index_meta,
config,
dataset,
false,
std::vector<int64_t>{label.get(), score.get()});
auto* growing = dynamic_cast<SegmentGrowingImpl*>(segment.get());
ASSERT_NE(growing, nullptr);
growing->FillAbsentFields();
ASSERT_EQ(growing->get_row_count(), row_count);
auto offsets = growing->GetArrayOffsets(label);
ASSERT_NE(offsets, nullptr);
auto score_offsets = growing->GetArrayOffsets(score);
ASSERT_NE(score_offsets, nullptr);
EXPECT_EQ(offsets.get(), score_offsets.get());
EXPECT_EQ(offsets->GetRowCount(), row_count);
EXPECT_EQ(offsets->GetTotalElementCount(), 0);
for (int64_t i = 0; i <= row_count; ++i) {
auto [start, end] = offsets->ElementIDRangeOfRow(i);
EXPECT_EQ(start, 0);
EXPECT_EQ(end, 0);
}
}
class GrowingTest
: public ::testing::TestWithParam<
std::tuple</*index type*/ std::string, knowhere::MetricType>> {
internal/core/unittest/test_loading.cpp
+3 -2
View File
@@ -267,7 +267,8 @@ static const auto kIndexLoadTestValues = ::testing::Values(
1UL * 1024 * 1024 * 1024,
1UL * 1024 * 1024 * 1024,
false}),
// VECTOR_ARRAY + HNSW (FLAT): has_raw_data = true
// VECTOR_ARRAY + HNSW (FLAT): keep field data resident for struct offsets
// and row validity even when the index carries raw payloads.
std::pair<std::map<std::string, std::string>, LoadResourceRequest>(
{{"index_type", "HNSW"},
{"metric_type", "MAX_SIM"},
@@ -276,7 +277,7 @@ static const auto kIndexLoadTestValues = ::testing::Values(
{"mmap", "false"},
{"field_type", "vector_array"},
{"element_type", "vector_float"}},
{2UL * 1024 * 1024 * 1024, 0UL, 1UL * 1024 * 1024 * 1024, 0UL, true}),
{2UL * 1024 * 1024 * 1024, 0UL, 1UL * 1024 * 1024 * 1024, 0UL, false}),
// VECTOR_ARRAY + HNSW_SQ (SQ8, fp32): has_raw_data = false (lossy)
std::pair<std::map<std::string, std::string>, LoadResourceRequest>(
{{"index_type", "HNSW_SQ"},
internal/core/unittest/test_sealed.cpp
+5 -4
View File
@@ -3025,7 +3025,7 @@ TEST_P(SealedVectorArrayTest, SearchVectorArray) {
}
}
TEST_P(SealedVectorArrayTest, BulkSubscriptVectorArrayFromIndex) {
TEST_P(SealedVectorArrayTest, DISABLED_BulkSubscriptVectorArrayFromIndex) {
int64_t collection_id = 1;
int64_t partition_id = 2;
int64_t segment_id = 3;
@@ -3216,7 +3216,7 @@ TEST(SealedVectorArrayNullable, BulkSubscriptEmptyThenSingleVectorArrayRows) {
}
TEST(SealedVectorArrayNullable,
BulkSubscriptVectorArrayFromIndexKeepsLogicalRows) {
DISABLED_BulkSubscriptVectorArrayFromIndexKeepsLogicalRows) {
int64_t collection_id = 1;
int64_t partition_id = 2;
int64_t segment_id = 3;
@@ -3355,7 +3355,8 @@ TEST(SealedVectorArrayNullable,
}
}
TEST(SealedVectorArrayNullable, BulkSubscriptAllNullVectorArrayFromIndex) {
TEST(SealedVectorArrayNullable,
DISABLED_BulkSubscriptAllNullVectorArrayFromIndex) {
int64_t collection_id = 1;
int64_t partition_id = 2;
int64_t segment_id = 3;
@@ -3611,7 +3612,7 @@ TEST(SealedVectorArrayFallback,
}
#ifdef BUILD_DISK_ANN
TEST_P(SealedVectorArrayTest, BulkSubscriptVectorArrayFromDiskIndex) {
TEST_P(SealedVectorArrayTest, DISABLED_BulkSubscriptVectorArrayFromDiskIndex) {
// DiskANN only registers fp32, fp16, bf16
if (element_type == DataType::VECTOR_BINARY ||
element_type == DataType::VECTOR_INT8) {
internal/parser/planparserv2/parser_visitor.go
+1
View File
@@ -2090,6 +2090,7 @@ func (v *ParserVisitor) VisitStructField(ctx *parser.StructFieldContext) interfa
FieldId: field.FieldID,
DataType: field.DataType,
ElementType: field.GetElementType(),
Nullable: field.GetNullable(),
},
},
},
internal/parser/planparserv2/plan_parser_v2_test.go
+37
View File
@@ -2551,6 +2551,43 @@ func Test_JSONContainsAll(t *testing.T) {
}
}
func Test_JSONContainsStructFieldUsesSubFieldNullable(t *testing.T) {
for _, testcase := range []struct {
name string
parentNullable bool
childNullable bool
wantNullable bool
}{
{name: "child nullable", childNullable: true, wantNullable: true},
{name: "parent nullable only", parentNullable: true, wantNullable: false},
} {
t.Run(testcase.name, func(t *testing.T) {
schema := newTestSchema(true)
schema.StructArrayFields[0].Nullable = testcase.parentNullable
schema.StructArrayFields[0].Fields[1].Nullable = testcase.childNullable
helper, err := typeutil.CreateSchemaHelper(schema)
require.NoError(t, err)
plan, err := CreateSearchPlan(helper, `json_contains_all(struct_array[sub_int], [])`, "FloatVectorField", &planpb.QueryInfo{
Topk: 0,
MetricType: "",
SearchParams: "",
RoundDecimal: 0,
}, nil, nil)
require.NoError(t, err)
jsonContainsExpr := plan.GetVectorAnns().GetPredicates().GetJsonContainsExpr()
require.NotNil(t, jsonContainsExpr)
columnInfo := jsonContainsExpr.GetColumnInfo()
require.NotNil(t, columnInfo)
assert.Equal(t, int64(134), columnInfo.GetFieldId())
assert.Equal(t, schemapb.DataType_Array, columnInfo.GetDataType())
assert.Equal(t, schemapb.DataType_Int32, columnInfo.GetElementType())
assert.Equal(t, testcase.wantNullable, columnInfo.GetNullable())
})
}
}
func Test_JSONContainsAny(t *testing.T) {
schema := newTestSchemaHelper(t)
expr := ""
tests/python_client/milvus_client/test_milvus_client_struct_array_nullable.py
+5 -2
View File
@@ -965,7 +965,9 @@ def gt_struct_array_expression_rows(
threshold = int(raw_threshold) if raw_threshold is not None else None
results = []
for row in scoped_rows:
struct_array = row.get(STRUCT_FIELD) or []
struct_array = row.get(STRUCT_FIELD)
if struct_array is None:
continue
match_count = sum(1 for element in struct_array if _eval_struct_element_condition(element, condition))
total_count = len(struct_array)
matched = (
@@ -1887,7 +1889,8 @@ class TestMilvusClientStructArraySchemaEvolution(TestMilvusClientV2Base):
method: create nullable scalar Struct Array rows covering explicit null, omitted, empty, one-match,
two-match, and zero-match profiles in both sealed and growing segments, then query every MATCH operator
expected: expected result sets are computed from source data and expression; MATCH_ANY/LEAST only return rows
with enough matching elements, and MATCH_ALL/MOST/EXACT treat null/empty profiles as zero-element inputs
with enough matching elements, null/omitted profiles evaluate to unknown, and MATCH_ALL/MOST/EXACT treat
empty profiles as zero-element inputs
"""
collection_name = cf.gen_unique_str(f"{prefix}_nullable_struct_match_null_expr")
client = self._client()