[improvement](inverted index) Push the candidate row bitmap down into phrase queries

be/src/common/config.cpp

@@ -25,6 +25,7 @@
 #include <lz4/lz4hc.h>
 
 #include <cerrno> // IWYU pragma: keep
+#include <cmath>
 #include <cstdlib>
 #include <cstring>
 #include <fstream> // IWYU pragma: keep
@@ -1319,6 +1320,9 @@ DEFINE_Bool(enable_inverted_index_cache_check_timestamp, "true");
 DEFINE_mBool(enable_inverted_index_correct_term_write, "true");
 DEFINE_Int32(inverted_index_fd_number_limit_percent, "20"); // 20%
 DEFINE_Int32(inverted_index_query_cache_shards, "256");
+DEFINE_mDouble(inverted_index_candidate_pushdown_ratio, "0.3");
+DEFINE_Validator(inverted_index_candidate_pushdown_ratio,
+                 [](const double v) -> bool { return std::isfinite(v) && v <= 1.0; });
 
 // inverted index match bitmap cache size
 DEFINE_String(inverted_index_query_cache_limit, "10%");

be/src/common/config.h

@@ -1372,6 +1372,11 @@ DECLARE_Bool(enable_inverted_index_cache_check_timestamp);
 DECLARE_mBool(enable_inverted_index_correct_term_write);
 DECLARE_Int32(inverted_index_fd_number_limit_percent); // 50%
 DECLARE_Int32(inverted_index_query_cache_shards);
+// When the candidate row bitmap of a segment scan is smaller than
+// num_rows * this ratio, it is pushed down into inverted index queries so
+// doc-list intersection and verification run only over the candidates
+// (see IndexQueryContext::candidate_rows). <= 0 disables the pushdown.
+DECLARE_mDouble(inverted_index_candidate_pushdown_ratio);
 
 // inverted index match bitmap cache size
 DECLARE_String(inverted_index_query_cache_limit);

be/src/storage/index/index_query_context.h

