From f022540ccce00621aa008a59753e9588d46c1862 Mon Sep 17 00:00:00 2001 From: Spade A <71589810+SpadeA-Tang@users.noreply.github.com> Date: Wed, 21 Jan 2026 15:21:30 +0800 Subject: [PATCH] enhance: optimize ngram inverted index by executing phase 1/2 in batch expressions (#46999) 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 image After this PR image --------- Signed-off-by: SpadeA --- .../core/src/exec/expression/ConjunctExpr.cpp | 61 +- .../core/src/exec/expression/ConjunctExpr.h | 40 +- internal/core/src/exec/expression/Expr.cpp | 108 +++- internal/core/src/exec/expression/Expr.h | 39 +- .../src/exec/expression/LikeConjunctExpr.cpp | 103 ++++ .../src/exec/expression/LikeConjunctExpr.h | 62 ++ .../exec/expression/LikeConjunctExprTest.cpp | 332 ++++++++++ .../core/src/exec/expression/UnaryExpr.cpp | 95 ++- internal/core/src/exec/expression/UnaryExpr.h | 32 +- .../core/src/index/NgramInvertedIndex.cpp | 576 +++++++++--------- internal/core/src/index/NgramInvertedIndex.h | 51 +- .../core/src/index/NgramInvertedIndexTest.cpp | 2 +- 12 files changed, 1134 insertions(+), 367 deletions(-) create mode 100644 internal/core/src/exec/expression/LikeConjunctExpr.cpp create mode 100644 internal/core/src/exec/expression/LikeConjunctExpr.h create mode 100644 internal/core/src/exec/expression/LikeConjunctExprTest.cpp diff --git a/internal/core/src/exec/expression/ConjunctExpr.cpp b/internal/core/src/exec/expression/ConjunctExpr.cpp index 26666002d2..a794914321 100644 --- a/internal/core/src/exec/expression/ConjunctExpr.cpp +++ b/internal/core/src/exec/expression/ConjunctExpr.cpp @@ -15,6 +15,10 @@ // limitations under the License. #include "ConjunctExpr.h" +#include "UnaryExpr.h" +#include "LikeConjunctExpr.h" + +#include 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> ngram_exprs; + for (size_t idx : like_indices_) { + auto unary_expr = + std::dynamic_pointer_cast( + 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( + 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); diff --git a/internal/core/src/exec/expression/ConjunctExpr.h b/internal/core/src/exec/expression/ConjunctExpr.h index 249db77637..62f05fd8b1 100644 --- a/internal/core/src/exec/expression/ConjunctExpr.h +++ b/internal/core/src/exec/expression/ConjunctExpr.h @@ -17,6 +17,7 @@ #pragma once #include +#include #include "common/EasyAssert.h" #include "common/OpContext.h" @@ -118,15 +119,25 @@ class PhyConjunctFilterExpr : public Expr { ToString() const { if (!input_order_.empty()) { std::vector 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 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) { + 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&& indices) { + like_indices_ = std::move(indices); + } + + const std::vector& + 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 input_order_; + // Indices of LIKE expressions for potential batch ngram optimization (AND only) + std::vector 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 batch_ngram_indices_; }; } //namespace exec } // namespace milvus diff --git a/internal/core/src/exec/expression/Expr.cpp b/internal/core/src/exec/expression/Expr.cpp index 58ba20e168..1e2e610fd7 100644 --- a/internal/core/src/exec/expression/Expr.cpp +++ b/internal/core/src/exec/expression/Expr.cpp @@ -138,19 +138,75 @@ CompileExpressions(const std::vector& sources, return exprs; } +static std::optional +ShouldFlatten(const expr::TypedExprPtr& expr, + const std::unordered_set& flat_candidates = {}) { + if (auto call = + std::dynamic_pointer_cast(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(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& 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 CompileInputs(const expr::TypedExprPtr& expr, QueryContext* context, const std::unordered_set& flatten_cadidates) { std::vector compiled_inputs; + auto flatten = ShouldFlatten(expr); for (auto& input : expr->inputs()) { if (dynamic_cast(input.get())) { AssertInfo( dynamic_cast(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 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& expr, std::vector other_expr; std::vector heavy_conjunct_expr; std::vector light_conjunct_expr; + // Record all LIKE expression indices for potential batch ngram optimization + std::vector 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& 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& 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& 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& expr, if (input->name() == "PhyConjunctFilterExpr") { bool sub_expr_heavy = false; - auto expr = std::static_pointer_cast(input); - ReorderConjunctExpr(expr, context, sub_expr_heavy); + auto sub_expr = + std::static_pointer_cast(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& 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& 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); } diff --git a/internal/core/src/exec/expression/Expr.h b/internal/core/src/exec/expression/Expr.h index bb917e06ca..febd554056 100644 --- a/internal/core/src/exec/expression/Expr.h +++ b/internal/core/src/exec/expression/Expr.h @@ -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 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 || std::is_same_v || std::is_same_v, - "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( 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 cached_match_res_{nullptr}; int32_t consistency_level_{0}; - // Cache for ngram match. - std::shared_ptr cached_ngram_match_res_{nullptr}; + // Cache for ngram Phase1 result (index query, before post-filter). + std::shared_ptr cached_phase1_res_{nullptr}; }; bool diff --git a/internal/core/src/exec/expression/LikeConjunctExpr.cpp b/internal/core/src/exec/expression/LikeConjunctExpr.cpp new file mode 100644 index 0000000000..bfad8c9bf5 --- /dev/null +++ b/internal/core/src/exec/expression/LikeConjunctExpr.cpp @@ -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> 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(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(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(std::move(batch_candidates), + std::move(valid_result)); +} +} // namespace exec +} // namespace milvus diff --git a/internal/core/src/exec/expression/LikeConjunctExpr.h b/internal/core/src/exec/expression/LikeConjunctExpr.h new file mode 100644 index 0000000000..3a88cbce35 --- /dev/null +++ b/internal/core/src/exec/expression/LikeConjunctExpr.h @@ -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 +#include + +#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> 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> ngram_exprs_; + // Cached Phase1 result (segment-level ngram index query result) + std::shared_ptr cached_phase1_res_{nullptr}; + + int64_t active_count_{0}; + int64_t batch_size_{0}; + int64_t current_pos_{0}; +}; + +} // namespace exec +} // namespace milvus diff --git a/internal/core/src/exec/expression/LikeConjunctExprTest.cpp b/internal/core/src/exec/expression/LikeConjunctExprTest.cpp new file mode 100644 index 0000000000..608abaa095 --- /dev/null +++ b/internal/core/src/exec/expression/LikeConjunctExprTest.cpp @@ -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 +#include +#include +#include +#include + +#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 +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(); + 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 pk_data = {0, 1, 2, 3, 4, 5, 6, 7}; + boost::container::vector 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 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 row_ids(nb); + std::vector 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& 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(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{log_path}; + + auto ngram_params = index::NgramParams{ + .loading_index = false, + .min_gram = 2, + .max_gram = 4, + }; + auto index = + std::make_shared(ctx, ngram_params); + index->Build(config); + auto create_index_result = index->Upload(); + auto index_files = create_index_result->GetIndexFiles(); + + std::map 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(&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::ColumnInfo(title_fid, DataType::VARCHAR), + proto::plan::OpType::InnerMatch, + val_database, + std::vector{}); + + auto expr_title_design = + std::make_shared( + milvus::expr::ColumnInfo(title_fid, DataType::VARCHAR), + proto::plan::OpType::InnerMatch, + val_design, + std::vector{}); + + auto expr_content_sql = + std::make_shared( + milvus::expr::ColumnInfo(content_fid, DataType::VARCHAR), + proto::plan::OpType::InnerMatch, + val_sql, + std::vector{}); + + auto expr_content_query = + std::make_shared( + milvus::expr::ColumnInfo(content_fid, DataType::VARCHAR), + proto::plan::OpType::InnerMatch, + val_query, + std::vector{}); + + // Create nested AND expression: ((A AND B) AND C) AND D + auto and_expr_1 = std::make_shared( + milvus::expr::LogicalBinaryExpr::OpType::And, + expr_title_database, + expr_title_design); + auto and_expr_2 = std::make_shared( + milvus::expr::LogicalBinaryExpr::OpType::And, + and_expr_1, + expr_content_sql); + auto and_expr = std::make_shared( + milvus::expr::LogicalBinaryExpr::OpType::And, + and_expr_2, + expr_content_query); + + // Create RetrievePlan + auto plan = std::make_unique(schema); + plan->plan_node_ = std::make_unique(); + 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 expected_pks = {0, 2, 6}; + std::set 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); +} diff --git a/internal/core/src/exec/expression/UnaryExpr.cpp b/internal/core/src/exec/expression/UnaryExpr.cpp index 10557e2197..228208e512 100644 --- a/internal/core/src/exec/expression/UnaryExpr.cpp +++ b/internal/core/src/exec/expression/UnaryExpr.cpp @@ -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(expr_->val_); + return pinned_ngram_index_.get()->CanHandleLiteral(literal, + expr_->op_type_); +} + +void +PhyUnaryRangeFilterExpr::ExecuteNgramPhase1(TargetBitmap& candidates) { + if (!arg_inited_) { + value_arg_.SetValue(expr_->val_); + arg_inited_ = true; + } + + auto literal = value_arg_.GetValue(); + 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(expr_->val_); + arg_inited_ = true; + } + + auto literal = value_arg_.GetValue(); + 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 @@ -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(std::move(res_opt.value())); + + auto total_count = static_cast(index->Count()); + TargetBitmap candidates(total_count, true); + index->ExecutePhase1(literal, expr_->op_type_, candidates); + cached_phase1_res_ = + std::make_shared(std::move(candidates)); cached_index_chunk_valid_res_ = std::make_shared(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(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(std::move(result), + return std::make_shared(std::move(batch_candidates), std::move(valid_result)); } diff --git a/internal/core/src/exec/expression/UnaryExpr.h b/internal/core/src/exec/expression/UnaryExpr.h index 03232ae1ea..50a2148589 100644 --- a/internal/core/src/exec/expression/UnaryExpr.h +++ b/internal/core/src/exec/expression/UnaryExpr.h @@ -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 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 ExecNgramMatch(EvalCtx& context); diff --git a/internal/core/src/index/NgramInvertedIndex.cpp b/internal/core/src/index/NgramInvertedIndex.cpp index b8cfc01806..802057bc6c 100644 --- a/internal/core/src/index/NgramInvertedIndex.cpp +++ b/internal/core/src/index/NgramInvertedIndex.cpp @@ -273,134 +273,68 @@ NgramInvertedIndex::Load(milvus::tracer::TraceContext ctx, "load ngram index done for field id:{} with dir:{}", field_id_, path_); } -std::optional -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 +split_by_wildcard(const std::string& literal) { + std::vector 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(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(this->nested_path_); - if (x.error()) { - return false; - } - auto data_val = x.value(); - return data_val.find(literal) != std::string::npos; - }; - - return ExecuteQueryWithPredicate( - 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( - 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(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(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( - 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( - 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(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(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( - 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( - 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 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 -std::optional -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(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 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 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(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(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(op_type)); + root_span->SetAttribute("phase1_literal_length", + static_cast(literal.length())); + root_span->SetAttribute("phase1_total_count", + static_cast(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(std::move(bitset)); } -std::vector -split_by_wildcard(const std::string& literal) { - std::vector 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(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( + data, size, res, predicate); + }; + segment->template ProcessDataChunkForRange( + 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(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(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(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(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( + data, size, res, predicate); + }; + segment->template ProcessDataChunkForRange( + 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 -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(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(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(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(data, size, res, predicate); - }; - - segment->template ProcessAllDataChunkBatched( - 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(data, size, res, predicate); - }; - - segment->template ProcessAllDataChunkBatched( - 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(total_count)); - } - - return std::optional(std::move(bitset)); + return std::optional(std::move(candidates)); } } // namespace milvus::index diff --git a/internal/core/src/index/NgramInvertedIndex.h b/internal/core/src/index/NgramInvertedIndex.h index d256e0e855..03adfd2483 100644 --- a/internal/core/src/index/NgramInvertedIndex.h +++ b/internal/core/src/index/NgramInvertedIndex.h @@ -55,11 +55,39 @@ class NgramInvertedIndex : public InvertedIndexTantivy { void BuildWithJsonFieldData(const std::vector& datas); + // For unit tests only - combines Phase1 and Phase2 in one call std::optional - 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 { } private: - template - std::optional - 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 - MatchQuery(const std::string& literal, - exec::SegmentExpr* segment, - const TargetBitmap* pre_filter); - void ApplyIterativeNgramFilter(const std::vector& sorted_terms, size_t total_count, @@ -99,7 +113,6 @@ class NgramInvertedIndex : public InvertedIndexTantivy { bool ShouldUseBatchStrategy(double pre_filter_hit_rate) const; - private: uintptr_t min_gram_{0}; uintptr_t max_gram_{0}; int64_t field_id_{0}; diff --git a/internal/core/src/index/NgramInvertedIndexTest.cpp b/internal/core/src/index/NgramInvertedIndexTest.cpp index 6f980b6dbd..c83a571bf5 100644 --- a/internal/core/src/index/NgramInvertedIndexTest.cpp +++ b/internal/core/src/index/NgramInvertedIndexTest.cpp @@ -153,7 +153,7 @@ test_ngram_with_data(const boost::container::vector& data, 0); if (op_type != proto::plan::OpType::Equal) { std::optional 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 {