Lightweight Updates v2 - #103182

`MutationsInterpreter::execute` installs a `CheckSortedTransform` at the
end of the pipeline to assert that the mutated stream is monotonic by the
storage's sorting key. The gate was `source.getMergeTreeData()`, which
matched any `MergeTree` source — including lightweight updates that route
through `storage->read(...)`.

When `max_threads > 1` that read dispatches to `readFromPool`, which hands
mark ranges to streams dynamically: each pipeline port sees a
non-monotonic sequence of granules and thus a non-monotonic sequence of
sort-key values, tripping `LOGICAL_ERROR: Sort order of blocks violated
...`. v1 lightweight updates didn't surface this because their mutation
pipeline only reads virtual columns (`_part_offset`, `_block_number`,
`_block_offset`, `_data_version`), so the sort-key column stayed out of
the header and `getStorageSortDescriptionIfPossible` returned empty.
v2 reads the physical source columns of the sort-key expression so they
can be persisted on every patch row, which put the sort-key column back
into the header and armed the check.

The check was also semantically unnecessary for that path: the
patch-part writer sorts its own output by `(sort_key..., _block_number,
_block_offset)`. Tighten the gate to `source.isMutatingDataPart()` so
it only fires for the single-part `MutateTask` flow — where inputs come
from `MergeTreeSequentialSource` and each stream is genuinely
sort-key-monotonic — and skip for lightweight updates and for other
storages (e.g. `Iceberg`) that expose a sorting key but read multiple
sorted files in parallel.

Regression test `04103_lightweight_update_v2_parallel_read_sort_check`
reproduces the failing `DELETE ... WHERE (a % 2) = 0` pattern from the
report with a `MergeTree` table small enough for CI but still wide
enough (500K rows, 1024-row granularity, `max_threads = 8`) to force
`readFromPool` to split ranges across ports.

`MergeTreePatchReaderMergeOnKey::readPatches` kept reading and appending
patch blocks until `needNewPatch` saw the last block's max sort-key catch
up to the current main block's max. In the common case where the caller's
deque was already drained — `MergeTreeReadersChain` evicted everything on
the previous main-cursor advance — the loop started from a null
`last_read_patch` and walked *every* range whose max was still below
`main_max`, pushing each into `results`. Every such block was already
useless for the current iteration (its max sort-key lies below
`main_min`, so `needOldPatch` would return `false` and the deque would
evict it on the very next call), but during this single `readPatches`
call the full catch-up stack lived in memory at once.

For a 5 M-row patch with 8-granule task ranges that is ~200 MiB of
pinned patch data per patch per worker thread; the reporter hit 60 GiB
peak on 96 patch parts covering 200 M rows, and a repro with 5 patches
over 10 M rows / `max_threads = 8` shows 1.65 GiB before the fix.

Inline the same predicate that `MergeTreeReadersChain::readPatches`
applies on the next call: after reading each block ask `needOldPatch`
whether it would be retained. If not — discard it. We still need the
block to anchor the `last_read_patch` pointer fed to `needNewPatch` on
the next loop iteration, so keep the most recent discarded block alive
in a local holder that outlives only this function call. Once the loop
finds a block whose max sort-key reaches the current main range, that
block is retained and the loop exits via `needNewPatch` as before.

Memory after the fix (same `SELECT count()` on the 10 M / 5-patch
repro):

  max_threads = 1   258 MiB  →  8.5 MiB
  max_threads = 4   1017 MiB →   31 MiB
  max_threads = 8   1.65 GiB →   63 MiB

Regression test `04104_lightweight_update_v2_eager_patch_discard` drives
the catch-up shape (5 disjoint `INSERT`s, one DELETE, `SELECT count()`
under a 64 MiB `max_memory_usage` cap) — the pre-fix code blows through
the cap; the fixed path stays well under it.

Perf top on a 10 M-row / 5-patch `SELECT count()` showed ~58 % of CPU
burned inside `std::unordered_map` machinery powering `Block::getByName`:

    45.17 % std::__hash_const_iterator<...>
     8.99 % DB::Block::findPositionByName(std::string_view)
     8.52 % std::__murmur2_or_cityhash
     6.83 % compareSortKeyRows
     4.33 % DB::Block::getPositionByName(...)
     4.10 % DB::Block::getByName(...)

`compareSortKeyRows` called `a_block.getByName(name)` and
`b_block.getByName(name)` **per column, per row comparison** — one full
`unordered_map` lookup for each sort-key value the merge touched. For an
`ORDER BY a` table the two-pointer merge issues one comparator call per
advanced cursor, so the hash-table lookups dominate the whole loop.

Resolve the sort-key column pointers once per block (`SortKeyColumns`),
then compare through raw `IColumn *` in the inner loop. The comparator
function is now an `ALWAYS_INLINE` helper over the pre-resolved cursor.

Throughput on the 10 M-row / 5-patch repro, `SELECT count()`:

                 before fix   after fix
  max_threads=1   500 K r/s   8.1 M r/s    (16×)
  max_threads=8                16  M r/s

