…el scanners of a tablet #69352 (#69418)

### What problem does this PR solve?

Issue Number: None

Related PR: #69352

Problem Summary:

Backport of #69352 to branch-4.1.

A parallel scan (`ParallelScannerBuilder`) splits one tablet into
several scanners. Each `OlapScanner` then calls
`CollectionStatistics::collect()` with only the `rs_splits` of its own
split, so the BM25 doc count, avgdl and term doc frequencies come from a
subset of the tablet's rowsets. Rows with identical text get different
scores depending on which scanner reads them. Because the split depends
on the tablet's row count and the core count, a row's score and its rank
in `ORDER BY score()` change when unrelated rows are loaded, and
pagination over a score-ordered search returns repeated and missing
rows.

The reproduction in #69352 (8 rowsets with a different share of matching
rows, forced split with `parallel_scan_min_rows_per_scanner=1024`) gives
the 2,280 identical matching rows **15 distinct scores (1.34 to 5.81)**
on 4.1.4, 4.1.5 and branch-4.1. With `enable_parallel_scan=false` they
all get 3.647686.

Fix (same design as #69352, adapted to branch-4.1):

- `TabletCollectionStatistics` (in
`storage/compaction/collection_statistics.h`) holds the `rs_splits` of a
tablet's entire read source. The first scanner of the tablet to prepare
collects the statistics once with `CollectionStatistics::collect()` over
all of them, and every scanner of that tablet copies them into its own
`CollectionStatistics`. Later scanners wait for the first collection and
get its result or error.
- `CollectionStatistics::copy_collected_from()` copies the collected
counters and resets the derived idf/avgdl caches, so no cache is shared
between scanner threads.
- `ParallelScannerBuilder` creates one holder per tablet, only when the
scan has a `score()` runtime. Both the by-rowid and by-segment build
paths pass it to their scanners through `OlapScanner::Params`.
- The non-parallel path is unchanged. Each of its scanners already
receives the tablet's full read source.

Differences from #69352: on branch-4.1, `CollectionStatistics` lives in
`storage/compaction/`, has no `SearchPredicateCollector` constructor
argument and no `collect_full_collection()`, so the holder keeps the
read source's `rs_splits` and calls `collect()` on them.
`OlapScanner::Params` is initialized positionally on this branch, so the
new field is appended at the end.

Statistics remain per tablet. Making them consistent across tablets and
BEs is out of scope.

### Release note

Fix BM25 `score()` returning different scores for identical rows of the
same tablet when a parallel scan split the tablet into several scanners.

### Check List (For Author)

- Test
- [x] Regression test:
`inverted_index_p0/test_bm25_score_parallel_scan_statistics` (same suite
as #69352). Passes on branch-4.1 with this change.
- [x] Unit Test:
`CollectionStatisticsDetailedTest.CopyCollectedFromScoresLikeSource`,
`CollectionStatisticsTest.TabletCollectionStatisticsCollectsOnceAndCopiesToEveryCaller`
(not run locally; left to CI)
- [x] Manual test: ran the reproduction from #69352 against branch-4.1
builds on macOS arm64.

| Build | `enable_parallel_scan=true` (split forced) |
`enable_parallel_scan=false` |
      |---|---|---|
| branch-4.1, without this change | 15 distinct scores (1.34 to 5.81) |
1 score (3.647686) |
| branch-4.1 `95846ddf` with this change | 1 score (3.647686), same over
5 repeated runs | 1 score (3.647686) |
    - [ ] No need to test or manual test.

- Behavior changed:
- [x] Yes. Score queries that use a parallel scan now score every row of
a tablet against statistics of the whole tablet. Scores can differ from
earlier releases but are now independent of the scanner split.

- Does this need documentation?
    - [x] No.

### Check List (For Reviewer who merge this PR)

- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label

---------

Co-authored-by: Shivam Gupta <[email protected]>

…el scanners of a tablet #69352 (#69440)

### What problem does this PR solve?

Issue Number: None

Related PR: #69352

Problem Summary:

Backport of #69352 to branch-4.2.

A parallel scan (`ParallelScannerBuilder`) splits one tablet into
several scanners. Each `OlapScanner` then calls
`CollectionStatistics::collect()` with only the `rs_splits` of its own
split, so the BM25 doc count, avgdl and term doc frequencies come from a
subset of the tablet's rowsets. Rows with identical text get different
scores depending on which scanner reads them. Because the split depends
on the tablet's row count and the core count, a row's score and its rank
in `ORDER BY score()` change when unrelated rows are loaded, and
pagination over a score-ordered search returns repeated and missing
rows.

Fix (same as #69352):

- `TabletCollectionStatistics` holds the rowsets of a tablet's entire
read source. The first scanner of the tablet to prepare collects the
statistics once with `collect_full_collection()`, and every scanner of
that tablet copies them into its own `CollectionStatistics`. Later
scanners wait for the first collection and get its result or error.
- `CollectionStatistics::copy_collected_from()` copies the collected
counters and resets the derived idf/avgdl caches, so no cache is shared
between scanner threads.
- `ParallelScannerBuilder` creates one holder per tablet, only when the
scan has a `score()` runtime. Both the by-rowid and by-segment build
paths pass it to their scanners through `OlapScanner::Params`.
- The non-parallel path is unchanged. Each of its scanners already
receives the tablet's full read source.

Differences from #69352: branch-4.2 does not have the changelog, TSO and
bucket fields of `OlapScanner::Params` or the `scan_range` argument of
`_build_scanner`, so the scanner changes were applied by hand on top of
the branch-4.2 code. `OlapScanner::Params` is initialized positionally
on this branch, so the new field is appended at the end. branch-4.2 has
no test that calls `_build_scanner` directly, so the follow-up test fix
from #69352 is not needed here.

Statistics remain per tablet. Making them consistent across tablets and
BEs is out of scope.

### Release note

Fix BM25 `score()` returning different scores for identical rows of the
same tablet when a parallel scan split the tablet into several scanners.

### Check List (For Author)

- Test
- [x] Regression test:
`inverted_index_p0/test_bm25_score_parallel_scan_statistics` (same suite
as #69352). Passes on branch-4.2 with this change.
- [x] Unit Test:
`CollectionStatisticsDetailedTest.CopyCollectedFromScoresLikeSource`,
`TabletCollectionStatisticsTest.CollectsOnceAndCopiesToEveryCaller`
(compiled locally; run left to CI)
- [x] Manual test: ran the reproduction from #69352 against branch-4.2
builds on macOS arm64.

| Build | `enable_parallel_scan=true` (split forced) |
`enable_parallel_scan=false` |
      |---|---|---|
| branch-4.2 `a5d2661d`, without this change | 15 distinct scores (1.34
to 5.81) | 1 score (3.647686) |
| branch-4.2 `a5d2661d` with this change | 1 score (3.647686), same over
5 repeated runs | 1 score (3.647686) |
    - [ ] No need to test or manual test.

- Behavior changed:
- [x] Yes. Score queries that use a parallel scan now score every row of
a tablet against statistics of the whole tablet. Scores can differ from
earlier releases but are now independent of the scanner split.

- Does this need documentation?
    - [x] No.

### Check List (For Reviewer who merge this PR)

- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label

Co-authored-by: Shivam Gupta <[email protected]>

…erge-on-write tables (#69438)

### What problem does this PR solve?

Issue Number: None

Related PR: #69352 (BM25 statistics shared across the parallel scanners
of a tablet; this PR is its follow-up for replicas), #69313 (BM25 token
totals after CLucene compaction)

Problem Summary:

On a merge-on-write table, BM25 `score()` gives the same row a different
score on each replica of its tablet. A query picks one replica per
tablet, so with `replication_num = 3` a row has up to three scores.
Pagination over `ORDER BY score()` then repeats and skips rows from one
page to the next. Each replica is stable on its own (`use_fix_replica`
pins the scores), but the replicas disagree.

`CollectionStatistics::process_segment()` takes the document count, the
token count and each term's document frequency straight from every
segment's inverted index. Under merge-on-write, a row overwritten or
deleted by a later load stays in its segment's index, marked only in the
delete bitmap. A row written with the delete sign, which CDC deletes and
`DELETE` produce, stays as a normal row that queries filter out by
predicate. Both kinds are counted until compaction removes them, and
each replica compacts on its own schedule. So each replica counts a
different set of rows nobody can see.

Reproduction (3 BEs, `replication_num = 3`, 8 buckets, auto compaction
off): 8 loads, 3 rounds of overwriting a quarter of the rows, a
`DELETE`, and full rows inserted with `__DORIS_DELETE_SIGN__ = 1`. Then
a full compaction (`POST /api/compaction/run?compact_type=full`) of the
tablets on one BE only. On master with #69352, the query below changed
the score of all 3,799 matching rows from run to run: 1,900 distinct
scores instead of 950. With `use_fix_replica = 0/1/2` each replica was
stable, but the three disagreed.

```sql
SELECT id, score() FROM rt WHERE SEARCH('content:jobs OR content:try') ORDER BY score() DESC LIMIT 100000;
```

The token total has a second, replica-independent problem: CLucene index
compaction (`inverted_index_compaction_enable`, on by default) copies
the source token totals even when it drops rows, which #69313 addresses.
On master, the same rows scored differently in a compacted table and in
a table freshly loaded with its visible rows.

Fix:

When the scan reads a delete bitmap (merge-on-write),
`CollectionStatistics` describes only the rows the scan returns:

- `OlapScanner` passes the scan's delete bitmap and read version
(`MowHiddenRows`) to `collect()`, or to
`TabletCollectionStatistics::copy_to()` for a split tablet. Every
scanner of a tablet reads with the same bitmap and version, so the
statistics shared by #69352 stay correct.
- For each segment, the hidden rows are the rows the delete bitmap marks
at the read version (`get_agg`, the same call the rowset reader uses),
plus the rows whose hidden `__DORIS_DELETE_SIGN__` column is set. Every
unique-key table has that column, and `DELETE` on a merge-on-write table
writes such rows unless `enable_mow_light_delete` is on. The delete-sign
column is read only for segments whose zone map shows a non-zero sign
(or cannot tell), so a segment without CDC deletes costs nothing extra.
The read follows the query's `enable_segment_cache` and
`enable_page_cache`.
- Document count: the index's document count minus the hidden rows.
- Term document frequency: `IndexSource::scoring_doc_freq_among()`
counts how many hidden rows a term's postings list. When the hidden rows
are much rarer than the term's postings, it leaps between them with
`seek_block`. Otherwise it decodes the postings once and walks the
hidden rows alongside them, with no allocation, and a term without
postings costs nothing. SNII overrides it so that reserved-namespace
scoring terms resolve as `scoring_doc_freq` resolves them.
- Token count: the sum of the document lengths the norms of the returned
non-NULL rows stand for, or the returned row count for an index without
norms. The exact length of a hidden row is not stored anywhere: CLucene
and SNII keep only the encoded norm (exact up to 40 tokens for CLucene,
up to 254 for SNII). Subtracting decoded lengths from the exact
`sumTotalTermFreq` would still leave replicas apart. A row's norm is the
same in every segment it is written to, so summing norms gives every
replica the same count. It is also the same quantity the BM25 tf
normalization already divides, `dl / avgdl`, with `dl` taken from the
same norm.
- A NULL row has no tokens, but its norm stands for one (CLucene's
default norm, SNII's `encode_norm(0)`). Its length is left out, using
the index's null bitmap, and it still counts as a document, as before.
So on CLucene a table whose texts all have at most 40 tokens, and none
of which is an empty string, scores exactly as a duplicate-key table
with the same rows.
- The statistics read CLucene norms into a buffer of their own, one
segment at a time, and keep none. Asking a CLucene reader for a field's
norms makes it keep them for the searcher's lifetime, uncharged to the
searcher cache, and the statistics visit every segment of the read set,
including those the scan prunes and never scores. Reading per segment
also avoids the copy of all segments' norms a multi-segment reader would
keep.
- What the statistics keep instead is small and kept with the cached
searcher: the sum of a field's non-NULL norm lengths, and the sum of the
hidden rows' lengths for the last few sets of hidden rows of a field
(queries at one read version hide the same rows). So queries at an
unchanged version read no norms; a segment's norms are read again only
when its hidden rows change. The hidden and NULL row ID lists the
statistics build while collecting are charged to the query's memory.

Why the token count has to change too: a variant that left out hidden
rows only from the document count and the document frequencies, keeping
today's token total, made the replicas agree only while every replica
compacted through inverted index compaction, which carries dropped rows'
token counts forward. With one replica compacted with
`inverted_index_compaction_enable = false` (the index rebuilt from
column data), every row again scored differently on that replica. Its
scores were also 12–31 % off a fresh load of the visible rows on
average, because its token total still counted the hidden rows.

A tablet whose rows are all hidden (every row deleted or overwritten,
not yet compacted) has no document a query can match. Its statistics
report zero documents instead of failing the query with "No data
available for SimilarityCollector".

Tables without a delete bitmap (duplicate key, merge-on-read, or
`skip_delete_bitmap`) collect the same statistics as before.

For every `score()` query, with or without a delete bitmap:

- The statistics reads follow the query's `disable_file_cache` and
`inverted_index_snii_read_no_write_file_cache`, as the scan's reads do.
They run before the scan, so they are the reads that fill the file
cache. CLucene reads norms without a per-read context, as scoring does.
- A CLucene searcher the statistics open drops the query's I/O context
once its meta is read, as the query path already does
(`create_index_searcher`), and a CLucene error while opening it fails
the query; before, it escaped the scanner thread uncaught and aborted
the backend. Before, a searcher the statistics put into the searcher
cache kept the opening query's I/O context, including its file cache
statistics pointer, for later queries to read through.

Not covered:

- Merge-on-read unique tables and duplicate tables with `DELETE`
predicates also count rows that compaction later drops. They need merge
or predicate evaluation to tell which rows are visible, and are left for
a follow-up.
- With `skip_delete_sign = true` or `show_hidden_columns = true`, rows
written with the delete sign are returned but still left out of the
statistics. Rows deleted by a delete predicate
(`enable_mow_light_delete`) are still counted, as the delete bitmap does
not mark them.
- A row indexed without tokens that is not NULL keeps a norm that cannot
be told from a one-token row's, so on a merge-on-write table it counts
as one token: an analyzed empty string, which both formats index like a
NULL but leave out of the null bitmap, and on SNII any text that yields
no token (for example `'!!!'`, or text without ASCII letters or digits
under the `english` parser), whose norm SNII stores as that of one
token. On CLucene such a text has a zero norm and counts as zero tokens.
Telling these rows apart would mean reading the column.
- On SNII, the sum of the norm lengths is not cached yet: each
merge-on-write query reads the segment's norms.
- Leaving NULL rows out relies on the index's null bitmap, which CLucene
index compaction carries over; if reading a source bitmap fails there,
the compacted index loses it (existing behaviour of the vendored CLucene
compaction).
- Statistics stay per tablet, as before.

Interaction with #69313: that PR makes the token total of a CLucene
index built by compaction exact. With this PR, merge-on-write scans no
longer read that total, so their scores do not depend on it. Scans of
other tables still do. The two changes are independent.

…tistics of merge-on-write tables #69438 (#69752)

### What problem does this PR solve?

Issue Number: None

Related PR: #69438 (this is its backport to branch-4.2), #69352 and its
branch-4.2 pick #69440 (BM25 statistics shared across the parallel
scanners of a tablet)

Problem Summary:

Backport of #69438 to branch-4.2.

On a merge-on-write table, BM25 `score()` gives the same row a different
score on each replica of its tablet. A query picks one replica per
tablet, so with `replication_num = 3` a row has up to three scores, and
pagination over `ORDER BY score()` repeats and skips rows from one page
to the next. `CollectionStatistics::process_segment()` takes the
document count, the token count and each term's document frequency
straight from every segment's inverted index. Rows a later load
overwrote or deleted (marked only in the delete bitmap) and rows written
with the delete sign (CDC deletes, `DELETE`) stay in the index and are
counted until compaction removes them, and each replica compacts on its
own schedule.

On branch-4.2 (3 BEs, `replication_num = 3`, 8 buckets, CDC-style
upserts and delete-sign deletes, the replicas on one BE fully
compacted), all 1,731 matching rows of a `SEARCH(... OR ...)` query got
more than one score over 15 runs and differed between `use_fix_replica =
0/1/2`.

Fix: the same change as #69438. When the scan reads a delete bitmap
(merge-on-write), `CollectionStatistics` describes only the rows the
scan returns:

- `OlapScanner` passes the scan's delete bitmap and read version
(`MowHiddenRows`) to `collect()`, or to
`TabletCollectionStatistics::copy_to()` for a split tablet (#69440).
- Per segment, the hidden rows are the rows the delete bitmap marks at
the read version plus the rows whose `__DORIS_DELETE_SIGN__` is set. The
delete-sign column is read only for segments whose zone map shows a
non-zero sign (or cannot tell), following the query's
`enable_segment_cache` and `enable_page_cache`.
- Document count: the index's document count minus the hidden rows. Term
document frequency: `IndexSource::scoring_doc_freq_among()` subtracts
the hidden rows a term's postings list, leaping between them when they
are much rarer than the postings, otherwise walking them alongside the
decoded postings.
- Token count: the sum of the document lengths the norms of the returned
non-NULL rows stand for (NULL rows left out through the index's null
bitmap), or the returned row count for an index without norms. Every
replica counts the same, compacted or not.
- CLucene norms are read one segment at a time into a buffer of the
statistics' own and not kept; the non-NULL norm length sum of a field
and the length sums of its last four sets of hidden rows are kept with
the cached searcher (`CluceneSearcherCacheValue`).
- A tablet whose rows are all hidden reports zero documents instead of
failing the query.

Tables without a delete bitmap collect the same statistics as before.
For every `score()` query, the statistics' I/O context follows
`disable_file_cache` and `inverted_index_snii_read_no_write_file_cache`,
and a CLucene searcher the statistics open drops the query's I/O context
once its meta is read; a CLucene error while opening it fails the query.

Differences from #69438, for branch-4.2: segments are identified by
their id instead of `RowsetSegmentView`, and the delete-sign column
reader comes from `Segment::get_column_reader()` (branch-4.2 has no
`get_column_reader_for_pruning`), a segment without the column reading
as no deletes. The tests are adapted the same way.

Not covered (as in #69438): merge-on-read unique tables and duplicate
tables with `DELETE` predicates; rows deleted by a delete predicate
(`enable_mow_light_delete`); rows written with the delete sign that the
query returns (`skip_delete_sign`, `show_hidden_columns`) are still left
out; an analyzed empty string, and on SNII any text that yields no
token, counts as one token on a merge-on-write table; SNII norm length
sums are not cached. Statistics stay per tablet.

### Release note

Fix BM25 `score()` on merge-on-write tables differing between the
replicas of a tablet, and before and after compaction. Statistics now
count only the rows a query can return, and the average document length
comes from the indexed norms.

### Check List (For Author)

- Test
    - [x] Regression test:
- New: `inverted_index_p0/test_bm25_score_replica_statistics` and
`fault_injection_p0/test_bm25_score_statistics_searcher_io_context`
(same suites as #69438). Updated `.out`: `test_bm25_score.out`,
`test_score_filter_matrix.out` (V2, V3 and SNII merge-on-write cases),
same values as #69438.
- All 26 branch-4.2 suites that use `score()` (`inverted_index_p0`,
`search`, `ann_index_p0`, `variant_p0`, `nereids_rules_p0`,
`fault_injection_p0`) pass with this change on a 3-BE cluster. Without
it, the replica suite, the fault-injection suite (the query fails with
`[E-6002] The 'is_index_data' flag should be false for non-meta inverted
index files.`) and the two updated suites fail.
- [x] Unit Test: the tests of #69438
(`CollectionStatisticsTest.HiddenRowsAreCollectedLikeCompactedAway`,
`AllRowsHiddenCollectsEmptyStatistics`,
`NullRowsCountAsDocumentsWithoutTokens`,
`CollectionStatisticsHiddenRowsTest.*` for `count_listed_docs`,
`exclude_hidden_docs` and `exclude_hidden_term_docs`,
`CluceneSearcherCacheValueTest.*`,
`MultiSegmentCollectorTest.NormLengthSumsReadEachSegment`,
`OlapScannerTest.BuildScoreRuntimeCollectionIoContextFollowsQueryFileCacheOptions`),
with the existing collection statistics, BM25, searcher cache,
multi-segment, query engine, SNII and `SEARCH` function tests: 427 tests
in 35 suites pass in an ASAN build on macOS arm64.
- [x] Manual test (3 BEs, macOS arm64, `replication_num = 3`, auto
compaction off, `enable_sql_cache = false`), base = branch-4.2 without
this change, same data and compaction state for both builds:

        | Check | branch-4.2 | with this change |
        |---|---|---|
| CDC-style table (8 buckets, upserts, delete-sign deletes, NULLs, 4
fields), one BE's replicas fully compacted with inverted index
compaction off: rows with more than one score over 15 runs / differing
between replicas 0, 1, 2 (of 1,731) | 1,731 / 1,731 | 0 / 0 |
| Same, scores differing from a freshly loaded copy of the visible rows
| — | 0 of 1,731 |
| Partial-column updates, sequence column, stream load with the delete
sign and `merge_type: DELETE`, group commit, keyword and
lowercase-normalizer indexes, partitions, variant, SNII: rows differing
between replicas after one replica compacted | 84–790 per table (8 of 9
checks) | 0 in every check, all equal to a fresh copy |
| Merge-on-write vs duplicate table with 50 % NULL rows, V3 and SNII | 0
of 250 differ | 0 of 250 differ |

Cost, query `SEARCH('content:jobs OR content:try') ORDER BY score() DESC
LIMIT 100`, one bucket, median of 15 warm runs; `StatisticsCollectTime`
with one scanner, query total of `SEARCH('content:jobs')` with the
default parallel scan:

        | Table | Statistics | Query total |
        |---|---|---|
| V2, 8 M rows + 200 k overwritten + 50 k delete-sign rows | 0.3 ms →
6.7 ms | 68 ms → 76 ms |
| V3, 6 M rows + 2 M overwritten + 200 k delete-sign rows (2.2 M hidden,
6 % NULL) | 0.3 ms → 55 ms | 61 ms → 113 ms |
| V2, 6 M rows + 2 M overwritten, no NULL rows | 0.3 ms → 7.2 ms | 56 ms
→ 61 ms |
| SNII, 2 M rows + 500 k overwritten + 200 k delete-sign rows | 0.2 ms →
23 ms | 49 ms → 71 ms |

The cost grows with the rows compaction has not removed yet. With NULL
rows, most of it is leaving the NULL rows out of the hidden rows on
every query (`exclude_hidden_docs`), as in #69438.

Memory (BE allocated bytes, sum of 3 BEs, a 4-field merge-on-write table
of 41 segments): no growth beyond the noise of the measurement (after 4
queries +26 MB → +25 MB with key pruning; +142 MB → below the starting
value scoring all segments).

- Behavior changed:
- [x] Yes. On merge-on-write tables, BM25 statistics leave out rows
hidden by the delete bitmap and rows with the delete sign, and avgdl is
computed from the norms of the visible rows, so scores no longer depend
on the replica or on compaction. Scores of merge-on-write tables shift
once, also without deletes, because norms store long lengths
approximately.

- Does this need documentation?
    - [x] No.

### Check List (For Reviewer who merge this PR)

- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label

Co-authored-by: Shivam Gupta <[email protected]>

…tistics of merge-on-write tables #69438 (#69751)

### What problem does this PR solve?

Issue Number: None

Related PR: #69438 (this is its backport to branch-4.1), #69352 and its
branch-4.1 pick #69418 (BM25 statistics shared across the parallel
scanners of a tablet)

Problem Summary:

Backport of #69438 to branch-4.1.

On a merge-on-write table, BM25 `score()` gives the same row a different
score on each replica of its tablet. A query picks one replica per
tablet, so with `replication_num = 3` a row has up to three scores, and
pagination over `ORDER BY score()` repeats and skips rows from one page
to the next. `CollectionStatistics::process_segment()` takes the
document count, the token count and each term's document frequency
straight from every segment's inverted index. Rows a later load
overwrote or deleted (marked only in the delete bitmap) and rows written
with the delete sign (CDC deletes, `DELETE`) stay in the index and are
counted until compaction removes them, and each replica compacts on its
own schedule.

On branch-4.1 (3 BEs, `replication_num = 3`, 8 buckets, CDC-style
upserts and delete-sign deletes, the replicas on one BE fully
compacted), the 1,731 matching rows of a `SEARCH(... OR ...)` query had
1,294 rows with more than one score over 15 runs, and all 1,731 differed
between `use_fix_replica = 0/1/2`.

Fix (same as #69438):

When the scan reads a delete bitmap (merge-on-write),
`CollectionStatistics` describes only the rows the scan returns:

- `OlapScanner` passes the scan's delete bitmap and read version
(`MowHiddenRows`) to `collect()`, or to
`TabletCollectionStatistics::copy_to()` for a split tablet (#69418).
Every scanner of a tablet reads with the same bitmap and version.
- For each segment, the hidden rows are the rows the delete bitmap marks
at the read version (`get_agg`), plus the rows whose
`__DORIS_DELETE_SIGN__` is set. The delete-sign column is read only for
segments whose zone map shows a non-zero sign (or cannot tell),
following the query's `enable_segment_cache` and `enable_page_cache`.
- Document count: the index's document count minus the hidden rows.
- Term document frequency: the hidden rows a term's postings list are
subtracted. When the hidden rows are much rarer than the postings (32
times), the collector leaps from hidden row to hidden row; otherwise it
decodes the postings once and walks the hidden rows alongside, without a
bitset.
- Token count: the sum of the document lengths the norms of the returned
non-NULL rows stand for, or the returned row count for an index without
norms. A row's norm is the same in every segment it is written to, so
every replica counts the same, compacted or not. NULL rows are left out
of the token count through the index's null bitmap and still count as
documents.
- The statistics read CLucene norms one CLucene segment at a time into a
buffer of their own (`IndexReader::norms(field, bytes)`) and keep none.
What they keep is small and lives with the cached searcher: a field's
non-NULL norm length sum, and the length sums of the last four sets of
hidden rows of a field. The hidden and NULL row ID lists are charged to
the query's memory.
- A tablet whose rows are all hidden reports zero documents instead of
failing the query with "No data available for SimilarityCollector".

Tables without a delete bitmap (duplicate key, merge-on-read,
`skip_delete_bitmap`) collect the same statistics as before.

For every `score()` query: the statistics' I/O context follows the
query's `disable_file_cache`, and a CLucene searcher the statistics open
drops the query's I/O context once its meta is read, as the query path
does. A CLucene error while opening it now fails the query; before, it
escaped the scanner thread and aborted the backend (reproduced on
branch-4.1 by
`fault_injection_p0/test_bm25_score_statistics_searcher_io_context`).

Differences from #69438, for branch-4.1:

- branch-4.1 has no SNII format and no `IndexSource` layer. The
hidden-row counting works on `TermIterator` (`read_range` / `advance`)
and the norm reading on the CLucene reader directly; both are helpers in
`storage/compaction/collection_statistics.cpp`
(`collection_statistics_detail`).
- The norm length sums are kept on
`InvertedIndexSearcherCache::CacheValue`; branch-4.1 has no
`CluceneSearcherCacheValue`.
- `InvertedIndexReader::read_null_bitmap` is split so that the
statistics can read the null bitmap through the query cache without an
`InvertedIndexReader` (`read_inverted_index_null_bitmap()`).
- `inverted_index_snii_read_no_write_file_cache` does not exist on
branch-4.1, so only `disable_file_cache` is followed.

Not covered (as in #69438): merge-on-read unique tables and duplicate
tables with `DELETE` predicates; rows deleted by a delete predicate
(`enable_mow_light_delete`); rows written with the delete sign that the
query returns (`skip_delete_sign`, `show_hidden_columns`) are still left
out of the statistics; an analyzed empty string counts as one token on a
merge-on-write table. Statistics stay per tablet.

### Release note

Fix BM25 `score()` on merge-on-write tables differing between the
replicas of a tablet, and before and after compaction. Statistics now
count only the rows a query can return, and the average document length
comes from the indexed norms.

### Check List (For Author)

- Test
    - [x] Regression test:
- New: `inverted_index_p0/test_bm25_score_replica_statistics` (3
replicas, overwrites, `DELETE`, delete-sign rows, one replica fully
compacted, scores compared with a freshly loaded table; NULL rows vs a
duplicate table; all rows deleted) and
`fault_injection_p0/test_bm25_score_statistics_searcher_io_context`
(same suites as #69438).
- Updated `.out`: `test_bm25_score.out` and the V2/V3 merge-on-write
cases of `test_score_filter_matrix.out`, to the same values as #69438.
- All 23 branch-4.1 suites that use `score()` (`inverted_index_p0`,
`search`, `ann_index_p0`, `variant_p0`, `nereids_rules_p0`,
`fault_injection_p0`) pass with this change on a 3-BE cluster. Without
it, the two new suites and the two updated ones fail; the
fault-injection suite aborts two of the three backends with an uncaught
`CLuceneError`.
- [x] Unit Test:
`CollectionStatisticsTest.HiddenRowsAreCollectedLikeCompactedAway` (real
CLucene segments: a segment with a hidden 45-token row collects the same
doc count, token count, df, avgdl and idf as the segment without it),
`CollectionStatisticsTest.AllRowsHiddenCollectsEmptyStatistics`,
`CollectionStatisticsTest.NullRowsCountAsDocumentsWithoutTokens` (real
segment with NULL rows, with and without a delete bitmap),
`CollectionStatisticsTest.CountListedDocsLeapsOrScansOverHiddenRows`
(both counting paths, dense blocks, norm length sums),
`CollectionStatisticsHiddenRowsTest.*` (`exclude_hidden_docs` with and
without norms and NULL rows, rows beyond the index),
`InvertedIndexSearcherCacheValueTest.*` (norm length sums computed once
per field and per set of hidden rows, failures not kept, oldest set
evicted), `MultiSegmentCollectorTest.NormLengthSumsReadEachSegment` (an
index of three CLucene segments sums its norms per segment, equal to the
combined norms),
`OlapScannerTest.BuildScoreRuntimeCollectionIoContextFollowsQueryFileCacheOptions`.
With the existing collection statistics, collection similarity, BM25,
searcher cache, searcher builder, olap scanner, multi-segment, boolean
query, null predicate and `SEARCH` function tests: 233 tests in 26
suites pass in an ASAN build on macOS arm64.
- [x] Manual test (3 BEs, macOS arm64, `replication_num = 3`, auto
compaction off, `enable_sql_cache = false`), base = branch-4.1 without
this change, same data and compaction state for both builds:

        | Check | branch-4.1 | with this change |
        |---|---|---|
| CDC-style table (8 buckets, upserts, delete-sign deletes, NULLs, 4
fields), one BE's replicas fully compacted with inverted index
compaction off: rows with more than one score over 15 runs / differing
between replicas 0, 1, 2 (of 1,731) | 1,294 / 1,731 | 0 / 0 |
| Same, scores differing from a freshly loaded copy of the visible rows
| — | 0 of 1,731 |
| Partial-column updates, sequence column, stream load with the delete
sign and `merge_type: DELETE`, group commit, keyword and
lowercase-normalizer indexes, partitions, variant: rows differing
between replicas after one replica compacted | 84–790 per table (7 of 8
checks) | 0 in every check, all equal to a fresh copy |
| Merge-on-write vs duplicate table with 50 % NULL rows (V3) | 0 of 250
differ | 0 of 250 differ |

Cost, query `SEARCH('content:jobs OR content:try') ORDER BY score() DESC
LIMIT 100`, one bucket, median of 15 warm runs:

| Table | `StatisticsCollectTime`, one scanner | Query total, default
parallel scan |
        |---|---|---|
| V2, 8 M rows + 200 k overwritten + 50 k delete-sign rows | 0.3 ms →
7.7 ms | 78 ms → 82 ms |
| V3, 6 M rows + 2 M overwritten + 200 k delete-sign rows (2.2 M hidden)
| 0.3 ms → 62 ms | 71 ms → 125 ms |

The first query after a restart also sums each segment's norms (8 M
rows: 5 ms → 32 ms; 2.2 M hidden: 6 ms → 99 ms). The cost grows with the
rows compaction has not removed yet.

Memory (BE allocated bytes, sum of 3 BEs, a 4-field merge-on-write table
of 41 segments): after 4 queries +16 MB → +22 MB with key pruning and
+109 MB → +75 MB scoring all segments, within the noise of the
measurement.

- Behavior changed:
- [x] Yes. On merge-on-write tables, BM25 statistics leave out rows
hidden by the delete bitmap and rows with the delete sign, and avgdl is
computed from the norms of the visible rows, so scores no longer depend
on the replica or on compaction. Scores of merge-on-write tables shift
once, also without deletes, because norms store lengths above 40 tokens
approximately.

- Does this need documentation?
    - [x] No.

### Check List (For Reviewer who merge this PR)

- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label

---------

Co-authored-by: Shivam Gupta <[email protected]>

### What problem does this PR solve?

Issue Number: None

Related PR: apache#69352, apache#69438, apache#69403

Problem Summary:
MoW light deletes filter rows through versioned DELETE predicates, but BM25
statistics only excluded delete-bitmap and delete-sign rows. Reuse DeleteHandler
with the captured read version and historical column identities to collect the
additional hidden row IDs. Read only predicate columns in bounded batches and
merge disjoint, tracked row-ID lists into the existing statistics path.
Keep the upstream tablet sharing, per-scanner caches and norms-based avgdl.

### Release note

Correct BM25 statistics after merge-on-write light deletes without waiting for
compaction.

### Check List (For Author)

- Test: 93 selected ASAN tests passed, including three real-segment predicate
  tests. Full-master SQL passed V2/V3 with phrase true/false and SNII with phrase
  true, including mutation phases and forced scanner splitting. Existing parallel
  scan and replica-statistics suites passed. The unfixed query path fails the new
  regression's initial score comparison. Format, hygiene and changed-line source
  and test clang-tidy passed.
- Behavior changed: Yes. Light-deleted rows no longer affect MoW BM25 statistics.
- Does this need documentation: No new configuration or syntax.