Draft: refactor Read path iterators - #69515

…r read mode

Issue Number: N/A

Related PR: N/A

Problem Summary:

BlockReader picked one of six `_next_block` member functions through a function pointer and
kept the state of all of them in the same members, so one field meant different things in
different read modes: `_stored_data_columns` was the aggregate value buffer of AGG_KEYS reads
and the two-row key-group buffer of MIN_DELTA scans; `_agg_columns_idx` was both "the columns
to aggregate" and "the columns `_insert_data_normal` skips"; `_next_row` was the prefetched
merge row of the merging modes and a dangling reference in direct mode; `_eof` stood for three
different end conditions; and the members that only AGG, UNIQUE or changelog reads use were
visible from every path. Reading one mode meant reading them all.

Every (table model, read mode) pair now has its own iterator that owns only its own state:

  DirectBlockIterator            DUP_KEYS, merge-on-write UNIQUE_KEYS queries, direct mode
  UniqueKeyBlockIterator         merge-on-read UNIQUE_KEYS without a sequence map
  SequenceMapBlockIterator       merge-on-read UNIQUE_KEYS with a sequence map
  AggKeyBlockIterator            AGG_KEYS (the value folding lives in AggValueAggregator)
  ChangelogMinDeltaBlockIterator changelog MIN_DELTA scans
  ChangelogDetailBlockIterator   changelog DETAIL scans

BlockReader keeps the rowset capture, the merge of the rows and the choice of the iterator
(`_init_block_iterator`, one documented table). The prefetched merge row the merging
iterators need is a `MergedRowCursor`, the output row/byte limits are an
`OutputBlockLimits`, and the UPDATE AFTER row the changelog scans carry across a Block
boundary is a `ChangelogPendingRow`. The row-level algorithms are unchanged.

Two error paths change: a non-EOF error while positioning on the first merged row is
returned instead of swallowed, and an unknown keys type returns an InternalError instead of
failing a DCHECK.

None

- Test: Unit Test
    - New block_reader_table_models_test.cpp writes real segments for every table model and
      read mode (DUP query; AGG query, direct mode and compaction; UNIQUE merge-on-read query
      and base compaction with delete sign; merge-on-write query with a delete bitmap;
      sequence-map query; changelog MIN_DELTA and DETAIL scans) and checks both the iterator
      BlockReader picks and the rows it returns; new agg_value_aggregator_test.cpp; the
      existing BlockReader tests drive the iterator classes directly.
    - Ran BlockReader*, AggValueAggregator*, *Changelog*, TestRowIdConversion*,
      VerticalCompactionTest*, OrderedDataCompactionTest*, SegCompactionTest*,
      VerticalBlockReaderTest*, CollectedEnough* and EstimateCollectedEnough*: 199 tests pass.
- Behavior changed: No
- Does this need documentation: No

Co-Authored-By: Claude Fable 5.1 <[email protected]>

… segment iterators

Issue Number: N/A

Related PR: apache#4958, apache#7785, apache#35432

Problem Summary:

Between BlockReader and SegmentIterator a read went through seven objects of four kinds, and
the merge of the rows was done twice: BetaRowsetReader merged the overlapping segments of each
rowset with a VMergeIterator, then VCollectIterator merged the rowsets with a two-level heap
(the rowset with the most rows against a sub-heap of the others, PR apache#4958, a row-storage era
optimization). Every layer had its own way of saying "same key as the previous row" (the
segment merge set it on the loser of a comparison and handed it up as a BlockWithSameBit; the
rowset merge set it again; Level0Iterator carried it across Blocks), its own tie-break between
versions of a key (data_id in the segment merge, the rowset version in the rowset merge), its
own end-of-data signal (EndOfFile with rows still in the Block, OK with no rows, row_pos == -1,
_rs_reader == nullptr, _cur_child == nullptr) and its own mode flags (Level0Iterator read by
Block or by BlockView depending on the child, Level1Iterator merged or concatenated depending
on a flag). The three bugs fixed by the overlapping-segments PR all sat on the hand-overs
between these layers.

The read path is now one iterator per layer, each with one contract:

  BlockReader                 captures the rowsets, builds the rows of the read, picks the
                              TabletBlockIterator of the read mode
   └ TabletBlockIterator      folds the rows into output Blocks: Direct, UniqueKey,
                              SequenceMap, AggKey, ChangelogMinDelta, ChangelogDetail, TopN
      └ VMergeIterator        one heap over the segment iterators of every rowset (each
        or VUnionIterator     segment of an overlapping rowset, one VUnionIterator for a
                              rowset whose segments do not overlap), in (rowset version,
                              segment) order
         └ LazyInitSegmentIterator -> SegmentIterator

