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

fix: support null predicates on struct array parent fields (#50436)

Browse Source 
issue: https://github.com/milvus-io/milvus/issues/50081
ref: https://github.com/milvus-io/milvus/issues/42148

Signed-off-by: SpadeA <tangchenjie1210@gmail.com>
This commit is contained in:
Spade A
2026-06-11 18:08:22 +08:00
committed by GitHub
parent 65f248bd73
commit f3b89c72c1
11 changed files with 666 additions and 154 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/core/src/exec/expression/Expr.h
+124 -148
View File
@@ -1668,139 +1668,138 @@ class SegmentExpr : public Expr {
template <typename T>
TargetBitmap
ProcessChunksForValid(bool use_index) {
if (use_index) {
// when T is ArrayView, the ScalarIndex<T> shall be ScalarIndex<ElementType>
// NOT ScalarIndex<ArrayView>
if (std::is_same_v<T, ArrayView>) {
auto element_type =
segment_->get_schema()[field_id_].get_element_type();
switch (element_type) {
case DataType::BOOL: {
return ProcessIndexChunksForValid<bool>();
}
case DataType::INT8: {
return ProcessIndexChunksForValid<int8_t>();
}
case DataType::INT16: {
return ProcessIndexChunksForValid<int16_t>();
}
case DataType::INT32: {
return ProcessIndexChunksForValid<int32_t>();
}
case DataType::INT64: {
return ProcessIndexChunksForValid<int64_t>();
}
case DataType::FLOAT: {
return ProcessIndexChunksForValid<float>();
}
case DataType::DOUBLE: {
return ProcessIndexChunksForValid<double>();
}
case DataType::STRING:
case DataType::VARCHAR: {
return ProcessIndexChunksForValid<std::string>();
}
case DataType::GEOMETRY: {
return ProcessIndexChunksForValid<std::string>();
}
default:
ThrowInfo(DataTypeInvalid,
"unsupported element type: {}",
element_type);
}
}
return ProcessIndexChunksForValid<T>();
} else {
if constexpr (std::is_same_v<T, VectorArray>) {
return ProcessDataChunksForValid<T>();
} else {
if (use_index) {
// when T is ArrayView, the ScalarIndex<T> shall be ScalarIndex<ElementType>
// NOT ScalarIndex<ArrayView>
if (std::is_same_v<T, ArrayView>) {
auto element_type =
segment_->get_schema()[field_id_].get_element_type();
switch (element_type) {
case DataType::BOOL: {
return ProcessIndexChunksForValid<bool>();
}
case DataType::INT8: {
return ProcessIndexChunksForValid<int8_t>();
}
case DataType::INT16: {
return ProcessIndexChunksForValid<int16_t>();
}
case DataType::INT32: {
return ProcessIndexChunksForValid<int32_t>();
}
case DataType::INT64: {
return ProcessIndexChunksForValid<int64_t>();
}
case DataType::FLOAT: {
return ProcessIndexChunksForValid<float>();
}
case DataType::DOUBLE: {
return ProcessIndexChunksForValid<double>();
}
case DataType::STRING:
case DataType::VARCHAR: {
return ProcessIndexChunksForValid<std::string>();
}
case DataType::GEOMETRY: {
return ProcessIndexChunksForValid<std::string>();
}
default:
ThrowInfo(DataTypeInvalid,
"unsupported element type: {}",
element_type);
}
}
return ProcessIndexChunksForValid<T>();
} else {
return ProcessDataChunksForValid<T>();
}
}
}
template <typename T>
TargetBitmap
ProcessChunksForValidByOffsets(bool use_index, const OffsetVector& input) {
typedef std::
conditional_t<std::is_same_v<T, std::string_view>, std::string, T>
IndexInnerType;
using Index = index::ScalarIndex<IndexInnerType>;
auto batch_size = input.size();
TargetBitmap valid_result(batch_size);
valid_result.set();
if (use_index) {
// when T is ArrayView, the ScalarIndex<T> shall be ScalarIndex<ElementType>
// NOT ScalarIndex<ArrayView>
if (std::is_same_v<T, ArrayView>) {
auto element_type =
segment_->get_schema()[field_id_].get_element_type();
switch (element_type) {
case DataType::BOOL: {
return ProcessChunksForValidByOffsets<bool>(use_index,
input);
}
case DataType::INT8: {
return ProcessChunksForValidByOffsets<int8_t>(use_index,
input);
}
case DataType::INT16: {
return ProcessChunksForValidByOffsets<int16_t>(
use_index, input);
}
case DataType::INT32: {
return ProcessChunksForValidByOffsets<int32_t>(
use_index, input);
}
case DataType::INT64: {
return ProcessChunksForValidByOffsets<int64_t>(
use_index, input);
}
case DataType::FLOAT: {
return ProcessChunksForValidByOffsets<float>(use_index,
input);
}
case DataType::DOUBLE: {
return ProcessChunksForValidByOffsets<double>(use_index,
input);
}
case DataType::STRING:
case DataType::VARCHAR: {
return ProcessChunksForValidByOffsets<std::string>(
use_index, input);
}
default:
ThrowInfo(DataTypeInvalid,
"unsupported element type: {}",
element_type);
}
}
auto scalar_index =
dynamic_cast<const Index*>(pinned_index_[0].get());
auto* index_ptr = const_cast<Index*>(scalar_index);
const auto& res = index_ptr->IsNotNull();
for (auto i = 0; i < batch_size; ++i) {
valid_result[i] = res[input[i]];
}
auto apply_field_valid_data = [&]() {
std::vector<int64_t> offsets(input.begin(), input.end());
segment_->ApplyFieldValidDataByOffsets(
op_ctx_,
field_id_,
offsets.data(),
batch_size,
TargetBitmapView(valid_result));
};
if constexpr (std::is_same_v<T, VectorArray>) {
apply_field_valid_data();
} else {
for (auto i = 0; i < batch_size; ++i) {
auto offset = input[i];
auto [chunk_id,
chunk_offset] = [&]() -> std::pair<int64_t, int64_t> {
if (segment_->type() == SegmentType::Growing) {
return {offset / size_per_chunk_,
offset % size_per_chunk_};
} else if (segment_->is_chunked()) {
return segment_->get_chunk_by_offset(field_id_, offset);
} else {
return {0, offset};
typedef std::conditional_t<std::is_same_v<T, std::string_view>,
std::string,
T>
IndexInnerType;
using Index = index::ScalarIndex<IndexInnerType>;
if (use_index) {
// when T is ArrayView, the ScalarIndex<T> shall be ScalarIndex<ElementType>
// NOT ScalarIndex<ArrayView>
if (std::is_same_v<T, ArrayView>) {
auto element_type =
segment_->get_schema()[field_id_].get_element_type();
switch (element_type) {
case DataType::BOOL: {
return ProcessChunksForValidByOffsets<bool>(
use_index, input);
}
case DataType::INT8: {
return ProcessChunksForValidByOffsets<int8_t>(
use_index, input);
}
case DataType::INT16: {
return ProcessChunksForValidByOffsets<int16_t>(
use_index, input);
}
case DataType::INT32: {
return ProcessChunksForValidByOffsets<int32_t>(
use_index, input);
}
case DataType::INT64: {
return ProcessChunksForValidByOffsets<int64_t>(
use_index, input);
}
case DataType::FLOAT: {
return ProcessChunksForValidByOffsets<float>(
use_index, input);
}
case DataType::DOUBLE: {
return ProcessChunksForValidByOffsets<double>(
use_index, input);
}
case DataType::STRING:
case DataType::VARCHAR: {
return ProcessChunksForValidByOffsets<std::string>(
use_index, input);
}
default:
ThrowInfo(DataTypeInvalid,
"unsupported element type: {}",
element_type);
}
}();
auto pw = segment_->chunk_data<T>(op_ctx_, field_id_, chunk_id);
auto chunk = pw.get();
const bool* valid_data = chunk.valid_data();
if (valid_data != nullptr) {
valid_result[i] = valid_data[chunk_offset];
} else {
break;
}
auto scalar_index =
dynamic_cast<const Index*>(pinned_index_[0].get());
auto* index_ptr = const_cast<Index*>(scalar_index);
const auto& res = index_ptr->IsNotNull();
for (auto i = 0; i < batch_size; ++i) {
valid_result[i] = res[input[i]];
}
} else {
apply_field_valid_data();
}
}
return valid_result;
@@ -1832,35 +1831,12 @@ class SegmentExpr : public Expr {
size = std::min(size, batch_size_ - processed_size);
if (size == 0)
continue; //do not go empty-loop at the bound of the chunk
bool access_sealed_variable_column = false;
if constexpr (std::is_same_v<T, std::string_view> ||
std::is_same_v<T, Json> ||
std::is_same_v<T, ArrayView>) {
if (segment_->type() == SegmentType::Sealed) {
auto pw = segment_->get_batch_views<T>(
op_ctx_, field_id_, i, data_pos, size);
auto [data_vec, valid_data] = pw.get();
ApplyValidData(valid_data.data(),
valid_result + processed_size,
valid_result + processed_size,
size);
access_sealed_variable_column = true;
}
}
if (!access_sealed_variable_column) {
auto pw = segment_->chunk_data<T>(op_ctx_, field_id_, i);
auto chunk = pw.get();
const bool* valid_data = chunk.valid_data();
if (valid_data == nullptr) {
return valid_result;
}
valid_data += data_pos;
ApplyValidData(valid_data,
valid_result + processed_size,
valid_result + processed_size,
size);
}
segment_->ApplyFieldValidData(op_ctx_,
field_id_,
i,
data_pos,
size,
valid_result + processed_size);
processed_size += size;
if (processed_size >= batch_size_) {
internal/core/src/exec/expression/ExprArrayTest.cpp
+220 -1
View File
@@ -18,6 +18,7 @@
#include <folly/FBVector.h>
#include <gtest/gtest.h>
#include <algorithm>
#include <array>
#include <cstdint>
#include <functional>
#include <map>
@@ -550,6 +551,224 @@ TEST(Expr, TestArrayNullExpr) {
}
}
TEST(Expr, TestStructArrayParentNullExprUsesRepresentativeSubField) {
auto schema = std::make_shared<Schema>();
auto fakevec_fid = schema->AddDebugField(
"fakevec", DataType::VECTOR_FLOAT, 16, knowhere::metric::L2);
auto i64_fid = schema->AddDebugField("id", DataType::INT64);
auto profile_history_fid = schema->AddDebugField(
"profile[history]", DataType::ARRAY, DataType::INT64, true);
schema->set_primary_field_id(i64_fid);
constexpr int N = 128;
auto raw_data = DataGen(schema, N, 43, 0, 1, 2);
auto valid_data = raw_data.get_col_valid(profile_history_fid);
auto profile_history_col =
raw_data.get_col<ScalarFieldProto>(profile_history_fid);
auto growing = CreateGrowingSegment(schema, empty_index_meta);
auto offset = growing->PreInsert(N);
growing->Insert(offset,
N,
raw_data.row_ids_.data(),
raw_data.timestamps_.data(),
raw_data.raw_);
auto growing_segment = dynamic_cast<SegmentGrowingImpl*>(growing.get());
ASSERT_NE(growing_segment, nullptr);
FixedVector<Array> arrays;
arrays.reserve(N);
std::vector<uint8_t> valid_bitmap((N + 7) / 8, 0);
for (int i = 0; i < N; ++i) {
arrays.emplace_back(profile_history_col[i]);
if (valid_data[i]) {
valid_bitmap[i >> 3] |= 1 << (i & 0x07);
}
}
auto field_data =
storage::CreateFieldData(DataType::ARRAY, DataType::INT64, true);
field_data->FillFieldData(arrays.data(), valid_bitmap.data(), N, 0);
auto cm = storage::RemoteChunkManagerSingleton::GetInstance()
.GetRemoteChunkManager();
auto sealed_segment = CreateSealedSegment(schema);
auto field_data_info =
PrepareSingleFieldInsertBinlog(kCollectionID,
kPartitionID,
kSegmentID,
profile_history_fid.get(),
{field_data},
cm);
sealed_segment->LoadFieldData(field_data_info);
auto make_plan = [&](proto::plan::NullExpr_NullOp op) {
auto null_expr = std::make_shared<expr::NullExpr>(
expr::ColumnInfo(profile_history_fid,
DataType::ARRAY,
DataType::INT64,
{},
true),
op);
return std::make_shared<plan::FilterBitsNode>(DEFAULT_PLANNODE_ID,
null_expr);
};
std::vector<
std::pair<proto::plan::NullExpr_NullOp, std::function<bool(bool)>>>
testcases = {
{proto::plan::NullExpr_NullOp_IsNull,
[](bool valid) { return !valid; }},
{proto::plan::NullExpr_NullOp_IsNotNull,
[](bool valid) { return valid; }},
};
for (auto [op, ref_func] : testcases) {
std::array<const SegmentInternalInterface*, 2> segments = {
static_cast<const SegmentInternalInterface*>(growing_segment),
static_cast<const SegmentInternalInterface*>(sealed_segment.get())};
for (auto* segment : segments) {
auto plan = make_plan(op);
auto final = ExecuteQueryExpr(plan, segment, N, MAX_TIMESTAMP);
ASSERT_EQ(final.size(), N);
for (int i = 0; i < N; ++i) {
ASSERT_EQ(final[i], ref_func(valid_data[i]))
<< "segment type " << segment->type() << ", row " << i;
}
milvus::exec::OffsetVector offsets;
offsets.reserve(N / 2);
for (int i = 0; i < N; ++i) {
if (i % 2 == 0) {
offsets.emplace_back(i);
}
}
auto col_vec = milvus::test::gen_filter_res(
plan.get(), segment, N, MAX_TIMESTAMP, &offsets);
BitsetTypeView view(col_vec->GetRawData(), col_vec->size());
ASSERT_EQ(view.size(), offsets.size());
for (int i = 0; i < offsets.size(); ++i) {
ASSERT_EQ(view[i], ref_func(valid_data[offsets[i]]))
<< "segment type " << segment->type() << ", offset row "
<< offsets[i];
}
}
}
(void)fakevec_fid;
}
TEST(Expr, TestVectorArrayNullExpr) {
auto schema = std::make_shared<Schema>();
auto fakevec_fid = schema->AddDebugField(
"fakevec", DataType::VECTOR_FLOAT, 16, knowhere::metric::L2);
auto i64_fid = schema->AddDebugField("id", DataType::INT64);
auto vector_array_fid =
schema->AddDebugVectorArrayField("array_float_vector",
DataType::VECTOR_FLOAT,
4,
knowhere::metric::L2,
true);
schema->set_primary_field_id(i64_fid);
constexpr int N = 128;
auto raw_data = DataGen(schema, N, 42, 0, 1, 2);
auto valid_data = raw_data.get_col_valid(vector_array_fid);
auto vector_array_col =
raw_data.get_col<VectorFieldProto>(vector_array_fid);
auto growing = CreateGrowingSegment(schema, empty_index_meta);
auto offset = growing->PreInsert(N);
growing->Insert(offset,
N,
raw_data.row_ids_.data(),
raw_data.timestamps_.data(),
raw_data.raw_);
auto growing_segment = dynamic_cast<SegmentGrowingImpl*>(growing.get());
ASSERT_NE(growing_segment, nullptr);
std::vector<uint8_t> valid_bitmap((N + 7) / 8, 0);
std::vector<milvus::VectorArray> vector_arrays;
vector_arrays.reserve(N);
for (int i = 0; i < N; ++i) {
if (valid_data[i]) {
valid_bitmap[i >> 3] |= 1 << (i & 0x07);
vector_arrays.emplace_back(vector_array_col[i]);
}
}
auto field_data = storage::CreateFieldData(
DataType::VECTOR_ARRAY, DataType::VECTOR_FLOAT, true, 4);
field_data->FillFieldData(vector_arrays.data(), valid_bitmap.data(), N, 0);
auto cm = storage::RemoteChunkManagerSingleton::GetInstance()
.GetRemoteChunkManager();
auto sealed_segment = CreateSealedSegment(schema);
auto field_data_info =
PrepareSingleFieldInsertBinlog(kCollectionID,
kPartitionID,
kSegmentID,
vector_array_fid.get(),
{field_data},
cm);
sealed_segment->LoadFieldData(field_data_info);
auto make_plan = [&](proto::plan::NullExpr_NullOp op) {
auto null_expr = std::make_shared<expr::NullExpr>(
expr::ColumnInfo(vector_array_fid,
DataType::VECTOR_ARRAY,
DataType::VECTOR_FLOAT,
{},
true),
op);
return std::make_shared<plan::FilterBitsNode>(DEFAULT_PLANNODE_ID,
null_expr);
};
std::vector<
std::pair<proto::plan::NullExpr_NullOp, std::function<bool(bool)>>>
testcases = {
{proto::plan::NullExpr_NullOp_IsNull,
[](bool valid) { return !valid; }},
{proto::plan::NullExpr_NullOp_IsNotNull,
[](bool valid) { return valid; }},
};
for (auto [op, ref_func] : testcases) {
std::array<const SegmentInternalInterface*, 2> segments = {
static_cast<const SegmentInternalInterface*>(growing_segment),
static_cast<const SegmentInternalInterface*>(sealed_segment.get())};
for (auto* segment : segments) {
auto plan = make_plan(op);
auto final = ExecuteQueryExpr(plan, segment, N, MAX_TIMESTAMP);
EXPECT_EQ(final.size(), N);
for (int i = 0; i < N; ++i) {
ASSERT_EQ(final[i], ref_func(valid_data[i]))
<< "segment type " << segment->type() << ", row " << i;
}
milvus::exec::OffsetVector offsets;
offsets.reserve(N / 2);
for (int i = 0; i < N; ++i) {
if (i % 2 == 0) {
offsets.emplace_back(i);
}
}
auto col_vec = milvus::test::gen_filter_res(
plan.get(), segment, N, MAX_TIMESTAMP, &offsets);
BitsetTypeView view(col_vec->GetRawData(), col_vec->size());
ASSERT_EQ(view.size(), offsets.size());
for (int i = 0; i < offsets.size(); ++i) {
ASSERT_EQ(view[i], ref_func(valid_data[offsets[i]]))
<< "segment type " << segment->type() << ", offset row "
<< offsets[i];
}
}
}
(void)fakevec_fid;
}
TEST(Expr, PraseArrayContainsExpr) {
// Test expressions for array_contains operations
std::vector<const char*> exprs{
@@ -2356,4 +2575,4 @@ TEST(Expr, TestArrayContainsForStruct) {
<< "Distances should be sorted in ascending order (with index)";
}
}
}
}
internal/core/src/exec/expression/NullExpr.cpp
+13
View File
@@ -99,6 +99,10 @@ PhyNullExpr::Eval(EvalCtx& context, VectorPtr& result) {
result = ExecVisitorImpl<ArrayView>(input);
break;
}
case DataType::VECTOR_ARRAY: {
result = ExecVisitorImpl<VectorArray>(input);
break;
}
case DataType::GEOMETRY: {
if (segment_->type() == SegmentType::Growing &&
!storage::MmapManager::GetInstance()
@@ -117,6 +121,15 @@ PhyNullExpr::Eval(EvalCtx& context, VectorPtr& result) {
}
}
void
PhyNullExpr::DetermineExecPath() {
if (expr_->column_.data_type_ == DataType::VECTOR_ARRAY) {
exec_path_ = ExprExecPath::RawData;
return;
}
SegmentExpr::DetermineExecPath();
}
template <typename T>
VectorPtr
PhyNullExpr::ExecVisitorImpl(OffsetVector* input) {
internal/core/src/exec/expression/NullExpr.h
+4 -1
View File
@@ -55,6 +55,9 @@ class PhyNullExpr : public SegmentExpr {
void
Eval(EvalCtx& context, VectorPtr& result) override;
void
DetermineExecPath() override;
std::string
ToString() const override {
return fmt::format("{}", expr_->ToString());
@@ -83,4 +86,4 @@ class PhyNullExpr : public SegmentExpr {
int64_t precheck_pos_{0};
};
} //namespace exec
} // namespace milvus
} // namespace milvus
internal/core/src/segcore/ChunkedSegmentSealedImpl.cpp
+97
View File
@@ -1360,6 +1360,103 @@ ChunkedSegmentSealedImpl::prefetch_chunks(
}
}
void
ChunkedSegmentSealedImpl::ApplyFieldValidData(
milvus::OpContext* op_ctx,
FieldId field_id,
int64_t chunk_id,
int64_t offset,
int64_t size,
TargetBitmapView valid_result) const {
if (size == 0) {
return;
}
std::shared_ptr<ChunkedColumnInterface> column;
{
std::shared_lock lck(mutex_);
AssertInfo(
get_bit(field_data_ready_bitset_, field_id),
"Can't get bitset element at " + std::to_string(field_id.get()));
column = get_column(field_id);
AssertInfo(column != nullptr,
"field {} must exist when applying valid data",
field_id.get());
}
if (!column->IsNullable()) {
return;
}
auto data_type = schema_->operator[](field_id).get_data_type();
if (ChunkedColumnInterface::IsPrimitiveDataType(data_type)) {
auto pw = column->Span(op_ctx, chunk_id);
auto span = pw.get();
const bool* valid_data = span.valid_data();
if (valid_data == nullptr) {
return;
}
valid_data += offset;
for (int64_t i = 0; i < size; ++i) {
if (!valid_data[i]) {
valid_result[i] = false;
}
}
return;
}
auto row_offset = column->GetNumRowsUntilChunk(chunk_id) + offset;
std::vector<int64_t> offsets(size);
for (int64_t i = 0; i < size; ++i) {
offsets[i] = row_offset + i;
}
column->BulkIsValid(
op_ctx,
[&valid_result](bool is_valid, size_t i) {
if (!is_valid) {
valid_result[i] = false;
}
},
offsets.data(),
size);
}
void
ChunkedSegmentSealedImpl::ApplyFieldValidDataByOffsets(
milvus::OpContext* op_ctx,
FieldId field_id,
const int64_t* offsets,
int64_t count,
TargetBitmapView valid_result) const {
if (count == 0) {
return;
}
std::shared_ptr<ChunkedColumnInterface> column;
{
std::shared_lock lck(mutex_);
AssertInfo(
get_bit(field_data_ready_bitset_, field_id),
"Can't get bitset element at " + std::to_string(field_id.get()));
column = get_column(field_id);
AssertInfo(column != nullptr,
"field {} must exist when applying valid data",
field_id.get());
}
if (!column->IsNullable()) {
return;
}
column->BulkIsValid(
op_ctx,
[&valid_result](bool is_valid, size_t i) {
if (!is_valid) {
valid_result[i] = false;
}
},
offsets,
count);
}
PinWrapper<SpanBase>
ChunkedSegmentSealedImpl::chunk_data_impl(milvus::OpContext* op_ctx,
FieldId field_id,
internal/core/src/segcore/ChunkedSegmentSealedImpl.h
+15
View File
@@ -498,6 +498,21 @@ class ChunkedSegmentSealedImpl : public SegmentSealed {
FieldId field_id,
const std::vector<int64_t>& chunk_ids) const override;
void
ApplyFieldValidData(milvus::OpContext* op_ctx,
FieldId field_id,
int64_t chunk_id,
int64_t offset,
int64_t size,
TargetBitmapView valid_result) const override;
void
ApplyFieldValidDataByOffsets(milvus::OpContext* op_ctx,
FieldId field_id,
const int64_t* offsets,
int64_t count,
TargetBitmapView valid_result) const override;
protected:
// blob and row_count
PinWrapper<SpanBase>
internal/core/src/segcore/SegmentGrowingImpl.cpp
+51
View File
@@ -1207,6 +1207,57 @@ SegmentGrowingImpl::chunk_data_impl(milvus::OpContext* op_ctx,
get_insert_record().get_span_base(field_id, chunk_id));
}
void
SegmentGrowingImpl::ApplyFieldValidData(milvus::OpContext* op_ctx,
FieldId field_id,
int64_t chunk_id,
int64_t offset,
int64_t size,
TargetBitmapView valid_result) const {
(void)op_ctx;
if (size == 0) {
return;
}
auto& field_meta = schema_->operator[](field_id);
if (!field_meta.is_nullable()) {
return;
}
auto valid_vec_ptr = insert_record_.get_valid_data(field_id);
auto data = insert_record_.get_data_base(field_id);
auto row_offset = data->get_size_per_chunk() * chunk_id + offset;
auto valid_data = valid_vec_ptr->get_chunk_data(row_offset);
for (int64_t i = 0; i < size; ++i) {
if (!valid_data[i]) {
valid_result[i] = false;
}
}
}
void
SegmentGrowingImpl::ApplyFieldValidDataByOffsets(
milvus::OpContext* op_ctx,
FieldId field_id,
const int64_t* offsets,
int64_t count,
TargetBitmapView valid_result) const {
(void)op_ctx;
if (count == 0) {
return;
}
auto& field_meta = schema_->operator[](field_id);
if (!field_meta.is_nullable()) {
return;
}
auto valid_vec_ptr = insert_record_.get_valid_data(field_id);
for (int64_t i = 0; i < count; ++i) {
if (!valid_vec_ptr->is_valid(offsets[i])) {
valid_result[i] = false;
}
}
}
PinWrapper<std::pair<std::vector<std::string_view>, FixedVector<bool>>>
SegmentGrowingImpl::chunk_string_view_impl(
milvus::OpContext* op_ctx,
internal/core/src/segcore/SegmentGrowingImpl.h
+15
View File
@@ -662,6 +662,21 @@ class SegmentGrowingImpl : public SegmentGrowing {
ResourceUsage
EstimateSegmentResourceUsage() const;
void
ApplyFieldValidData(milvus::OpContext* op_ctx,
FieldId field_id,
int64_t chunk_id,
int64_t offset,
int64_t size,
TargetBitmapView valid_result) const override;
void
ApplyFieldValidDataByOffsets(milvus::OpContext* op_ctx,
FieldId field_id,
const int64_t* offsets,
int64_t count,
TargetBitmapView valid_result) const override;
protected:
int64_t
num_chunk(FieldId field_id) const override;
internal/core/src/segcore/SegmentInterface.h
+18
View File
@@ -353,6 +353,24 @@ class SegmentInternalInterface : public SegmentInterface {
// do nothing
}
// Apply field nullability to an already-initialized valid_result bitmap.
// Implementations only clear invalid rows and leave valid rows unchanged.
virtual void
ApplyFieldValidData(milvus::OpContext* op_ctx,
FieldId field_id,
int64_t chunk_id,
int64_t offset,
int64_t size,
TargetBitmapView valid_result) const = 0;
// Offsets are segment-level row offsets. valid_result must have count bits.
virtual void
ApplyFieldValidDataByOffsets(milvus::OpContext* op_ctx,
FieldId field_id,
const int64_t* offsets,
int64_t count,
TargetBitmapView valid_result) const = 0;
template <typename T>
PinWrapper<Span<T>>
chunk_data(milvus::OpContext* op_ctx,
internal/parser/planparserv2/parser_visitor.go
+43 -4
View File
@@ -1210,6 +1210,45 @@ func (v *ParserVisitor) getChildColumnInfo(identifier, child, structSubField, st
return v.getColumnInfoFromJSONIdentifier(child.GetText())
}
func (v *ParserVisitor) getStructArrayParentColumnInfo(fieldName string) (*planpb.ColumnInfo, bool, error) {
fieldName = decodeUnicode(fieldName)
if _, err := v.schema.GetFieldFromName(fieldName); err == nil {
return nil, false, nil
}
structField := v.schema.GetStructArrayFieldFromName(fieldName)
if structField == nil {
return nil, false, nil
}
subFields := structField.GetFields()
if len(subFields) == 0 {
return nil, true, merr.WrapErrParameterInvalidMsg(
"struct array field %s has no sub-fields", fieldName)
}
subField := subFields[0]
return &planpb.ColumnInfo{
FieldId: subField.GetFieldID(),
DataType: subField.GetDataType(),
ElementType: subField.GetElementType(),
Nullable: structField.GetNullable() || subField.GetNullable(),
}, true, nil
}
func (v *ParserVisitor) getNullExprColumnInfo(identifier, child antlr.TerminalNode) (*planpb.ColumnInfo, error) {
if identifier != nil {
// try struct first
if columnInfo, ok, err := v.getStructArrayParentColumnInfo(identifier.GetText()); ok || err != nil {
return columnInfo, err
}
}
return v.getChildColumnInfo(identifier, child, nil, nil)
}
func isUnsupportedNullExprVectorType(dataType schemapb.DataType) bool {
return typeutil.IsVectorType(dataType) && !typeutil.IsVectorArrayType(dataType)
}
// VisitCall parses the expr to call plan.
func (v *ParserVisitor) VisitCall(ctx *parser.CallContext) interface{} {
functionName := strings.ToLower(ctx.Identifier().GetText())
@@ -1960,12 +1999,12 @@ func (v *ParserVisitor) VisitEmptyArray(ctx *parser.EmptyArrayContext) interface
}
func (v *ParserVisitor) VisitIsNotNull(ctx *parser.IsNotNullContext) interface{} {
column, err := v.getChildColumnInfo(ctx.Identifier(), ctx.JSONIdentifier(), nil, nil)
column, err := v.getNullExprColumnInfo(ctx.Identifier(), ctx.JSONIdentifier())
if err != nil {
return err
}
if typeutil.IsVectorType(column.DataType) {
if isUnsupportedNullExprVectorType(column.DataType) {
return merr.WrapErrParameterInvalidMsg("IsNull/IsNotNull operations are not supported on vector fields")
}
@@ -2007,12 +2046,12 @@ func (v *ParserVisitor) VisitIsNotNull(ctx *parser.IsNotNullContext) interface{}
}
func (v *ParserVisitor) VisitIsNull(ctx *parser.IsNullContext) interface{} {
column, err := v.getChildColumnInfo(ctx.Identifier(), ctx.JSONIdentifier(), nil, nil)
column, err := v.getNullExprColumnInfo(ctx.Identifier(), ctx.JSONIdentifier())
if err != nil {
return err
}
if typeutil.IsVectorType(column.DataType) {
if isUnsupportedNullExprVectorType(column.DataType) {
return merr.WrapErrParameterInvalidMsg("IsNull/IsNotNull operations are not supported on vector fields")
}
internal/parser/planparserv2/plan_parser_v2_test.go
+66
View File
@@ -952,6 +952,72 @@ func TestExpr_IsNull(t *testing.T) {
}
}
func TestExpr_StructArrayParentIsNull(t *testing.T) {
schema := newTestSchema(true)
schema.StructArrayFields[0].Nullable = true
helper, err := typeutil.CreateSchemaHelper(schema)
require.NoError(t, err)
expr, err := ParseExpr(helper, `struct_array is null`, nil)
require.NoError(t, err)
nullExpr := expr.GetNullExpr()
require.NotNil(t, nullExpr)
assert.Equal(t, planpb.NullExpr_IsNull, nullExpr.GetOp())
assert.Equal(t, int64(133), nullExpr.GetColumnInfo().GetFieldId())
assert.Equal(t, schemapb.DataType_Array, nullExpr.GetColumnInfo().GetDataType())
assert.Equal(t, schemapb.DataType_VarChar, nullExpr.GetColumnInfo().GetElementType())
assert.True(t, nullExpr.GetColumnInfo().GetNullable())
expr, err = ParseExpr(helper, `struct_array is not null`, nil)
require.NoError(t, err)
nullExpr = expr.GetNullExpr()
require.NotNil(t, nullExpr)
assert.Equal(t, planpb.NullExpr_IsNotNull, nullExpr.GetOp())
assert.Equal(t, int64(133), nullExpr.GetColumnInfo().GetFieldId())
assert.Equal(t, schemapb.DataType_Array, nullExpr.GetColumnInfo().GetDataType())
assert.True(t, nullExpr.GetColumnInfo().GetNullable())
}
func TestExpr_VectorArrayOnlyStructParentIsNull(t *testing.T) {
schema := newTestSchema(false)
schema.StructArrayFields = append(schema.StructArrayFields, &schemapb.StructArrayFieldSchema{
FieldID: 10000,
Name: "vector_struct",
Nullable: true,
Fields: []*schemapb.FieldSchema{
{
FieldID: 10001,
Name: "vector_struct[embeddings]",
DataType: schemapb.DataType_ArrayOfVector,
ElementType: schemapb.DataType_FloatVector,
TypeParams: []*commonpb.KeyValuePair{
{Key: common.DimKey, Value: "4"},
},
},
},
})
helper, err := typeutil.CreateSchemaHelper(schema)
require.NoError(t, err)
for _, testcase := range []struct {
expr string
op planpb.NullExpr_NullOp
}{
{expr: `vector_struct is null`, op: planpb.NullExpr_IsNull},
{expr: `vector_struct is not null`, op: planpb.NullExpr_IsNotNull},
} {
expr, err := ParseExpr(helper, testcase.expr, nil)
require.NoError(t, err, testcase.expr)
nullExpr := expr.GetNullExpr()
require.NotNil(t, nullExpr, testcase.expr)
assert.Equal(t, testcase.op, nullExpr.GetOp(), testcase.expr)
assert.Equal(t, int64(10001), nullExpr.GetColumnInfo().GetFieldId(), testcase.expr)
assert.Equal(t, schemapb.DataType_ArrayOfVector, nullExpr.GetColumnInfo().GetDataType(), testcase.expr)
assert.Equal(t, schemapb.DataType_FloatVector, nullExpr.GetColumnInfo().GetElementType(), testcase.expr)
assert.True(t, nullExpr.GetColumnInfo().GetNullable(), testcase.expr)
}
}
func TestExpr_IsNotNull(t *testing.T) {
schema := newTestSchema(false)
helper, err := typeutil.CreateSchemaHelper(schema)