…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]>