- RowwiseIterator has two ways to read: next_batch(Block*) and next_row(IteratorRowRef*).
  next_batch returns OK only with rows and EndOfFile only with none; next_row hands out one
  row whose reference lives until the next call, with is_same set when the key repeats the
  previous row's. BlockWithSameBit, BlockView, is_merge_iterator(), data_id() and empty()
  are gone.
- VMergeIterator is rewritten: a flat binary heap of child indexes, the child order
  (rowset version, then segment) as the only tie-break after the sequence column, and run
  detection: the top child advances without touching the heap while its next row still sorts
  before the best row of any other child. The same-key bit and skip_same are computed in one
  place, against the row handed out before. next_batch copies runs with insert_range_from.
- VCollectIterator, Level0Iterator and Level1Iterator are deleted. The scanner-local TopN
  read moves into TopNBlockIterator, reading one key-ordered iterator per rowset; the best n
  rows seen so far live in its Candidates helper (offer a batch, trim to n, worst value for
  the TopN runtime predicate), the same logic as _topn_next had.
- RowsetReader is a factory of segment iterators (get_segment_iterators, new_iterator);
  the per-rowset fields it wrote into the shared RowsetReaderContext (rowset_id, merged_rows)
  are gone. next_batch stays for the readers of one rowset (schema change).
- SegmentIterator keeps the batch limit of the adaptive batch size in its own member instead
  of rewriting _opts.block_row_max, reads on over batches a predicate filtered completely so
  next_batch never returns OK without rows (checking for cancellation on the way), and
  asserts it is not read after EndOfFile.
- TabletReader::next_block_with_aggregation sets eof only with an empty Block; the scanner's
  workaround for an eof with rows is removed.

Every read returns the same rows in the same order. What changes around them: the byte
budget of an output Block is checked after every copied run and every 64 rows instead of
being predicted per row, so a Block may end a little later; eof is reported with an empty
Block instead of together with the last rows; a LazyInitSegmentIterator opens its segment with
the io context of its rowset reader. The profile counters keep their names.

None

- Test: Unit Test / Regression test
    - BE UT: 531 tests of the storage read path (BlockReader*, VGenericIteratorsTest*,
      *Changelog*, VerticalBlockReaderTest*, VerticalCompactionTest*, SegCompactionTest*,
      TestRowIdConversion*, BetaRowsetTest*, *SchemaChange*, CumulativeCompaction*,
      TabletReaderTest*, SegmentIterator*, SegmentsKeyBoundsTruncation*, IndexBuilder*, ...).
      Rewritten: vgeneric_iterators_test (merge/union tests on the new interfaces, plus
      next_row, skip_same, row-limit and union next_row tests), block_reader_changelog_merge_test
      (a real VMergeIterator of two rowsets under the changelog iterators),
      block_reader_change_next_block_test (fake rows through next_row, eof only with an empty
      Block). Kept: block_reader_table_models_test (every table model and read mode on real
      segments), block_reader_overlapping_segments_test, agg_value_aggregator_test and the
      compaction, schema change, rowid conversion and segcompaction suites. The key bounds,
      expr zone map and index builder tests follow the renamed or removed members.
    - Regression on a one-BE cluster with the p0 pipeline settings: fault_injection_p0
      (test_overlapping_segments_merge, test_topn_pruned_struct_overlapping_segments),
      unique_with_mor_p0, unique_seq_map_p0, topn_optimize, delete_p0, data_model_p0,
      changelog_p0 (25/25), schema_change_p0 (125/126), compaction (56/58), unique_with_mow_p0
      (107/109); the 6 suites not passed all need an S3 bucket for a broker load or for cold
      storage, which this environment has no access to.
- Behavior changed: No
- Does this need documentation: No

Co-Authored-By: Claude Fable 5.1 <[email protected]>

Issue Number: N/A

Related PR: N/A

Problem Summary:

The parameters of a read go through three structs, ReaderParams -> RowsetReaderContext ->
StorageReadOptions, and the decisions about the merge were made again at every layer, each
under its own name:

- Whether the rows must come out in key order was `read_orderby_key`,
  `force_key_ordered_read`, `need_ordered_result` and BlockReader's own `_needs_merge`.
- Whether the older versions of a key are dropped was `is_unique`, computed in five places
  with four different answers: TabletReader excluded sequence-map tables, the two schema
  changes and the vertical reader did not, and the cluster-key schema change set it to false.
  The segment merge applied it to whatever columns it compared by, so a merge-on-write TopN
  read ordered by a key prefix dropped rows whose full key differed, and a schema change of a
  sequence-map table with overlapping segments lost column groups.
