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

feat: impl StructArray -- support creating index for scalar fields (#46599)

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

This PR enable struct sub scalar fields to create nested index. The
nested means treating elements in array (we know that all fields in
array of struct are actually array) as separate documents.

The index is used by Struct to speed up Match expression as well as
Element-filter expression. This PR only enable Match expression to be
optmized by index and leave Element-filter expression in the following
PR.
Now, MatchExpr executes sub-expr by using offset inputs. To make it
support index, it also needs to support brute force without offset
inputs, so this PR:
1. enable struct scalar fields to create nested index
2. enable match expr to support brute force without offset input
3. enable match expr to support index

---------

Signed-off-by: SpadeA <tangchenjie1210@gmail.com>
Signed-off-by: SpadeA-Tang <tangchenjie1210@gmail.com>
This commit is contained in:
Spade A
2026-01-14 16:53:27 +08:00
committed by GitHub
parent eb63a363bd
commit 24c87e93c0
17 changed files with 2677 additions and 384 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/core/src/common/ArrayOffsets.cpp
+55
View File
@@ -113,6 +113,32 @@ ArrayOffsetsSealed::RowBitsetToElementOffsets(
return element_offsets;
}
FixedVector<int32_t>
ArrayOffsetsSealed::RowOffsetsToElementOffsets(
const FixedVector<int32_t>& row_offsets) const {
FixedVector<int32_t> element_offsets;
if (row_offsets.empty()) {
return element_offsets;
}
int32_t row_count = GetRowCount();
int64_t avg_elem_per_row =
(row_count > 0)
? static_cast<int64_t>(element_row_ids_.size()) / row_count
: 1;
element_offsets.reserve(row_offsets.size() * avg_elem_per_row);
for (auto row_id : row_offsets) {
assert(row_id >= 0 && row_id < row_count);
int32_t first_elem = row_to_element_start_[row_id];
int32_t last_elem = row_to_element_start_[row_id + 1];
for (int32_t elem_id = first_elem; elem_id < last_elem; ++elem_id) {
element_offsets.push_back(elem_id);
}
}
return element_offsets;
}
std::shared_ptr<ArrayOffsetsSealed>
ArrayOffsetsSealed::BuildFromSegment(const void* segment,
const FieldMeta& field_meta) {
@@ -308,6 +334,35 @@ ArrayOffsetsGrowing::RowBitsetToElementOffsets(
return element_offsets;
}
FixedVector<int32_t>
ArrayOffsetsGrowing::RowOffsetsToElementOffsets(
const FixedVector<int32_t>& row_offsets) const {
std::shared_lock lock(mutex_);
FixedVector<int32_t> element_offsets;
if (row_offsets.empty()) {
return element_offsets;
}
int64_t avg_elem_per_row =
(committed_row_count_ > 0)
? static_cast<int64_t>(element_row_ids_.size()) /
committed_row_count_
: 1;
element_offsets.reserve(row_offsets.size() * avg_elem_per_row);
for (auto row_id : row_offsets) {
assert(row_id >= 0 && row_id < committed_row_count_);
int32_t first_elem = row_to_element_start_[row_id];
int32_t last_elem = row_to_element_start_[row_id + 1];
for (int32_t elem_id = first_elem; elem_id < last_elem; ++elem_id) {
element_offsets.push_back(elem_id);
}
}
return element_offsets;
}
void
ArrayOffsetsGrowing::Insert(int64_t row_id_start,
const int32_t* array_lengths,
internal/core/src/common/ArrayOffsets.h
+14
View File
@@ -62,6 +62,12 @@ class IArrayOffsets {
virtual FixedVector<int32_t>
RowBitsetToElementOffsets(const TargetBitmapView& row_bitset,
int64_t row_start) const = 0;
// Convert row offsets to element offsets
// Returns element IDs for all specified rows
virtual FixedVector<int32_t>
RowOffsetsToElementOffsets(
const FixedVector<int32_t>& row_offsets) const = 0;
};
class ArrayOffsetsSealed : public IArrayOffsets {
@@ -109,6 +115,10 @@ class ArrayOffsetsSealed : public IArrayOffsets {
RowBitsetToElementOffsets(const TargetBitmapView& row_bitset,
int64_t row_start) const override;
FixedVector<int32_t>
RowOffsetsToElementOffsets(
const FixedVector<int32_t>& row_offsets) const override;
static std::shared_ptr<ArrayOffsetsSealed>
BuildFromSegment(const void* segment, const FieldMeta& field_meta);
@@ -152,6 +162,10 @@ class ArrayOffsetsGrowing : public IArrayOffsets {
RowBitsetToElementOffsets(const TargetBitmapView& row_bitset,
int64_t row_start) const override;
FixedVector<int32_t>
RowOffsetsToElementOffsets(
const FixedVector<int32_t>& row_offsets) const override;
private:
struct PendingRow {
int64_t row_id;
internal/core/src/common/Utils.h
+5
View File
@@ -572,4 +572,9 @@ Align(int32_t number, int32_t alignment) {
return (number + alignment - 1) & ~(alignment - 1);
}
inline bool
IsStructSubField(const std::string& fieldName) {
return fieldName.find('[') != std::string::npos;
}
} // namespace milvus
internal/core/src/exec/expression/BinaryArithOpEvalRangeExpr.cpp
+21 -11
View File
@@ -917,7 +917,7 @@ PhyBinaryArithOpEvalRangeExpr::ExecRangeVisitorImplForIndex(
T>
HighPrecisionType;
auto real_batch_size =
has_offset_input_ ? input->size() : GetNextBatchSize();
GetNextRealBatchSize(input, expr_->column_.element_level_);
if (real_batch_size == 0) {
return nullptr;
}
@@ -1437,8 +1437,9 @@ PhyBinaryArithOpEvalRangeExpr::ExecRangeVisitorImplForData(
int64_t,
T>
HighPrecisionType;
auto real_batch_size =
has_offset_input_ ? input->size() : GetNextBatchSize();
GetNextRealBatchSize(input, expr_->column_.element_level_);
if (real_batch_size == 0) {
return nullptr;
}
@@ -1845,7 +1846,7 @@ PhyBinaryArithOpEvalRangeExpr::ExecRangeVisitorImplForData(
int64_t processed_size;
if (has_offset_input_) {
if (expr_->column_.element_level_) {
// For element-level filtering
// For element-level filtering with offset input
processed_size = ProcessElementLevelByOffsets<T>(execute_sub_batch,
skip_index_func,
input,
@@ -1863,14 +1864,23 @@ PhyBinaryArithOpEvalRangeExpr::ExecRangeVisitorImplForData(
right_operand);
}
} else {
AssertInfo(!expr_->column_.element_level_,
"Element-level filtering is not supported without offsets");
processed_size = ProcessDataChunks<T>(execute_sub_batch,
skip_index_func,
res,
valid_res,
value,
right_operand);
if (expr_->column_.element_level_) {
// For element-level filtering without offset input (brute force)
processed_size =
ProcessDataChunksForElementLevel<T>(execute_sub_batch,
skip_index_func,
res,
valid_res,
value,
right_operand);
} else {
processed_size = ProcessDataChunks<T>(execute_sub_batch,
skip_index_func,
res,
valid_res,
value,
right_operand);
}
}
AssertInfo(processed_size == real_batch_size,
"internal error: expr processed rows {} not equal "
internal/core/src/exec/expression/BinaryRangeExpr.cpp
+12 -7
View File
@@ -281,7 +281,8 @@ PhyBinaryRangeFilterExpr::ExecRangeVisitorImplForIndex() {
return res;
}
auto real_batch_size = GetNextBatchSize();
auto real_batch_size =
GetNextRealBatchSize(nullptr, expr_->column_.element_level_);
if (real_batch_size == 0) {
return nullptr;
}
@@ -327,7 +328,7 @@ PhyBinaryRangeFilterExpr::ExecRangeVisitorImplForData(EvalCtx& context) {
}
auto real_batch_size =
has_offset_input_ ? input->size() : GetNextBatchSize();
GetNextRealBatchSize(input, expr_->column_.element_level_);
if (real_batch_size == 0) {
return nullptr;
}
@@ -433,7 +434,7 @@ PhyBinaryRangeFilterExpr::ExecRangeVisitorImplForData(EvalCtx& context) {
int64_t processed_size;
if (has_offset_input_) {
if (expr_->column_.element_level_) {
// For element-level filtering
// For element-level filtering with offset input
processed_size = ProcessElementLevelByOffsets<T>(execute_sub_batch,
skip_index_func,
input,
@@ -452,10 +453,14 @@ PhyBinaryRangeFilterExpr::ExecRangeVisitorImplForData(EvalCtx& context) {
val2);
}
} else {
AssertInfo(!expr_->column_.element_level_,
"Element-level filtering is not supported without offsets");
processed_size = ProcessDataChunks<T>(
execute_sub_batch, skip_index_func, res, valid_res, val1, val2);
if (expr_->column_.element_level_) {
// For element-level filtering without offset input (brute force)
processed_size = ProcessDataChunksForElementLevel<T>(
execute_sub_batch, skip_index_func, res, valid_res, val1, val2);
} else {
processed_size = ProcessDataChunks<T>(
execute_sub_batch, skip_index_func, res, valid_res, val1, val2);
}
}
AssertInfo(processed_size == real_batch_size,
"internal error: expr processed rows {} not equal "
internal/core/src/exec/expression/Expr.h
+443 -157
View File
@@ -351,6 +351,63 @@ class SegmentExpr : public Expr {
: batch_size_;
}
int64_t
GetNextRealBatchSize(const OffsetVector* input, bool element_level) {
if (input != nullptr) {
return input->size();
} else if (element_level) {
auto [_, elem_count] = GetNextBatchSizeForElementLevel();
return elem_count;
}
return GetNextBatchSize();
}
// Get the next batch size for element-level processing
// Returns: (batch_rows, elem_count) where batch_rows is number of rows to process
// and elem_count is the total number of elements in those rows
std::pair<int64_t, int64_t>
GetNextBatchSizeForElementLevel() {
auto array_offsets = segment_->GetArrayOffsets(field_id_);
AssertInfo(array_offsets != nullptr,
"ArrayOffsets not found for field {}",
field_id_.get());
// Use index path or data path based on whether index is being used
auto current_chunk = SegmentExpr::CanUseIndex() && use_index_
? current_index_chunk_
: current_data_chunk_;
auto current_chunk_pos = SegmentExpr::CanUseIndex() && use_index_
? current_index_chunk_pos_
: current_data_chunk_pos_;
int64_t current_rows = 0;
if (SegmentExpr::CanUseIndex() && use_index_ &&
segment_->type() == SegmentType::Sealed) {
// For sealed segment with index, position is already global
current_rows = current_chunk_pos;
} else if (segment_->is_chunked()) {
current_rows =
segment_->num_rows_until_chunk(field_id_, current_chunk) +
current_chunk_pos;
} else {
current_rows = current_chunk * size_per_chunk_ + current_chunk_pos;
}
auto batch_rows = std::min(batch_size_, active_count_ - current_rows);
if (batch_rows == 0) {
return {0, 0};
}
// Calculate elem_count based on global row positions
auto [elem_start, _] = array_offsets->ElementIDRangeOfRow(current_rows);
auto [elem_end, __] =
array_offsets->ElementIDRangeOfRow(current_rows + batch_rows);
auto elem_count = elem_end - elem_start;
return {batch_rows, elem_count};
}
// used for processing raw data expr for sealed segments.
// now only used for std::string_view && json
// TODO: support more types
@@ -848,6 +905,254 @@ class SegmentExpr : public Expr {
}
}
// Process element-level data without offset input
// This is the counterpart of ProcessDataChunks for element-level expressions
// Iterates over rows in batch, but returns element-level results
// The caller must pre-allocate res/valid_res with elem_count size (from GetNextBatchSizeForElementLevel)
template <typename ElementType, typename FUNC, typename... ValTypes>
int64_t
ProcessDataChunksForElementLevel(
FUNC func,
std::function<bool(const milvus::SkipIndex&, FieldId, int)> skip_func,
TargetBitmapView res,
TargetBitmapView valid_res,
const ValTypes&... values) {
static_assert(!std::is_same_v<ElementType, Json>,
"Json element type is not supported for "
"element-level filtering");
int64_t processed_rows = 0;
int64_t processed_elems = 0;
// Prefetch chunks to reduce cache miss latency
if (!prefetched_) {
std::vector<int64_t> pf_chunk_ids;
pf_chunk_ids.reserve(num_data_chunk_ - current_data_chunk_);
for (size_t i = current_data_chunk_; i < num_data_chunk_; i++) {
pf_chunk_ids.push_back(i);
}
segment_->prefetch_chunks(op_ctx_, field_id_, pf_chunk_ids);
prefetched_ = true;
}
for (size_t i = current_data_chunk_; i < num_data_chunk_; i++) {
auto data_pos =
i == current_data_chunk_ ? current_data_chunk_pos_ : 0;
int64_t size = segment_->chunk_size(field_id_, i) - data_pos;
size = std::min(size, batch_size_ - processed_rows);
if (size == 0)
continue;
auto& skip_index = segment_->GetSkipIndex();
if ((!skip_func || !skip_func(skip_index, field_id_, i)) &&
(!namespace_skip_func_.has_value() ||
!namespace_skip_func_.value()(i))) {
if (segment_->type() == SegmentType::Sealed) {
auto pw = segment_->get_batch_views<ArrayView>(
op_ctx_, field_id_, i, data_pos, size);
auto [data_vec, valid_data] = pw.get();
for (size_t j = 0; j < static_cast<size_t>(size); j++) {
auto elem_count = data_vec[j].length();
bool is_row_valid = !valid_data.data() || valid_data[j];
if (!is_row_valid) {
// Row is invalid, mark all elements as false
for (size_t k = 0; k < elem_count; k++) {
res[processed_elems + k] =
valid_res[processed_elems + k] = false;
}
} else {
// Row is valid, process array elements
if constexpr (std::is_same_v<ElementType,
std::string_view> ||
std::is_same_v<ElementType,
std::string>) {
// String type: extract one by one
for (size_t k = 0; k < elem_count; k++) {
auto str_view =
data_vec[j]
.template get_data<
std::string_view>(k);
ElementType str_val(str_view);
func(&str_val,
nullptr,
nullptr,
1,
res + processed_elems + k,
valid_res + processed_elems + k,
values...);
}
} else {
// Fixed-length numeric types
// Note: int8_t/int16_t are stored as int32_t in Array
using StorageType = std::conditional_t<
std::is_same_v<ElementType, int8_t> ||
std::is_same_v<ElementType, int16_t>,
int32_t,
ElementType>;
auto* raw_data =
reinterpret_cast<const StorageType*>(
data_vec[j].data());
if constexpr (std::is_same_v<StorageType,
ElementType>) {
// Types match, batch process
func(raw_data,
nullptr,
nullptr,
elem_count,
res + processed_elems,
valid_res + processed_elems,
values...);
} else {
// int8_t/int16_t: need conversion
for (size_t k = 0; k < elem_count; k++) {
ElementType val =
static_cast<ElementType>(
raw_data[k]);
func(&val,
nullptr,
nullptr,
1,
res + processed_elems + k,
valid_res + processed_elems + k,
values...);
}
}
}
}
processed_elems += elem_count;
}
} else {
// Growing segment: use Array
auto pw =
segment_->chunk_data<Array>(op_ctx_, field_id_, i);
auto chunk = pw.get();
const Array* data = chunk.data() + data_pos;
const bool* valid_data = chunk.valid_data();
if (valid_data != nullptr) {
valid_data += data_pos;
}
for (size_t j = 0; j < static_cast<size_t>(size); j++) {
auto elem_count = data[j].length();
bool is_row_valid = !valid_data || valid_data[j];
if (!is_row_valid) {
// Row is invalid, mark all elements as false
for (size_t k = 0; k < elem_count; k++) {
res[processed_elems + k] =
valid_res[processed_elems + k] = false;
}
} else {
// Row is valid, process array elements
if constexpr (std::is_same_v<ElementType,
std::string_view> ||
std::is_same_v<ElementType,
std::string>) {
// String type: extract one by one
for (size_t k = 0; k < elem_count; k++) {
auto str_view =
data[j]
.template get_data<
std::string_view>(k);
ElementType str_val(str_view);
func(&str_val,
nullptr,
nullptr,
1,
res + processed_elems + k,
valid_res + processed_elems + k,
values...);
}
} else {
// Fixed-length numeric types
// Note: int8_t/int16_t are stored as int32_t in Array
using StorageType = std::conditional_t<
std::is_same_v<ElementType, int8_t> ||
std::is_same_v<ElementType, int16_t>,
int32_t,
ElementType>;
auto* raw_data =
reinterpret_cast<const StorageType*>(
data[j].data());
if constexpr (std::is_same_v<StorageType,
ElementType>) {
// Types match, batch process
func(raw_data,
nullptr,
nullptr,
elem_count,
res + processed_elems,
valid_res + processed_elems,
values...);
} else {
// int8_t/int16_t: need conversion
for (size_t k = 0; k < elem_count; k++) {
ElementType val =
static_cast<ElementType>(
raw_data[k]);
func(&val,
nullptr,
nullptr,
1,
res + processed_elems + k,
valid_res + processed_elems + k,
values...);
}
}
}
}
processed_elems += elem_count;
}
}
} else {
// Chunk is skipped, mark all elements as false
if (segment_->type() == SegmentType::Sealed) {
auto pw = segment_->get_batch_views<ArrayView>(
op_ctx_, field_id_, i, data_pos, size);
auto [data_vec, valid_data] = pw.get();
for (size_t j = 0; j < static_cast<size_t>(size); j++) {
auto elem_count = data_vec[j].length();
for (size_t k = 0; k < elem_count; k++) {
res[processed_elems + k] =
valid_res[processed_elems + k] = false;
}
processed_elems += elem_count;
}
} else {
auto pw =
segment_->chunk_data<Array>(op_ctx_, field_id_, i);
auto chunk = pw.get();
const Array* data = chunk.data() + data_pos;
for (size_t j = 0; j < static_cast<size_t>(size); j++) {
auto elem_count = data[j].length();
for (size_t k = 0; k < elem_count; k++) {
res[processed_elems + k] =
valid_res[processed_elems + k] = false;
}
processed_elems += elem_count;
}
}
}
processed_rows += size;
if (processed_rows >= batch_size_) {
current_data_chunk_ = i;
current_data_chunk_pos_ = data_pos + size;
break;
}
}
return processed_elems;
}
// Template parameter to control whether segment offsets are needed (for GIS functions)
template <typename T,
bool NeedSegmentOffsets = false,
@@ -1198,68 +1503,6 @@ class SegmentExpr : public Expr {
return processed_size;
}
int
ProcessIndexOneChunk(TargetBitmap& result,
TargetBitmap& valid_result,
size_t chunk_id,
const TargetBitmap& chunk_res,
const TargetBitmap& chunk_valid_res,
int processed_rows) {
auto data_pos =
chunk_id == current_index_chunk_ ? current_index_chunk_pos_ : 0;
auto size = std::min(
std::min(size_per_chunk_ - data_pos, batch_size_ - processed_rows),
int64_t(chunk_res.size()));
// result.insert(result.end(),
// chunk_res.begin() + data_pos,
// chunk_res.begin() + data_pos + size);
result.append(chunk_res, data_pos, size);
valid_result.append(chunk_valid_res, data_pos, size);
return size;
}
// Helper function to get index pointer for JSON and non-JSON types
template <typename IndexInnerType>
struct IndexPtrResult {
index::ScalarIndex<IndexInnerType>* index_ptr;
std::shared_ptr<index::JsonFlatIndexQueryExecutor<IndexInnerType>>
executor;
};
template <typename IndexInnerType>
IndexPtrResult<IndexInnerType>
GetIndexPtrForChunk(size_t chunk_id) {
using Index = index::ScalarIndex<IndexInnerType>;
IndexPtrResult<IndexInnerType> result{nullptr, nullptr};
if (field_type_ == DataType::JSON) {
auto pointer = milvus::Json::pointer(nested_path_);
PinWrapper<const index::IndexBase*> json_pw =
pinned_index_[chunk_id];
// check if it is a json flat index, if so, create a json flat index query executor
auto json_flat_index =
dynamic_cast<const index::JsonFlatIndex*>(json_pw.get());
if (json_flat_index) {
auto index_path = json_flat_index->GetNestedPath();
result.executor =
json_flat_index->template create_executor<IndexInnerType>(
pointer.substr(index_path.size()));
result.index_ptr = result.executor.get();
} else {
auto json_index = const_cast<index::IndexBase*>(json_pw.get());
result.index_ptr = dynamic_cast<Index*>(json_index);
}
} else {
auto scalar_index =
dynamic_cast<const Index*>(pinned_index_[chunk_id].get());
result.index_ptr = const_cast<Index*>(scalar_index);
}
return result;
}
template <typename T, typename FUNC, typename... ValTypes>
VectorPtr
ProcessIndexChunks(FUNC func, const ValTypes&... values) {
@@ -1267,41 +1510,82 @@ class SegmentExpr : public Expr {
conditional_t<std::is_same_v<T, std::string_view>, std::string, T>
IndexInnerType;
using Index = index::ScalarIndex<IndexInnerType>;
AssertInfo(num_index_chunk_ == 1,
"scalar index should have exactly 1 chunk, got {}",
num_index_chunk_);
// Cache index result (execute only once)
if (cached_index_chunk_id_ != 0) {
Index* index_ptr = nullptr;
PinWrapper<const index::IndexBase*> json_pw;
// Executor for JsonFlatIndex. Must outlive index_ptr. Only used for JSON type.
std::shared_ptr<index::JsonFlatIndexQueryExecutor<IndexInnerType>>
executor;
if (field_type_ == DataType::JSON) {
auto pointer = milvus::Json::pointer(nested_path_);
json_pw = pinned_index_[0];
auto json_flat_index =
dynamic_cast<const index::JsonFlatIndex*>(json_pw.get());
if (json_flat_index) {
auto index_path = json_flat_index->GetNestedPath();
executor = json_flat_index
->template create_executor<IndexInnerType>(
pointer.substr(index_path.size()));
index_ptr = executor.get();
} else {
auto json_index =
const_cast<index::IndexBase*>(json_pw.get());
index_ptr = dynamic_cast<Index*>(json_index);
}
} else {
auto scalar_index =
dynamic_cast<const Index*>(pinned_index_[0].get());
index_ptr = const_cast<Index*>(scalar_index);
}
cached_index_chunk_res_ = std::make_shared<TargetBitmap>(
std::move(func(index_ptr, values...)));
cached_index_chunk_valid_res_ =
std::make_shared<TargetBitmap>(index_ptr->IsNotNull());
cached_index_chunk_id_ = 0;
cached_is_nested_index_ = index_ptr->IsNestedIndex();
}
TargetBitmap result;
TargetBitmap valid_result;
int processed_rows = 0;
for (size_t i = current_index_chunk_; i < num_index_chunk_; i++) {
// This cache result help getting result for every batch loop.
// It avoids indexing execute for every batch because indexing
// executing costs quite much time.
if (cached_index_chunk_id_ != i) {
auto index_result = GetIndexPtrForChunk<IndexInnerType>(i);
Index* index_ptr = index_result.index_ptr;
if (cached_is_nested_index_) {
// Nested index: batch by rows, return corresponding elements
auto array_offsets = segment_->GetArrayOffsets(field_id_);
cached_index_chunk_res_ = std::make_shared<TargetBitmap>(
std::move(func(index_ptr, values...)));
auto valid_result = index_ptr->IsNotNull();
cached_index_chunk_valid_res_ =
std::make_shared<TargetBitmap>(std::move(valid_result));
cached_index_chunk_id_ = i;
}
auto data_pos = current_index_chunk_pos_;
auto batch_rows = std::min(batch_size_, active_count_ - data_pos);
auto size = ProcessIndexOneChunk(result,
valid_result,
i,
*cached_index_chunk_res_,
*cached_index_chunk_valid_res_,
processed_rows);
// Calculate corresponding element range
auto [elem_start, _] = array_offsets->ElementIDRangeOfRow(data_pos);
auto [elem_end, __] =
array_offsets->ElementIDRangeOfRow(data_pos + batch_rows);
auto elem_count = elem_end - elem_start;
if (processed_rows + size >= batch_size_) {
current_index_chunk_ = i;
current_index_chunk_pos_ = i == current_index_chunk_
? current_index_chunk_pos_ + size
: size;
break;
}
processed_rows += size;
result.append(*cached_index_chunk_res_, elem_start, elem_count);
valid_result.append(
*cached_index_chunk_valid_res_, elem_start, elem_count);
current_index_chunk_pos_ = data_pos + batch_rows;
} else {
// Normal index: batch by rows
auto data_pos = current_index_chunk_pos_;
auto size =
std::min(std::min(size_per_chunk_ - data_pos, batch_size_),
int64_t(cached_index_chunk_res_->size()));
result.append(*cached_index_chunk_res_, data_pos, size);
valid_result.append(*cached_index_chunk_valid_res_, data_pos, size);
current_index_chunk_pos_ = data_pos + size;
}
return std::make_shared<ColumnVector>(std::move(result),
@@ -1515,25 +1799,6 @@ class SegmentExpr : public Expr {
return valid_result;
}
int
ProcessIndexOneChunkForValid(TargetBitmap& valid_result,
size_t chunk_id,
const TargetBitmap& chunk_valid_res,
int processed_rows) {
auto data_pos =
chunk_id == current_index_chunk_ ? current_index_chunk_pos_ : 0;
auto size = std::min(
std::min(size_per_chunk_ - data_pos, batch_size_ - processed_rows),
int64_t(chunk_valid_res.size()));
if (field_type_ == DataType::GEOMETRY &&
segment_->type() == SegmentType::Growing) {
size = std::min(batch_size_ - processed_rows,
int64_t(chunk_valid_res.size()) - data_pos);
}
valid_result.append(chunk_valid_res, data_pos, size);
return size;
}
template <typename T>
TargetBitmap
ProcessIndexChunksForValid() {
@@ -1543,42 +1808,59 @@ class SegmentExpr : public Expr {
std::string,
T>;
using Index = index::ScalarIndex<IndexInnerType>;
int processed_rows = 0;
AssertInfo(num_index_chunk_ == 1,
"scalar index should have exactly 1 chunk, got {}",
num_index_chunk_);
// Cache valid result (execute only once)
if (cached_index_chunk_id_ != 0) {
Index* index_ptr = nullptr;
PinWrapper<const index::IndexBase*> json_pw;
// Executor for JsonFlatIndex. Must outlive index_ptr. Only used for JSON type.
std::shared_ptr<index::JsonFlatIndexQueryExecutor<IndexInnerType>>
executor;
if (field_type_ == DataType::JSON) {
auto pointer = milvus::Json::pointer(nested_path_);
json_pw = pinned_index_[0];
auto json_flat_index =
dynamic_cast<const index::JsonFlatIndex*>(json_pw.get());
if (json_flat_index) {
auto index_path = json_flat_index->GetNestedPath();
executor = json_flat_index
->template create_executor<IndexInnerType>(
pointer.substr(index_path.size()));
index_ptr = executor.get();
} else {
auto json_index =
const_cast<index::IndexBase*>(json_pw.get());
index_ptr = dynamic_cast<Index*>(json_index);
}
} else {
auto scalar_index =
dynamic_cast<const Index*>(pinned_index_[0].get());
index_ptr = const_cast<Index*>(scalar_index);
}
cached_index_chunk_valid_res_ =
std::make_shared<TargetBitmap>(index_ptr->IsNotNull());
cached_index_chunk_id_ = 0;
}
// Process current batch
TargetBitmap valid_result;
valid_result.set();
for (size_t i = current_index_chunk_; i < num_index_chunk_; i++) {
// This cache result help getting result for every batch loop.
// It avoids indexing execute for every batch because indexing
// executing costs quite much time.
if (cached_index_chunk_id_ != i) {
auto index_result = GetIndexPtrForChunk<IndexInnerType>(i);
Index* index_ptr = index_result.index_ptr;
auto data_pos = current_index_chunk_pos_;
auto size = std::min(std::min(size_per_chunk_ - data_pos, batch_size_),
int64_t(cached_index_chunk_valid_res_->size()));
auto execute_sub_batch = [](Index* index_ptr) {
TargetBitmap res = index_ptr->IsNotNull();
return res;
};
cached_index_chunk_valid_res_ = std::make_shared<TargetBitmap>(
std::move(execute_sub_batch(index_ptr)));
cached_index_chunk_id_ = i;
}
valid_result.append(*cached_index_chunk_valid_res_, data_pos, size);
auto size =
ProcessIndexOneChunkForValid(valid_result,
i,
*cached_index_chunk_valid_res_,
processed_rows);
current_index_chunk_pos_ = data_pos + size;
if (processed_rows + size >= batch_size_) {
current_index_chunk_ = i;
current_index_chunk_pos_ = i == current_index_chunk_
? current_index_chunk_pos_ + size
: size;
break;
}
processed_rows += size;
}
return valid_result;
}
@@ -1590,12 +1872,14 @@ class SegmentExpr : public Expr {
IndexInnerType;
using Index = index::ScalarIndex<IndexInnerType>;
for (size_t i = current_index_chunk_; i < num_index_chunk_; i++) {
auto scalar_index =
dynamic_cast<const Index*>(pinned_index_[i].get());
auto* index_ptr = const_cast<Index*>(scalar_index);
func(index_ptr, values...);
}
// For scalar index, num_index_chunk_ can only be 1
AssertInfo(num_index_chunk_ == 1,
"scalar index should have exactly 1 chunk, got {}",
num_index_chunk_);
auto scalar_index = dynamic_cast<const Index*>(pinned_index_[0].get());
auto* index_ptr = const_cast<Index*>(scalar_index);
func(index_ptr, values...);
}
bool
@@ -1670,8 +1954,11 @@ class SegmentExpr : public Expr {
using Index = index::ScalarIndex<IndexInnerType>;
if (op == OpType::Match || op == OpType::InnerMatch ||
op == OpType::PostfixMatch) {
auto scalar_index = dynamic_cast<const Index*>(
pinned_index_[current_index_chunk_].get());
AssertInfo(num_index_chunk_ == 1,
"scalar index should have exactly 1 chunk, got {}",
num_index_chunk_);
auto scalar_index =
dynamic_cast<const Index*>(pinned_index_[0].get());
auto* index_ptr = const_cast<Index*>(scalar_index);
// 1, index support regex query and try use it, then index handles the query;
// 2, index has raw data, then call index.Reverse_Lookup to handle the query;
@@ -1690,16 +1977,13 @@ class SegmentExpr : public Expr {
IndexInnerType;
using Index = index::ScalarIndex<IndexInnerType>;
for (size_t i = current_index_chunk_; i < num_index_chunk_; i++) {
auto scalar_index =
dynamic_cast<const Index*>(pinned_index_[i].get());
auto* index_ptr = const_cast<Index*>(scalar_index);
if (!index_ptr->HasRawData()) {
return false;
}
}
return true;
AssertInfo(num_index_chunk_ == 1,
"scalar index should have exactly 1 chunk, got {}",
num_index_chunk_);
auto scalar_index = dynamic_cast<const Index*>(pinned_index_[0].get());
auto* index_ptr = const_cast<Index*>(scalar_index);
return index_ptr->HasRawData();
}
void
@@ -1786,6 +2070,8 @@ class SegmentExpr : public Expr {
std::shared_ptr<TargetBitmap> cached_index_chunk_res_{nullptr};
// Cache for chunk valid res.
std::shared_ptr<TargetBitmap> cached_index_chunk_valid_res_{nullptr};
// Cache whether index is nested index
bool cached_is_nested_index_{false};
// Cache for text match.
std::shared_ptr<TargetBitmap> cached_match_res_{nullptr};
internal/core/src/exec/expression/GISFunctionFilterExpr.cpp
+8 -7
View File
@@ -551,14 +551,15 @@ PhyGISFunctionFilterExpr::EvalForIndexSegment() {
}
if (segment_->type() == SegmentType::Sealed) {
auto size = ProcessIndexOneChunk(batch_result,
batch_valid,
0,
*cached_index_chunk_res_,
coarse_valid_global_,
processed_rows);
auto data_pos = current_index_chunk_pos_;
auto size = std::min(
std::min(size_per_chunk_ - data_pos, batch_size_ - processed_rows),
int64_t(cached_index_chunk_res_->size()));
batch_result.append(*cached_index_chunk_res_, data_pos, size);
batch_valid.append(coarse_valid_global_, data_pos, size);
processed_rows += size;
current_index_chunk_pos_ = current_index_chunk_pos_ + size;
current_index_chunk_pos_ += size;
} else {
for (size_t i = current_data_chunk_; i < num_data_chunk_; i++) {
auto data_pos =
internal/core/src/exec/expression/MatchExpr.cpp
+262 -180
View File
@@ -15,7 +15,6 @@
// limitations under the License.
#include "MatchExpr.h"
#include <numeric>
#include <utility>
#include "common/Tracer.h"
#include "common/Types.h"
@@ -25,129 +24,169 @@ namespace exec {
using MatchType = milvus::expr::MatchType;
// Core matching logic for a single row's elements
// Returns true if the row matches the condition
template <MatchType match_type, bool all_valid>
bool
MatchSingleRow(int64_t bitset_start,
int64_t row_elem_count,
const TargetBitmapView& match_bitset,
const TargetBitmapView& valid_bitset,
int64_t threshold) {
int64_t hit_count = 0;
int64_t element_count = row_elem_count;
if constexpr (all_valid) {
for (int64_t j = 0; j < row_elem_count; ++j) {
bool matched = match_bitset[bitset_start + j];
if (matched) {
++hit_count;
}
// Early exit conditions
if constexpr (match_type == MatchType::MatchAny) {
if (hit_count > 0)
return true;
} else if constexpr (match_type == MatchType::MatchAll) {
if (!matched)
return false;
} else if constexpr (match_type == MatchType::MatchLeast) {
if (hit_count >= threshold)
return true;
} else if constexpr (match_type == MatchType::MatchMost ||
match_type == MatchType::MatchExact) {
if (hit_count > threshold)
return false;
}
}
} else {
element_count = 0;
for (int64_t j = 0; j < row_elem_count; ++j) {
if (!valid_bitset[bitset_start + j]) {
continue;
}
++element_count;
bool matched = match_bitset[bitset_start + j];
if (matched) {
++hit_count;
}
// Early exit conditions
if constexpr (match_type == MatchType::MatchAny) {
if (hit_count > 0)
return true;
} else if constexpr (match_type == MatchType::MatchAll) {
if (!matched)
return false;
} else if constexpr (match_type == MatchType::MatchLeast) {
if (hit_count >= threshold)
return true;
} else if constexpr (match_type == MatchType::MatchMost ||
match_type == MatchType::MatchExact) {
if (hit_count > threshold)
return false;
}
}
}
// Final match decision
if constexpr (match_type == MatchType::MatchAny) {
return hit_count > 0;
} else if constexpr (match_type == MatchType::MatchAll) {
// Empty array returns true (vacuous truth)
return hit_count == element_count;
} else if constexpr (match_type == MatchType::MatchLeast) {
return hit_count >= threshold;
} else if constexpr (match_type == MatchType::MatchMost) {
return hit_count <= threshold;
} else if constexpr (match_type == MatchType::MatchExact) {
return hit_count == threshold;
}
return false;
}
// Process contiguous rows [row_start, row_start + row_count)
template <MatchType match_type, bool all_valid>
void
ProcessMatchRows(int64_t row_count,
const IArrayOffsets* array_offsets,
const TargetBitmapView& match_bitset,
const TargetBitmapView& valid_bitset,
TargetBitmapView& result_bitset,
int64_t threshold) {
ProcessContiguousRows(int64_t row_count,
int64_t row_start,
int64_t elem_start,
const IArrayOffsets* array_offsets,
const TargetBitmapView& match_bitset,
const TargetBitmapView& valid_bitset,
TargetBitmapView& result_bitset,
int64_t threshold) {
for (int64_t i = 0; i < row_count; ++i) {
auto [first_elem, last_elem] = array_offsets->ElementIDRangeOfRow(i);
int64_t hit_count = 0;
int64_t element_count = last_elem - first_elem;
bool early_fail = false;
auto [first_elem, last_elem] =
array_offsets->ElementIDRangeOfRow(row_start + i);
int64_t bitset_start = first_elem - elem_start;
int64_t row_elem_count = last_elem - first_elem;
if constexpr (all_valid) {
for (auto j = first_elem; j < last_elem; ++j) {
bool matched = match_bitset[j];
if (matched) {
++hit_count;
}
if constexpr (match_type == MatchType::MatchAny) {
if (hit_count > 0) {
break;
}
} else if constexpr (match_type == MatchType::MatchAll) {
if (!matched) {
early_fail = true;
break;
}
} else if constexpr (match_type == MatchType::MatchLeast) {
if (hit_count >= threshold) {
break;
}
} else if constexpr (match_type == MatchType::MatchMost ||
match_type == MatchType::MatchExact) {
if (hit_count > threshold) {
early_fail = true;
break;
}
}
}
} else {
element_count = 0;
for (auto j = first_elem; j < last_elem; ++j) {
if (!valid_bitset[j]) {
continue;
}
++element_count;
bool matched = match_bitset[j];
if (matched) {
++hit_count;
}
if constexpr (match_type == MatchType::MatchAny) {
if (hit_count > 0) {
break;
}
} else if constexpr (match_type == MatchType::MatchAll) {
if (!matched) {
early_fail = true;
break;
}
} else if constexpr (match_type == MatchType::MatchLeast) {
if (hit_count >= threshold) {
break;
}
} else if constexpr (match_type == MatchType::MatchMost ||
match_type == MatchType::MatchExact) {
if (hit_count > threshold) {
early_fail = true;
break;
}
}
}
}
if (early_fail) {
continue;
}
bool is_match = false;
if constexpr (match_type == MatchType::MatchAny) {
is_match = hit_count > 0;
} else if constexpr (match_type == MatchType::MatchAll) {
is_match = hit_count == element_count;
} else if constexpr (match_type == MatchType::MatchLeast) {
is_match = hit_count >= threshold;
} else if constexpr (match_type == MatchType::MatchMost) {
is_match = hit_count <= threshold;
} else if constexpr (match_type == MatchType::MatchExact) {
is_match = hit_count == threshold;
}
if (is_match) {
bool matched = MatchSingleRow<match_type, all_valid>(bitset_start,
row_elem_count,
match_bitset,
valid_bitset,
threshold);
if (matched) {
result_bitset[i] = true;
}
}
}
template <MatchType match_type>
// Process non-contiguous rows specified by offset_input
template <MatchType match_type, bool all_valid>
void
DispatchByValidity(bool all_valid,
int64_t row_count,
const IArrayOffsets* array_offsets,
const TargetBitmapView& match_bitset,
const TargetBitmapView& valid_bitset,
TargetBitmapView& result_bitset,
int64_t threshold) {
if (all_valid) {
ProcessMatchRows<match_type, true>(row_count,
array_offsets,
match_bitset,
valid_bitset,
result_bitset,
threshold);
ProcessOffsetRows(const OffsetVector* row_offsets,
const IArrayOffsets* array_offsets,
const TargetBitmapView& match_bitset,
const TargetBitmapView& valid_bitset,
TargetBitmapView& result_bitset,
int64_t threshold) {
int64_t elem_cursor = 0;
for (size_t i = 0; i < row_offsets->size(); ++i) {
auto [first_elem, last_elem] =
array_offsets->ElementIDRangeOfRow((*row_offsets)[i]);
int64_t row_elem_count = last_elem - first_elem;
if (MatchSingleRow<match_type, all_valid>(elem_cursor,
row_elem_count,
match_bitset,
valid_bitset,
threshold)) {
result_bitset[i] = true;
}
elem_cursor += row_elem_count;
}
}
template <MatchType match_type, bool all_valid>
void
DispatchMatchProcessing(bool use_offset_input,
int64_t row_count,
int64_t row_start,
int64_t elem_start,
const OffsetVector* row_offsets,
const IArrayOffsets* array_offsets,
const TargetBitmapView& match_bitset,
const TargetBitmapView& valid_bitset,
TargetBitmapView& result_bitset,
int64_t threshold) {
if (use_offset_input) {
ProcessOffsetRows<match_type, all_valid>(row_offsets,
array_offsets,
match_bitset,
valid_bitset,
result_bitset,
threshold);
} else {
ProcessMatchRows<match_type, false>(row_count,
array_offsets,
match_bitset,
valid_bitset,
result_bitset,
threshold);
ProcessContiguousRows<match_type, all_valid>(row_count,
row_start,
elem_start,
array_offsets,
match_bitset,
valid_bitset,
result_bitset,
threshold);
}
}
@@ -156,8 +195,7 @@ PhyMatchFilterExpr::Eval(EvalCtx& context, VectorPtr& result) {
tracer::AutoSpan span("PhyMatchFilterExpr::Eval", tracer::GetRootSpan());
auto input = context.get_offset_input();
AssertInfo(input == nullptr,
"Offset input in match filter expr is not implemented now");
SetHasOffsetInput(input != nullptr);
auto schema = segment_->get_schema();
auto field_meta =
@@ -166,25 +204,37 @@ PhyMatchFilterExpr::Eval(EvalCtx& context, VectorPtr& result) {
auto array_offsets = segment_->GetArrayOffsets(field_meta.get_id());
AssertInfo(array_offsets != nullptr, "Array offsets not available");
int64_t row_count =
context.get_exec_context()->get_query_context()->get_active_count();
result = std::make_shared<ColumnVector>(TargetBitmap(row_count, false),
TargetBitmap(row_count, true));
int64_t batch_rows;
int64_t elem_start;
FixedVector<int32_t> element_offsets_storage;
EvalCtx eval_ctx(context.get_exec_context());
if (has_offset_input_) {
// offset_input mode: process all input row_ids at once
batch_rows = input->size();
element_offsets_storage =
array_offsets->RowOffsetsToElementOffsets(*input);
eval_ctx.set_offset_input(&element_offsets_storage);
elem_start = 0;
} else {
// Sequential batch mode
batch_rows = std::min(batch_size_, active_count_ - current_pos_);
auto [start, _] = array_offsets->ElementIDRangeOfRow(current_pos_);
elem_start = start;
}
if (batch_rows <= 0) {
result = nullptr;
return;
}
result = std::make_shared<ColumnVector>(TargetBitmap(batch_rows, false),
TargetBitmap(batch_rows, true));
auto col_vec = std::dynamic_pointer_cast<ColumnVector>(result);
AssertInfo(col_vec != nullptr, "Result should be ColumnVector");
AssertInfo(col_vec->IsBitmap(), "Result should be bitmap");
auto col_vec_size = col_vec->size();
TargetBitmapView bitset_view(col_vec->GetRawData(), col_vec_size);
auto [total_elements, _] = array_offsets->ElementIDRangeOfRow(row_count);
FixedVector<int32_t> element_offsets(total_elements);
std::iota(element_offsets.begin(), element_offsets.end(), 0);
EvalCtx eval_ctx(context.get_exec_context(), &element_offsets);
TargetBitmapView bitset_view(col_vec->GetRawData(), col_vec->size());
VectorPtr match_result;
// TODO(SpadeA): can be executed in batch
inputs_[0]->Eval(eval_ctx, match_result);
auto match_result_col_vec =
std::dynamic_pointer_cast<ColumnVector>(match_result);
@@ -201,56 +251,88 @@ PhyMatchFilterExpr::Eval(EvalCtx& context, VectorPtr& result) {
auto match_type = expr_->get_match_type();
int64_t threshold = expr_->get_count();
switch (match_type) {
case MatchType::MatchAny:
DispatchByValidity<MatchType::MatchAny>(all_valid,
row_count,
array_offsets.get(),
match_result_bitset_view,
match_result_valid_view,
bitset_view,
threshold);
break;
case MatchType::MatchAll:
DispatchByValidity<MatchType::MatchAll>(all_valid,
row_count,
array_offsets.get(),
match_result_bitset_view,
match_result_valid_view,
bitset_view,
threshold);
break;
case MatchType::MatchLeast:
DispatchByValidity<MatchType::MatchLeast>(all_valid,
row_count,
array_offsets.get(),
match_result_bitset_view,
match_result_valid_view,
bitset_view,
threshold);
break;
case MatchType::MatchMost:
DispatchByValidity<MatchType::MatchMost>(all_valid,
row_count,
array_offsets.get(),
match_result_bitset_view,
match_result_valid_view,
bitset_view,
threshold);
break;
case MatchType::MatchExact:
DispatchByValidity<MatchType::MatchExact>(all_valid,
row_count,
array_offsets.get(),
match_result_bitset_view,
match_result_valid_view,
bitset_view,
threshold);
break;
default:
ThrowInfo(OpTypeInvalid,
"Unsupported match type: {}",
static_cast<int>(match_type));
auto dispatch = [&]<bool all_valid_v>() {
switch (match_type) {
case MatchType::MatchAny:
DispatchMatchProcessing<MatchType::MatchAny, all_valid_v>(
has_offset_input_,
batch_rows,
current_pos_,
elem_start,
input,
array_offsets.get(),
match_result_bitset_view,
match_result_valid_view,
bitset_view,
threshold);
break;
case MatchType::MatchAll:
DispatchMatchProcessing<MatchType::MatchAll, all_valid_v>(
has_offset_input_,
batch_rows,
current_pos_,
elem_start,
input,
array_offsets.get(),
match_result_bitset_view,
match_result_valid_view,
bitset_view,
threshold);
break;
case MatchType::MatchLeast:
DispatchMatchProcessing<MatchType::MatchLeast, all_valid_v>(
has_offset_input_,
batch_rows,
current_pos_,
elem_start,
input,
array_offsets.get(),
match_result_bitset_view,
match_result_valid_view,
bitset_view,
threshold);
break;
case MatchType::MatchMost:
DispatchMatchProcessing<MatchType::MatchMost, all_valid_v>(
has_offset_input_,
batch_rows,
current_pos_,
elem_start,
input,
array_offsets.get(),
match_result_bitset_view,
match_result_valid_view,
bitset_view,
threshold);
break;
case MatchType::MatchExact:
DispatchMatchProcessing<MatchType::MatchExact, all_valid_v>(
has_offset_input_,
batch_rows,
current_pos_,
elem_start,
input,
array_offsets.get(),
match_result_bitset_view,
match_result_valid_view,
bitset_view,
threshold);
break;
default:
ThrowInfo(OpTypeInvalid,
"Unsupported match type: {}",
static_cast<int>(match_type));
}
};
if (all_valid) {
dispatch.template operator()<true>();
} else {
dispatch.template operator()<false>();
}
if (!has_offset_input_) {
current_pos_ += batch_rows;
}
}
internal/core/src/exec/expression/MatchExpr.h
+2
View File
@@ -43,6 +43,7 @@ class PhyMatchFilterExpr : public Expr {
segment_(segment),
active_count_(active_count),
batch_size_(batch_size) {
size_per_chunk_ = segment_->size_per_chunk();
}
void
@@ -80,6 +81,7 @@ class PhyMatchFilterExpr : public Expr {
int64_t active_count_;
int64_t current_pos_{0};
int64_t batch_size_;
int64_t size_per_chunk_;
};
} // namespace exec
internal/core/src/exec/expression/MatchExprTest.cpp
+1664
View File
File diff suppressed because it is too large Load Diff
internal/core/src/exec/expression/TermExpr.cpp
+15 -8
View File
@@ -859,7 +859,8 @@ PhyTermFilterExpr::ExecVisitorImplForIndex() {
conditional_t<std::is_same_v<T, std::string_view>, std::string, T>
IndexInnerType;
using Index = index::ScalarIndex<IndexInnerType>;
auto real_batch_size = GetNextBatchSize();
auto real_batch_size =
GetNextRealBatchSize(nullptr, expr_->column_.element_level_);
if (real_batch_size == 0) {
return nullptr;
}
@@ -906,7 +907,8 @@ template <>
VectorPtr
PhyTermFilterExpr::ExecVisitorImplForIndex<bool>() {
using Index = index::ScalarIndex<bool>;
auto real_batch_size = GetNextBatchSize();
auto real_batch_size =
GetNextRealBatchSize(nullptr, expr_->column_.element_level_);
if (real_batch_size == 0) {
return nullptr;
}
@@ -934,8 +936,9 @@ VectorPtr
PhyTermFilterExpr::ExecVisitorImplForData(EvalCtx& context) {
auto* input = context.get_offset_input();
const auto& bitmap_input = context.get_bitmap_input();
auto real_batch_size =
has_offset_input_ ? input->size() : GetNextBatchSize();
GetNextRealBatchSize(input, expr_->column_.element_level_);
if (real_batch_size == 0) {
return nullptr;
}
@@ -1006,7 +1009,7 @@ PhyTermFilterExpr::ExecVisitorImplForData(EvalCtx& context) {
int64_t processed_size;
if (has_offset_input_) {
if (expr_->column_.element_level_) {
// For element-level filtering
// For element-level filtering with offset input
processed_size = ProcessElementLevelByOffsets<T>(execute_sub_batch,
skip_index_func,
input,
@@ -1022,10 +1025,14 @@ PhyTermFilterExpr::ExecVisitorImplForData(EvalCtx& context) {
arg_set_);
}
} else {
AssertInfo(!expr_->column_.element_level_,
"Element-level filtering is not supported without offsets");
processed_size = ProcessDataChunks<T>(
execute_sub_batch, skip_index_func, res, valid_res, arg_set_);
if (expr_->column_.element_level_) {
// For element-level filtering without offset input (brute force)
processed_size = ProcessDataChunksForElementLevel<T>(
execute_sub_batch, skip_index_func, res, valid_res, arg_set_);
} else {
processed_size = ProcessDataChunks<T>(
execute_sub_batch, skip_index_func, res, valid_res, arg_set_);
}
}
AssertInfo(processed_size == real_batch_size,
"internal error: expr processed rows {} not equal "
internal/core/src/exec/expression/UnaryExpr.cpp
+12 -7
View File
@@ -1384,7 +1384,8 @@ PhyUnaryRangeFilterExpr::ExecRangeVisitorImplForIndex() {
return res;
}
auto real_batch_size = GetNextBatchSize();
auto real_batch_size =
GetNextRealBatchSize(nullptr, expr_->column_.element_level_);
if (real_batch_size == 0) {
return nullptr;
}
@@ -1538,7 +1539,7 @@ PhyUnaryRangeFilterExpr::ExecRangeVisitorImplForData(EvalCtx& context) {
}
auto real_batch_size =
has_offset_input_ ? input->size() : GetNextBatchSize();
GetNextRealBatchSize(input, expr_->column_.element_level_);
if (real_batch_size == 0) {
return nullptr;
}
@@ -1722,7 +1723,7 @@ PhyUnaryRangeFilterExpr::ExecRangeVisitorImplForData(EvalCtx& context) {
int64_t processed_size;
if (has_offset_input_) {
if (expr_->column_.element_level_) {
// For element-level filtering
// For element-level filtering with offset input
processed_size = ProcessElementLevelByOffsets<T>(
execute_sub_batch, skip_index_func, input, res, valid_res, val);
} else {
@@ -1730,10 +1731,14 @@ PhyUnaryRangeFilterExpr::ExecRangeVisitorImplForData(EvalCtx& context) {
execute_sub_batch, skip_index_func, input, res, valid_res, val);
}
} else {
AssertInfo(!expr_->column_.element_level_,
"Element-level filtering is not supported without offsets");
processed_size = ProcessDataChunks<T>(
execute_sub_batch, skip_index_func, res, valid_res, val);
if (expr_->column_.element_level_) {
// For element-level filtering without offset input (brute force)
processed_size = ProcessDataChunksForElementLevel<T>(
execute_sub_batch, skip_index_func, res, valid_res, val);
} else {
processed_size = ProcessDataChunks<T>(
execute_sub_batch, skip_index_func, res, valid_res, val);
}
}
AssertInfo(processed_size == real_batch_size,
"internal error: expr processed rows {} not equal "
internal/core/src/index/IndexFactory.cpp
+52
View File
@@ -533,11 +533,63 @@ IndexFactory::CreateGeometryIndex(
return std::make_unique<RTreeIndex<std::string>>(file_manager_context);
}
IndexBasePtr
IndexFactory::CreateNestedIndex(
IndexType index_type,
int32_t tantivy_index_version,
const storage::FileManagerContext& file_manager_context) {
AssertInfo(index_type == INVERTED_INDEX_TYPE,
"Nested index only supports inverted index for now");
DataType element_type = static_cast<DataType>(
file_manager_context.fieldDataMeta.field_schema.element_type());
switch (element_type) {
case DataType::BOOL:
return std::make_unique<InvertedIndexTantivy<bool>>(
tantivy_index_version,
file_manager_context,
false, // inverted_index_single_segment
true, // user_specified_doc_id
true); // is_nested_index
case DataType::INT8:
return std::make_unique<InvertedIndexTantivy<int8_t>>(
tantivy_index_version, file_manager_context, false, true, true);
case DataType::INT16:
return std::make_unique<InvertedIndexTantivy<int16_t>>(
tantivy_index_version, file_manager_context, false, true, true);
case DataType::INT32:
return std::make_unique<InvertedIndexTantivy<int32_t>>(
tantivy_index_version, file_manager_context, false, true, true);
case DataType::INT64:
return std::make_unique<InvertedIndexTantivy<int64_t>>(
tantivy_index_version, file_manager_context, false, true, true);
case DataType::FLOAT:
return std::make_unique<InvertedIndexTantivy<float>>(
tantivy_index_version, file_manager_context, false, true, true);
case DataType::DOUBLE:
return std::make_unique<InvertedIndexTantivy<double>>(
tantivy_index_version, file_manager_context, false, true, true);
case DataType::STRING:
case DataType::VARCHAR:
return std::make_unique<InvertedIndexTantivy<std::string>>(
tantivy_index_version, file_manager_context, false, true, true);
default:
ThrowInfo(DataTypeInvalid, "Invalid data type:{}", element_type);
}
}
IndexBasePtr
IndexFactory::CreateScalarIndex(
const CreateIndexInfo& create_index_info,
const storage::FileManagerContext& file_manager_context) {
auto data_type = create_index_info.field_type;
if (IsStructSubField(create_index_info.field_name)) {
assert(data_type == DataType::ARRAY);
return CreateNestedIndex(create_index_info.index_type,
create_index_info.tantivy_index_version,
file_manager_context);
}
switch (data_type) {
case DataType::BOOL:
case DataType::INT8:
internal/core/src/index/IndexFactory.h
+6
View File
@@ -132,6 +132,12 @@ class IndexFactory {
const storage::FileManagerContext& file_manager_context =
storage::FileManagerContext());
IndexBasePtr
CreateNestedIndex(IndexType index_type,
int32_t tantivy_index_version,
const storage::FileManagerContext& file_manager_context =
storage::FileManagerContext());
IndexBasePtr
CreateScalarIndex(const CreateIndexInfo& create_index_info,
const storage::FileManagerContext& file_manager_context =
internal/core/src/index/InvertedIndexTantivy.cpp
+86 -6
View File
@@ -68,12 +68,14 @@ InvertedIndexTantivy<T>::InvertedIndexTantivy(
uint32_t tantivy_index_version,
const storage::FileManagerContext& ctx,
bool inverted_index_single_segment,
bool user_specified_doc_id)
bool user_specified_doc_id,
bool is_nested_index)
: ScalarIndex<T>(INVERTED_INDEX_TYPE),
schema_(ctx.fieldDataMeta.field_schema),
tantivy_index_version_(tantivy_index_version),
inverted_index_single_segment_(inverted_index_single_segment),
user_specified_doc_id_(user_specified_doc_id) {
user_specified_doc_id_(user_specified_doc_id),
is_nested_index_(is_nested_index) {
mem_file_manager_ = std::make_shared<MemFileManager>(ctx);
disk_file_manager_ = std::make_shared<DiskFileManager>(ctx);
// push init wrapper to load process
@@ -548,13 +550,25 @@ InvertedIndexTantivy<T>::BuildWithRawDataForUT(size_t n,
tantivy_index_version_,
inverted_index_single_segment_);
}
bool is_nested_index = config.find("is_nested_index") != config.end();
if (!inverted_index_single_segment_) {
if (config.find("is_array") != config.end()) {
// only used in ut.
auto arr = static_cast<const boost::container::vector<T>*>(values);
for (size_t i = 0; i < n; i++) {
wrapper_->template add_array_data(
arr[i].data(), arr[i].size(), i);
if (is_nested_index) {
is_nested_index_ = true;
// For nested index, each array element is a separate document
int64_t offset = 0;
for (size_t i = 0; i < n; i++) {
wrapper_->template add_data<T>(
arr[i].data(), arr[i].size(), offset);
offset += arr[i].size();
}
} else {
for (size_t i = 0; i < n; i++) {
wrapper_->template add_array_data(
arr[i].data(), arr[i].size(), i);
}
}
} else {
wrapper_->add_data<T>(static_cast<const T*>(values), n, 0);
@@ -647,7 +661,11 @@ InvertedIndexTantivy<T>::BuildWithFieldData(
}
case proto::schema::DataType::Array: {
build_index_for_array(field_datas);
if (is_nested_index_) {
build_index_for_array_nested(field_datas);
} else {
build_index_for_array(field_datas);
}
break;
}
@@ -730,6 +748,68 @@ InvertedIndexTantivy<std::string>::build_index_for_array(
}
}
template <typename T>
void
InvertedIndexTantivy<T>::build_index_for_array_nested(
const std::vector<std::shared_ptr<FieldDataBase>>& field_datas) {
using ElementType = std::conditional_t<std::is_same<T, int8_t>::value ||
std::is_same<T, int16_t>::value,
int32_t,
T>;
int64_t offset = 0;
int64_t row_offset = 0;
for (const auto& data : field_datas) {
auto n = data->get_num_rows();
auto array_column = static_cast<const Array*>(data->Data());
for (int64_t i = 0; i < n; i++, row_offset++) {
if (schema_.nullable() && !data->is_valid(i)) {
// Record null row offset, no elements to add
null_offset_.push_back(row_offset);
continue;
}
auto length = array_column[i].length();
wrapper_->template add_data<ElementType>(
reinterpret_cast<const ElementType*>(array_column[i].data()),
length,
offset);
offset += length;
}
}
}
template <>
void
InvertedIndexTantivy<std::string>::build_index_for_array_nested(
const std::vector<std::shared_ptr<FieldDataBase>>& field_datas) {
int64_t offset = 0;
int64_t row_offset = 0;
for (const auto& data : field_datas) {
auto n = data->get_num_rows();
auto array_column = static_cast<const Array*>(data->Data());
for (int64_t i = 0; i < n; i++, row_offset++) {
if (schema_.nullable() && !data->is_valid(i)) {
// Record null row offset, no elements to add
null_offset_.push_back(row_offset);
continue;
}
Assert(IsStringDataType(array_column[i].get_element_type()));
Assert(IsStringDataType(
static_cast<DataType>(schema_.element_type())));
std::vector<std::string> output;
auto length = array_column[i].length();
output.reserve(length);
for (int64_t j = 0; j < length; j++) {
output.push_back(
array_column[i].template get_data<std::string>(j));
}
wrapper_->add_data(output.data(), length, offset);
offset += length;
}
}
}
template class InvertedIndexTantivy<bool>;
template class InvertedIndexTantivy<int8_t>;
template class InvertedIndexTantivy<int16_t>;
internal/core/src/index/InvertedIndexTantivy.h
+15 -1
View File
@@ -79,7 +79,8 @@ class InvertedIndexTantivy : public ScalarIndex<T> {
explicit InvertedIndexTantivy(uint32_t tantivy_index_version,
const storage::FileManagerContext& ctx,
bool inverted_index_single_segment = false,
bool user_specified_doc_id = true);
bool user_specified_doc_id = true,
bool is_nested_index = false);
~InvertedIndexTantivy();
@@ -204,6 +205,11 @@ class InvertedIndexTantivy : public ScalarIndex<T> {
this->cached_byte_size_ = total;
}
bool
IsNestedIndex() const override {
return is_nested_index_;
}
virtual const TargetBitmap
PrefixMatch(const std::string_view prefix);
@@ -274,6 +280,10 @@ class InvertedIndexTantivy : public ScalarIndex<T> {
build_index_for_array(
const std::vector<std::shared_ptr<FieldDataBase>>& field_datas);
void
build_index_for_array_nested(
const std::vector<std::shared_ptr<FieldDataBase>>& field_datas);
virtual void
build_index_for_json(
const std::vector<std::shared_ptr<FieldDataBase>>& field_datas) {
@@ -337,5 +347,9 @@ class InvertedIndexTantivy : public ScalarIndex<T> {
// for now, only TextMatchIndex can be built for growing segment,
// and can read and insert concurrently.
bool is_growing_{false};
// `is_nested_index_` can only be true for array data type. When it's true,
// every element in the array is treated as a separate document in the index.
bool is_nested_index_{false};
};
} // namespace milvus::index
internal/core/src/index/ScalarIndex.h
+5
View File
@@ -151,6 +151,11 @@ class ScalarIndex : public IndexBase {
index_type_ == milvus::index::ASCENDING_SORT;
}
virtual bool
IsNestedIndex() const {
return false;
}
virtual int64_t
Size() = 0;