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 {