- `read_orderby_key_columns` was both the sort columns of a TopN read and the compare
  columns of a cluster-key schema change; `use_insert_order_when_same` and
  `use_sequence_column_for_merge_order` were the third and fourth tie-break knobs.
- `version` meant the version read in ReaderParams and the context, and the rowset's own
  version in StorageReadOptions.
- BetaRowsetReader wrote `condition_cache_digest` back into the context shared by every rowset
  of the read, digesting the key ranges once more per rowset, and every segment carried its
  own copy of the ~40-field StorageReadOptions even before it was opened.

Now:

- `MergeDecision` (storage/iterators.h) names the merge of a read once: `key_ordered`,
  `compare_ordinals` with `compare_whole_key`, the sequence or TSO column and its direction,
  `reverse`, `older_first` and `drop_older_versions`. `TabletReader::decide_merge` is the only
  place that computes it; the schema changes call it too. The rowset readers, the merge
  iterator, the TopN read and the vertical merge consume it. `drop_older_versions` is never set
  for a merge compared by a key prefix, which fixes the merge-on-write TopN read, and never
  for a sequence-map table whatever the caller, which fixes the schema change.
- RowsetReaderContext is read-only for the rowset readers (`const RowsetReaderContext*`
  everywhere) and loses `need_ordered_result`, `is_unique`, `use_insert_order_when_same`,
  `read_orderby_key_columns`, `read_orderby_key_reverse`, `use_sequence_column_for_merge_order`
  and `is_key_column_group`. The condition cache digest is computed into the segment options.
- ReaderParams loses `aggregation` (direct_mode already implied it) and
  `force_key_ordered_read`; `read_orderby_key` is `force_key_order`, and the changelog
  MIN_DELTA / DETAIL scans are key-ordered by TabletReader instead of by a scanner flag.
- StorageReadOptions loses the merge knobs (`use_insert_order_when_same`,
  `read_orderby_key_columns`, `changelog_tso_idx`); `version` is `rowset_version`. The segments
  of a rowset share one `shared_ptr<const StorageReadOptions>`; a LazyInitSegmentIterator only
  copies it when it reads a split of its segment.

Fix `ORDER BY <key prefix> LIMIT n` on a merge-on-write UNIQUE KEY table returning too few
rows when one load wrote the same prefix into several segments of one rowset.

- Test: Unit Test / Regression test
    - New tablet_reader_decide_merge_test.cpp: the decision table of decide_merge (whole key,
      sequence column, sequence map, key prefix, cluster key, AGG_KEYS, changelog TSO, flag
      pass-through). New regression fault_injection_p0/test_mow_topn_key_prefix_overlapping_segments:
      ORDER BY k1 LIMIT 10 on a merge-on-write table whose rowset has two overlapping segments
      returned k1 = 0..9 once each before, k1 = 0..4 twice each now.
    - BE UT: 685 tests of the read path (BlockReader*, VGenericIteratorsTest*, *Changelog*,
      VerticalBlockReaderTest*, VerticalCompactionTest*, SegCompactionTest*,
      TestRowIdConversion*, BetaRowsetTest*, *SchemaChange*, CumulativeCompaction*,
      TabletReaderTest*, SegmentIterator*, IndexBuilder*, *IndexStorage*, GroupRowsetWriter*,
      *MowTransform*, *VariantDocMode*, *VariantColumnWriterReader*, ...): 680 pass, 5 are
      SKIPPED/DISABLED by the tests themselves.
    - Regression on a one-BE cluster with the p0 pipeline settings: fault_injection_p0 (3),
      unique_with_mor_p0, unique_seq_map_p0, topn_optimize, changelog_p0 (25/25), delete_p0,
      data_model_p0, schema_change_p0 (125/126), compaction (56/58), unique_with_mow_p0
      (107/109); the suites not passed all need an S3 bucket for a broker load or cold storage.
- Behavior changed: Yes. A merge-on-write TopN read ordered by a key prefix keeps every row
  with that prefix; a schema change of a sequence-map table keeps every version of a key;
  the condition cache key of a rowset no longer depends on how many rowsets precede it.
- Does this need documentation: No

Co-Authored-By: Claude Fable 5.1 <[email protected]>

…ase merged-out children

### What problem does this PR solve?

Issue Number: N/A

Related PR: N/A

Problem Summary:

Follow-ups of the self-review of the read path refactor:

