[improvement](scan) Align scanner split to segments for MATCH

`SegmentIterator::_lazy_init` runs the inverted index query over the whole
segment first and only then intersects the result with `_opts.row_ranges`.
The cost of a MATCH predicate is therefore priced per segment, while
`ParallelScannerBuilder` splits by rows. When a segment ends up shared by two
scanners, its posting lists are walked once per scanner.

Add `_build_scanners_by_match_expr`, which collects whole segments and closes
a split once `_rows_per_scanner` is reached, so split boundaries only fall
between segments. A segment larger than the threshold takes a scanner of its
own. `segment_row_ranges` is left empty so the downstream reader scans whole
segments instead of intersecting a range covering the entire segment.

The path is taken only for DUP/MOW tables in cloud mode when the expressions
pushed down to the storage layer contain MATCH; `_build_scanners_by_rowid` and
`_build_scanners_by_per_segment` are unchanged, and a query without MATCH
behaves exactly as before. The `tablet_hotspot().count()` call shared by all
three strategies is hoisted into `build_scanners()`.

Measured on a cloud cluster, 20M rows in one tablet over 8 segments, cold
query (file cache cleared and BE restarted between runs), 3 runs each:

                                  before     after
  NumScanners                         10         5
  InvertedIndexQueryTime          5299ms    2576ms   -51%
  ScannerCpuTime                  5408ms    2796ms   -48%
  InvertedIndexSearcherOpenTime    144ms      58ms   -60%
  ScannerGetBlockTime             1529ms    1761ms   +15%
  Total                            931ms     821ms

This trades scan parallelism for total CPU: the number of scanners drops, so
`ScannerGetBlockTime` grows, while the index query work roughly halves.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VgV4Zy5NaEYpUP4Yrh1yWn

Author: csun5285

be/src/exec/operator/olap_scan_operator.cpp

@@ -460,6 +460,22 @@ static bool contains_expr_node_type(const VExprSPtr& expr, TExprNodeType::type n
     });
 }
 
