[fix](inverted index) Fix compound candidate safety and phrase-prefix scoring

### What problem does this PR solve?

Issue Number: None

Related PR: #67180

Problem Summary: A VirtualSlotRef-wrapped compound expression bypassed the
candidate suppression used to protect SQL three-valued bitmap evaluation. A
nullable NOT(A AND B) expression could therefore short-circuit on a partial
TRUE bitmap and silently drop a row for which NOT(NULL AND FALSE) is TRUE.
Resolve the effective root through VirtualSlotRef before deciding whether to
suppress the candidate.

Scoring a multi-term phrase-prefix query could also choose a low-frequency
UnionTermIterator as the norm source and fail with "UnionTermIterator does not
support scoring". Restrict the norm source to the exact TermPositionsIterator,
which owns the per-document norm. A no-candidate run reproduces the same
failure, confirming that this scoring edge predates candidate pushdown.

### Release note

Fix nullable compound inverted-index filtering through virtual columns and
multi-term phrase-prefix scoring failures.

### Check List (For Author)

- Test: Unit Test
    - Added RED/GREEN tests for VirtualSlotRef-wrapped nullable three-valued logic.
    - Added RED/GREEN real-index scoring tests with and without candidate rows.
    - Ran 15 related ASAN BE unit tests.
    - Ran ./build.sh --be with BUILD_TYPE=ASAN.
- Behavior changed: Yes. Correct query results are preserved and affected scoring queries no longer fail.
- Does this need documentation: No

Author: airborne12

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

@@ -74,8 +74,8 @@ void PhraseQuery::add(const InvertedIndexQueryInfo& query_info) {
         _others.emplace_back(&_iterators[i]);
     }
     for (auto& iter : _iterators) {
-        if (!std::holds_alternative<RoaringDocIdIterPtr>(iter)) {
-            _norm_source = &iter;
+        if (const auto* term_iter = std::get_if<TermPositionsIterPtr>(&iter)) {
+            _norm_source = term_iter->get();
             break;
         }
     }
@@ -188,7 +188,7 @@ void PhraseQuery::search_by_skiplist(roaring::Roaring& roaring) {
                 continue;
             }
             roaring.add(doc);
-            int32_t norm = visit_node(*_norm_source, 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);
@@ -305,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,9 +74,9 @@ class PhraseQuery : public Query {
 
     DISI* _lead1 = nullptr;
     DISI* _lead2 = nullptr;
-    // Norm source for scoring: always a real postings iterator, never the
-    // pushed-down candidate bitmap (whose norm is a meaningless constant).
-    DISI* _norm_source = 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;
 
@@ -85,4 +85,4 @@ class PhraseQuery : public Query {
     SimilarityPtr _phrase_similarity;
 };
 
-} // namespace doris::segment_v2
+} // namespace doris::segment_v2

be/src/storage/segment/segment_iterator.cpp