- BlockReader now keeps the RowsetReaders of the read for as long as their segment iterators
  live. A RowsetReader holds a reference on its rowset (Rowset::acquire / release); the
  scanner drops its rs_splits right after init, so the reference was gone while
  LazyInitSegmentIterators still had the rowset to open.
- The profile counter CollectIteratorMergeTime was never filled since the Level1Iterator that
  filled it is gone; BlockReader::next_block_with_aggregation times the fold under the same
  counters as before (merge / normal).
- VMergeIterator releases the segment iterator and the Block of a child as soon as the child
  has no more rows, instead of at the end of the merge; the previous implementation did too.
- A rowset with one segment is one child of the merge directly, without a VUnionIterator
  around it.
- The order of the rs_splits is the same-key tie-break of the merge; _init_rows_iter checks
  that they come in version order.
- AggValueAggregator documents the invariant its flush point relies on: an AGG_KEYS merge
  hands out every row of a Block.

### Release note

None

### Check List (For Author)

- Test: Unit Test
    - BlockReader*, VGenericIteratorsTest*, TabletReader*, *Changelog*, VerticalCompactionTest*,
      SegCompactionTest*, TestRowIdConversion*.
- Behavior changed: No
- Does this need documentation: No

Co-Authored-By: Claude Fable 5.1 <[email protected]>

### What problem does this PR solve?

Issue Number: None

Related PR: None

Problem Summary:

AggKeyBlockIterator took the merged rows of an AGG_KEYS read one at a time
(VMergeIterator::next_row). AggValueAggregator buffered the row references and copied them
column by column into per-column buffers. Because the merge refills a child's Block once its
last row is handed out, AggValueAggregator had to flush on the last row of every source
Block: a lifetime contract between two objects that only a comment stated. Every key group
then paid add_batch_range, insert_result_into and reset per value column, one-row groups
included, and the REPLACE reader function copied each group's value into a Field.

This commit switches the AGG_KEYS fold to Blocks, the way StarRocks pairs its heap merge with
an aggregate iterator:
- RowwiseIterator::next_batch(Block*, same_key): next_batch plus one flag per row, set when
  the row repeats the key of the row before it, across batches too. VMergeIterator sets the
  flag of the first row of every run from the remembered key; the other rows of a run are new
  keys because a child holds every key once. A union sets every flag to 0, as its next_row
  never flags a row.
- VMergeIterator::next_batch (both overloads) records the runs it takes and copies them column
  by column, a one-row run with insert_from. The recorded runs point into the children's
  Blocks, so they are copied before the top child's Block is refilled or released, before the
  byte budget is measured and at the end of the batch. Copying run by run with every column
  walks children x columns memory streams at once when keys interleave row by row; on 16
  value columns and 8 rowsets the copy cost about twice as much (prototype benchmark below).
  next_batch(Block*) without flags (DUP_KEYS compactions, schema changes) gets the same copy.
- AggKeyBlockIterator folds the row ranges of each merged Block. Key columns and REPLACE
  columns are copied from the group's first row, which is exactly the row the REPLACE reader
  function keeps (IsFirst, NULLs included; REPLACE_IF_NOT_NULL skips NULLs and is still
  folded). The other value columns are folded by AggValueAggregator with one add_batch_range
  per group and merged Block; a group may stay open across merged Blocks. A stretch of
  one-row groups is copied to the output as is when every folded function returns its single
  input unchanged (SUM, MIN, MAX, REPLACE, REPLACE_IF_NOT_NULL, HLL_UNION, BITMAP_UNION);
  QUANTILE_UNION and the generic AGG_STATE union keep folding. Output Blocks still end only
  in front of a new key and hold at most batch_size groups.
- AggValueAggregator no longer buffers references or flushes. Its aggregate states live in an
  Arena, aligned by align_of_data(), and a throwing create() no longer leaves the vector of
  functions and the vector of states with different lengths.

Prototype benchmark, in memory, RELEASE (artifacts in the PR description): the merge and fold
of 119 shapes (2-64 rowsets, 0-95% repeated keys, run lengths 1-1024, narrow and wide rows,
nullable columns, six key columns, long strings, HLL and BITMAP, batch and child Block sizes),
7-15 interleaved repetitions, median CPU time. The prototype of this design against the
prototype of the previous AGG path (a merge faithful to next_row plus the real
AggValueAggregator), geometric mean of the speedup:
- all shapes 2.48x; under 1.2 rows per key 3.84x; 1.2-3 rows per key 1.76x; 3 or more rows
  per key 1.15x;