Correctness-preserving: the comparator semantics (NULL-aware,
nan-last, per-column `reverse_flags`) are unchanged — only the lookup
scheme changed. The three existing v2 regression tests
(`04102_smoke`, `04103_parallel_read_sort_check`,
`04104_eager_patch_discard`) continue to pass.

`getRangesInPatchPartMergeOnKey` previously returned every patch mark
range of the v2 patch, relying on the streaming `needOldPatch` /
`needNewPatch` comparator inside the reader loop to skip useless ones
at read time. That works for memory (blocks are evicted before they
stack up) but is catastrophic for I/O and CPU: each read task handed
the whole patch must still *decompress and deserialize* every mark
range before deciding it's below main's cursor.

On a 200 M-row `SELECT count()` with a 100 M-row patch and
`max_threads = 8` this produced:

  PatchesReadRows                 50_432_368_896   ~500× the patch
  PatchesReadUncompressedBytes    1.47 TiB         against a ~1 GiB table
  Query duration                  49 s             at ~4 M rows/s

The amplification is `tasks_per_thread × #patch_ranges` because each
read task has its own `MergeTreePatchReaders`, and each of them needs
to walk the whole patch to cover the task's main range.

Fix at range-enumeration time. Both main and patch carry the sort-key
expression's result columns as column 0 of the primary index. Fold the
task's main `MarkRanges` down to `[min_granule, max_granule_end)`,
binary-search the patch's primary index for the granule window whose
sort-key spans `[main_sk[min_granule], main_sk[max_granule_end])`, and
return only that window. When main reads all the way to the end of the
part (`max_granule_end == main_marks_count`) there is no index row that
bounds main's max, so the upper bound is left open — every remaining
patch granule might still match.

Falls through to "all ranges" in edge cases where we can't determine a
bound (empty index, mismatched column types). Correctness is preserved;
only the perf benefit is lost.

Results on the 10 M-row / 8-patch `SELECT count()` repro, `max_threads = 1`:

  PatchesReadRows                 14_464_400  →  5_152_576  (~2.8×)
  PatchesReadUncompressedBytes    430 MiB     →  157 MiB    (~2.7×)
  Query duration                  1247 ms     →  1004 ms

And at `max_threads = 8`:

  PatchesReadRows                  89_271_826  →  8_104_768  (~11×)
  PatchesReadUncompressedBytes     2.66 GiB    →  247 MiB    (~11×)
  Query duration                   933 ms      →  176 ms     (~5×)

On the reporter's 200 M-row / 1-patch shape at `max_threads = 8` the
per-task read set is bounded by the task's main-range intersection
with the patch's sort-key range, eliminating the `tasks × patch-size`
multiplication that was responsible for the 50 B-row / 1.5 TiB read.

All existing v2 regression tests (`04102`, `04103`, `04104`) still
pass; the change is a pure read-set tightening and does not alter
patch-apply semantics.

After the per-task sort-key pruning in `getRangesInPatchPartMergeOnKey`,
`RangesInPatchParts::getRanges` still piped the result through
`getIntersectingRanges`, which snapped the tight range up to the
pre-computed 8-mark chunks in `ranges_by_name`. The snapping existed
for v1 Merge/Join, where the raw ranges come from a `_part_offset`
interval intersection that itself has no chunking. For v2 MergeOnKey
the raw ranges are already *granule-tight* — returning a 1-mark range
means the reader only needs 1 mark (~8192 rows), but
`getIntersectingRanges` turned that into an 8-mark chunk (~65k rows).

In a repro at 30 M rows × 15 patches with `max_threads = 8`:

  before: PatchesReadRows = 180 M  (~12× the total 15 M patch rows)
  after:  PatchesReadRows =  32 M  (~2.1×)
  peak memory 570 MiB → 148 MiB, query 1.9 s → 0.7 s

For MergeOnKey we now re-split the sort-key-tight range into chunks of
`max_granules_in_range` directly, so the reader still gets chunk-sized
read units but never reaches into marks outside the actual overlap.

Merge/Join paths are unchanged — they still consult
`getIntersectingRanges` because their raw ranges are not pre-trimmed
to granule boundaries.

Drop the serialized sort-key expression list + DESC flags from
`SourcePartsSetForPatch` (and the SQL-to-AST parser in `PatchPartsUtils`);
readers now rebuild the v2 patch's `KeyDescription` from the target table's
current in-memory metadata snapshot in `getPatchPartMetadataV2` and
`getAlterConversionsForPart`. The sinks carry a single `is_v2_format`
boolean instead of the SQL text and reverse-flag vector. Net -120 lines
of plumbing with no change in apply behavior.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

The v2 patch part schema drops `_part_offset` from stored columns (it
was v1's sort-key tie-breaker and is dead weight under v2's two-cursor
merge). But `MutationsInterpreter` unconditionally emits `_part_offset`
in the pipeline block — it's one of the mandatory patch system columns
injected alongside `_block_number` / `_block_offset` / `_part` — so the
pipeline output is always one column wider than the sink's header.

