fix: dedupe unmatched merge insert source rows

[lance-gatekeeper]

Summary

• Track non-null join keys for unmatched source rows in FirstSeen mode.
• Apply the same behavior to the v2 and indexed-scan merge paths across source batches.
• Preserve SQL NULL equality semantics and clarify the public documentation.

Root cause

FirstSeen deduplication tracked matched target row IDs only. Unmatched insert actions have no target row ID, so duplicate source keys bypassed the tracker and every row was inserted.

Validation

• cargo test -p lance test_first_seen_dedupes_unmatched_source_rows
• cargo test -p lance source_dedupe
• cargo fmt --all -- --check
• cargo clippy --all --tests --benches -- -D warnings

Fixes #7907

[lance-gatekeeper]

Gate recommendation: request changes. The exact non-null-key state fixes the reported cross-batch gap in both merge paths, but logical Arrow NULLs can still enter the seen set and be discarded as duplicates. Detect logical NULL after scalar extraction before recording a key so the implementation matches the documented NULL-distinct contract.

[lance-gatekeeper]

Array::is_null checks only the physical top-level null buffer, so it returns false for logical NULLs such as NullArray and dictionary/run/union child-null values. Those rows become the same ScalarValue::Null; HashSet::insert then rejects the second row, silently dropping a source row despite the documented SQL NULL semantics. Extract the scalar first and use ScalarValue::is_null() (or equivalent logical-null detection) before adding the key.

Reproducer run against this head

#[test]
fn test_inserted_key_tracker_preserves_logical_nulls() {
    let batch = RecordBatch::try_new(
        Arc::new(Schema::new(vec![Field::new("id", DataType::Null, true)])),
        vec![Arc::new(arrow_array::NullArray::new(2))],
    )
    .unwrap();
    let mut tracker = InsertedKeyTracker::default();
    let on = ["id".to_string()];

    assert!(tracker.insert(&batch, 0, &on).unwrap());
    assert!(tracker.insert(&batch, 1, &on).unwrap(), "SQL NULL keys must remain distinct");
}

cargo test -p lance test_inserted_key_tracker_preserves_logical_nulls --lib fails the second assertion on 256b97f0813ee8b1cefd3736e5fcd27d6449001f; both calls must return true.

[lance-gatekeeper]

Gate recommendation: request changes. The logical-NULL fix now matches SQL NULL semantics, but FirstSeen still does not reliably keep the first source row on the v2 path: processing can reverse ordered input batches, contradicting the public contract and its guidance to sort the source. Preserve source order through v2 before deduplication, or make execution-order semantics an explicit maintainer contract decision.

[lance-gatekeeper]

FirstSeen still does not preserve the documented source order on the v2 path. Making both batches identical avoids observing the winner, but an ordered reader with (108, 1) followed by (108, 2) retains value 2; sorting the reader therefore cannot select the documented first row. Preserve source order through the v2 merge before applying deduplication, or make execution-order semantics an explicit maintainer contract decision, and keep a distinct-value cross-batch regression.

Reproducer run against this head

I restored the prior distinct-batch inputs in a clean isolated checkout of a7e94563b6bd33185834f75b6fd8787839f6f9a1 (first batch value 1, second batch value 2) and ran:

cargo test -p lance test_first_seen_dedupes_unmatched_source_rows -- --nocapture

The v2 case failed at the final assertion: expected 1, observed 2. The indexed case passed.

[lance-gatekeeper]

❌ Gate recommendation: request changes.

FirstSeen still does not reliably preserve source order on the v2 path, and the post-join sort introduces a full-stream memory barrier in the current execution context.

Prefer deduplicating the sequential source stream before the unordered join; that fixes the winner at the contract boundary, reduces join work, and avoids sorting the full joined payload.

[lance-gatekeeper]

This post-join sort still does not make FirstSeen deterministic: the current distinct-batch regression remains flaky on v2, so a later source value can still win. It also runs under TaskContext::default(); that runtime uses an unbounded memory pool, so DataFusion’s external sorter never gets a failed reservation that would trigger spilling and can retain the full joined payload before emitting. Deduplicate the sequential source stream before the unordered join instead; that fixes the winner before reordering, reduces join work, and avoids this global memory barrier.

Reproducer run against this head

for review_run in {1..20}; do
  cargo test -p lance test_first_seen_dedupes_unmatched_source_rows --lib -- --test-threads=1 || exit 1
done

On e3bd090d3906df8c541e7556de12b72f1cb8592e, run 1 passed and run 2 failed case_1_v2 at the final assertion: expected retained value 1, observed 2. The indexed case passed.

[lance-gatekeeper]

⚠️ Gate recommendation: approve with a non-blocking risk.

The source-side deduplication fixes FirstSeen winner selection before join reordering and removes the full joined-payload sort. Exact key tracking remains an adoption consideration for very high-cardinality sources; a bounded or spillable key store is the longer-term mitigation.

[lance-gatekeeper]

Exact deduplication retains one Vec<ScalarValue> per unique non-null key for the lifetime of the source scan, so memory is O(unique-key bytes) and is not governed by DataFusion's memory pool. This is materially smaller than sorting the full joined payload and is non-blocking, but very high-cardinality sources can still exhaust memory. A bounded/spillable key store (or a documented cardinality limit) would make this operationally predictable.

[lance-gatekeeper]

⚠️ Gate recommendation: approve with a non-blocking risk.

The checked compare-exchange replacement preserves skipped-duplicate accounting and does not change the source-side ordering fix. The existing high-cardinality key-memory risk remains; a bounded or spillable key store is still the longer-term mitigation.

[codecov]

Codecov Report

❌ Patch coverage is 75.56818% with 43 lines in your changes missing coverage. Please review.

Files with missing lines	Patch %	Lines
rust/lance/src/dataset/write/merge_insert.rs	77.77%	18 Missing and 12 partials ⚠️
...lance/src/dataset/write/merge_insert/exec/write.rs	61.90%	6 Missing and 2 partials ⚠️
...ance/src/dataset/write/merge_insert/exec/delete.rs	58.33%	4 Missing and 1 partial ⚠️

📢 Thoughts on this report? Let us know!

[lance-gatekeeper]

⚠️ Gate recommendation: approve with a non-blocking risk.

The merge from main preserves the source-side ordering fix and checked duplicate accounting, including their interaction with the newer indexed and full-write paths. The existing high-cardinality key-memory risk remains; a bounded or spillable key store is still the longer-term mitigation.