- slower in 8 of 119 shapes: HLL (1.12-1.26x, the 16 KB HLLs are copied into the merged Block
  and again for one-row groups), nullable 16-column SUM at 95% repeated keys (1.23-1.25x),
  and four shapes at 1.03-1.09x.
The prototypes read in-memory Blocks without segment IO, decoding or predicates, so a whole
query gains less.

### Release note

None

### Check List (For Author)

- Test: Unit Test
    - New: VGenericIteratorsTest.MergeNextBatchFlagsRepeatedKeysAcrossBlocks,
      UnionNextBatchFlagsNoRow; BlockReaderTableModelsTest.AggKeysCopyOneRowGroupsAndFoldTheOthers
      (one-row and several-row groups, SUM/MIN/MAX/REPLACE/REPLACE_IF_NOT_NULL, batch sizes
      1-100, query and cumulative compaction); agg_value_aggregator_test rewritten for the
      new interface (row ranges, a group across Blocks, nullable SUM, which aggregations
      copy one-row groups).
    - Read path UT set (AggValueAggregatorTest, VGenericIteratorsTest, BlockReader*,
      *Changelog*, VerticalBlockReaderTest, VerticalCompactionTest, SegCompactionTest,
      TestRowIdConversion*, BetaRowsetTest, *SchemaChange*, CumulativeCompaction*,
      TabletReaderTest*, SegmentIterator*, IndexBuilder*, ...): 547 pass, 5 are skipped or
      disabled by the tests themselves.
- Behavior changed: No. The output rows are the same; HLL and BITMAP results of one-row groups
  are the stored values instead of a union of one value, which holds the same values.
- Does this need documentation: No

Co-Authored-By: Claude Opus 5.5 <[email protected]>

…plit vgeneric_iterators

### What problem does this PR solve?

Issue Number: None

Related PR: None

Problem Summary:

