[opt](scan) Extend sequence mapping candidate scan to NGRAM bloom filters

Issue Number: close #66879

Related PR: #66881

Problem Summary:

Allow positive LIKE predicates backed by NGRAM_BF indexes to drive sequence-mapping candidate scans. Reuse key-prefix ranges during candidate collection while retaining the full-key fallback, and skip synthetic whole-value key ranges.

Add bloom-filter profile accounting and regression coverage for matching, stale versions, short patterns, NOT LIKE, disabled inverted-index queries, and composite key prefixes.

Author: zhaixiangli

be/src/exec/operator/olap_scan_operator.cpp

@@ -304,6 +304,8 @@ Status OlapScanLocalState::_init_profile() {
             ADD_COUNTER(_segment_profile, "SeqMapCandidateScanBytes", TUnit::BYTES);
     _seq_map_candidate_index_filtered_rows_counter =
             ADD_COUNTER(_segment_profile, "SeqMapCandidateIndexFilteredRows", TUnit::UNIT);
+    _seq_map_candidate_bloom_filter_filtered_rows_counter =
+            ADD_COUNTER(_segment_profile, "SeqMapCandidateBloomFilterFilteredRows", TUnit::UNIT);
     _seq_map_candidate_index_downgrades_counter =
             ADD_COUNTER(_segment_profile, "SeqMapCandidateIndexDowngrades", TUnit::UNIT);
     _seq_map_candidate_index_lookup_timer =
@@ -1232,6 +1234,10 @@ Status OlapScanLocalState::_build_key_ranges_and_filters() {
             }
             DCHECK(_slot_id_to_predicates.count(iter->first) > 0);
             const auto& value_range = iter->second;
+            if (std::visit([](const auto& range) { return range.is_whole_value_range(); },
+                           value_range)) {
+                break;
+            }
 
             std::optional<int> key_to_erase;
             bool is_fixed_value_range = false;

be/src/exec/operator/olap_scan_operator.h

@@ -265,6 +265,7 @@ class OlapScanLocalState final : public ScanLocalState<OlapScanLocalState> {
     RuntimeProfile::Counter* _seq_map_candidate_scan_rows_counter = nullptr;
     RuntimeProfile::Counter* _seq_map_candidate_scan_bytes_counter = nullptr;
     RuntimeProfile::Counter* _seq_map_candidate_index_filtered_rows_counter = nullptr;
+    RuntimeProfile::Counter* _seq_map_candidate_bloom_filter_filtered_rows_counter = nullptr;
     RuntimeProfile::Counter* _seq_map_candidate_index_downgrades_counter = nullptr;
     RuntimeProfile::Counter* _seq_map_candidate_index_lookup_timer = nullptr;
     RuntimeProfile::Counter* _seq_map_candidate_cache_local_bytes_counter = nullptr;

be/src/exec/scan/olap_scanner.cpp

@@ -65,6 +65,7 @@
 #include "storage/index/inverted/inverted_index_profile.h"
 #include "storage/index/inverted/similarity/collection_statistics.h"
 #include "storage/iterator/block_reader.h"
+#include "storage/itoken_extractor.h"
 #include "storage/olap_common.h"
 #include "storage/olap_tuple.h"
 #include "storage/olap_utils.h"
@@ -344,6 +345,7 @@ void OlapScanner::_merge_seq_map_candidate_stats(const OlapReaderStatistics& can
     total_stats->seq_map_candidate_scan_bytes += candidate_stats.uncompressed_bytes_read;
     total_stats->seq_map_candidate_index_filtered_rows +=
             candidate_stats.rows_inverted_index_filtered;
+    total_stats->seq_map_candidate_bloom_filter_filtered_rows += candidate_stats.rows_bf_filtered;
     total_stats->seq_map_candidate_index_downgrades +=
             candidate_stats.inverted_index_downgrade_count;
     total_stats->seq_map_candidate_index_lookup_ns += candidate_stats.inverted_index_lookup_timer;
@@ -363,6 +365,7 @@ void OlapScanner::_merge_seq_map_candidate_stats(const OlapReaderStatistics& can
 
 Status OlapScanner::_collect_seq_map_candidate_keys(
         const std::vector<std::shared_ptr<ColumnPredicate>>& driver_predicates,
+        const std::vector<FunctionFilter>& driver_function_filters,
         const std::vector<std::shared_ptr<ColumnPredicate>>& key_predicates,
         int64_t previous_candidate_scan_rows, bool price_point_lookups, int64_t max_candidate_keys,
         size_t max_candidate_bytes, size_t candidate_workspace_bytes,
@@ -391,7 +394,7 @@ Status OlapScanner::_collect_seq_map_candidate_keys(
     for (const auto& predicate : driver_predicates) {
         candidate_params.predicates.emplace_back(predicate->clone(predicate->column_id()));
     }
-    candidate_params.function_filters.clear();
+    candidate_params.function_filters = driver_function_filters;
     candidate_params.all_access_paths.clear();
     candidate_params.predicate_access_paths.clear();
     candidate_params.output_columns.clear();
@@ -422,7 +425,7 @@ Status OlapScanner::_collect_seq_map_candidate_keys(
 
     std::vector<ColumnId> candidate_columns;
     candidate_columns.reserve(_tablet_reader_params.tablet_schema->num_key_columns() +
-                              driver_predicates.size());
+                              driver_predicates.size() + driver_function_filters.size());
     for (uint32_t cid = 0; cid < _tablet_reader_params.tablet_schema->num_key_columns(); ++cid) {
         candidate_columns.push_back(cid);
     }
@@ -432,6 +435,14 @@ Status OlapScanner::_collect_seq_map_candidate_keys(
             candidate_columns.push_back(predicate->column_id());
         }
     }
+    for (const auto& filter : driver_function_filters) {
+        const int32_t cid = candidate_params.tablet_schema->field_index(filter._col_name);
+        DORIS_CHECK_GE(cid, 0);
+        if (std::find(candidate_columns.begin(), candidate_columns.end(), cid) ==
+            candidate_columns.end()) {
+            candidate_columns.push_back(cid);
+        }
+    }
     candidate_params.return_columns = candidate_columns;
     candidate_params.origin_return_columns = &candidate_columns;
     candidate_params.tablet_columns_convert_to_null_set = nullptr;
@@ -534,9 +545,18 @@ void OlapScanner::_record_seq_map_candidate_fallback_reason(RuntimeProfile* prof
     profile->add_info_string("SeqMapCandidateFallbackReason." + fallback_reason, fallback_reason);
 }
 
+bool OlapScanner::_has_usable_ngram_bf_pattern(const FunctionFilter& filter, size_t gram_size) {
+    NgramTokenExtractor extractor(gram_size);
+    size_t position = 0;
+    std::string token;
+    return extractor.next_in_string_like(filter._string_param.data, filter._string_param.size,
+                                         &position, token);
+}
+
 Status OlapScanner::_build_seq_map_candidate_keys(
         const std::vector<std::shared_ptr<ColumnPredicate>>& key_predicates,
         const std::map<uint32_t, std::vector<std::shared_ptr<ColumnPredicate>>>& group_drivers,
+        const std::map<uint32_t, std::vector<FunctionFilter>>& group_function_drivers,
         int64_t max_candidate_keys, const CandidateMemoryBudget& memory_budget,
         const CandidateScanCostLimit& cost_limit) {
     auto& params = _tablet_reader_params;
@@ -560,6 +580,11 @@ Status OlapScanner::_build_seq_map_candidate_keys(
     size_t final_key_bytes = 0;
     bool first_group = true;
     for (const auto& [seq_col, predicates] : group_drivers) {
+        const auto function_it = group_function_drivers.find(seq_col);
+        const std::vector<FunctionFilter> empty_function_filters;
+        const auto& function_filters = function_it == group_function_drivers.end()
+                                               ? empty_function_filters
+                                               : function_it->second;
         CandidateKeyMap group_keys;
         size_t group_key_bytes = 0;
         bool limit_exceeded = false;
@@ -568,7 +593,7 @@ Status OlapScanner::_build_seq_map_candidate_keys(
         bool cost_exceeded = false;
         const size_t remaining_candidate_bytes = memory_budget.key_bytes - final_key_bytes;
         RETURN_IF_ERROR(_collect_seq_map_candidate_keys(
-                predicates, key_predicates, stats->seq_map_candidate_scan_rows,
+                predicates, function_filters, key_predicates, stats->seq_map_candidate_scan_rows,
                 group_drivers.size() == 1, max_candidate_keys, remaining_candidate_bytes,
                 memory_budget.workspace_bytes, cost_limit, &group_keys, &group_key_bytes,
                 &limit_exceeded, &bytes_exceeded, &reservation_exceeded, &cost_exceeded));
@@ -671,14 +696,10 @@ Status OlapScanner::_prepare_seq_map_candidate_keys() {
         _seq_map_candidate_fallback_reason = "candidate_key_bytes_limit";
         return Status::OK();
     }
-    if (!query_options.enable_inverted_index_query) {
-        ++stats->seq_map_candidate_fallbacks;
-        _seq_map_candidate_fallback_reason = "inverted_index_query_disabled";
-        return Status::OK();
-    }
-
     std::vector<std::shared_ptr<ColumnPredicate>> key_predicates;
     std::map<uint32_t, std::vector<std::shared_ptr<ColumnPredicate>>> group_drivers;
+    std::map<uint32_t, std::vector<FunctionFilter>> group_function_drivers;
+    bool has_disabled_inverted_driver = false;
     const auto& value_to_seq = schema->value_col_idx_to_seq_col_idx();
     for (const auto& predicate : params.predicates) {
         const auto cid = predicate->column_id();
@@ -697,17 +718,55 @@ Status OlapScanner::_prepare_seq_map_candidate_keys() {
         if (!positive_driver || schema->inverted_indexs(column).empty()) {
             continue;
         }
+        if (!query_options.enable_inverted_index_query) {
+            has_disabled_inverted_driver = true;
+            continue;
+        }
         group_drivers[seq_it->second].push_back(predicate);
         ++stats->seq_map_candidate_driver_predicates;
     }
 
+    for (const auto& filter : params.function_filters) {
+        if (filter._opposite || !config::enable_query_like_bloom_filter) {
+            continue;
+        }
+        const int32_t cid = schema->field_index(filter._col_name);
+        DORIS_CHECK_GE(cid, 0);
+        const auto& column = schema->column(cid);
+        if (column.is_key()) {
+            continue;
+        }
+        const auto seq_it = value_to_seq.find(cid);
+        if (seq_it == value_to_seq.end()) {
+            continue;
+        }
+        const auto* ngram_index = schema->get_ngram_bf_index(column.unique_id());
+        if (ngram_index == nullptr ||
+            !_has_usable_ngram_bf_pattern(filter, ngram_index->get_gram_size())) {
+            continue;
+        }
+        group_drivers.try_emplace(seq_it->second);
+        group_function_drivers[seq_it->second].push_back(filter);
+        ++stats->seq_map_candidate_driver_predicates;
+    }
+
     if (group_drivers.empty()) {
         ++stats->seq_map_candidate_fallbacks;
-        _seq_map_candidate_fallback_reason = "no_indexed_positive_driver";
+        _seq_map_candidate_fallback_reason = has_disabled_inverted_driver
+                                                     ? "inverted_index_query_disabled"
+                                                     : "no_indexed_positive_driver";
         return Status::OK();
     }
     stats->seq_map_candidate_driver_groups = group_drivers.size();
-    if (!params.start_key.empty() || !params.end_key.empty()) {
+    // Full-key ranges are already selective. Prefix ranges such as advertiser_id = ?
+    // still leave enough rows for a value-column candidate scan to be useful.
+    const auto range_uses_full_key =
+            [key_column_count = schema->num_key_columns()](const std::vector<OlapTuple>& keys) {
+                return std::ranges::any_of(keys, [key_column_count](const OlapTuple& key) {
+                    return key.size() >= key_column_count;
+                });
+            };
+    if (range_uses_full_key(params.start_key) || range_uses_full_key(params.end_key)) {
         ++stats->seq_map_candidate_fallbacks;
         _seq_map_candidate_fallback_reason = "key_range_present";
         return Status::OK();
@@ -725,8 +784,9 @@ Status OlapScanner::_prepare_seq_map_candidate_keys() {
     try {
         ++enable_thread_catch_bad_alloc;
         Defer restore_bad_alloc_catch {[&] { --enable_thread_catch_bad_alloc; }};
-        build_status = _build_seq_map_candidate_keys(key_predicates, group_drivers,
-                                                     max_candidate_keys, memory_budget, cost_limit);
+        build_status =
+                _build_seq_map_candidate_keys(key_predicates, group_drivers, group_function_drivers,
+                                              max_candidate_keys, memory_budget, cost_limit);
     } catch (const Exception& exception) {
         build_status = exception.code() == ErrorCode::MEM_ALLOC_FAILED
                                ? Status::MemoryLimitExceeded(exception.to_string())
@@ -1635,6 +1695,8 @@ void OlapScanner::_collect_profile_before_close() {
                    stats.seq_map_candidate_scan_bytes);
     COUNTER_UPDATE(local_state->_seq_map_candidate_index_filtered_rows_counter,
                    stats.seq_map_candidate_index_filtered_rows);
+    COUNTER_UPDATE(local_state->_seq_map_candidate_bloom_filter_filtered_rows_counter,
+                   stats.seq_map_candidate_bloom_filter_filtered_rows);
     COUNTER_UPDATE(local_state->_seq_map_candidate_index_downgrades_counter,
                    stats.seq_map_candidate_index_downgrades);
     COUNTER_UPDATE(local_state->_seq_map_candidate_index_lookup_timer,

be/src/exec/scan/olap_scanner.h

@@ -147,10 +147,12 @@ class OlapScanner : public Scanner {
     [[nodiscard]] Status _build_seq_map_candidate_keys(
             const std::vector<std::shared_ptr<ColumnPredicate>>& key_predicates,
             const std::map<uint32_t, std::vector<std::shared_ptr<ColumnPredicate>>>& group_drivers,
+            const std::map<uint32_t, std::vector<FunctionFilter>>& group_function_drivers,
             int64_t max_candidate_keys, const CandidateMemoryBudget& memory_budget,
             const CandidateScanCostLimit& cost_limit);
     [[nodiscard]] Status _collect_seq_map_candidate_keys(
             const std::vector<std::shared_ptr<ColumnPredicate>>& driver_predicates,
+            const std::vector<FunctionFilter>& driver_function_filters,
             const std::vector<std::shared_ptr<ColumnPredicate>>& key_predicates,
             int64_t previous_candidate_scan_rows, bool price_point_lookups,
             int64_t max_candidate_keys, size_t max_candidate_bytes,
@@ -176,6 +178,8 @@ class OlapScanner : public Scanner {
     [[nodiscard]] static bool _is_candidate_memory_failure(const Status& status);
     static void _record_seq_map_candidate_fallback_reason(RuntimeProfile* profile,
                                                           const std::string& fallback_reason);
+    [[nodiscard]] static bool _has_usable_ngram_bf_pattern(const FunctionFilter& filter,
+                                                           size_t gram_size);
     [[nodiscard]] static size_t _estimate_candidate_key_bytes(const std::string& encoded_key,
                                                               size_t key_column_count);
     [[nodiscard]] size_t _estimate_candidate_map_bytes(const CandidateKeyMap& candidate_keys) const;

be/src/storage/olap_common.h

@@ -295,6 +295,7 @@ struct OlapReaderStatistics {
     int64_t seq_map_candidate_scan_rows = 0;
     int64_t seq_map_candidate_scan_bytes = 0;
     int64_t seq_map_candidate_index_filtered_rows = 0;
+    int64_t seq_map_candidate_bloom_filter_filtered_rows = 0;
     int64_t seq_map_candidate_index_downgrades = 0;
     int64_t seq_map_candidate_index_lookup_ns = 0;
     int64_t seq_map_candidate_cache_local_bytes = 0;

be/test/exec/scan/olap_scanner_test.cpp

@@ -24,6 +24,7 @@
 
 #include "cloud/config.h"
 #include "common/config.h"
+#include "exprs/function_filter.h"
 #include "io/io_common.h"
 #include "runtime/runtime_profile.h"
 #include "testutil/mock/mock_runtime_state.h"
@@ -250,6 +251,16 @@ TEST(OlapScannerTest, CandidateMemoryFailuresAreClassifiedForFallback) {
     EXPECT_FALSE(OlapScanner::_is_candidate_memory_failure(Status::Cancelled("query cancelled")));
 }
 
+TEST(OlapScannerTest, CandidateNgramPatternMustContainCompleteGram) {
+    FunctionFilter usable(false, "title", nullptr, StringRef("%have%"));
+    FunctionFilter too_short(false, "title", nullptr, StringRef("%ha%"));
+    FunctionFilter split_by_wildcard(false, "title", nullptr, StringRef("%h_ve%"));
+
+    EXPECT_TRUE(OlapScanner::_has_usable_ngram_bf_pattern(usable, 3));
+    EXPECT_FALSE(OlapScanner::_has_usable_ngram_bf_pattern(too_short, 3));
+    EXPECT_FALSE(OlapScanner::_has_usable_ngram_bf_pattern(split_by_wildcard, 3));
+}
+
 TEST(OlapScannerTest, CandidateFallbackReasonsDoNotOverwriteEachOther) {
     RuntimeProfile profile("candidate fallback reasons");
 
@@ -271,6 +282,7 @@ TEST(OlapScannerTest, CandidateStatsMergeFullFileCacheAccounting) {
     candidate_stats.raw_rows_read = 11;
     candidate_stats.uncompressed_bytes_read = 12;
     candidate_stats.rows_inverted_index_filtered = 13;
+    candidate_stats.rows_bf_filtered = 14;
     candidate_stats.inverted_index_downgrade_count = 14;
     candidate_stats.inverted_index_lookup_timer = 15;
     candidate_stats.io_ns = 16;
@@ -293,6 +305,7 @@ TEST(OlapScannerTest, CandidateStatsMergeFullFileCacheAccounting) {
     EXPECT_EQ(11, total_stats.seq_map_candidate_scan_rows);
     EXPECT_EQ(12, total_stats.seq_map_candidate_scan_bytes);
     EXPECT_EQ(13, total_stats.seq_map_candidate_index_filtered_rows);
+    EXPECT_EQ(14, total_stats.seq_map_candidate_bloom_filter_filtered_rows);
     EXPECT_EQ(14, total_stats.seq_map_candidate_index_downgrades);
     EXPECT_EQ(15, total_stats.seq_map_candidate_index_lookup_ns);
     EXPECT_EQ(20, total_stats.seq_map_candidate_cache_local_bytes);

regression-test/data/unique_seq_map_p0/test_seq_map_candidate_key_scan.out

@@ -24,10 +24,22 @@
 1
 3
 
+-- !candidate_ngram_like --
+1
+
+-- !candidate_ngram_short_pattern --
+1
+
+-- !candidate_ngram_not_like --
+5002
+
 -- !candidate_index_disabled --
 1
 3
 
+-- !candidate_ngram_with_inverted_disabled --
+1
+
 -- !candidate_no_driver --
 2