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

enhance: optimize ngram inverted index by executing phase 1/2 in batch expressions (#46999)

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

After this PR, if there are some LIKE exprs connected by `and` and ngram
index is enabled, all phase 1 of them will be executed before phase 2.

This PR significantly improve the performance of ngram index in the
situation of multiple LIKE.
Before this PR, LIKE in ngram index is under performed in the case of
multiple LIKE connected with `and` when the intersection of these
results are small. This is because brute force can take advantage of
rows that are filtered while ngram always do all the things which is
solved by this PR.

The following test shows the effect where the dataset is wiki with 20k
rows, and there are 4, 6, 8 LIKE with `and`:
Before this PR
<img width="805" height="678" alt="image"
src="https://github.com/user-attachments/assets/202ea67d-2ed0-400a-b128-529695e438ab"
/>
After this PR
<img width="720" height="707" alt="image"
src="https://github.com/user-attachments/assets/1865e2ad-f1d6-498c-aaf3-652445e7fc6e"
/>

---------

Signed-off-by: SpadeA <tangchenjie1210@gmail.com>
This commit is contained in:
Spade A
2026-01-21 15:21:30 +08:00
committed by GitHub
parent 8fb3fc927e
commit f022540ccc
12 changed files with 1134 additions and 367 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/core/src/exec/expression/ConjunctExpr.cpp
+58 -3
View File
@@ -15,6 +15,10 @@
// limitations under the License.
#include "ConjunctExpr.h"
#include "UnaryExpr.h"
#include "LikeConjunctExpr.h"
#include <algorithm>
namespace milvus {
namespace exec {
@@ -100,11 +104,62 @@ PhyConjunctFilterExpr::Eval(EvalCtx& context, VectorPtr& result) {
input_order_[i] = i;
}
}
for (int i = 0; i < input_order_.size(); ++i) {
auto has_input_offset = context.get_offset_input() != nullptr;
if (!has_input_offset && !like_batch_initialized_ && is_and_ &&
like_indices_.size() > 1) {
like_batch_initialized_ = true;
// Collect LIKE expressions that can use ngram index at runtime
std::vector<std::shared_ptr<PhyUnaryRangeFilterExpr>> ngram_exprs;
for (size_t idx : like_indices_) {
auto unary_expr =
std::dynamic_pointer_cast<PhyUnaryRangeFilterExpr>(
inputs_[idx]);
if (unary_expr && unary_expr->CanUseNgramIndex()) {
ngram_exprs.push_back(unary_expr);
batch_ngram_indices_.insert(idx);
}
}
// Create PhyLikeConjunctExpr and add to inputs_ if we have >= 2 eligible
if (ngram_exprs.size() >= 2) {
auto active_count = ngram_exprs[0]->GetActiveCount();
auto like_conjunct = std::make_shared<PhyLikeConjunctExpr>(
std::move(ngram_exprs),
op_ctx_,
active_count,
context.get_query_config()->get_expr_batch_size());
inputs_.push_back(like_conjunct);
} else {
batch_ngram_indices_.clear();
// Remove the like_conjunct index from input_order_ since we're not
// creating the batch expression. The index was reserved at compile
// time but the PhyLikeConjunctExpr is not being created at runtime.
auto original_size = inputs_.size();
input_order_.erase(std::remove_if(input_order_.begin(),
input_order_.end(),
[original_size](size_t idx) {
return idx >= original_size;
}),
input_order_.end());
}
}
bool has_result = false;
for (size_t i = 0; i < input_order_.size(); ++i) {
size_t idx = input_order_[i];
// Skip expressions already executed via batch ngram
if (batch_ngram_indices_.count(idx)) {
continue;
}
VectorPtr input_result;
inputs_[input_order_[i]]->Eval(context, input_result);
if (i == 0) {
inputs_[idx]->Eval(context, input_result);
if (!has_result) {
result = input_result;
has_result = true;
auto all_flat_result = GetColumnVector(result);
if (CanSkipFollowingExprs(all_flat_result)) {
SkipFollowingExprs(i + 1);
internal/core/src/exec/expression/ConjunctExpr.h
+37 -3
View File
@@ -17,6 +17,7 @@
#pragma once
#include <fmt/core.h>
#include <set>
#include "common/EasyAssert.h"
#include "common/OpContext.h"
@@ -118,15 +119,25 @@ class PhyConjunctFilterExpr : public Expr {
ToString() const {
if (!input_order_.empty()) {
std::vector<std::string> inputs;
for (auto& i : input_order_) {
inputs.push_back(inputs_[i]->ToString());
inputs.reserve(input_order_.size());
for (const auto& i : input_order_) {
if (i < inputs_.size()) {
inputs.push_back(inputs_[i]->ToString());
} else {
// Reserved position for runtime-created expressions (e.g., PhyLikeConjunctExpr)
// This can happen during optimization phase before the expression is actually created
inputs.push_back(
fmt::format("[RuntimeExpr:index={}:pending]", i));
}
}
std::string input_str =
is_and_ ? Join(inputs, " && ") : Join(inputs, " || ");
return fmt::format("[ConjuctExpr:{}]", input_str);
}
// Fallback: no reordering applied yet
std::vector<std::string> inputs;
for (auto& in : inputs_) {
inputs.reserve(inputs_.size());
for (const auto& in : inputs_) {
inputs.push_back(in->ToString());
}
std::string input_str =
@@ -154,6 +165,13 @@ class PhyConjunctFilterExpr : public Expr {
return input_order_;
}
// Add a new expression to inputs and return its index
size_t
AddInput(std::shared_ptr<Expr> expr) {
inputs_.push_back(std::move(expr));
return inputs_.size() - 1;
}
void
SetNextExprBitmapInput(const ColumnVectorPtr& vec, EvalCtx& context) {
TargetBitmapView last_res_bitmap(vec->GetRawData(), vec->size());
@@ -181,6 +199,16 @@ class PhyConjunctFilterExpr : public Expr {
return !is_and_;
}
void
SetLikeIndices(std::vector<size_t>&& indices) {
like_indices_ = std::move(indices);
}
const std::vector<size_t>&
GetLikeIndices() const {
return like_indices_;
}
private:
int64_t
UpdateResult(ColumnVectorPtr& input_result,
@@ -198,6 +226,12 @@ class PhyConjunctFilterExpr : public Expr {
// true if conjunction (and), false if disjunction (or).
bool is_and_;
std::vector<size_t> input_order_;
// Indices of LIKE expressions for potential batch ngram optimization (AND only)
std::vector<size_t> like_indices_;
// Flag to indicate if batch ngram optimization has been initialized
bool like_batch_initialized_{false};
// Indices of expressions executed via batch ngram (to skip in normal iteration)
std::set<size_t> batch_ngram_indices_;
};
} //namespace exec
} // namespace milvus
internal/core/src/exec/expression/Expr.cpp
+91 -17
View File
@@ -138,19 +138,75 @@ CompileExpressions(const std::vector<expr::TypedExprPtr>& sources,
return exprs;
}
static std::optional<std::string>
ShouldFlatten(const expr::TypedExprPtr& expr,
const std::unordered_set<std::string>& flat_candidates = {}) {
if (auto call =
std::dynamic_pointer_cast<const expr::LogicalBinaryExpr>(expr)) {
if (call->op_type_ == expr::LogicalBinaryExpr::OpType::And ||
call->op_type_ == expr::LogicalBinaryExpr::OpType::Or) {
return call->name();
}
}
return std::nullopt;
}
static bool
IsCall(const expr::TypedExprPtr& expr, const std::string& name) {
if (auto call =
std::dynamic_pointer_cast<const expr::LogicalBinaryExpr>(expr)) {
return call->name() == name;
}
return false;
}
static bool
AllInputTypeEqual(const expr::TypedExprPtr& expr) {
const auto& inputs = expr->inputs();
for (int i = 1; i < inputs.size(); i++) {
if (inputs[0]->type() != inputs[i]->type()) {
return false;
}
}
return true;
}
static void
FlattenInput(const expr::TypedExprPtr& input,
const std::string& flatten_call,
std::vector<expr::TypedExprPtr>& flat) {
if (IsCall(input, flatten_call) && AllInputTypeEqual(input)) {
for (auto& child : input->inputs()) {
FlattenInput(child, flatten_call, flat);
}
} else {
flat.emplace_back(input);
}
}
std::vector<ExprPtr>
CompileInputs(const expr::TypedExprPtr& expr,
QueryContext* context,
const std::unordered_set<std::string>& flatten_cadidates) {
std::vector<ExprPtr> compiled_inputs;
auto flatten = ShouldFlatten(expr);
for (auto& input : expr->inputs()) {
if (dynamic_cast<const expr::InputTypeExpr*>(input.get())) {
AssertInfo(
dynamic_cast<const expr::FieldAccessTypeExpr*>(expr.get()),
"An InputReference can only occur under a FieldReference");
} else {
compiled_inputs.push_back(
CompileExpression(input, context, flatten_cadidates, false));
if (flatten.has_value()) {
std::vector<expr::TypedExprPtr> flat_exprs;
FlattenInput(input, flatten.value(), flat_exprs);
for (auto& flat_input : flat_exprs) {
compiled_inputs.push_back(CompileExpression(
flat_input, context, flatten_cadidates, false));
}
} else {
compiled_inputs.push_back(CompileExpression(
input, context, flatten_cadidates, false));
}
}
}
return compiled_inputs;
@@ -406,8 +462,11 @@ ReorderConjunctExpr(std::shared_ptr<milvus::exec::PhyConjunctFilterExpr>& expr,
std::vector<size_t> other_expr;
std::vector<size_t> heavy_conjunct_expr;
std::vector<size_t> light_conjunct_expr;
// Record all LIKE expression indices for potential batch ngram optimization
std::vector<size_t> like_indices;
const auto& inputs = expr->GetInputsRef();
bool and_conjunction = expr->IsAnd();
for (int i = 0; i < inputs.size(); i++) {
auto input = inputs[i];
@@ -417,16 +476,18 @@ ReorderConjunctExpr(std::shared_ptr<milvus::exec::PhyConjunctFilterExpr>& expr,
numeric_expr.push_back(i);
continue;
}
if (segment->HasIndex(column.field_id_)) {
if (segment->HasIndex(column.field_id_) && !IsLikeExpr(input)) {
indexed_expr.push_back(i);
continue;
}
if (IsStringDataType(column.data_type_)) {
auto is_like_expr = IsLikeExpr(input);
if (is_like_expr) {
str_like_expr.push_back(i);
if (IsLikeExpr(input)) {
has_heavy_operation = true;
str_like_expr.push_back(i);
if (and_conjunction) {
like_indices.push_back(i);
}
} else {
string_expr.push_back(i);
}
@@ -434,10 +495,12 @@ ReorderConjunctExpr(std::shared_ptr<milvus::exec::PhyConjunctFilterExpr>& expr,
}
if (IsArrayDataType(column.data_type_)) {
auto is_like_expr = IsLikeExpr(input);
if (is_like_expr) {
array_like_expr.push_back(i);
if (IsLikeExpr(input)) {
has_heavy_operation = true;
array_like_expr.push_back(i);
if (and_conjunction) {
like_indices.push_back(i);
}
} else {
array_expr.push_back(i);
}
@@ -445,9 +508,11 @@ ReorderConjunctExpr(std::shared_ptr<milvus::exec::PhyConjunctFilterExpr>& expr,
}
if (IsJsonDataType(column.data_type_)) {
auto is_like_expr = IsLikeExpr(input);
if (is_like_expr) {
if (IsLikeExpr(input)) {
json_like_expr.push_back(i);
if (and_conjunction) {
like_indices.push_back(i);
}
} else {
json_expr.push_back(i);
}
@@ -458,8 +523,9 @@ ReorderConjunctExpr(std::shared_ptr<milvus::exec::PhyConjunctFilterExpr>& expr,
if (input->name() == "PhyConjunctFilterExpr") {
bool sub_expr_heavy = false;
auto expr = std::static_pointer_cast<PhyConjunctFilterExpr>(input);
ReorderConjunctExpr(expr, context, sub_expr_heavy);
auto sub_expr =
std::static_pointer_cast<PhyConjunctFilterExpr>(input);
ReorderConjunctExpr(sub_expr, context, sub_expr_heavy);
has_heavy_operation |= sub_expr_heavy;
if (sub_expr_heavy) {
heavy_conjunct_expr.push_back(i);
@@ -478,7 +544,6 @@ ReorderConjunctExpr(std::shared_ptr<milvus::exec::PhyConjunctFilterExpr>& expr,
other_expr.push_back(i);
}
reorder.reserve(inputs.size());
// Final reorder sequence:
// 1. Numeric column expressions (fastest to evaluate)
// 2. Indexed column expressions (can use index for efficient filtering)
@@ -504,14 +569,23 @@ ReorderConjunctExpr(std::shared_ptr<milvus::exec::PhyConjunctFilterExpr>& expr,
reorder.end(), array_like_expr.begin(), array_like_expr.end());
reorder.insert(reorder.end(), json_expr.begin(), json_expr.end());
reorder.insert(reorder.end(), json_like_expr.begin(), json_like_expr.end());
// Reserve position for like_conjunct (will be added to inputs_ at runtime)
bool has_batch_like = like_indices.size() > 1;
if (has_batch_like) {
reorder.push_back(
inputs.size()); // inputs.size() will be like_conjunct's index
expr->SetLikeIndices(std::move(like_indices));
}
reorder.insert(
reorder.end(), heavy_conjunct_expr.begin(), heavy_conjunct_expr.end());
reorder.insert(reorder.end(), compare_expr.begin(), compare_expr.end());
AssertInfo(reorder.size() == inputs.size(),
"reorder size:{} but input size:{}",
size_t expected_size = inputs.size() + (has_batch_like ? 1 : 0);
AssertInfo(reorder.size() == expected_size,
"reorder size:{} but expected size:{}",
reorder.size(),
inputs.size());
expected_size);
expr->Reorder(reorder);
}
internal/core/src/exec/expression/Expr.h
+26 -13
View File
@@ -1462,32 +1462,44 @@ class SegmentExpr : public Expr {
}
}
// Specialized method for ngram post-filter: processes all data in batches
// - Starts from position 0
// Specialized method for ngram post-filter: processes data in a specific range
// - Starts from segment_offset (global offset across all chunks)
// - Processes exactly 'size' rows
// - Does NOT modify segment state variables (current_data_chunk_, etc.)
template <typename T, typename FUNC>
int64_t
ProcessAllDataChunkBatched(FUNC func, TargetBitmapView res) {
ProcessDataChunkForRange(FUNC func,
TargetBitmapView res,
int64_t segment_offset,
int64_t size) {
static_assert(std::is_same_v<T, std::string_view> ||
std::is_same_v<T, Json> ||
std::is_same_v<T, ArrayView>,
"ProcessAllDataChunkBatched only supports string_view, "
"ProcessDataChunkForRange only supports string_view, "
"Json, and ArrayView types");
AssertInfo(segment_->is_chunked(),
"ProcessAllDataChunkBatched requires chunked segment");
"ProcessDataChunkForRange requires chunked segment");
AssertInfo(segment_->type() == SegmentType::Sealed,
"ProcessAllDataChunkBatched requires sealed segment");
"ProcessDataChunkForRange requires sealed segment");
int64_t processed_size = 0;
int64_t remaining = size;
for (size_t chunk_id = 0; chunk_id < num_data_chunk_; chunk_id++) {
// Find starting chunk and offset
auto [start_chunk_id, start_chunk_offset] =
segment_->get_chunk_by_offset(field_id_, segment_offset);
for (size_t chunk_id = start_chunk_id;
chunk_id < num_data_chunk_ && remaining > 0;
chunk_id++) {
int64_t chunk_size = segment_->chunk_size(field_id_, chunk_id);
int64_t chunk_offset = 0;
int64_t chunk_offset =
(chunk_id == start_chunk_id) ? start_chunk_offset : 0;
while (chunk_offset < chunk_size) {
int64_t batch_size =
std::min(batch_size_, chunk_size - chunk_offset);
while (chunk_offset < chunk_size && remaining > 0) {
int64_t batch_size = std::min(
{batch_size_, chunk_size - chunk_offset, remaining});
auto pw = segment_->get_batch_views<T>(
op_ctx_, field_id_, chunk_id, chunk_offset, batch_size);
@@ -1497,6 +1509,7 @@ class SegmentExpr : public Expr {
chunk_offset += batch_size;
processed_size += batch_size;
remaining -= batch_size;
}
}
@@ -2077,8 +2090,8 @@ class SegmentExpr : public Expr {
std::shared_ptr<TargetBitmap> cached_match_res_{nullptr};
int32_t consistency_level_{0};
// Cache for ngram match.
std::shared_ptr<TargetBitmap> cached_ngram_match_res_{nullptr};
// Cache for ngram Phase1 result (index query, before post-filter).
std::shared_ptr<TargetBitmap> cached_phase1_res_{nullptr};
};
bool
internal/core/src/exec/expression/LikeConjunctExpr.cpp
+103
View File
@@ -0,0 +1,103 @@
// Licensed to the LF AI & Data foundation under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License. You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
#include "exec/expression/LikeConjunctExpr.h"
#include "exec/expression/UnaryExpr.h"
namespace milvus {
namespace exec {
PhyLikeConjunctExpr::PhyLikeConjunctExpr(
std::vector<std::shared_ptr<PhyUnaryRangeFilterExpr>> ngram_exprs,
milvus::OpContext* op_ctx,
int64_t active_count,
int64_t batch_size)
: Expr(DataType::BOOL, {}, "PhyLikeConjunctExpr", op_ctx),
ngram_exprs_(std::move(ngram_exprs)),
active_count_(active_count),
batch_size_(batch_size) {
}
int64_t
PhyLikeConjunctExpr::GetNextBatchSize() {
return current_pos_ + batch_size_ >= active_count_
? active_count_ - current_pos_
: batch_size_;
}
void
PhyLikeConjunctExpr::Eval(EvalCtx& context, VectorPtr& result) {
AssertInfo(context.get_offset_input() == nullptr,
"Offset input is not supported for PhyLikeConjunctExpr");
AssertInfo(!ngram_exprs_.empty(),
"PhyLikeConjunctExpr requires at least one expression");
auto real_batch_size = GetNextBatchSize();
if (real_batch_size == 0) {
result = nullptr;
return;
}
// Phase 1: Execute once for entire segment (no bitmap_input needed)
if (cached_phase1_res_ == nullptr) {
TargetBitmap candidates(active_count_, true);
// Execute all ngram index queries, AND merge results
for (auto& expr : ngram_exprs_) {
expr->ExecuteNgramPhase1(candidates);
if (candidates.none()) {
break;
}
}
cached_phase1_res_ =
std::make_shared<TargetBitmap>(std::move(candidates));
}
// Phase 2: Execute per batch with batch-level bitmap_input
// Extract current batch from Phase1 result
TargetBitmap batch_candidates;
batch_candidates.append(*cached_phase1_res_, current_pos_, real_batch_size);
// Apply batch-level pre_filter from previous expressions in conjunction
const auto& bitmap_input = context.get_bitmap_input();
if (!bitmap_input.empty()) {
AssertInfo(static_cast<int64_t>(bitmap_input.size()) == real_batch_size,
"bitmap_input size {} != real_batch_size {}",
bitmap_input.size(),
real_batch_size);
batch_candidates &= bitmap_input;
}
// Execute Phase2 (post-filter) on this batch
if (!batch_candidates.none()) {
for (auto& expr : ngram_exprs_) {
expr->ExecuteNgramPhase2(
batch_candidates, current_pos_, real_batch_size);
if (batch_candidates.none()) {
break;
}
}
}
// For ngram like expression, the valid result is always true as result has all information
TargetBitmap valid_result(real_batch_size, true);
current_pos_ += real_batch_size;
result = std::make_shared<ColumnVector>(std::move(batch_candidates),
std::move(valid_result));
}
} // namespace exec
} // namespace milvus
internal/core/src/exec/expression/LikeConjunctExpr.h
+62
View File
@@ -0,0 +1,62 @@
// Licensed to the LF AI & Data foundation under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License. You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
#pragma once
#include <memory>
#include <vector>
#include "exec/expression/Expr.h"
namespace milvus {
namespace exec {
class PhyUnaryRangeFilterExpr;
// PhyLikeConjunctExpr optimizes multiple LIKE expressions with ngram index.
// Phase1 (ngram index query) executes once for the entire segment.
// Phase2 (post-filter) executes per batch with batch-level pre_filter.
class PhyLikeConjunctExpr : public Expr {
public:
PhyLikeConjunctExpr(
std::vector<std::shared_ptr<PhyUnaryRangeFilterExpr>> ngram_exprs,
milvus::OpContext* op_ctx,
int64_t active_count,
int64_t batch_size);
void
Eval(EvalCtx& context, VectorPtr& result) override;
std::string
ToString() const override {
return "PhyLikeConjunctExpr";
}
private:
int64_t
GetNextBatchSize();
std::vector<std::shared_ptr<PhyUnaryRangeFilterExpr>> ngram_exprs_;
// Cached Phase1 result (segment-level ngram index query result)
std::shared_ptr<TargetBitmap> cached_phase1_res_{nullptr};
int64_t active_count_{0};
int64_t batch_size_{0};
int64_t current_pos_{0};
};
} // namespace exec
} // namespace milvus
internal/core/src/exec/expression/LikeConjunctExprTest.cpp
+332
View File
@@ -0,0 +1,332 @@
// Copyright (C) 2019-2020 Zilliz. All rights reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance
// with the License. You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software distributed under the License
// is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
// or implied. See the License for the specific language governing permissions and limitations under the License
#include <gtest/gtest.h>
#include <memory>
#include <numeric>
#include <string>
#include <vector>
#include "common/Schema.h"
#include "common/Consts.h"
#include "test_utils/GenExprProto.h"
#include "query/PlanProto.h"
#include "query/ExecPlanNodeVisitor.h"
#include "query/Plan.h"
#include "expr/ITypeExpr.h"
#include "test_utils/storage_test_utils.h"
#include "index/IndexFactory.h"
#include "index/NgramInvertedIndex.h"
#include "segcore/load_index_c.h"
#include "test_utils/cachinglayer_test_utils.h"
#include "milvus-storage/filesystem/fs.h"
using namespace milvus;
using namespace milvus::query;
using namespace milvus::segcore;
using namespace milvus::exec;
std::unique_ptr<proto::segcore::RetrieveResults>
RetrieveWithPlan(SegmentInterface* segment,
const query::RetrievePlan* plan,
Timestamp timestamp) {
return segment->Retrieve(
nullptr, plan, timestamp, DEFAULT_MAX_OUTPUT_SIZE, false);
}
// Test multiple LIKE conditions on multiple fields with AND using ngram index
// 2 fields x 2 LIKE conditions each = 4 total LIKE conditions
// title LIKE '%Database%' AND title LIKE '%Design%' AND content LIKE '%SQL%' AND content LIKE '%query%'
TEST(LikeConjunctExpr, TestMultiFieldMultiLikeWithRetrieve) {
int64_t collection_id = 1;
int64_t partition_id = 2;
int64_t segment_id = 3;
int64_t index_version = 4000;
EXEC_EVAL_EXPR_BATCH_SIZE.store(3);
auto schema = std::make_shared<Schema>();
auto pk_fid = schema->AddDebugField("pk", DataType::INT64);
auto title_fid = schema->AddDebugField("title", DataType::VARCHAR);
auto content_fid = schema->AddDebugField("content", DataType::VARCHAR);
schema->set_primary_field_id(pk_fid);
// Use TestLocalPath to match LocalChunkManagerSingleton initialized in init_gtest.cpp
auto storage_config = gen_local_storage_config(TestLocalPath);
auto cm = CreateChunkManager(storage_config);
auto fs = storage::InitArrowFileSystem(storage_config);
// Initialize ArrowFileSystemSingleton for UpdateSealedSegmentIndex
milvus_storage::ArrowFileSystemConfig arrow_conf;
arrow_conf.storage_type = "local";
arrow_conf.root_path = storage_config.root_path;
milvus_storage::ArrowFileSystemSingleton::GetInstance().Init(arrow_conf);
// Test data: 8 rows
// title conditions: "Database" AND "Design"
// content conditions: "SQL" AND "query"
std::vector<int64_t> pk_data = {0, 1, 2, 3, 4, 5, 6, 7};
boost::container::vector<std::string> title_data = {
"Database Design Patterns", // 0: has "Database" + "Design" ✓
"Machine Learning Fundamentals", // 1: no match
"Advanced Database Design Guide", // 2: has "Database" + "Design" ✓
"Web Development Design", // 3: has "Design" only
"Database Performance Tuning", // 4: has "Database" only
"Cloud Computing Basics", // 5: no match
"Distributed Database Design Systems", // 6: has "Database" + "Design" ✓
"Software Design Principles", // 7: has "Design" only
};
boost::container::vector<std::string> content_data = {
"SQL query optimization techniques", // 0: has "SQL" + "query" ✓
"Deep learning and neural networks", // 1: no match
"Advanced SQL query patterns", // 2: has "SQL" + "query" ✓
"HTML CSS JavaScript tutorial", // 3: no match
"SQL tuning and monitoring", // 4: has "SQL" only
"AWS Azure GCP comparison", // 5: no match
"Complex SQL query design", // 6: has "SQL" + "query" ✓
"Python Java C++ examples", // 7: no match
};
// Expected matches (all 4 conditions): rows 0, 2, 6
auto segment = CreateSealedSegment(schema);
// Load system fields (row_id and timestamp) for proper Retrieve support
size_t nb = pk_data.size();
std::vector<int64_t> row_ids(nb);
std::vector<int64_t> timestamps(nb);
std::iota(row_ids.begin(), row_ids.end(), 0);
std::iota(timestamps.begin(), timestamps.end(), 0);
{
auto row_id_field_data =
storage::CreateFieldData(DataType::INT64, DataType::NONE, false);
row_id_field_data->FillFieldData(row_ids.data(), row_ids.size());
auto row_id_field_info =
PrepareSingleFieldInsertBinlog(collection_id,
partition_id,
segment_id,
RowFieldID.get(),
{row_id_field_data},
cm);
segment->LoadFieldData(row_id_field_info);
}
{
auto ts_field_data =
storage::CreateFieldData(DataType::INT64, DataType::NONE, false);
ts_field_data->FillFieldData(timestamps.data(), timestamps.size());
auto ts_field_info =
PrepareSingleFieldInsertBinlog(collection_id,
partition_id,
segment_id,
TimestampFieldID.get(),
{ts_field_data},
cm);
segment->LoadFieldData(ts_field_info);
}
// Create ChunkManagerWrapper at test level for cleanup after test ends
auto cm_w = ChunkManagerWrapper(cm);
// Helper to load field with ngram index
auto load_varchar_field_with_ngram =
[&](FieldId field_id,
const boost::container::vector<std::string>& data,
int64_t index_id,
int64_t field_index_build_id) {
auto field_meta = gen_field_meta(collection_id,
partition_id,
segment_id,
field_id.get(),
DataType::VARCHAR,
DataType::NONE,
false);
auto index_meta = gen_index_meta(segment_id,
field_id.get(),
field_index_build_id,
index_version);
auto field_data = storage::CreateFieldData(
DataType::VARCHAR, DataType::NONE, false);
field_data->FillFieldData(data.data(), data.size());
auto field_data_info =
PrepareSingleFieldInsertBinlog(collection_id,
partition_id,
segment_id,
field_id.get(),
{field_data},
cm);
segment->LoadFieldData(field_data_info);
auto payload_reader =
std::make_shared<milvus::storage::PayloadReader>(field_data);
storage::InsertData insert_data(payload_reader);
insert_data.SetFieldDataMeta(field_meta);
insert_data.SetTimestamps(0, 100);
auto serialized_bytes = insert_data.Serialize(storage::Remote);
auto log_path = fmt::format("{}/{}/{}/{}/{}",
collection_id,
partition_id,
segment_id,
field_id.get(),
0);
cm_w.Write(
log_path, serialized_bytes.data(), serialized_bytes.size());
storage::FileManagerContext ctx(field_meta, index_meta, cm, fs);
Config config;
config[milvus::index::INDEX_TYPE] =
milvus::index::INVERTED_INDEX_TYPE;
config[INSERT_FILES_KEY] = std::vector<std::string>{log_path};
auto ngram_params = index::NgramParams{
.loading_index = false,
.min_gram = 2,
.max_gram = 4,
};
auto index =
std::make_shared<index::NgramInvertedIndex>(ctx, ngram_params);
index->Build(config);
auto create_index_result = index->Upload();
auto index_files = create_index_result->GetIndexFiles();
std::map<std::string, std::string> index_params{
{milvus::index::INDEX_TYPE, milvus::index::NGRAM_INDEX_TYPE},
{milvus::index::MIN_GRAM, "2"},
{milvus::index::MAX_GRAM, "4"},
{milvus::LOAD_PRIORITY, "HIGH"},
};
LoadIndexInfo load_index_info;
load_index_info.collection_id = collection_id;
load_index_info.partition_id = partition_id;
load_index_info.segment_id = segment_id;
load_index_info.field_id = field_id.get();
load_index_info.field_type = DataType::VARCHAR;
load_index_info.enable_mmap = true;
load_index_info.mmap_dir_path = "/tmp/test-like-conjunct-mmap-dir";
load_index_info.index_id = index_id;
load_index_info.index_build_id = field_index_build_id;
load_index_info.index_version = index_version;
load_index_info.index_params = index_params;
load_index_info.index_files = index_files;
load_index_info.schema = field_meta.field_schema;
load_index_info.index_size = 1024 * 1024;
uint8_t trace_id[16] = {0};
uint8_t span_id[8] = {0};
CTraceContext trace{
.traceID = trace_id,
.spanID = span_id,
.traceFlags = 0,
};
auto cload_index_info =
static_cast<CLoadIndexInfo>(&load_index_info);
AppendIndexV2(trace, cload_index_info);
UpdateSealedSegmentIndex(segment.get(), cload_index_info);
};
// Load pk field
{
auto pk_field_data =
storage::CreateFieldData(DataType::INT64, DataType::NONE, false);
pk_field_data->FillFieldData(pk_data.data(), pk_data.size());
auto field_data_info = PrepareSingleFieldInsertBinlog(collection_id,
partition_id,
segment_id,
pk_fid.get(),
{pk_field_data},
cm);
segment->LoadFieldData(field_data_info);
}
// Load title and content fields with ngram index (use different index_build_id for each)
load_varchar_field_with_ngram(title_fid, title_data, 5001, 4001);
load_varchar_field_with_ngram(content_fid, content_data, 5002, 4002);
// Create 4 LIKE expressions:
// title LIKE '%Database%' AND title LIKE '%Design%' AND content LIKE '%SQL%' AND content LIKE '%query%'
proto::plan::GenericValue val_database, val_design, val_sql, val_query;
val_database.set_string_val("Database");
val_design.set_string_val("Design");
val_sql.set_string_val("SQL");
val_query.set_string_val("query");
auto expr_title_database =
std::make_shared<milvus::expr::UnaryRangeFilterExpr>(
milvus::expr::ColumnInfo(title_fid, DataType::VARCHAR),
proto::plan::OpType::InnerMatch,
val_database,
std::vector<proto::plan::GenericValue>{});
auto expr_title_design =
std::make_shared<milvus::expr::UnaryRangeFilterExpr>(
milvus::expr::ColumnInfo(title_fid, DataType::VARCHAR),
proto::plan::OpType::InnerMatch,
val_design,
std::vector<proto::plan::GenericValue>{});
auto expr_content_sql =
std::make_shared<milvus::expr::UnaryRangeFilterExpr>(
milvus::expr::ColumnInfo(content_fid, DataType::VARCHAR),
proto::plan::OpType::InnerMatch,
val_sql,
std::vector<proto::plan::GenericValue>{});
auto expr_content_query =
std::make_shared<milvus::expr::UnaryRangeFilterExpr>(
milvus::expr::ColumnInfo(content_fid, DataType::VARCHAR),
proto::plan::OpType::InnerMatch,
val_query,
std::vector<proto::plan::GenericValue>{});
// Create nested AND expression: ((A AND B) AND C) AND D
auto and_expr_1 = std::make_shared<milvus::expr::LogicalBinaryExpr>(
milvus::expr::LogicalBinaryExpr::OpType::And,
expr_title_database,
expr_title_design);
auto and_expr_2 = std::make_shared<milvus::expr::LogicalBinaryExpr>(
milvus::expr::LogicalBinaryExpr::OpType::And,
and_expr_1,
expr_content_sql);
auto and_expr = std::make_shared<milvus::expr::LogicalBinaryExpr>(
milvus::expr::LogicalBinaryExpr::OpType::And,
and_expr_2,
expr_content_query);
// Create RetrievePlan
auto plan = std::make_unique<query::RetrievePlan>(schema);
plan->plan_node_ = std::make_unique<query::RetrievePlanNode>();
plan->plan_node_->plannodes_ =
milvus::test::CreateRetrievePlanByExpr(and_expr);
plan->field_ids_ = {pk_fid};
// Execute retrieve
auto results = RetrieveWithPlan(segment.get(), plan.get(), MAX_TIMESTAMP);
// Verify results: rows 0, 2, 6 should match (all 4 LIKE conditions)
ASSERT_EQ(results->fields_data_size(), 1);
auto& field_data = results->fields_data(0);
ASSERT_TRUE(field_data.has_scalars());
auto& pk_results = field_data.scalars().long_data();
std::set<int64_t> expected_pks = {0, 2, 6};
std::set<int64_t> actual_pks;
for (int i = 0; i < pk_results.data_size(); i++) {
actual_pks.insert(pk_results.data(i));
}
EXPECT_EQ(actual_pks, expected_pks);
EXEC_EVAL_EXPR_BATCH_SIZE.store(DEFAULT_EXEC_EVAL_EXPR_BATCH_SIZE);
}
internal/core/src/exec/expression/UnaryExpr.cpp
+80 -15
View File
@@ -1895,7 +1895,47 @@ PhyUnaryRangeFilterExpr::ExecTextMatch() {
bool
PhyUnaryRangeFilterExpr::CanUseNgramIndex() const {
return pinned_ngram_index_.get() != nullptr && !has_offset_input_;
if (pinned_ngram_index_.get() == nullptr || has_offset_input_) {
return false;
}
auto literal = GetValueFromProto<std::string>(expr_->val_);
return pinned_ngram_index_.get()->CanHandleLiteral(literal,
expr_->op_type_);
}
void
PhyUnaryRangeFilterExpr::ExecuteNgramPhase1(TargetBitmap& candidates) {
if (!arg_inited_) {
value_arg_.SetValue<std::string>(expr_->val_);
arg_inited_ = true;
}
auto literal = value_arg_.GetValue<std::string>();
auto index = pinned_ngram_index_.get();
AssertInfo(index != nullptr,
"ngram index should not be null, field_id: {}",
field_id_.get());
index->ExecutePhase1(literal, expr_->op_type_, candidates);
}
void
PhyUnaryRangeFilterExpr::ExecuteNgramPhase2(TargetBitmap& candidates,
int64_t segment_offset,
int64_t batch_size) {
if (!arg_inited_) {
value_arg_.SetValue<std::string>(expr_->val_);
arg_inited_ = true;
}
auto literal = value_arg_.GetValue<std::string>();
auto index = pinned_ngram_index_.get();
AssertInfo(index != nullptr,
"ngram index should not be null, field_id: {}",
field_id_.get());
index->ExecutePhase2(
literal, expr_->op_type_, this, candidates, segment_offset, batch_size);
}
std::optional<VectorPtr>
@@ -1911,32 +1951,57 @@ PhyUnaryRangeFilterExpr::ExecNgramMatch(EvalCtx& context) {
return std::nullopt;
}
if (cached_ngram_match_res_ == nullptr) {
auto index = pinned_ngram_index_.get();
AssertInfo(index != nullptr,
"ngram index should not be null, field_id: {}",
field_id_.get());
auto index = pinned_ngram_index_.get();
AssertInfo(index != nullptr,
"ngram index should not be null, field_id: {}",
field_id_.get());
auto res_opt =
index->ExecuteQuery(literal, expr_->op_type_, this, nullptr);
if (!res_opt.has_value()) {
// Phase 1: Execute once for entire segment (no bitmap_input needed)
if (cached_phase1_res_ == nullptr) {
if (!index->CanHandleLiteral(literal, expr_->op_type_)) {
return std::nullopt;
}
cached_ngram_match_res_ =
std::make_shared<TargetBitmap>(std::move(res_opt.value()));
auto total_count = static_cast<size_t>(index->Count());
TargetBitmap candidates(total_count, true);
index->ExecutePhase1(literal, expr_->op_type_, candidates);
cached_phase1_res_ =
std::make_shared<TargetBitmap>(std::move(candidates));
cached_index_chunk_valid_res_ =
std::make_shared<TargetBitmap>(std::move(index->IsNotNull()));
}
TargetBitmap result;
// Phase 2: Execute per batch with batch-level bitmap_input
TargetBitmap batch_candidates;
batch_candidates.append(
*cached_phase1_res_, current_data_global_pos_, real_batch_size);
// Apply batch-level pre_filter from previous expressions in conjunction
const auto& bitmap_input = context.get_bitmap_input();
if (!bitmap_input.empty()) {
AssertInfo(static_cast<int64_t>(bitmap_input.size()) == real_batch_size,
"bitmap_input size {} != real_batch_size {}",
bitmap_input.size(),
real_batch_size);
batch_candidates &= bitmap_input;
}
// Execute Phase2 (post-filter) on this batch
if (!batch_candidates.none()) {
index->ExecutePhase2(literal,
expr_->op_type_,
this,
batch_candidates,
current_data_global_pos_,
real_batch_size);
}
TargetBitmap valid_result;
result.append(
*cached_ngram_match_res_, current_data_global_pos_, real_batch_size);
valid_result.append(*cached_index_chunk_valid_res_,
current_data_global_pos_,
real_batch_size);
MoveCursor();
return std::make_shared<ColumnVector>(std::move(result),
return std::make_shared<ColumnVector>(std::move(batch_candidates),
std::move(valid_result));
}
internal/core/src/exec/expression/UnaryExpr.h
+31 -1
View File
@@ -35,6 +35,7 @@
#include "common/bson_view.h"
#include "index/json_stats/bson_inverted.h"
#include "cachinglayer/CacheSlot.h"
#include "index/NgramInvertedIndex.h"
namespace milvus {
namespace exec {
@@ -825,6 +826,32 @@ class PhyUnaryRangeFilterExpr : public SegmentExpr {
return expr_->column_.data_type_;
}
int64_t
GetActiveCount() const {
return active_count_;
}
// Check if ngram index can be used (index exists + literal is valid + no offset input)
bool
CanUseNgramIndex() const override;
// Execute ngram Phase1 only (index query), ANDs result into candidates
// Requires: CanUseNgramIndex() == true
// Requires: candidates must be non-empty (caller initializes with all-true,
// then ANDs with pre_filter/offset_input before calling)
void
ExecuteNgramPhase1(TargetBitmap& candidates);
// Execute ngram Phase2 (post-filter verification) on candidate bitset
// - segment_offset: starting position in segment
// - batch_size: number of rows to process
// - candidates: bitmap of size batch_size
// Requires: CanUseNgramIndex() == true
void
ExecuteNgramPhase2(TargetBitmap& candidates,
int64_t segment_offset,
int64_t batch_size);
private:
template <typename T>
VectorPtr
@@ -881,8 +908,11 @@ class PhyUnaryRangeFilterExpr : public SegmentExpr {
VectorPtr
ExecTextMatch();
// Check if ngram index exists
bool
CanUseNgramIndex() const override;
HasNgramIndex() const {
return pinned_ngram_index_.get() != nullptr;
}
std::optional<VectorPtr>
ExecNgramMatch(EvalCtx& context);
internal/core/src/index/NgramInvertedIndex.cpp
+281 -295
View File
@@ -273,134 +273,68 @@ NgramInvertedIndex::Load(milvus::tracer::TraceContext ctx,
"load ngram index done for field id:{} with dir:{}", field_id_, path_);
}
std::optional<TargetBitmap>
NgramInvertedIndex::ExecuteQuery(const std::string& literal,
proto::plan::OpType op_type,
exec::SegmentExpr* segment,
const TargetBitmap* pre_filter) {
tracer::AutoSpan span(
"NgramInvertedIndex::ExecuteQuery", tracer::GetRootSpan(), true);
if (pre_filter != nullptr && pre_filter->none()) {
return TargetBitmap(pre_filter->size(), false);
std::vector<std::string>
split_by_wildcard(const std::string& literal) {
std::vector<std::string> result;
std::string r;
r.reserve(literal.size());
bool escape_mode = false;
for (char c : literal) {
if (escape_mode) {
r += c;
escape_mode = false;
} else {
if (c == '\\') {
// consider case "\\%", we should reserve %
escape_mode = true;
} else if (c == '%' || c == '_') {
if (r.length() > 0) {
result.push_back(r);
r.clear();
}
} else {
r += c;
}
}
}
if (literal.length() < min_gram_) {
return std::nullopt;
}
if (Count() == 0) {
return TargetBitmap{};
if (r.length() > 0) {
result.push_back(r);
}
return result;
}
bool
NgramInvertedIndex::CanHandleLiteral(const std::string& literal,
proto::plan::OpType op_type) const {
switch (op_type) {
case proto::plan::OpType::Match:
return MatchQuery(literal, segment, pre_filter);
case proto::plan::OpType::InnerMatch: {
span.GetSpan()->SetAttribute("op_type", "InnerMatch");
span.GetSpan()->SetAttribute("query_literal_length",
static_cast<int>(literal.length()));
span.GetSpan()->SetAttribute("min_gram", min_gram_);
span.GetSpan()->SetAttribute("max_gram", max_gram_);
bool need_post_filter = literal.length() > max_gram_;
if (schema_.data_type() == proto::schema::DataType::JSON) {
auto predicate = [&literal, this](const milvus::Json& data) {
auto x =
data.template at<std::string_view>(this->nested_path_);
if (x.error()) {
return false;
}
auto data_val = x.value();
return data_val.find(literal) != std::string::npos;
};
return ExecuteQueryWithPredicate<milvus::Json>(
literal, segment, predicate, need_post_filter, pre_filter);
} else {
auto predicate = [&literal](const std::string_view& data) {
return data.find(literal) != std::string::npos;
};
return ExecuteQueryWithPredicate<std::string_view>(
literal, segment, predicate, need_post_filter, pre_filter);
case proto::plan::OpType::Match: {
// For Match (LIKE pattern), check all parts after splitting by wildcard
auto literals = split_by_wildcard(literal);
if (literals.empty()) {
return false;
}
}
case proto::plan::OpType::PrefixMatch: {
span.GetSpan()->SetAttribute("op_type", "PrefixMatch");
span.GetSpan()->SetAttribute("query_literal_length",
static_cast<int>(literal.length()));
span.GetSpan()->SetAttribute("min_gram", min_gram_);
span.GetSpan()->SetAttribute("max_gram", max_gram_);
if (schema_.data_type() == proto::schema::DataType::JSON) {
auto predicate = [&literal, this](const milvus::Json& data) {
auto x =
data.template at<std::string_view>(this->nested_path_);
if (x.error()) {
return false;
}
auto data_val = x.value();
return data_val.length() >= literal.length() &&
std::equal(literal.begin(),
literal.end(),
data_val.begin());
};
return ExecuteQueryWithPredicate<milvus::Json>(
literal, segment, predicate, true, pre_filter);
} else {
auto predicate = [&literal](const std::string_view& data) {
return data.length() >= literal.length() &&
std::equal(
literal.begin(), literal.end(), data.begin());
};
return ExecuteQueryWithPredicate<std::string_view>(
literal, segment, predicate, true, pre_filter);
}
}
case proto::plan::OpType::PostfixMatch: {
span.GetSpan()->SetAttribute("op_type", "PostfixMatch");
span.GetSpan()->SetAttribute("query_literal_length",
static_cast<int>(literal.length()));
span.GetSpan()->SetAttribute("min_gram", min_gram_);
span.GetSpan()->SetAttribute("max_gram", max_gram_);
if (schema_.data_type() == proto::schema::DataType::JSON) {
auto predicate = [&literal, this](const milvus::Json& data) {
auto x =
data.template at<std::string_view>(this->nested_path_);
if (x.error()) {
return false;
}
auto data_val = x.value();
return data_val.length() >= literal.length() &&
std::equal(literal.rbegin(),
literal.rend(),
data_val.rbegin());
};
return ExecuteQueryWithPredicate<milvus::Json>(
literal, segment, predicate, true, pre_filter);
} else {
auto predicate = [&literal](const std::string_view& data) {
return data.length() >= literal.length() &&
std::equal(
literal.rbegin(), literal.rend(), data.rbegin());
};
return ExecuteQueryWithPredicate<std::string_view>(
literal, segment, predicate, true, pre_filter);
for (const auto& l : literals) {
if (l.length() < min_gram_) {
return false;
}
}
return true;
}
case proto::plan::OpType::InnerMatch:
case proto::plan::OpType::PrefixMatch:
case proto::plan::OpType::PostfixMatch:
return literal.length() >= min_gram_;
default:
LOG_WARN("unsupported op type for ngram index: {}", op_type);
return std::nullopt;
return false;
}
}
template <typename T, typename Predicate>
inline void
handle_batch(const T* data,
const int64_t size,
TargetBitmapView res,
Predicate&& predicate) {
apply_predicate_on_batch(const T* data,
const int64_t size,
TargetBitmapView res,
Predicate&& predicate) {
auto next_off_option = res.find_first();
while (next_off_option.has_value()) {
auto next_off = next_off_option.value();
@@ -456,218 +390,270 @@ NgramInvertedIndex::ShouldUseBatchStrategy(double pre_filter_hit_rate) const {
pre_filter_hit_rate > kPreFilterHitRateThreshold);
}
template <typename T, typename Predicate>
std::optional<TargetBitmap>
NgramInvertedIndex::ExecuteQueryWithPredicate(const std::string& literal,
exec::SegmentExpr* segment,
Predicate&& predicate,
bool need_post_filter,
const TargetBitmap* pre_filter) {
void
NgramInvertedIndex::ExecutePhase1(const std::string& literal,
proto::plan::OpType op_type,
TargetBitmap& candidates) {
tracer::AutoSpan span(
"NgramInvertedIndex::ExecutePhase1", tracer::GetRootSpan(), true);
auto total_count = static_cast<size_t>(Count());
AssertInfo(total_count > 0, "ExecutePhase1: total_count must be > 0");
AssertInfo(!candidates.empty(),
"ExecutePhase1: candidates must be non-empty");
AssertInfo(candidates.size() == total_count,
"ExecutePhase1: candidates size {} != total_count {}",
candidates.size(),
total_count);
AssertInfo(total_count > 0, "total_count should be greater than 0");
// Calculate pre_filter stats for strategy selection
size_t candidate_count = (pre_filter != nullptr && !pre_filter->empty())
? pre_filter->count()
: total_count;
double pre_filter_hit_rate = 1.0 * candidate_count / total_count;
// Phase 1: ngram index query with adaptive strategy
TargetBitmap bitset(total_count, true);
if (pre_filter != nullptr && !pre_filter->empty()) {
bitset &= *pre_filter;
// If candidates has no bits set, return immediately
if (candidates.none()) {
return;
}
if (ShouldUseBatchStrategy(pre_filter_hit_rate)) {
TargetBitmap ngram_bitset{total_count};
wrapper_->ngram_match_query(
literal, min_gram_, max_gram_, &ngram_bitset);
bitset &= ngram_bitset;
// Use candidates for strategy selection
size_t pre_count = candidates.count();
double candidates_hit_rate = 1.0 * pre_count / total_count;
// Get literals to query
std::vector<std::string> literals_vec;
if (op_type == proto::plan::OpType::Match) {
literals_vec = split_by_wildcard(literal);
AssertInfo(!literals_vec.empty(),
"ExecutePhase1: Match pattern must have non-empty parts");
for (const auto& l : literals_vec) {
AssertInfo(l.length() >= min_gram_,
"ExecutePhase1: part length {} < min_gram {}",
l.length(),
min_gram_);
}
} else {
std::vector<std::string> literals_vec = {literal};
AssertInfo(literal.length() >= min_gram_,
"ExecutePhase1: literal length {} < min_gram {}",
literal.length(),
min_gram_);
literals_vec.push_back(literal);
}
bool use_batch_strategy = ShouldUseBatchStrategy(candidates_hit_rate);
// Choose strategy and execute, AND results into candidates
if (use_batch_strategy) {
// Batch strategy: query all ngram terms at once
for (const auto& l : literals_vec) {
TargetBitmap ngram_bitset{total_count};
wrapper_->ngram_match_query(l, min_gram_, max_gram_, &ngram_bitset);
candidates &= ngram_bitset;
}
} else {
// Iterative strategy: query terms one by one, sorted by doc_freq
auto sorted_terms =
wrapper_->ngram_tokenize(literals_vec, min_gram_, max_gram_);
AssertInfo(!sorted_terms.empty(),
"ngram_tokenize should not return empty for valid literal");
ApplyIterativeNgramFilter(sorted_terms, total_count, bitset);
ApplyIterativeNgramFilter(sorted_terms, total_count, candidates);
}
if (bitset.none()) {
return std::move(bitset);
}
auto after_pre_filter_count = bitset.count();
auto final_result_count = after_pre_filter_count;
// Phase 2: post-filter
if (need_post_filter) {
TargetBitmapView res(bitset);
auto execute_batch = [&predicate](const T* data,
const int64_t size,
TargetBitmapView res) {
handle_batch(data, size, res, predicate);
};
segment->template ProcessAllDataChunkBatched<T>(execute_batch, res);
final_result_count = bitset.count();
}
// Set tracing attributes
if (auto root_span = tracer::GetRootSpan()) {
double hit_rate_before_phase2 =
1.0 * after_pre_filter_count / total_count;
double final_hit_rate = 1.0 * final_result_count / total_count;
root_span->SetAttribute("need_post_filter", need_post_filter);
root_span->SetAttribute("pre_filter_hit_rate", pre_filter_hit_rate);
root_span->SetAttribute("hit_rate_before_phase2",
hit_rate_before_phase2);
root_span->SetAttribute("final_hit_rate", final_hit_rate);
root_span->SetAttribute("total_count", static_cast<int>(total_count));
size_t post_count = candidates.count();
double pre_hit_rate = 1.0 * pre_count / total_count;
double post_hit_rate = 1.0 * post_count / total_count;
root_span->SetAttribute("phase1_op_type", static_cast<int>(op_type));
root_span->SetAttribute("phase1_literal_length",
static_cast<int>(literal.length()));
root_span->SetAttribute("phase1_total_count",
static_cast<int>(total_count));
root_span->SetAttribute("phase1_pre_hit_rate", pre_hit_rate);
root_span->SetAttribute("phase1_post_hit_rate", post_hit_rate);
root_span->SetAttribute("phase1_use_batch_strategy",
use_batch_strategy);
}
return std::optional<TargetBitmap>(std::move(bitset));
}
std::vector<std::string>
split_by_wildcard(const std::string& literal) {
std::vector<std::string> result;
std::string r;
r.reserve(literal.size());
bool escape_mode = false;
for (char c : literal) {
if (escape_mode) {
r += c;
escape_mode = false;
} else {
if (c == '\\') {
// consider case "\\%", we should reserve %
escape_mode = true;
} else if (c == '%' || c == '_') {
if (r.length() > 0) {
result.push_back(r);
r.clear();
}
} else {
r += c;
void
NgramInvertedIndex::ExecutePhase2(const std::string& literal,
proto::plan::OpType op_type,
exec::SegmentExpr* segment,
TargetBitmap& candidates,
int64_t segment_offset,
int64_t batch_size) {
// InnerMatch with short literal doesn't need post-filter
if (op_type == proto::plan::OpType::InnerMatch &&
literal.length() <= max_gram_) {
return;
}
if (candidates.none()) {
return;
}
AssertInfo(static_cast<int64_t>(candidates.size()) == batch_size,
"candidates size {} != batch_size {}",
candidates.size(),
batch_size);
size_t pre_count = candidates.count();
TargetBitmapView res(candidates);
if (schema_.data_type() == proto::schema::DataType::JSON) {
// JSON type handling
auto apply_predicate = [&](auto&& predicate) {
auto execute_batch = [&predicate](const milvus::Json* data,
const int64_t size,
TargetBitmapView res) {
apply_predicate_on_batch<milvus::Json>(
data, size, res, predicate);
};
segment->template ProcessDataChunkForRange<milvus::Json>(
execute_batch, res, segment_offset, batch_size);
};
switch (op_type) {
case proto::plan::OpType::InnerMatch: {
apply_predicate([&literal, this](const milvus::Json& data) {
auto x =
data.template at<std::string_view>(this->nested_path_);
if (x.error()) {
return false;
}
return x.value().find(literal) != std::string::npos;
});
break;
}
case proto::plan::OpType::PrefixMatch: {
apply_predicate([&literal, this](const milvus::Json& data) {
auto x =
data.template at<std::string_view>(this->nested_path_);
if (x.error()) {
return false;
}
auto data_val = x.value();
return data_val.length() >= literal.length() &&
std::equal(literal.begin(),
literal.end(),
data_val.begin());
});
break;
}
case proto::plan::OpType::PostfixMatch: {
apply_predicate([&literal, this](const milvus::Json& data) {
auto x =
data.template at<std::string_view>(this->nested_path_);
if (x.error()) {
return false;
}
auto data_val = x.value();
return data_val.length() >= literal.length() &&
std::equal(literal.rbegin(),
literal.rend(),
data_val.rbegin());
});
break;
}
case proto::plan::OpType::Match: {
PatternMatchTranslator translator;
auto regex_pattern = translator(literal);
RegexMatcher matcher(regex_pattern);
apply_predicate([&matcher, this](const milvus::Json& data) {
auto x =
data.template at<std::string_view>(this->nested_path_);
if (x.error()) {
return false;
}
return matcher(x.value());
});
break;
}
default:
break;
}
} else {
// String/Varchar type handling
auto apply_predicate = [&](auto&& predicate) {
auto execute_batch = [&predicate](const std::string_view* data,
const int64_t size,
TargetBitmapView res) {
apply_predicate_on_batch<std::string_view>(
data, size, res, predicate);
};
segment->template ProcessDataChunkForRange<std::string_view>(
execute_batch, res, segment_offset, batch_size);
};
switch (op_type) {
case proto::plan::OpType::InnerMatch: {
apply_predicate([&literal](const std::string_view& data) {
return data.find(literal) != std::string::npos;
});
break;
}
case proto::plan::OpType::PrefixMatch: {
apply_predicate([&literal](const std::string_view& data) {
return data.length() >= literal.length() &&
std::equal(
literal.begin(), literal.end(), data.begin());
});
break;
}
case proto::plan::OpType::PostfixMatch: {
apply_predicate([&literal](const std::string_view& data) {
return data.length() >= literal.length() &&
std::equal(
literal.rbegin(), literal.rend(), data.rbegin());
});
break;
}
case proto::plan::OpType::Match: {
PatternMatchTranslator translator;
auto regex_pattern = translator(literal);
RegexMatcher matcher(regex_pattern);
apply_predicate([&matcher](const std::string_view& data) {
return matcher(data);
});
break;
}
default:
break;
}
}
if (r.length() > 0) {
result.push_back(r);
}
return result;
}
std::optional<TargetBitmap>
NgramInvertedIndex::MatchQuery(const std::string& literal,
exec::SegmentExpr* segment,
const TargetBitmap* pre_filter) {
if (auto root_span = tracer::GetRootSpan()) {
root_span->SetAttribute("match_query_literal_length",
static_cast<int>(literal.length()));
root_span->SetAttribute("match_query_min_gram", min_gram_);
root_span->SetAttribute("match_query_max_gram", max_gram_);
NgramInvertedIndex::ExecuteQueryForUT(const std::string& literal,
proto::plan::OpType op_type,
exec::SegmentExpr* segment,
const TargetBitmap* pre_filter) {
if (!CanHandleLiteral(literal, op_type)) {
return std::nullopt;
}
auto total_count = static_cast<size_t>(Count());
AssertInfo(total_count > 0, "total_count should be greater than 0");
if (total_count == 0) {
return TargetBitmap{};
}
auto literals = split_by_wildcard(literal);
for (const auto& l : literals) {
if (l.length() < min_gram_) {
return std::nullopt;
// Initialize candidates: start with all-true, then AND with pre_filter
TargetBitmap candidates(total_count, true);
if (pre_filter != nullptr) {
candidates &= *pre_filter;
if (candidates.none()) {
return std::move(candidates);
}
}
// Calculate pre_filter stats for strategy selection
size_t candidate_count = (pre_filter != nullptr && !pre_filter->empty())
? pre_filter->count()
: total_count;
double pre_filter_hit_rate = 1.0 * candidate_count / total_count;
// Phase 1: ngram index query
ExecutePhase1(literal, op_type, candidates);
TargetBitmap bitset(total_count, true);
if (pre_filter != nullptr && !pre_filter->empty()) {
bitset &= *pre_filter;
if (candidates.none()) {
return std::move(candidates);
}
if (ShouldUseBatchStrategy(pre_filter_hit_rate)) {
for (const auto& l : literals) {
TargetBitmap tmp_bitset{total_count};
wrapper_->ngram_match_query(l, min_gram_, max_gram_, &tmp_bitset);
bitset &= tmp_bitset;
}
} else {
auto sorted_terms =
wrapper_->ngram_tokenize(literals, min_gram_, max_gram_);
AssertInfo(!sorted_terms.empty(),
"ngram_tokenize returned empty sorted_terms for iterative "
"strategy");
ApplyIterativeNgramFilter(sorted_terms, total_count, bitset);
}
// Phase 2: post-filter verification (full segment)
ExecutePhase2(literal, op_type, segment, candidates, 0, total_count);
if (bitset.none()) {
return std::move(bitset);
}
auto after_pre_filter_count = bitset.count();
TargetBitmapView res(bitset);
PatternMatchTranslator translator;
auto regex_pattern = translator(literal);
RegexMatcher matcher(regex_pattern);
if (schema_.data_type() == proto::schema::DataType::JSON) {
auto predicate = [&matcher, this](const milvus::Json& data) {
auto x = data.template at<std::string_view>(this->nested_path_);
if (x.error()) {
return false;
}
return matcher(x.value());
};
auto execute_batch = [&predicate](const milvus::Json* data,
const int64_t size,
TargetBitmapView res) {
handle_batch<milvus::Json>(data, size, res, predicate);
};
segment->template ProcessAllDataChunkBatched<milvus::Json>(
execute_batch, res);
} else {
auto predicate = [&matcher](const std::string_view& data) {
return matcher(data);
};
auto execute_batch = [&predicate](const std::string_view* data,
const int64_t size,
TargetBitmapView res) {
handle_batch<std::string_view>(data, size, res, predicate);
};
segment->template ProcessAllDataChunkBatched<std::string_view>(
execute_batch, res);
}
auto final_result_count = bitset.count();
if (auto root_span = tracer::GetRootSpan()) {
double hit_rate_before_phase2 =
1.0 * after_pre_filter_count / total_count;
double final_hit_rate = 1.0 * final_result_count / total_count;
root_span->SetAttribute("match_pre_filter_hit_rate",
pre_filter_hit_rate);
root_span->SetAttribute("match_hit_rate_before_phase2",
hit_rate_before_phase2);
root_span->SetAttribute("match_final_hit_rate", final_hit_rate);
root_span->SetAttribute("match_total_count",
static_cast<int>(total_count));
}
return std::optional<TargetBitmap>(std::move(bitset));
return std::optional<TargetBitmap>(std::move(candidates));
}
} // namespace milvus::index
internal/core/src/index/NgramInvertedIndex.h
+32 -19
View File
@@ -55,11 +55,39 @@ class NgramInvertedIndex : public InvertedIndexTantivy<std::string> {
void
BuildWithJsonFieldData(const std::vector<FieldDataPtr>& datas);
// For unit tests only - combines Phase1 and Phase2 in one call
std::optional<TargetBitmap>
ExecuteQuery(const std::string& literal,
proto::plan::OpType op_type,
exec::SegmentExpr* segment,
const TargetBitmap* pre_filter = nullptr);
ExecuteQueryForUT(const std::string& literal,
proto::plan::OpType op_type,
exec::SegmentExpr* segment,
const TargetBitmap* pre_filter = nullptr);
// Check if literal can be handled by ngram index (length >= min_gram)
bool
CanHandleLiteral(const std::string& literal,
proto::plan::OpType op_type) const;
// Phase1: Execute ngram index query, AND-merge result into candidates
// Requires: CanHandleLiteral(literal, op_type) == true
// Requires: candidates must be non-empty (caller initializes it)
// Ngram query result is AND-merged with existing candidates.
void
ExecutePhase1(const std::string& literal,
proto::plan::OpType op_type,
TargetBitmap& candidates);
// Phase2: Execute post-filter verification on a specific range
// - segment_offset: starting position in segment
// - batch_size: number of rows to process
// - candidates: bitmap of size batch_size (relative to the range)
// Requires: CanHandleLiteral(literal, op_type) == true
void
ExecutePhase2(const std::string& literal,
proto::plan::OpType op_type,
exec::SegmentExpr* segment,
TargetBitmap& candidates,
int64_t segment_offset,
int64_t batch_size);
ScalarIndexType
GetIndexType() const override {
@@ -77,20 +105,6 @@ class NgramInvertedIndex : public InvertedIndexTantivy<std::string> {
}
private:
template <typename T, typename Predicate>
std::optional<TargetBitmap>
ExecuteQueryWithPredicate(const std::string& literal,
exec::SegmentExpr* segment,
Predicate&& predicate,
bool need_post_filter,
const TargetBitmap* pre_filter);
// Match is something like xxx%xxx%xxx, xxx%xxx, %xxx%xxx, xxx_x etc.
std::optional<TargetBitmap>
MatchQuery(const std::string& literal,
exec::SegmentExpr* segment,
const TargetBitmap* pre_filter);
void
ApplyIterativeNgramFilter(const std::vector<std::string>& sorted_terms,
size_t total_count,
@@ -99,7 +113,6 @@ class NgramInvertedIndex : public InvertedIndexTantivy<std::string> {
bool
ShouldUseBatchStrategy(double pre_filter_hit_rate) const;
private:
uintptr_t min_gram_{0};
uintptr_t max_gram_{0};
int64_t field_id_{0};
internal/core/src/index/NgramInvertedIndexTest.cpp
+1 -1
View File
@@ -153,7 +153,7 @@ test_ngram_with_data(const boost::container::vector<std::string>& data,
0);
if (op_type != proto::plan::OpType::Equal) {
std::optional<TargetBitmap> bitset_opt =
index->ExecuteQuery(literal, op_type, &segment_expr);
index->ExecuteQueryForUT(literal, op_type, &segment_expr);
if (forward_to_br) {
ASSERT_TRUE(!bitset_opt.has_value());
} else {