storage/iterator/vgeneric_iterators.{h,cpp} held three unrelated iterators: the heap merge of a
read, the union of key-disjoint inputs and the statistics iterator of aggregate push-down. A
user of one of them included the headers all three need, column_reader.h and
PlanNodes_types.h among them. The merge class also kept the name of the per-rowset
VMergeIterator it replaced, although it is now the one heap over the segment iterators of every
rowset of a read (the counterpart of StarRocks' HeapMergeIterator).

- VMergeIterator is renamed VHeapMergeIterator, and new_merge_iterator new_heap_merge_iterator.
- The three iterators move to storage/iterator/heap_merge_iterator.{h,cpp},
  union_iterator.{h,cpp} and statistics_iterator.{h,cpp}; vgeneric_iterators.{h,cpp} are removed.
  heap_merge_iterator.h no longer includes column_reader.h or PlanNodes_types.h.
- The code moves unchanged. Comments that named VMergeIterator follow the rename, and a test
  comment that still described the deleted VMergeIteratorContext::block_reset describes the
  current Block creation.

### Release note

None

### Check List (For Author)

- Test: Unit Test
    - Read path UT set (VGenericIteratorsTest, AggValueAggregatorTest, BlockReader*,
      *Changelog*, VerticalBlockReaderTest, VerticalCompactionTest, SegCompactionTest,
      TestRowIdConversion*, BetaRowsetTest, *SchemaChange*, CumulativeCompaction*,
      TabletReaderTest*, SegmentIterator*, IndexBuilder*, ...): 547 pass, 5 are skipped or
      disabled by the tests themselves. clang-tidy: no finding in the new files.
- Behavior changed: No
- Does this need documentation: No

Co-Authored-By: Claude Opus 5.5 <[email protected]>

… VHeapMergeIterator

Issue Number: None

Related PR: None

Problem Summary:

Vertical compaction kept a merge of its own next to the one of the read path:

- VerticalHeapMergeIterator, a std::priority_queue of per-segment contexts that pops and pushes
  once per row, and VerticalFifoMergeIterator for tables without keys. An iterator_init_flag
  vector told them which segments to open first, so that a rowset whose segments do not overlap
  only holds one segment at a time.
- VerticalBlockReader folded the value groups of AGG_KEYS tables with its own buffer of row
  references (_init_agg_state, _append_agg_data, _update_agg_data, _copy_agg_data,
  _update_agg_value).

The read path has the same pieces: VHeapMergeIterator, which only touches its heap at the end of
a run of rows, and AggKeyBlockIterator with AggValueAggregator.

- The key group merges with VHeapMergeIterator. Its new next_batch(Block*, runs) reports the
  child of every row taken, a row dropped by drop_older_versions included, and
  VerticalKeyMergeIterator records them in the RowSourcesBuffer. The MergeDecision is the one
  TabletReader::decide_merge makes for the key group (by the cluster key of a merge-on-write
  table that has one), with drop_older_versions for AGG_KEYS too, as before.
- The sources of the merge replace iterator_init_flag: a segment of a rowset whose segments
  overlap, one VUnionIterator for a rowset whose segments do not, one per group of a
  NONOVERLAPPING_WITHIN_GROUP rowset. The merge still holds one segment of such a rowset at a
  time. Every value group builds the same sources, and a row source names a source instead of a
  segment. A table without keys reads all its sources as one union, the oldest first, as
  VerticalFifoMergeIterator did.
- The value groups of AGG_KEYS tables fold with AggKeyBlockIterator over the new
  VerticalMaskMergeIterator::next_batch(Block*, same_key), whose flags are the agg flags of the
  row sources.
- VerticalMaskMergeIterator::next_batch returns EndOfFile only without rows, like the other
  iterators; next_row, which only the old AGG_KEYS fold used, is removed.
  VerticalMergeIteratorContext, now only used by the value groups, loses the comparison state and
  initializes its iterator, which a union needs.

The rows written are the same. Two internal values change: the key group samples the bytes per
row for the next batch size estimate from its first merged Block instead of the first Block of
every source opened at the start, and the active segment context counter only counts value group
sources.

None

- Test: Unit Test
    - New: VGenericIteratorsTest.MergeNextBatchReportsTheChildOfEveryRow,
      VerticalCompactionTest.TestAggKeyVerticalMergeFoldsEveryValueGroup (SUM, REPLACE and MAX
      in three value groups, keys repeated across rowsets and across the segments of an
      overlapping rowset); VerticalBlockReaderTest rewritten for the group sizes.
    - Vertical compaction and read path UT set (VerticalCompactionTest, VerticalBlockReaderTest,
      SegCompactionTest, TestRowIdConversion*, VGenericIteratorsTest, AggValueAggregatorTest,
      BlockReader*, *Compaction*, *Merger*, *SchemaChange*, SegmentIterator*, ...): 909 pass,
      5 are skipped or disabled by the tests themselves. clang-tidy: no finding on the changed
      lines.
- Behavior changed: No
- Does this need documentation: No

Co-Authored-By: Claude Opus 5.5 <[email protected]>

…s, the key merge and the mask merge

Issue Number: None

Related PR: None

Problem Summary:

storage/iterator/vertical_merge_iterator.{h,cpp} held every piece of a vertical compaction:
RowSource and RowSourcesBuffer, the key group's VerticalKeyMergeIterator, the value groups'
VerticalMaskMergeIterator with its VerticalMergeIteratorContext and RowBatch, and
VerticalCompactionContextStats. merger.cpp and segcompaction.cpp, which only need the row sources
buffer, included the heap merge through it.

- row_sources_buffer.{h,cpp}: RowSource and RowSourcesBuffer.
- vertical_key_merge_iterator.{h,cpp}: VerticalKeyMergeIterator, the only one that includes
  heap_merge_iterator.h.
- mask_merge_iterator.{h,cpp}: VerticalMaskMergeIterator, RowBatch and
  VerticalCompactionContextStats. VerticalMergeIteratorContext, used by nothing else, is defined
  in mask_merge_iterator.cpp; the header only declares it, and the constructor and destructor of
  VerticalMaskMergeIterator move to the .cpp, where it is complete.
- vertical_merge_iterator.{h,cpp} are removed, every includer includes what it uses, and
  vertical_merge_iterator_test.cpp, which tests RowBatch and the sparse column copies, becomes
  mask_merge_iterator_test.cpp.

The code moves unchanged.

None

- Test: Unit Test
    - Vertical compaction and read path UT set (VerticalCompactionTest, VerticalBlockReaderTest,
      SparseColumnOptimizationTest, SegCompactionTest, TestRowIdConversion*,
      VGenericIteratorsTest, *Compaction*, *Merger*, ...): 948 pass, 5 are skipped or disabled
      by the tests themselves.
- Behavior changed: No
- Does this need documentation: No

Co-Authored-By: Claude Opus 5.5 <[email protected]>

…on iterators

### What problem does this PR solve?

Issue Number: None

Related PR: None

Problem Summary:

The split of vertical_merge_iterator.cpp turned every line of row_sources_buffer.cpp and
mask_merge_iterator.cpp into an added line, so clang-tidy, which checks the changed lines, now
checks code it never checked before and reports six findings:

- three signed/unsigned comparisons: VerticalMergeIteratorContext::advance_by and
  VerticalMaskMergeIterator::_next_batch compare with std::cmp_less_equal and std::cmp_less, and
  RowSourcesBuffer::same_source_count counts in a size_t, the type it returns;
- a leak the static analyzer reports in VerticalMaskMergeIterator::_next_batch, which copies the
  EndOfFile Status in a conditional expression; it returns OK or the Status from two branches now;
- VerticalMaskMergeIterator::unique_key_next_batch is longer (100 lines) and more complex
  (cognitive complexity 56) than the thresholds: joining a run to the batches, a run continuing
  the last batch extending it, becomes append_to_batches(), and with it go the three variables
  of the batch being built and the flush of the last one after the loop.

No behavior changes.

### Release note

None

### Check List (For Author)

- Test: Unit Test
    - Vertical compaction and read path UT set (VerticalCompactionTest, VerticalBlockReaderTest,
      SparseColumnOptimizationTest, SegCompactionTest, TestRowIdConversion*, *Compaction*,
      *Merger*, ...): 948 pass, 5 are skipped or disabled by the tests themselves.
    - clang-tidy: row_sources_buffer.cpp and mask_merge_iterator.cpp have no finding left.
- Behavior changed: No
- Does this need documentation: No

Co-Authored-By: Claude Opus 5.5 <[email protected]>

…key apart above it

Issue Number: None

Related PR: apache#69515

Problem Summary:

VHeapMergeIterator did three jobs at once: it sorted the rows of the read, it decided which
rows share a key (a same-key flag kept per run, with the last key copied aside), and for
merge-on-read UNIQUE_KEYS it dropped the older versions itself (drop_older_versions). That
made the heap carry a second-best child (`_second`), a hand-written top-down sift (two
comparisons per level), a remembered key copied at the end of every run, four ways to consume
it (next_batch, next_batch with same-key flags, next_batch with runs, next_row) and the
assumption that a child never repeats a key, which only holds for some table models.

It is also slow where runs are short. On 8 overlapping rowsets whose keys interleave row by
row, every row ends a run, and each row cost about 2d+3 key comparisons (d = log2 of the
children) plus a copy of the key; master's pairing heap needs about 3.8 at 8 children. Measured
on RELEASE builds in low-load rounds: merge-on-read reads with 8 versions of every key
1.38-1.49x the CPU of master, vertical compactions of AGG/merge-on-read tables 1.17-1.27x, the
key group merge going from 2.0 to 4.3 CPU-s, half of it in the sift.

Now the merge only sorts, the way StarRocks' HeapMergeIterator does, and the readers above it
tell the rows of one key apart by comparing them:

- VHeapMergeIterator keeps the child whose row goes next (the lead) out of the heap, so the
  heap's top is the best row of every other child. After every row the lead's next row is
  compared with that top once; only when the top goes first are the two swapped with
  std::pop_heap/std::push_heap (bottom-up, one comparison per level). A run of rows from one
  child therefore costs one comparison per row, and rows that interleave cost about d+2. The
  consecutive rows taken from one child are still copied column by column. It hands out every
  row: next_batch(Block*), and next_batch_with_children() for the key group of a vertical
  compaction. `_second`, the sift, the remembered key, next_row, the same-key overload and
  MergedRun are gone.
- SameKeyIterator, above the merge in BlockReader and in BetaRowsetReader::new_iterator (TopN,
  schema change), works the same-key flags out from the key columns of the merged rows: a row
  is flagged when its key equals the key of the row before it, the last row of the previous
  Block for the first row of a Block. With drop_older_versions it keeps the first, newest, row
  of every key and counts the others in merged_rows(); otherwise it passes the rows through and
  answers next_batch(Block*, same_key) for the folds (AGG_KEYS, sequence map, changelog). The
  flags no longer depend on a child holding every key once.
- MergedRowCursor reads Blocks through next_batch(Block*, same_key) instead of next_row(), so
  RowwiseIterator::next_row and VUnionIterator::next_row are removed.
- UniqueKeyBlockIterator reads whole Blocks: the rows it gets already hold one row per key, so
  it only filters the delete sign, instead of copying row by row.
- VerticalKeyMergeIterator flags the merged key rows itself and, for UNIQUE_KEYS without a
  cluster key and AGG_KEYS, drops all but the first row of every key while recording every row
  in the row sources, the rule master's VerticalHeapMergeIterator used. Before this change the
  key group of a sequence-map table kept the repeated keys while their row sources were
  flagged, which the value groups skip, so key and value groups could disagree in a
  segcompaction of such a table; now both drop them, as on master.

Cost: an overlapping AGG_KEYS or merge-on-read read and the key group of a compaction pay one
key comparison per row for the flags, also within long runs, where the old heap knew there was
no repeated key without comparing; a sequence-map or changelog read copies every row once more
into the cursor's Block. Rowsets that do not overlap are read through a VUnionIterator and pay
nothing.

None

- Test: Unit Test
    - VGenericIteratorsTest updated (MergeDropOlderVersionsKeepsOneRowPerKey,
      MergeFlagsRepeatedKeysWithinOneChild, MergeNextBatchReportsTheChildOfEveryRow for
      next_batch_with_children); the fake rows of BlockReaderChangeNextBlockTest and the merge
      of BlockReaderChangelogMergeTest follow the new interfaces
    - 349 tests of the iterator, compaction, rowid conversion, tablet reader and schema change
      suites: 348 pass; SegCompactionTest.FatalSegcompactionFailsBuild fails only in that
      combined run and passes alone and with the segcompaction suites (13/13)
- Behavior changed: No (results are the same; the segcompaction of a sequence-map table goes
  back to master's behavior)
- Does this need documentation: No

Co-Authored-By: Claude Opus 5.5 <[email protected]>

…d since the branch point

### What problem does this PR solve?

Issue Number: None

Related PR: apache#69515

Problem Summary:

Rebasing the read-path refactor onto master brings in code written against the old names:

- The row-store read-ahead of base_tablet.cpp and the point-query read-ahead of
  point_query_executor.cpp set StorageReadOptions::version, which this branch renamed to
  rowset_version (it was always the rowset's own version, not the version read).
- segment_read_ahead.cpp calls config::is_cloud_mode() without including cloud/config.h; on
  master it arrived through headers whose includes this branch trimmed.
- block_reader_row_ttl_test.cpp, added by the row TTL change (apache#65858), marks the key group
  through RowsetReaderContext::is_key_column_group, which this branch removed; VerticalBlockReader
  reads its own _is_key_column_group.

### Release note

None

### Check List (For Author)

- Test: Unit Test (built and run together with the next commit)
- Behavior changed: No
- Does this need documentation: No

Co-Authored-By: Claude Opus 5.5 <[email protected]>

…eKeyIterator

### What problem does this PR solve?

Issue Number: None

Related PR: apache#69515

Problem Summary:

SameKeyIterator had two behaviors switched by MergeDecision::drop_older_versions: for
merge-on-read UNIQUE_KEYS it kept the first, newest, row of every key and dropped the others,
and for AGG_KEYS, a sequence map and changelog scans it flagged the rows of one key for the fold
above. One class with two modes needed a DORIS_CHECK to stop the flagging call in the dropping
mode, and its name only described the second one.

After the rebase onto master, UniqueKeyBlockIterator did nothing on its own either: the older
versions were dropped below it, and master's row TTL change (apache#65858) moved the delete-sign filter
of the compaction and checksum reads into BlockReader::next_block_with_aggregation. It was a
second DirectBlockIterator.

Now:

- UniqueKeyIterator (unique_key_iterator.h) keeps only the newest row of every key and counts the
  others in merged_rows(). BlockReader and BetaRowsetReader::new_iterator (schema change, TopN)
  put it above the merge when the read's drop_older_versions is set.
- SameKeyIterator only flags the rows of one key; it takes the key ordinals, not the decision.
- Both share SameKeyFlags, as the key group of a vertical compaction does.
- UniqueKeyBlockIterator is removed: a merge-on-read UNIQUE_KEYS read is a DirectBlockIterator
  over the UniqueKeyIterator, with the delete sign filtered in BlockReader as before.

So a merge-on-read UNIQUE_KEYS read is: VHeapMergeIterator sorts, UniqueKeyIterator keeps the
newest row of every key, DirectBlockIterator hands the Blocks out.

### Release note

None

### Check List (For Author)

- Test: Unit Test
    - BlockReaderUniqueKeysTest checks the DirectBlockIterator over a UniqueKeyIterator;
      VGenericIteratorsTest and BlockReaderChangelogMergeTest build the merge the way the readers do
    - 363 tests of the iterator, compaction, rowid conversion, tablet reader, schema change and
      row TTL suites pass (built with format/transformer/vparquet_writer_test.cpp set aside: it
      does not compile on master since apache#69392 changed VIcebergPartitionWriter's WriteInfo)
- Behavior changed: No
- Does this need documentation: No

Co-Authored-By: Claude Opus 5.5 <[email protected]>