[improvement](inverted index) Address review: consumption-keyed cache policy, compound three-valued safety

Round 2/3 review fixes:

- Cache inserts are now gated on actual candidate consumption instead of
  candidate publication. A reply-direction candidate_rows_consumed flag is
  set only when PhraseQuery joins the candidate into its leapfrog and is
  re-armed per search (including the untokenized range path that bypasses
  match_index_search), so non-consuming queries (MATCH_ANY/ALL, term,
  regexp, single-term phrase, range) keep caching their full-segment
  bitmaps while a candidate is engaged.
- A candidate-restricted TRUE bitmap paired with the full-segment null
  bitmap could spuriously trigger VCompoundPred's three-valued AND
  shortcut (NOT(A AND B) with nullable A: a candidate row's FALSE is
  mis-typed as NULL and the row is dropped). Compound roots are now
  evaluated with the candidate suppressed in both the conjunct and
  virtual-column projection loops; top-level single-predicate consumption
  keeps the restriction, which is exact within the candidate.
- The _norm_source invariant check moved out of the per-document scoring
  loop; it is validated once before the loop when scoring is active.

Author: airborne12

be/src/storage/index/index_query_context.h

@@ -77,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,9 +481,11 @@ 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();
-            // A bitmap produced under a candidate restriction is partial and
-            // must never be cached as the full-segment result.
-            if (context->candidate_rows == nullptr) {
+            // 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;
@@ -544,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;
@@ -602,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

@@ -59,6 +59,7 @@ void PhraseQuery::add(const InvertedIndexQueryInfo& query_info) {
     // 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) {
@@ -174,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) {
@@ -182,7 +188,6 @@ void PhraseQuery::search_by_skiplist(roaring::Roaring& roaring) {
                 continue;
             }
             roaring.add(doc);
-            DORIS_CHECK(_norm_source != nullptr);
             int32_t norm = visit_node(*_norm_source, Norm {});
             float score = _phrase_similarity->score(phrase_freq, static_cast<int64_t>(norm));
 

be/src/storage/segment/segment_iterator.cpp

@@ -1272,8 +1272,29 @@ Status SegmentIterator::_apply_index_expr() {
             !_opts.runtime_state->query_options().__isset.enable_ann_index_result_cache ||
             _opts.runtime_state->query_options().enable_ann_index_result_cache;
 
+    // Three-valued compound shortcuts (VCompoundPred) treat an empty TRUE
+    // bitmap as a whole-segment fact; a candidate-restricted TRUE bitmap can
+    // spuriously trigger them and mis-type candidate rows (NOT(A AND B) with
+    // nullable A: FALSE becomes NULL and the row is dropped). Compound roots
+    // are therefore evaluated without the candidate; the top-level
+    // single-predicate consumption stays exact within the candidate.
+    auto evaluate_without_candidate_for_compound = [&](const VExprContextSPtr& expr_ctx) {
+        const bool suppress = _index_query_context != nullptr &&
+                              _index_query_context->candidate_rows != nullptr &&
+                              expr_ctx->root()->node_type() == TExprNodeType::COMPOUND_PRED;
+        const roaring::Roaring* saved = suppress ? _index_query_context->candidate_rows : nullptr;
+        if (suppress) {
+            _index_query_context->candidate_rows = nullptr;
+        }
+        Status st = expr_ctx->evaluate_inverted_index(num_rows());
+        if (suppress) {
+            _index_query_context->candidate_rows = saved;
+        }
+        return st;
+    };
+
     for (const auto& expr_ctx : _common_expr_ctxs_push_down) {
-        if (Status st = expr_ctx->evaluate_inverted_index(num_rows()); !st.ok()) {
+        if (Status st = evaluate_without_candidate_for_compound(expr_ctx); !st.ok()) {
             if (_downgrade_without_index(st) || st.code() == ErrorCode::NOT_IMPLEMENTED_ERROR) {
                 continue;
             } else {
@@ -1293,7 +1314,7 @@ Status SegmentIterator::_apply_index_expr() {
         if (expr_ctx->get_index_context() == nullptr) {
             continue;
         }
-        if (Status st = expr_ctx->evaluate_inverted_index(num_rows()); !st.ok()) {
+        if (Status st = evaluate_without_candidate_for_compound(expr_ctx); !st.ok()) {
             if (_downgrade_without_index(st) || st.code() == ErrorCode::NOT_IMPLEMENTED_ERROR) {
                 continue;
             } else {

be/test/storage/segment/inverted_index_reader_test.cpp

@@ -2426,6 +2426,203 @@ class InvertedIndexReaderTest : public testing::Test {
         }
     }
 
+    // Candidate-pushdown cache policy: only a query that actually joined the
+    // candidate bitmap into its evaluation (multi-term phrase) produces a
+    // partial result that must stay out of the query cache. A query that never
+    // consumes the candidate (MATCH_ANY here) still computes the full-segment
+    // bitmap, and a cold miss must keep filling the cache even while
+    // candidate_rows is published on the context.
+    void test_candidate_pushdown_cache_policy() {
+        std::string_view rowset_id = "test_candidate_cache_policy";
+        int seg_id = 0;
+
+        std::vector<Slice> values = {
+                Slice("the quick brown fox jumps over the lazy dog"),
+                Slice("apache doris is a fast analytical database"),
+                Slice("inverted index provides fast text search capabilities")};
+
+        TabletIndex idx_meta;
+        auto index_meta_pb = std::make_unique<TabletIndexPB>();
+        index_meta_pb->set_index_type(IndexType::INVERTED);
+        index_meta_pb->set_index_id(1);
+        index_meta_pb->set_index_name("test_candidate_cache_policy");
+        index_meta_pb->clear_col_unique_id();
+        index_meta_pb->add_col_unique_id(1);
+        index_meta_pb->mutable_properties()->insert({"parser", "english"});
+        index_meta_pb->mutable_properties()->insert({"lower_case", "true"});
+        index_meta_pb->mutable_properties()->insert({"support_phrase", "true"});
+        idx_meta.init_from_pb(*index_meta_pb.get());
+
+        std::string index_path_prefix;
+        prepare_string_index(rowset_id, seg_id, values, &idx_meta, &index_path_prefix);
+
+        OlapReaderStatistics stats;
+        RuntimeState runtime_state;
+        TQueryOptions query_options;
+        query_options.enable_inverted_index_query_cache = true;
+        query_options.enable_inverted_index_searcher_cache = false;
+        query_options.inverted_index_max_expansions = 50;
+        runtime_state.set_query_options(query_options);
+
+        auto reader = std::make_shared<IndexFileReader>(
+                io::global_local_filesystem(), index_path_prefix, InvertedIndexStorageFormatPB::V2);
+        EXPECT_TRUE(reader->init().ok());
+        auto fulltext_reader = FullTextIndexReader::create_shared(&idx_meta, reader);
+        EXPECT_NE(fulltext_reader, nullptr);
+
+        io::IOContext io_ctx;
+        IndexQueryContextPtr context = std::make_shared<IndexQueryContext>();
+        context->io_ctx = &io_ctx;
+        context->stats = &stats;
+        context->runtime_state = &runtime_state;
+
+        roaring::Roaring candidate;
+        candidate.add(0);
+        candidate.add(1);
+        context->candidate_rows = &candidate;
+
+        // MATCH_ANY never consumes the candidate: full-segment result, cacheable.
+        {
+            Field qp = Field::create_field<TYPE_STRING>(std::string("quick database"));
+
+            std::shared_ptr<roaring::Roaring> first = std::make_shared<roaring::Roaring>();
+            auto status = fulltext_reader->query(context, "1", qp,
+                                                 InvertedIndexQueryType::MATCH_ANY_QUERY, first);
+            EXPECT_TRUE(status.ok()) << status;
+            EXPECT_GT(first->cardinality(), 0);
+
+            std::shared_ptr<roaring::Roaring> second = std::make_shared<roaring::Roaring>();
+            status = fulltext_reader->query(context, "1", qp,
+                                            InvertedIndexQueryType::MATCH_ANY_QUERY, second);
+            EXPECT_TRUE(status.ok()) << status;
+            EXPECT_EQ(stats.inverted_index_query_cache_hit, 1)
+                    << "the full-segment result of a non-consuming query must be cached "
+                       "even while candidate_rows is published";
+            EXPECT_EQ(*first, *second);
+        }
+
+        // A multi-term phrase joins the candidate into its leapfrog: its result
+        // is partial and must never be inserted into the cache.
+        {
+            Field qp = Field::create_field<TYPE_STRING>(std::string("quick brown"));
+
+            std::shared_ptr<roaring::Roaring> first = std::make_shared<roaring::Roaring>();
+            auto status = fulltext_reader->query(context, "1", qp,
+                                                 InvertedIndexQueryType::MATCH_PHRASE_QUERY, first);
+            EXPECT_TRUE(status.ok()) << status;
+            EXPECT_EQ(first->cardinality(), 1);
+            EXPECT_TRUE(first->contains(0));
+
+            std::shared_ptr<roaring::Roaring> second = std::make_shared<roaring::Roaring>();
+            status = fulltext_reader->query(context, "1", qp,
+                                            InvertedIndexQueryType::MATCH_PHRASE_QUERY, second);
+            EXPECT_TRUE(status.ok()) << status;
+            EXPECT_EQ(stats.inverted_index_query_cache_hit, 1)
+                    << "a candidate-restricted phrase result must not be served from or "
+                       "inserted into the query cache";
+            EXPECT_EQ(*first, *second);
+        }
+
+        context->candidate_rows = nullptr;
+    }
+
+    // The consumed flag must be re-armed per search: a range query on the
+    // untokenized reader never passes through match_index_search, so a stale
+    // flag left by an earlier candidate-consuming phrase search must not
+    // block its (full-segment) result from entering the cache.
+    void test_candidate_consumed_flag_reset_between_readers() {
+        std::vector<Slice> fulltext_values = {Slice("the quick brown fox")};
+        TabletIndex fulltext_meta;
+        auto fulltext_meta_pb = std::make_unique<TabletIndexPB>();
+        fulltext_meta_pb->set_index_type(IndexType::INVERTED);
+        fulltext_meta_pb->set_index_id(1);
+        fulltext_meta_pb->set_index_name("test_consumed_reset_ft");
+        fulltext_meta_pb->clear_col_unique_id();
+        fulltext_meta_pb->add_col_unique_id(1);
+        fulltext_meta_pb->mutable_properties()->insert({"parser", "english"});
+        fulltext_meta_pb->mutable_properties()->insert({"support_phrase", "true"});
+        fulltext_meta.init_from_pb(*fulltext_meta_pb.get());
+        std::string fulltext_prefix;
+        prepare_string_index("test_consumed_reset_ft", 0, fulltext_values, &fulltext_meta,
+                             &fulltext_prefix);
+
+        std::vector<Slice> plain_values = {Slice("alpha"), Slice("beta")};
+        TabletIndex plain_meta;
+        auto plain_meta_pb = std::make_unique<TabletIndexPB>();
+        plain_meta_pb->set_index_type(IndexType::INVERTED);
+        plain_meta_pb->set_index_id(2);
+        plain_meta_pb->set_index_name("test_consumed_reset_plain");
+        plain_meta_pb->clear_col_unique_id();
+        plain_meta_pb->add_col_unique_id(1);
+        plain_meta.init_from_pb(*plain_meta_pb.get());
+        std::string plain_prefix;
+        prepare_string_index("test_consumed_reset_plain", 0, plain_values, &plain_meta,
+                             &plain_prefix);
+
+        OlapReaderStatistics stats;
+        RuntimeState runtime_state;
+        TQueryOptions query_options;
+        query_options.enable_inverted_index_query_cache = true;
+        query_options.enable_inverted_index_searcher_cache = false;
+        query_options.inverted_index_max_expansions = 50;
+        runtime_state.set_query_options(query_options);
+
+        io::IOContext io_ctx;
+        IndexQueryContextPtr context = std::make_shared<IndexQueryContext>();
+        context->io_ctx = &io_ctx;
+        context->stats = &stats;
+        context->runtime_state = &runtime_state;
+
+        roaring::Roaring candidate;
+        candidate.add(0);
+        context->candidate_rows = &candidate;
+
+        // 1) A consuming phrase search on the fulltext reader sets the flag.
+        {
+            auto reader = std::make_shared<IndexFileReader>(io::global_local_filesystem(),
+                                                            fulltext_prefix,
+                                                            InvertedIndexStorageFormatPB::V2);
+            EXPECT_TRUE(reader->init().ok());
+            auto fulltext_reader = FullTextIndexReader::create_shared(&fulltext_meta, reader);
+            std::shared_ptr<roaring::Roaring> bitmap = std::make_shared<roaring::Roaring>();
+            Field qp = Field::create_field<TYPE_STRING>(std::string("quick brown"));
+            EXPECT_TRUE(fulltext_reader
+                                ->query(context, "1", qp,
+                                        InvertedIndexQueryType::MATCH_PHRASE_QUERY, bitmap)
+                                .ok());
+        }
+
+        // 2) A range query on the untokenized reader takes the switch branch
+        // that bypasses match_index_search; its full-segment result must
+        // still be cached (second run hits).
+        {
+            auto reader = std::make_shared<IndexFileReader>(
+                    io::global_local_filesystem(), plain_prefix, InvertedIndexStorageFormatPB::V2);
+            EXPECT_TRUE(reader->init().ok());
+            auto plain_reader = StringTypeInvertedIndexReader::create_shared(&plain_meta, reader);
+            Field qp = Field::create_field<TYPE_STRING>(std::string("alpha"));
+
+            std::shared_ptr<roaring::Roaring> first = std::make_shared<roaring::Roaring>();
+            EXPECT_TRUE(plain_reader
+                                ->query(context, "1", qp,
+                                        InvertedIndexQueryType::GREATER_EQUAL_QUERY, first)
+                                .ok());
+            EXPECT_EQ(first->cardinality(), 2);
+
+            std::shared_ptr<roaring::Roaring> second = std::make_shared<roaring::Roaring>();
+            EXPECT_TRUE(plain_reader
+                                ->query(context, "1", qp,
+                                        InvertedIndexQueryType::GREATER_EQUAL_QUERY, second)
+                                .ok());
+            EXPECT_EQ(stats.inverted_index_query_cache_hit, 1)
+                    << "a stale consumed flag from the earlier phrase search must not "
+                       "block caching of the range query's full-segment result";
+            EXPECT_EQ(*first, *second);
+        }
+
+        context->candidate_rows = nullptr;
+    }
+
     // Test fulltext index with comprehensive query types
     void test_fulltext_comprehensive_queries() {
         std::string_view rowset_id = "test_fulltext_comprehensive";
@@ -4320,6 +4517,14 @@ TEST_F(InvertedIndexReaderTest, UnsupportedDataTypes) {
     test_unsupported_data_types();
 }
 
+TEST_F(InvertedIndexReaderTest, CandidatePushdownCachePolicy) {
+    test_candidate_pushdown_cache_policy();
+}
+
+TEST_F(InvertedIndexReaderTest, CandidateConsumedFlagResetBetweenReaders) {
+    test_candidate_consumed_flag_reset_between_readers();
+}
+
 // Test InvertedIndexResultBitmap operator|= with NULL handling
 TEST_F(InvertedIndexReaderTest, ResultBitmapOrOperatorNullHandling) {
     // Test SQL three-valued logic for OR: