Track: native MERGE via lance-core MergeInsertBuilder (bypass Spark position-delta path)

[sezruby]

Tracking issue

Investigate adding a native MERGE strategy in lance-spark that dispatches
directly to lance-core's MergeInsertBuilder instead of going through
Spark's V2 row-level operations / position-delta path. Goal: measure
whether the native path is meaningfully faster on tables with many
fragments, then file upstream against lance-format/lance-spark if the
numbers justify it.

Today (lance-format/lance-spark @ HEAD)

MERGE INTO lance . ` SQL routes through:

• LancePositionDeltaDataset implements SupportsRowLevelOperations
• LancePositionDeltaOperation with representUpdateAsDeleteAndInsert = true
• rowId() = [_rowaddr], requiredMetadataAttributes() = [_fragid]

Catalyst's RewriteMergeIntoTable lowers MERGE into a position-delta
plan: WriteDelta(... Project(deletes+inserts) Join(targetScan, source, ON merge-condition) ...). The join, optional shuffle, and per-_fragid
write are all Spark stages.

LanceScan.planInputPartitions() emits one InputPartition per Lance
fragment, so target-scan and position-delta-write stages are
fragment-aligned. With a 2500-fragment table, several fragment-aligned
stages × 2500 = thousands of Spark tasks per MERGE.

FilterPushDown.isFilterSupported accepts only basic V2 Filter types
(EqualTo, In, range, null). The MERGE ON condition is evaluated as
a join predicate, not a pushable filter — so pushedFilters is empty
for the target scan, zonemap pruning skips, and all fragments are
scanned regardless of indexes.

MergeInsertBuilder from lance-core is not invoked anywhere in
lance-spark today.

Proposal

A planner strategy that intercepts MergeIntoTable nodes targeting Lance
tables and routes them through MergeInsertBuilder:

dataset.mergeInsertBuilder(onColumns)
  .whenMatchedUpdateAll()                  // or: ...UpdateAllIf(condition)
  .whenMatchedDelete()                     // or: ...DeleteIf(condition)
  .whenNotMatchedInsertAll()               // or: ...InsertAllIf(condition)
  .whenNotMatchedBySourceDelete()          // or: ...DeleteIf(condition)
  .execute(sourceRecordBatches);

Pros:

• Avoids Spark Join + shuffle + position-delta write (the per-_fragid
  rewrite stage).
• Lance-core can use BTREE/Bitmap indexes on the merge ON column
  for fragment pruning during the match phase.
• Per-arm predicates (whenMatchedUpdateAllIf etc.) map naturally onto
  the SQL WHEN MATCHED ... AND <cond> clauses.

Cons:

• Single-process: MergeInsertBuilder.execute doesn't shard across
  Spark executors. Source has to be materialized as RecordBatches on
  the driver (or one executor).
• Predicate translation: Spark Expression → Lance SQL string for the
  per-arm *If predicates.

Plan

1. Stand up a benchmark in lance-spark-knn_2.12/.../benchmark/ (same
   harness pattern as IndexedNearestJoinBenchmark):

   • Build a synthetic Lance table at varying scales (100, 500, 2500
     fragments).
   • Run MERGE INTO SQL via the current path.
   • Run an equivalent direct MergeInsertBuilder.execute from
     Java/Scala.
   • Validate equivalence (oracle check on a sample of merge keys).
   • Time both, report wall-clock + Spark task counts.

2. If MergeInsertBuilder wins meaningfully on the
   2500-fragment-with-small-source case (the workload shape seen in
   prod), file upstream as a feature request with the numbers attached.

3. If it doesn't (e.g. one-process source materialization dominates),
   document why and close.

Out of scope for this tracking issue

• Streaming source path (always materialize for the bench).
• Auto-routing heuristic between native and position-delta paths (will
  manifest as a config flag in the PoC).
• Full Catalyst MERGE rewrite — the PoC can call MergeInsertBuilder
  directly from a benchmark harness, no SQL parser changes needed
  for measurement.

Status

• Branch + benchmark scaffold
• Synthetic Lance table generator
• Position-delta path baseline run
• MergeInsertBuilder direct-call run
• Equivalence validation
• Numbers attached
• Upstream issue (filed only if numbers justify)

[sezruby]

Update: benchmark built + run — native path is faster at the one scale that completed

Benchmark is implemented and runs end-to-end against real pushshift reddit
data (public dataset). Caveat up front: only the 64K-source point completed.
Larger scales were repeatedly blocked by cluster instability, so the data here
is thin — one scale, measured twice.

Setup

• Target: pushshift_target, 63.78M rows / 89 fragments / ~11.9 GB, on
  object storage (Azure ADLS Gen2). max_rows_per_file = 1,000,000.
• Cluster: standalone Spark 3.5, 8 × (4 cores / 16 GB).
• Source built by sampling the target and bumping a column (so each iter is a
  pure upsert: matched-update + a tail of inserts).
• Each iter merges into a fresh clone of the seed; medians where >1 iter.

Native path (what we measure)

org.lance.Dataset.mergeInsert(MergeInsertParams, ArrowArrayStream) called
directly — no Spark in the merge step. Source is collected to the driver
and streamed into the C-data ArrowArrayStream in 10K-row batches
(StreamingArrowReader); lance-core does the join + merge in Rust.

Source materialization caveat: the bench still does
source.collectAsList() on the driver before streaming — the batching bounds
lance-core-side memory, not driver memory. True executor-side streaming
(mapPartitions + ArrowConverters.toBatchIterator) is still TODO.

Results — 64K source (the only scale that completed)

run	SQL (position-delta)	native (MergeInsertParams)	speedup
smoke, 1 iter	17.7 s	15.5 s	1.14x
sweep, 2-iter median	16.7 s	10.9 s	1.53x

Two runs of the same 64K workload gave 1.14x and 1.53x — the spread is cluster
variance, not a code difference. Streaming reader verified end-to-end
([native] streamed N batches). That's the whole verified dataset: native is
modestly faster at 64K; production-scale is unproven.

Why only one scale

The sweep (64K / 320K / 3.2M / 6.4M) never finished. 320K reached one SQL iter
(~11 s) before a worker died; 3.2M and 6.4M never ran. Failures were Spark
master restarts and worker-pod deaths on long jobs — an infra limitation, not
the bench. Re-running on a stable cluster is the main open item.

Indexed native merge — regression (do NOT enable index for native merge)

With a scalar index on the join key, the native path is slower and
memory-unbounded, the opposite of the intent:

• Root cause (confirmed on lance main): can_use_create_plan refuses the v2
  fast path when a scalar index would be used (&& !would_use_scalar_index),
  falling back to v1 create_indexed_scan_joined_stream, which forks the
  source with ReplayExec::new(Capacity::Unbounded, …) — buffers the entire
  source into memory.
• Observed qualitatively: OOM as source grew, far slower than unindexed at 64K.
  (No clean figures retained — flagging as observed behavior, not a measurement.)
• Upstream tracks the fix: lance-format/lance #6444 (v2 scalar-indexed
  joins) and #3480 (loads all target columns). use_index=false is the
  current workaround.
• Implication: a native-merge connector should default use_index=false for
  large sources until #6444 lands.

How the current (SQL) path actually runs — confirmed against upstream source

MERGE INTO → LancePositionDeltaOperation implements RowLevelOperation, SupportsDelta, representUpdateAsDeleteAndInsert = true. Catalyst joins
source↔target across executors and writes position deltas
(delete-by-_rowaddr + insert). MergeInsertBuilder is not invoked anywhere
in lance-spark. So the native path is genuinely new work, not a re-route.

Dependency landed upstream

A native merge needs a defined behavior for duplicate source keys (Spark's
MERGE raises MERGE_CARDINALITY_VIOLATION; lance-core defaults to Fail). The
Java SDK couldn't configure it, so exposed SourceDedupeBehavior
(Fail/FirstSeen) on the Java MergeInsertParams: lance-format/lance#7296
(open).

Status

• Branch + benchmark scaffold
• Synthetic + real (pushshift) Lance table generator
• Position-delta path baseline run
• MergeInsertBuilder direct-call run (streaming source)
• Equivalence validation
• Numbers attached — 64K only; larger scales infra-blocked
• Java prerequisite filed upstream (SourceDedupeBehavior → lance#7296)
• Executor-side streaming source (remove driver collectAsList)
• Larger-scale numbers on a stable cluster (the real blocker)
• Upstream lance-spark feature request / RFC for the native path