@@ -1279,9 +1279,18 @@ Status SegmentIterator::_apply_index_expr() {
     // 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 auto& root = expr_ctx->root();
+        DORIS_CHECK(root != nullptr);
+        const VExpr* effective_root = root.get();
+        if (root->is_virtual_slot_ref()) {
+            const auto& virtual_expr =
+                    assert_cast<const VirtualSlotRef*>(root.get())->get_virtual_column_expr();
+            DORIS_CHECK(virtual_expr != nullptr);
+            effective_root = virtual_expr.get();
+        }
         const bool suppress = _index_query_context != nullptr &&
                               _index_query_context->candidate_rows != nullptr &&
-                              expr_ctx->root()->node_type() == TExprNodeType::COMPOUND_PRED;
+                              effective_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;

be/test/storage/index/inverted/query/phrase_prefix_query_test.cpp

@@ -23,11 +23,13 @@
 
 #include "io/fs/local_file_system.h"
 #include "runtime/exec_env.h"
+#include "storage/compaction/collection_similarity.h"
 #include "storage/index/index_file_reader.h"
 #include "storage/index/index_file_writer.h"
 #include "storage/index/inverted/inverted_index_cache.h"
 #include "storage/index/inverted/inverted_index_searcher.h"
 #include "storage/index/inverted/inverted_index_writer.h"
+#include "storage/index/inverted/similarity/collection_statistics.h"
 #include "storage/tablet/tablet_schema.h"
 #include "util/slice.h"
 
@@ -311,6 +313,87 @@ TEST_F(PhrasePrefixQueryTest, test_multi_term_phrase_prefix_query) {
     EXPECT_GE(result.cardinality(), 0);
 }
 
+TEST_F(PhrasePrefixQueryTest, scoring_uses_norm_capable_exact_term_iterator) {
+    std::string_view rowset_id = "test_scoring_norm_source";
+    int seg_id = 0;
+
+    // "common" has df=5 while the rare* expansion union has df=2. With a
+    // one-row candidate the sorted DISI order is candidate, rare* union,
+    // common exact term, so choosing the first non-candidate iterator as the
+    // norm source selects a UnionTermIterator that cannot provide norms.
+    std::vector<Slice> values = {Slice("common rareone"), Slice("common raretwo"),
+                                 Slice("common filler"), Slice("common filler"),
+                                 Slice("common filler")};
+
+    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_scoring_norm_source");
+    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);
+
+    std::string index_path_prefix;
+    prepare_fulltext_index(rowset_id, seg_id, values, &idx_meta, &index_path_prefix);
+    auto searcher = create_searcher(index_path_prefix, idx_meta);
+    ASSERT_NE(searcher, nullptr);
+
+    class FixedStats : public CollectionStatistics {
+    public:
+        float get_or_calculate_idf(const std::wstring&, const std::wstring&) override {
+            return 1.0F;
+        }
+        float get_or_calculate_avg_dl(const std::wstring&) override { return 2.0F; }
+    };
+
+    RuntimeState runtime_state;
+    TQueryOptions query_options;
+    query_options.inverted_index_max_expansions = 50;
+    runtime_state.set_query_options(query_options);
+    io::IOContext io_ctx;
+    roaring::Roaring candidate;
+    candidate.add(0);
+
+    InvertedIndexQueryInfo query_info;
+    query_info.field_name = L"1";
+    query_info.term_infos.emplace_back("common", 0);
+    query_info.term_infos.emplace_back("rare", 1);
+    query_info.is_similarity_score = true;
+
+    auto run_scoring = [&](const roaring::Roaring* candidate_rows) {
+        IndexQueryContextPtr context = std::make_shared<IndexQueryContext>();
+        context->io_ctx = &io_ctx;
+        context->runtime_state = &runtime_state;
+        context->collection_statistics = std::make_shared<FixedStats>();
+        context->collection_similarity = std::make_shared<CollectionSimilarity>();
+        context->candidate_rows = candidate_rows;
+
+        PhrasePrefixQuery query(searcher, context);
+        query.add(query_info);
+        roaring::Roaring result;
+        query.search(result);
+        return std::pair {std::move(result), context->collection_similarity->release_scores()};
+    };
+
+    // The low-df union is also first without a candidate. This pins down that
+    // the scoring failure predates candidate pushdown rather than attributing
+    // it to the optimization merely because both share the DISI ordering.
+    auto [full_result, full_scores] = run_scoring(nullptr);
+    EXPECT_EQ(full_result.cardinality(), 2);
+    EXPECT_TRUE(full_result.contains(0));
+    EXPECT_TRUE(full_result.contains(1));
+    EXPECT_TRUE(full_scores.contains(0));
+    EXPECT_TRUE(full_scores.contains(1));
+
+    auto [restricted_result, restricted_scores] = run_scoring(&candidate);
+    EXPECT_EQ(restricted_result, candidate);
+    EXPECT_TRUE(restricted_scores.contains(0));
+}
+
 TEST_F(PhrasePrefixQueryTest, test_empty_terms_exception) {
     std::string_view rowset_id = "test_empty_terms";
     int seg_id = 0;
@@ -519,4 +602,4 @@ TEST_F(PhrasePrefixQueryTest, test_phrase_with_no_prefix_expansion) {
     EXPECT_GE(result.cardinality(), 0);
 }
 
-} // namespace doris::segment_v2
+} // namespace doris::segment_v2

