perf(merge_insert): defer reading non-source columns via a logical late-materialization rule

[wjones127]

Partial-schema upserts fill missing (non-source) columns from the target side of the join via col("target.<field>"), so projection pushdown keeps those columns in the target scan — reading wide columns for every target row even when only a few rows match.

This defers those reads to after the join shrinks the row count. It is done with a generic logical optimizer rule rather than a physical one, so the deferral is driven by which relation a column comes from (qualifier) and the scan-narrowing falls out of DataFusion's existing OptimizeProjections — no positional index remapping or hardcoded join side.

How it works

A new logical extension node, LateTakeNode, is inserted above the row-reducing join:

LateTake(deferred=[wide_col])           # re-fetches wide_col by _rowaddr, appends it
  Projection(carried columns only)      # drops deferred + the redundant equi-key
    Join(target, source)
      LanceRead(target, projection=[key, _rowaddr])   # wide_col no longer scanned
      <source>

• LateMaterializeJoin (the rule) finds a join with a Lance scan side that emits _rowaddr, picks the wide data columns that are only carried through (not join keys, not used in a join filter), and wraps the join. A width/storage gate (is_wide_column) only defers columns wide enough that re-fetching by address beats scanning them for every row.
• The node reports via necessary_children_exprs that it does not need the deferred columns from its child, so OptimizeProjections prunes them from the scan while keeping _rowaddr. Downstream references resolve the appended columns by name — the plan's output schema is unchanged.
• LateTakePlanner lowers the node to the existing physical TakeExec.
• The rule is applied only at the merge_insert call site (between two session_state.optimize passes), not the session-wide optimizer, to bound its blast radius.

A normalizing projection is inserted between the join and the take. Without it the opaque extension node forces HashJoinExec to emit its full left ++ right output, which for an equi-join on a shared key name yields two arrow fields named key (qualifiers are erased at the physical level) — a duplicate TakeExec cannot merge by name. The projection keeps only the carried columns, so the physical join gets a tight output projection.

TakeExec is now null-tolerant: outer-join insert rows have a null target._rowaddr, so the take fetches only the non-null addresses and scatters results back with NULLs at the insert positions. The taken fields are marked nullable when the take sits above an outer join.

Tests

• late_take.rs unit tests for the rule and node in isolation: wide-column deferred (scan narrowed, output schema unchanged), narrow/join-key/duplicate-name columns not deferred, qualifier-aware matching, outer-join nullable fields, necessary_children_exprs, the is_wide_column gate (local vs cloud thresholds, blob exclusion), and end-to-end execution parity vs the un-deferred plan (inner + outer with null _rowaddr scatter).
• merge_insert subcols tests: the target LanceRead excludes the deferred column and a Take appears above the join; on a selective update the wide column is materialized only for the matched rows (Take output_rows = matches, LanceRead still visits all rows); an UpdateIf condition on the deferred target column evaluates correctly; multi-fragment, multi-wide-column, and scalar-index cases pass.
• TakeExec null-address unit tests.

Out of scope

Write-side amplification (delete + append rewrites all columns of matched rows) stays on #4193.

Closes #7363

🤖 Generated with Claude Code

[codecov]

Codecov Report

❌ Patch coverage is 94.70990% with 62 lines in your changes missing coverage. Please review.

Files with missing lines	Patch %	Lines
rust/lance/src/io/exec/late_take.rs	93.58%	40 Missing and 8 partials ⚠️
rust/lance/src/dataset/write/merge_insert.rs	97.45%	0 Missing and 7 partials ⚠️
rust/lance/src/io/exec/take.rs	95.30%	2 Missing and 5 partials ⚠️

📢 Thoughts on this report? Let us know!

[wjones127]

Summary

This PR added a logical optimizer rule (late_take.rs) that defers reading wide, non-source columns in merge_insert's join until after matching, using a width-only gate (assuming a fixed 0.1% match rate, borrowed from the scanner's filter heuristic). We looked at how to make that gate cardinality-aware instead.

Building a cost model (Cost(take) = M·σ·L vs Cost(scan) = N·w/B) surfaced that the hash-join shuffle (σ) — not just selectivity — is the dominant, previously-unaccounted-for cost, since TakeExec receives row addresses in scrambled order.

That led to a better path: DataFusion's dynamic filter pushdown for hash joins can eliminate the problem at the source by skipping wide-column decode during the scan itself, with no shuffle penalty. Lance's FilteredReadExec doesn't yet implement the pushdown side (filed #7562). Tracing DataFusion's on_lr_is_preserved/join-swap logic shows that once that lands, plus the source-stats PR (#7368) and a DataFusion bump to 54.x, dynamic filtering will cover merge_insert's two dominant shapes: pure backfill (Inner) and upsert (Right→Left after the build-side swap).

It fundamentally cannot cover delete-tracking joins (Left/Full, target preserved) — dropping unmatched probe rows there would incorrectly discard legitimate delete candidates. For that case, the fix isn't a smarter width gate but extending LateTakeNode's existing necessary_children_exprs declaration to also require the opposite side's join key (a generic, sentinel-free match indicator), so TakeExec skips fetching unmatched rows there too.

Since the upstream work (#7562, #7368, the DataFusion bump) covers the common cases better than this PR's approach, closing this in favor of that, plus a smaller follow-up for the delete-tracking fix.