The `SinkToStorage` wiring refuses to connect mismatched headers and
bails with `Block structure mismatch in function connect between
AddDeduplicationInfoTransform and MergeTreeSinkPatch stream: different
number of columns`. The write writer downstream (`writeTempPartImpl`)
already filters incoming block columns against the metadata
(`getAllPhysical().filter(block.getNames())`), so the solution is to
make the *pipeline header* match the sink by projecting `_part_offset`
out before `AddDeduplicationInfoTransform`.

Done via an `ExpressionTransform` driven by a simple-identity
`ActionsDAG` with `_part_offset` removed from its outputs, gated on
`isV2LightweightUpdateUsable` and a `pipeline_header.has("_part_offset")`
guard so v1 paths (which do keep `_part_offset` on disk) are left
alone.

Verified on a fresh `DELETE FROM kek WHERE a % 2 = 0` at 5 M rows:
`columns.txt` now lists 6 columns (`a`, `_part`, `_part_data_version`,
`_row_exists`, `_block_number`, `_block_offset`) — `_part_offset` is
gone and no `_part_offset.bin` is written. `SELECT count()` over a 30 M
row + OPTIMIZE FINAL setup stays at ~2× read amplification and 139 MiB
peak memory; existing v2 regression tests (`04102`, `04103`, `04104`)
still pass.

`MergeTreeReadersChain::addPatchVirtuals(ReadResult&, const Block&)`
caches the patch-relevant columns from the first reader's result block
into `result.columns_for_patches`, so later read steps can fold them
back in at apply time. It was reading only from
`header.cloneWithColumns(result.columns)`, where `header` is the
first reader's post-prewhere `result_sample_block`.

When the query filters on a column used by the v2 `MergeOnKey` patch's
sort-key expression (`SELECT str_10p FROM t WHERE id = 500000`),
prewhere's ActionsDAG evaluates the filter and then projects the
sort-key-source column (`id`) out — it isn't referenced by any
downstream read step and doesn't survive the projection. The dropped
column is stashed in `result.additional_columns` by
`executePrewhereActionsAndFilterColumns`, but `addPatchVirtuals` never
looked there, so `result.columns_for_patches` ended up without `id`.
`applyPatchMergeOnKey` then hit `Not found column id in block` when
resolving the sort-key columns via `SortKeyColumns::resolve` on the
second reader's block.

Fix: before the inner `addPatchVirtuals` call, fold any entries in
`result.additional_columns` that are missing from `result_block` into
it. Row counts stay aligned because `result.additional_columns` is
filtered/shrunk in lockstep with `result.columns` (see `applyFilter`
and `shrink` in `MergeTreeRangeReader`).

Reproduced on a `MergeTree` table `ORDER BY id` with patch columns
updated via lightweight UPDATEs and `apply_patches_on_merge = 0`:

    SELECT str_10p FROM test_lwu_join WHERE id = 500000;

previously raised `Code: 10 NOT_FOUND_COLUMN_IN_BLOCK`; now returns
the patched value.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Write the semantic sort-key prefix length into source_parts.dat alongside
the format-version byte, so v2 patches no longer rebuild sort-key shape
from the current target-table metadata at every open. Slice the rebuilt
KeyDescription to the persisted length; this drops the n_full - n_appended
arithmetic from MergeTreeData::getAlterConversionsForPart and keeps
pre-ALTER patches' sort-key shape stable under schema drift (partition-id
hash isolates them regardless).

Move the v2 sort-key view (source/result column names, reverse flags,
expression) off PatchPartInfoBase's denormalized fields into a single
PatchSortKey struct populated once at construction from the patch part's
rebuilt metadata. Consumers (MergeTreePatchReaderMergeOnKey,
applyPatchMergeOnKey, getVirtualsRequiredForPatch,
MergeTreeReadersChain::addPatchVirtuals) read straight off
PatchPartInfo::sort_key.

Load source_parts.dat before primary.cidx in
IMergeTreeDataPart::loadColumnsChecksumsIndexes — the patch's metadata
rebuild now depends on getSortKeyPrefixSize(), and without the correct
prefix the index file read sees a too-short column count and throws
EXPECTED_END_OF_FILE on DETACH/ATTACH or server restart.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

getRangesInPatchPartMergeOnKey used a non-strict patch_sk[j] >= main_sk[end]
upper-bound compare, but MergeTreeDataPartWriterOnDisk writes the primary
index's final mark as the last row's value (not a past-the-end sentinel).
For a main part with a single row, main_sk[final_mark] == main_sk[0], and
patches whose first key equalled that value got excluded even though main
has a matching row. Switch to strict `>` via is_le so equal values are
kept; at worst we over-read one granule and the apply loop filters per row.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

v2 patch parts hash a different set of inputs (sort-key AST text + v2
marker, not just column names), so their partition-id hashes differ from
v1's. These tests were written with v1 defaults and hardcoded v1 hashes;
update references and any hardcoded OPTIMIZE ... PARTITION ID 'patch-<hash>'
calls to the v2 hashes.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>