be/test/storage/segment/segment_iterator_candidate_pushdown_test.cpp

@@ -32,8 +32,10 @@
 #include "common/config.h"
 #include "common/status.h"
 #include "core/data_type/data_type_number.h"
+#include "exprs/vcompound_pred.h"
 #include "exprs/vexpr.h"
 #include "exprs/vexpr_context.h"
+#include "exprs/virtual_slot_ref.h"
 #include "runtime/runtime_state.h"
 #include "storage/index/index_iterator.h"
 #include "storage/index/index_query_context.h"
@@ -96,6 +98,55 @@ class CapturingExpr : public VExpr {
     const roaring::Roaring* _captured_candidate = nullptr;
 };
 
+// Produces a deterministic inverted-index result while modeling the contract
+// of a candidate-consuming leaf: only its TRUE bitmap is candidate-restricted;
+// its NULL bitmap remains segment-wide so compound SQL three-valued logic can
+// distinguish FALSE from UNKNOWN.
+class CandidateRestrictedBitmapExpr : public VExpr {
+public:
+    CandidateRestrictedBitmapExpr(SegmentIterator* iter, std::initializer_list<uint32_t> true_rows,
+                                  std::initializer_list<uint32_t> null_rows)
+            : _iter(iter) {
+        _data_type = make_nullable(std::make_shared<DataTypeUInt8>());
+        for (uint32_t row : true_rows) {
+            _true_rows.add(row);
+        }
+        for (uint32_t row : null_rows) {
+            _null_rows.add(row);
+        }
+    }
+
+    const std::string& expr_name() const override {
+        static const std::string kName = "CandidateRestrictedBitmapExpr";
+        return kName;
+    }
+
+    Status execute_column_impl(VExprContext*, const Block*, const Selector*, size_t,
+                               ColumnPtr&) const override {
+        return Status::NotSupported("bitmap-only test expression");
+    }
+
+    Status evaluate_inverted_index(VExprContext* context, uint32_t) override {
+        _evaluated = true;
+        auto data = std::make_shared<roaring::Roaring>(_true_rows);
+        if (_iter->_index_query_context->candidate_rows != nullptr) {
+            *data &= *_iter->_index_query_context->candidate_rows;
+        }
+        context->get_index_context()->set_index_result_for_expr(
+                this, InvertedIndexResultBitmap(std::move(data),
+                                                std::make_shared<roaring::Roaring>(_null_rows)));
+        return Status::OK();
+    }
+
+    bool evaluated() const { return _evaluated; }
+
+private:
+    SegmentIterator* _iter;
+    roaring::Roaring _true_rows;
+    roaring::Roaring _null_rows;
+    bool _evaluated = false;
+};
+
 // An indexed predicate stub that shrinks the row bitmap to a fixed set,
 // standing in for a selective indexed equality applied before the expression
 // conjuncts (modeled on MockNestedPredicate of accept_null_predicate_test).
@@ -167,6 +218,44 @@ VExprContextSPtr make_capturing_ctx(const std::shared_ptr<CapturingExpr>& expr)
     return ctx;
 }
 