+// Find MATCH recursively; ones nested in AND / OR / NOT count too.
+static bool is_match_expr(const VExprSPtr& expr) {
+    if (expr == nullptr) {
+        return false;
+    }
+    if (expr->node_type() == TExprNodeType::MATCH_PRED ||
+        expr->node_type() == TExprNodeType::SEARCH_EXPR || expr->can_push_down_to_index()) {
+        return true;
+    }
+    if (is_match_expr(expr->get_impl())) {
+        return true;
+    }
+    return std::ranges::any_of(expr->children(),
+                               [](const auto& child) { return is_match_expr(child); });
+}
+
 static Status validate_residual_scan_conjuncts(RuntimeState* state,
                                                TPushAggOp::type push_down_agg_type,
                                                const VExprContextSPtrs& conjuncts) {
@@ -635,6 +651,13 @@ bool OlapScanLocalState::_is_binlog_merge_scan() const {
     return scan_type == TBinlogScanType::MIN_DELTA || scan_type == TBinlogScanType::DETAIL;
 }
 
+bool OlapScanLocalState::_has_match_in_expr() const {
+    // Look for MATCH in the expressions pushed down to the storage layer.
+    return std::ranges::any_of(_common_expr_ctxs_push_down, [](const auto& ctx) {
+        return ctx != nullptr && is_match_expr(ctx->root());
+    });
+}
+
 Status OlapScanLocalState::_init_scanners(std::list<ScannerSPtr>* scanners) {
     if (_scan_ranges.empty()) {
         _eos = true;
@@ -764,6 +787,7 @@ Status OlapScanLocalState::_init_scanners(std::list<ScannerSPtr>* scanners) {
                 std::max<int64_t>(1024, state()->parallel_scan_min_rows_per_scanner());
         scanner_builder.set_max_scanners_count(max_scanners_count);
         scanner_builder.set_min_rows_per_scanner(min_rows_per_scanner);
+        scanner_builder.set_has_match_in_expr(_has_match_in_expr());
         // If the session variable is set, force one scanner per segment.
         if (state()->query_options().__isset.optimize_index_scan_parallelism &&
             state()->query_options().optimize_index_scan_parallelism) {

be/src/exec/operator/olap_scan_operator.h

@@ -135,6 +135,10 @@ class OlapScanLocalState final : public ScanLocalState<OlapScanLocalState> {
 
     Status _init_scanners(std::list<ScannerSPtr>* scanners) override;
 
+    // Whether the expressions pushed down to the storage layer contain MATCH. If so, the scan
+    // cost is priced per segment, and the split no longer cuts a segment apart.
+    bool _has_match_in_expr() const;
+
     Status _build_key_ranges_and_filters();
 
     bool _is_tablet_pruned_by_runtime_filter(int64_t partition_id, int32_t bucket_seq,

be/src/exec/scan/parallel_scanner_builder.cpp

@@ -66,8 +66,20 @@ io::IOContext create_preload_io_context(RuntimeState* state, OlapReaderStatistic
 
 Status ParallelScannerBuilder::build_scanners(std::list<ScannerSPtr>& scanners) {
     RETURN_IF_ERROR(_load());
+
+    if (config::is_cloud_mode()) {
+        for (const auto& tablet_with_version : _tablets) {
+            // FIXME(plat1ko): Avoid pointer cast
+            ExecEnv::GetInstance()->storage_engine().to_cloud().tablet_hotspot().count(
+                    *tablet_with_version.tablet);
+        }
+    }
+
     if (_scan_parallelism_by_per_segment) {
         return _build_scanners_by_per_segment(scanners);
+    } else if (_is_dup_mow_key && config::is_cloud_mode() && _has_match_in_expr) {
+        // MATCH is priced per segment; cutting a segment apart runs its index query several times
+        return _build_scanners_by_match_expr(scanners);
     } else if (_is_dup_mow_key) {
         // Default strategy for DUP/MOW tables: split by rowids within segments
         return _build_scanners_by_rowid(scanners);
@@ -86,11 +98,6 @@ Status ParallelScannerBuilder::_build_scanners_by_rowid(std::list<ScannerSPtr>&
         DCHECK(_all_read_sources.contains(tablet->tablet_id()));
         auto& entire_read_source = _all_read_sources[tablet->tablet_id()];
 
-        if (config::is_cloud_mode()) {
-            // FIXME(plat1ko): Avoid pointer cast
-            ExecEnv::GetInstance()->storage_engine().to_cloud().tablet_hotspot().count(*tablet);
-        }
-
         // `rs_splits` in `entire read source` will be devided into several partitial read sources
         // to build several parallel scanners, based on segment rows number. All the partitial read sources
         // share the same delete predicates from their corresponding entire read source.
@@ -203,6 +210,88 @@ Status ParallelScannerBuilder::_build_scanners_by_rowid(std::list<ScannerSPtr>&
     return Status::OK();
 }
 
+// Split boundaries only fall between segments: collect whole segments, and close the split once
+// _rows_per_scanner is reached. A large segment exceeds the threshold as soon as it comes in, so it
+// takes a scanner of its own.
+//
+// Example, rows_per_scanner = 500, segment rows in units of 10k:
+//
+//   rs0=[100]   rs1=[1200]   rs2=[100, 1200, 100]   rs3=[200]
+//
+//     Scanner#1 = rs0[0,1) + rs1[0,1)   1300
+//     Scanner#2 = rs2[0,2)              1300
+//     Scanner#3 = rs2[2,3) + rs3[0,1)    300
+Status ParallelScannerBuilder::_build_scanners_by_match_expr(std::list<ScannerSPtr>& scanners) {
+    DCHECK_GE(_rows_per_scanner, _min_rows_per_scanner);
+    const auto rows_per_scanner = static_cast<int64_t>(_rows_per_scanner);
+
+    for (size_t tablet_idx = 0; tablet_idx < _tablets.size(); ++tablet_idx) {
+        auto&& [tablet, version] = _tablets[tablet_idx];
+        const auto& scan_range = *_scan_ranges[tablet_idx];
+        DCHECK(_all_read_sources.contains(tablet->tablet_id()));
+        auto& entire_read_source = _all_read_sources[tablet->tablet_id()];
+
+        TabletReadSource partitial_read_source;
+        int64_t rows_collected = 0;
+
+        // Hand the collected splits to a new scanner
+        auto flush_scanner = [&]() {
+            DCHECK(!partitial_read_source.rs_splits.empty());
+            scanners.emplace_back(
+                    _build_scanner(tablet, version, _key_ranges, scan_range,
+                                   {.rs_splits = std::move(partitial_read_source.rs_splits),
+                                    .delete_predicates = entire_read_source.delete_predicates,
+                                    .delete_bitmap = entire_read_source.delete_bitmap},
+                                   take_initial_file_cache_stats(&_tablet_preload_file_cache_stats,
+                                                                 tablet->tablet_id())));
+            partitial_read_source = {};
+            rows_collected = 0;
+        };
+
+        for (auto& rs_split : entire_read_source.rs_splits) {
+            auto reader = rs_split.rs_reader;
+            auto rowset = reader->rowset();
+            if (rowset->num_rows() == 0) {
+                continue;
+            }
+            const auto& segments_rows = _all_segments_rows[rowset->rowset_id()];
+
+            int64_t segment_start = 0;
+
+            // Leave segment_row_ranges empty; the downstream scans whole segments accordingly.
+            auto close_split = [&](int64_t end) {
+                if (end <= segment_start) {
+                    // The last segment just closed a split; no segment is left in this rowset
+                    return;
+                }
+                RowSetSplits split(reader->clone());
+                split.segment_offsets = {segment_start, end};
+                partitial_read_source.rs_splits.emplace_back(std::move(split));
+                segment_start = end;
+            };
+
+            for (size_t i = 0; i != segments_rows.size(); ++i) {
+                rows_collected += static_cast<int64_t>(segments_rows[i]);
+                if (rows_collected >= rows_per_scanner) {
+                    close_split(static_cast<int64_t>(i) + 1);
+                    flush_scanner();
+                }
+            }
+
+            // Segments at the tail that did not fill up wait for the next rowset: in the example
+            // rs0 is held here until rs1
+            close_split(static_cast<int64_t>(segments_rows.size()));
+        }
+
+        // Whatever did not fill up by the end still has to be sent: rs2[2,3) + rs3[0,1) in the example
+        if (rows_collected > 0) {
+            flush_scanner();
+        }
+    }
+
+    return Status::OK();
+}
+
 // Build scanners so that each segment is exclusively scanned by a single scanner.
 // This guarantees the number of scanners equals the number of segments across all rowsets
 // for the involved tablets. It preserves delete predicates and key ranges, and clones
@@ -216,11 +305,6 @@ Status ParallelScannerBuilder::_build_scanners_by_per_segment(std::list<ScannerS
         DCHECK(_all_read_sources.contains(tablet->tablet_id()));
         auto& entire_read_source = _all_read_sources[tablet->tablet_id()];
 
-        if (config::is_cloud_mode()) {
-            // FIXME(plat1ko): Avoid pointer cast
-            ExecEnv::GetInstance()->storage_engine().to_cloud().tablet_hotspot().count(*tablet);
-        }
-
         // For each RowSet split in the read source, split by segment id and build
         // one scanner per segment. Keep delete predicates shared.
         for (auto& rs_split : entire_read_source.rs_splits) {

be/src/exec/scan/parallel_scanner_builder.h

@@ -72,13 +72,17 @@ class ParallelScannerBuilder {
 
     void set_scan_parallelism_by_per_segment(bool v) { _scan_parallelism_by_per_segment = v; }
 
+    void set_has_match_in_expr(bool v) { _has_match_in_expr = v; }
+
     const OlapReaderStatistics* builder_stats() const { return &_builder_stats; }
 
 private:
     Status _load();
 
     Status _build_scanners_by_rowid(std::list<ScannerSPtr>& scanners);
 
+    Status _build_scanners_by_match_expr(std::list<ScannerSPtr>& scanners);
+
     // Build scanners so that each segment is handled by its own scanner.
     Status _build_scanners_by_per_segment(std::list<ScannerSPtr>& scanners);
 
@@ -106,6 +110,9 @@ class ParallelScannerBuilder {
     // Force building one scanner per segment when true.
     bool _scan_parallelism_by_per_segment {false};
 
+    // Set when the pushed-down expressions contain MATCH; filled in by the scan operator.
+    bool _has_match_in_expr {false};
+
     std::shared_ptr<RuntimeProfile> _scanner_profile;
     OlapReaderStatistics _builder_stats;
     RuntimeState* _state;