@@ -48,6 +48,17 @@ struct IndexQueryContext {
     // under a key a row-accurate query could hit.
     bool count_on_index_fastpath = false;
 
+    // Candidate-pushdown handshake. Set by SegmentIterator ONLY while it
+    // evaluates pushed-down conjuncts and the current candidate row bitmap is
+    // small enough (config::inverted_index_candidate_pushdown_ratio), reset
+    // right after. When set, an index query MAY restrict doc-list intersection
+    // and verification to this candidate set (PhraseQuery joins it into the
+    // leapfrog). A bitmap produced under a non-null candidate is PARTIAL and
+    // must never be inserted into the query cache; cache lookups stay valid
+    // (a cached full-segment bitmap intersected later is still correct). The
+    // pointee is owned by the caller and outlives the evaluation.
+    const roaring::Roaring* candidate_rows = nullptr;
+
     // ---- Reply direction: fields a READER writes and the CALLER reads back ----
     //
     // A caller that hands a reader a COPY of this context rather than the context itself must
@@ -66,10 +77,19 @@ struct IndexQueryContext {
     // remaining count as default rows without iterating the row bitmap.
     bool count_on_index_fastpath_hit = false;
 
+    // Reply direction of the candidate handshake. Set by a query iff it DID
+    // join candidate_rows into its evaluation (PhraseQuery's leapfrog), i.e.
+    // its result bitmap is partial; reset by the reader before each search.
+    // Only such a partial result must stay out of the query cache -- a query
+    // that never consumes the candidate (MATCH_ANY/ALL, term, regexp, single
+    // term phrase) still computes the full-segment bitmap and stays cacheable.
+    bool candidate_rows_consumed = false;
+
     // Folds the reply-direction fields a reader wrote on a copy of this context back into it.
     // Latching (never clearing) is what makes this safe to call for each of several readers.
     void merge_reader_outputs(const IndexQueryContext& reader_context) {
         count_on_index_fastpath_hit |= reader_context.count_on_index_fastpath_hit;
+        candidate_rows_consumed |= reader_context.candidate_rows_consumed;
     }
 };
 using IndexQueryContextPtr = std::shared_ptr<IndexQueryContext>;

be/src/storage/index/inverted/inverted_index_reader.cpp

@@ -347,6 +347,10 @@ Status InvertedIndexReader::match_index_search(
         context->runtime_state->query_options().inverted_index_compatible_read) {
         reader->setCompatibleRead(true);
     }
+    // Fresh per-search reply: only the query about to run decides whether it
+    // consumes the candidate set (and thus produces an uncacheable partial
+    // result); a consumed flag left by an earlier search must not leak in.
+    context->candidate_rows_consumed = false;
     try {
         SCOPED_RAW_TIMER(&context->stats->inverted_index_searcher_search_timer);
         auto query = QueryFactory::create(query_type, index_searcher, context);
@@ -477,7 +481,13 @@ Status FullTextIndexReader::query(const IndexQueryContextPtr& context,
             RETURN_IF_ERROR(match_index_search(context, query_type, query_info, *searcher_ptr,
                                                term_match_bitmap));
             term_match_bitmap->runOptimize();
-            cache->insert(cache_key, term_match_bitmap, &cache_handler);
+            // Only a bitmap whose query actually joined the candidate set is
+            // partial and must stay out of the cache; a non-consuming query
+            // (MATCH_ANY/ALL, term, regexp, single-term phrase) computed the
+            // full-segment result even while candidate_rows was published.
+            if (!context->candidate_rows_consumed) {
+                cache->insert(cache_key, term_match_bitmap, &cache_handler);
+            }
             bit_map = term_match_bitmap;
         }
         return Status::OK();
@@ -540,6 +550,10 @@ Status StringTypeInvertedIndexReader::query(const IndexQueryContextPtr& context,
         query_info.field_name = column_name_ws;
         query_info.term_infos.emplace_back(search_str, 0);
 
+        // Fresh per-search reply (the range-query cases below never pass
+        // through match_index_search, so a stale consumed flag from an
+        // earlier fulltext search must be cleared here too).
+        context->candidate_rows_consumed = false;
         auto result = std::make_shared<roaring::Roaring>();
         FulltextIndexSearcherPtr* searcher_ptr = nullptr;
         InvertedIndexCacheHandle inverted_index_cache_handle;
@@ -598,9 +612,11 @@ Status StringTypeInvertedIndexReader::query(const IndexQueryContextPtr& context,
                         "invalid query type when query untokenized inverted index");
             }
         }
-        // add to cache
+        // add to cache (unless a candidate-consuming query made it partial)
         result->runOptimize();
-        cache->insert(cache_key, result, &cache_handler);
+        if (!context->candidate_rows_consumed) {
+            cache->insert(cache_key, result, &cache_handler);
+        }
 
         bit_map = result;
         return Status::OK();

be/src/storage/index/inverted/query/phrase_query.cpp

@@ -53,6 +53,15 @@ void PhraseQuery::add(const InvertedIndexQueryInfo& query_info) {
         init_ordered_sloppy_phrase_matcher(query_info, is_similarity);
     }
 
+    // Two-phase evaluation with a pushed-down candidate set: the candidate
+    // bitmap joins the leapfrog intersection (restricting doc-list walking and
+    // position verification to candidates) but never a matcher's postings, so
+    // phrase semantics stay with the real term iterators.
+    if (_context->candidate_rows != nullptr) {
+        _iterators.emplace_back(std::make_shared<RoaringDocIdIterator>(_context->candidate_rows));
+        _context->candidate_rows_consumed = true;
+    }
+
     std::sort(_iterators.begin(), _iterators.end(), [](const DISI& a, const DISI& b) {
         int64_t freq1 = visit_node(a, DocFreq {});
         int64_t freq2 = visit_node(b, DocFreq {});
@@ -64,6 +73,12 @@ void PhraseQuery::add(const InvertedIndexQueryInfo& query_info) {
     for (int32_t i = 2; i < _iterators.size(); i++) {
         _others.emplace_back(&_iterators[i]);
     }
+    for (auto& iter : _iterators) {
+        if (const auto* term_iter = std::get_if<TermPositionsIterPtr>(&iter)) {
+            _norm_source = term_iter->get();
+            break;
+        }
+    }
 
     init_similarities(query_info.field_name, is_similarity);
 }
@@ -160,6 +175,11 @@ void PhraseQuery::search(roaring::Roaring& roaring) {
 }
 
 void PhraseQuery::search_by_skiplist(roaring::Roaring& roaring) {
+    if (_phrase_similarity) {
+        // _norm_source is fixed once in add(); validate it before the loop so
+        // the per-document path below stays free of release-mode checks.
+        DORIS_CHECK(_norm_source != nullptr);
+    }
     int32_t doc = 0;
     while ((doc = do_next(visit_node(*_lead1, NextDoc {}))) != INT32_MAX) {
         if (_phrase_similarity) {
@@ -168,7 +188,7 @@ void PhraseQuery::search_by_skiplist(roaring::Roaring& roaring) {
                 continue;
             }
             roaring.add(doc);
-            int32_t norm = visit_node(*_lead1, Norm {});
+            int32_t norm = _norm_source->norm();
             float score = _phrase_similarity->score(phrase_freq, static_cast<int64_t>(norm));
 
             _context->collection_similarity->collect(doc, score);
@@ -285,4 +305,4 @@ void PhraseQuery::parser_info(OlapReaderStatistics* stats, std::string& query,
     }
 }
 
-} // namespace doris::segment_v2
+} // namespace doris::segment_v2

be/src/storage/index/inverted/query/phrase_query.h

@@ -74,6 +74,9 @@ class PhraseQuery : public Query {
 
     DISI* _lead1 = nullptr;
     DISI* _lead2 = nullptr;
+    // Norm source for scoring: only an exact term iterator owns per-document
+    // norms. Candidate and multi-term union iterators remain approximations.
+    TermPositionsIterator* _norm_source = nullptr;
     std::vector<DISI*> _others;
     std::vector<DISI> _iterators;
 
@@ -82,4 +85,4 @@ class PhraseQuery : public Query {
     SimilarityPtr _phrase_similarity;
 };
 
-} // namespace doris::segment_v2
+} // namespace doris::segment_v2

be/src/storage/index/inverted/util/docid_set_iterator.h

@@ -21,11 +21,12 @@
 
 #include "common/exception.h"
 #include "storage/index/inverted/util/mock_iterator.h"
+#include "storage/index/inverted/util/roaring_docid_iterator.h"
 #include "storage/index/inverted/util/union_term_iterator.h"
 
 namespace doris::segment_v2 {
 
-using DISI = std::variant<TermPositionsIterPtr, UnionTermIterPtr, MockIterPtr>;
+using DISI = std::variant<TermPositionsIterPtr, UnionTermIterPtr, MockIterPtr, RoaringDocIdIterPtr>;
 
 template <typename DISIType, typename Func, typename... Args>
 auto visit_node(DISIType&& disi, Func&& func, Args&&... args) {

be/src/storage/index/inverted/util/roaring_docid_iterator.h

@@ -0,0 +1,88 @@
+// Licensed to the Apache Software Foundation (ASF) 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 <climits>
+#include <cstdint>
+#include <memory>
+
+#include "roaring/roaring.hh"
+
+namespace doris::segment_v2 {
+
+// Read-only DISI adapter over a candidate row bitmap
+// (IndexQueryContext::candidate_rows). Joining the leapfrog intersection of
+// PhraseQuery, it restricts doc-list intersection and position verification to
+// the candidate set (two-phase evaluation: this iterator drives the
+// approximation, real term iterators keep the position semantics). It never
+// joins a matcher's postings, so freq()/next_position()/norm() only satisfy
+// the DISI interface with neutral values.
+//
+// The underlying bitmap is NOT owned and must outlive the iterator; leapfrog
+// only moves forward, so advance() targets are monotonically non-decreasing.
+class RoaringDocIdIterator {
+public:
+    explicit RoaringDocIdIterator(const roaring::Roaring* rows)
+            : _rows(rows), _iter(rows->begin()) {}
+
+    // DISI convention: an iterator starts positioned BEFORE its first doc
+    // (search_by_skiplist opens with a NextDoc on the lead), so doc_id() is -1
+    // until the first next_doc()/advance() moves onto a real position.
+    int32_t doc_id() const {
+        if (!_started) {
+            return -1;
+        }
+        return _iter == _rows->end() ? INT_MAX : static_cast<int32_t>(*_iter);
+    }
+
+    int32_t freq() const { return 1; }
+
+    int32_t next_doc() {
+        if (!_started) {
+            _started = true;
+        } else if (_iter != _rows->end()) {
+            ++_iter;
+        }
+        return doc_id();
+    }
+
+    int32_t advance(int32_t target) {
+        _started = true;
+        if (target > 0) {
+            _iter.equalorlarger(static_cast<uint32_t>(target));
+        }
+        return doc_id();
+    }
+
+    int32_t doc_freq() const {
+        uint64_t cardinality = _rows->cardinality();
+        return cardinality > INT_MAX ? INT_MAX : static_cast<int32_t>(cardinality);
+    }
+
+    int32_t next_position() { return 0; }
+
+    int32_t norm() const { return 1; }
+
+private:
+    const roaring::Roaring* _rows;
+    roaring::Roaring::const_iterator _iter;
+    bool _started = false;
+};
+using RoaringDocIdIterPtr = std::shared_ptr<RoaringDocIdIterator>;
+
+} // namespace doris::segment_v2