+TExprNode make_compound_node(TExprOpcode::type opcode, int num_children) {
+    TExprNode node;
+    node.__set_type(create_type_desc(PrimitiveType::TYPE_BOOLEAN));
+    node.__set_node_type(TExprNodeType::COMPOUND_PRED);
+    node.__set_opcode(opcode);
+    node.__set_num_children(num_children);
+    node.__set_is_nullable(true);
+    return node;
+}
+
+VExprContextSPtr make_virtual_slot_ctx(const VExprSPtr& virtual_expr) {
+    TExprNode node;
+    node.__set_type(create_type_desc(PrimitiveType::TYPE_BOOLEAN));
+    node.__set_node_type(TExprNodeType::VIRTUAL_SLOT_REF);
+    node.__set_num_children(0);
+    node.__set_is_nullable(true);
+    node.__set_label("virtual_compound");
+    TSlotRef slot_ref;
+    slot_ref.__set_slot_id(-1);
+    slot_ref.__set_tuple_id(-1);
+    node.__set_slot_ref(slot_ref);
+
+    auto root = VirtualSlotRef::create_shared(node);
+    root->set_virtual_column_expr(virtual_expr);
+    static const std::string kColumnName = "virtual_compound";
+    root->set_column_name(&kColumnName);
+    root->set_column_data_type(make_nullable(std::make_shared<DataTypeUInt8>()));
+
+    auto ctx = std::make_shared<VExprContext>(root);
+    std::vector<std::unique_ptr<IndexIterator>> index_iters;
+    std::vector<IndexFieldNameAndTypePair> storage_types;
+    std::unordered_map<ColumnId, std::unordered_map<const VExpr*, bool>> status_map;
+    ColumnIteratorOptions column_iter_opts;
+    ctx->set_index_context(std::make_shared<IndexExecContext>(
+            index_iters, storage_types, status_map, nullptr, nullptr, column_iter_opts));
+    return ctx;
+}
+
 } // namespace
 
 class SegmentIteratorCandidatePushdownTest : public testing::Test {
@@ -280,6 +369,36 @@ TEST_F(SegmentIteratorCandidatePushdownTest, compound_root_evaluates_without_can
     EXPECT_EQ(_iter->_index_query_context->candidate_rows, nullptr);
 }
 
+TEST_F(SegmentIteratorCandidatePushdownTest,
+       virtual_slot_wrapped_compound_preserves_three_valued_logic) {
+    config::inverted_index_candidate_pushdown_ratio = 0.3;
+    _iter->_row_bitmap.add(0); // 1% of 100 rows: candidate engages
+
+    // At row 0, A is NULL and B is FALSE, so SQL requires
+    // NOT(A AND B) = NOT(FALSE) = TRUE. A also has a TRUE row outside the
+    // candidate so a candidate-restricted evaluation makes its TRUE bitmap
+    // empty and exposes VCompoundPred's invalid early exit.
+    auto nullable_a = std::make_shared<CandidateRestrictedBitmapExpr>(
+            _iter.get(), std::initializer_list<uint32_t> {50}, std::initializer_list<uint32_t> {0});
+    auto false_b = std::make_shared<CandidateRestrictedBitmapExpr>(
+            _iter.get(), std::initializer_list<uint32_t> {}, std::initializer_list<uint32_t> {});
+    auto and_expr = VCompoundPred::create_shared(make_compound_node(TExprOpcode::COMPOUND_AND, 2));
+    and_expr->add_child(nullable_a);
+    and_expr->add_child(false_b);
+    auto not_expr = VCompoundPred::create_shared(make_compound_node(TExprOpcode::COMPOUND_NOT, 1));
+    not_expr->add_child(and_expr);
+    auto ctx = make_virtual_slot_ctx(not_expr);
+    _iter->_common_expr_ctxs_push_down = {ctx};
+
+    ASSERT_TRUE(_iter->_get_row_ranges_by_column_conditions().ok());
+
+    const auto* result = ctx->get_index_context()->get_index_result_for_expr(not_expr.get());
+    ASSERT_NE(result, nullptr);
+    EXPECT_TRUE(false_b->evaluated()) << "AND must evaluate FALSE B after nullable A";
+    EXPECT_TRUE(result->get_data_bitmap()->contains(0))
+            << "NOT(NULL AND FALSE) must preserve candidate row 0";
+}
+
 // A non-finite configured ratio must never engage the pushdown (the multiply
 // and integer conversion would otherwise be undefined behavior).
 TEST_F(SegmentIteratorCandidatePushdownTest, non_finite_ratio_never_engages) {