[RLC] Case 1 follow-up — MERGE source-key read predicate as an exact semi-join (MERGE ‖ MERGE on disjoint keys)

[sezruby]

Stacked on #15 (conflict-data-skipping-delete-read → value-exact delta-io#7466 → min/max delta-io#7358). Implements the Case 1 follow-up from #4 / delta-io#7360: MERGE ‖ MERGE on disjoint keys.

Revision 2. The first commit used a literal IN list with a maxKeys fallback. Black-box tests of Databricks Runtime show it reconciles MERGE ‖ MERGE per key with no source-size cap (tested to 4M source rows), so the key check must be exact and threshold-free. The second commit switches to that: a pre-filter (InSet ≤ 10000 keys, else a min/max range) plus an exact left-semi join in the value-exact tiers. There is no cap and no fallback.

---

1. Problem

OSS MERGE records only the target-only conjuncts of ON as its read predicate
(getTargetOnlyPredicates). A key-only upsert ON t.id = s.id therefore reads true. On an
unpartitioned or liquid-clustered table, every concurrently added file conflicts, so two MERGEs on
disjoint keys always abort the loser with ConcurrentAppendException. Conflict-time data skipping
(Case 1) cannot help, because there is no predicate to skip with.

2. Reference behavior

OSS docs (docs.delta.io, Concurrency control): MERGE ‖ MERGE "can conflict", depending on
whether the operations touch the same files. The only recommended mitigation is partitioning plus an
explicit ON predicate.

Databricks Runtime (our black-box tests, DBR 16.4 / 18.2, unpartitioned DV table, WriteSerializable):

Scenario	DBR loser
disjoint updates + disjoint inserts, same file	commit
same NEW key inserted by both / same row updated by both	abort (ConcurrentAppendException)
6 writers, all disjoint keys	6/6 commit (1/6 without DV)
source 1k – 1M (+3M inserts), disjoint from the winner	commit
same, but the source includes 1 row the winner deleted	abort

The decision is per key, not per file or min/max range: a loser key inside the winner's new file's
range still commits. It does not degrade with the source size, so the key check is exact and has no
key-count cap. A key-set pre-filter can only affect how much of the winner's change is scanned.

3. Design

MergeIntoCommandBase.getSourceKeyReadPredicates returns, for a MERGE without NMBS whose ON has
usable equi-join keys, a set of predicates that are all ANDed in one read:

1. Pre-filters (one per key). These are cheap and evaluable everywhere.
   • At most inSetThreshold distinct key tuples (collected with distinct().limit(threshold+1)):
     InSet(t_i, non-null source values).
   • Above the threshold: t_i >= min(s_i) AND t_i <= max(s_i), from one aggregate.
   • A key with no non-null source value can match no row, so it becomes false.
2. MergeSourceKeyMatch(targetKeys, sourceKeys). A marker predicate whose children are the
   target keys and which carries the source DataFrame projected to the source keys. It evaluates to
   true in every generic path (file pruning, partition filters, stats skipping), where the
   pre-filters do the work. The conflict checker's value-exact tiers evaluate it exactly, as a
   left-semi join of the winner's rows against sourceKeys on the key tuple.

These are appended to the existing deltaTxn.filterFiles(...) call in
ClassicMergeExecutor.findTouchedFiles and in the InsertOnlyMergeExecutor path. The pre-filters
also prune the MERGE's own target scan and readFiles.

Key extraction. Each EqualTo(t, s) conjunct of ON qualifies when:

• t references only target columns and s only source columns, in either operand order;
• both sides are deterministic and have the same type;
• the type has binary equality.

Excluded:

• <=>, because NULL would match NULL;
• FLOAT/DOUBLE, because of -0.0 = 0.0 and NaN normalization;
• non-binary string collations.

For these exclusions, a pre-filter on the raw collected values could disagree with the join's =.

Conflict checker (ConflictDataSkippingReader.rowsMatchingAnyRead, used by both
value-exact tiers):

• If no read carries a MergeSourceKeyMatch, the code path is unchanged (a single where).
• Otherwise each read is evaluated as where(plain filters), then left_semi join with each
  marker's source keys. The reads are unioned and the result is checked for existence.
  • Tier 2, added files: !rows.isEmpty.
  • Tier 1b, delete/read: pre-image minus post-image weighted count, as before. A row matching
    several reads is counted once per read, and that multiplicity depends only on the row's values,
    so pre − post stays exact.
• Tier 1 (min/max stats) sees only the pre-filters. The marker builds no skipping filter, so it is
  dropped, which over-approximates.
• Any failure (e.g. a lost materialized block) is caught by the existing fail-safe and counted as a
  conflict.

Source lifetime. The commit runs inside runWithMaterializedSourceLostRetries, before the
materialized RDD is unpersisted, so the conflict check reuses it. A non-materialized (deterministic
Delta) source is re-read, exactly as MERGE itself reads its source twice.

Observability. Event delta.dml.merge.sourceKeyReadPredicate
{numKeyColumns, inSetThreshold, preFilter: inSet|range, numDistinctKeys (-1 above threshold)}.

4. Correctness

• Sound. A target row can satisfy ON only if its key tuple equals some source row's key
  tuple. So "key tuple ∈ source key tuples" ⊇ the rows the MERGE reads. This covers:

  • MATCHED rows;
  • rows matched by a MATCHED clause with a condition (the condition only narrows the joined rows);
  • the absence of a matching row, which lets NOT MATCHED insert: a winner row that would have
    matched has the same key tuple.

  Each pre-filter is implied by the tuple match, so pre-filters ∧ match equals the tuple match.

• Exact on tuples. The semi-join matches whole key tuples. A winner row (60, 5) against loser
  keys {(60, 0), (61, 5)} passes the per-column pre-filters but is not a conflict. This is tested.

• Same new key. The winner's inserted k ∈ S is found by the semi-join, so the loser aborts and
  there is no duplicate.

• Key-changing UPDATE. If the winner moves a key into S, the new image is an added file, so it
  is checked.

• Fail-safe. Flag off, NMBS, or no usable key gives today's predicate. Any scan or join error is
  treated as a conflict.

5. Size / threshold

There is no correctness threshold. inSetThreshold only selects the pre-filter. Changing it never
changes an outcome (tested with 1 and 0), only how much of the winner's change survives to the join.
Costs:

• MERGE side: one distinct().limit(threshold+1) job, plus one min/max aggregate above the
  threshold.
• Commit side: the semi-join of the winner's stats-surviving added rows (bounded by
  valueExact.maxBytes) against the source keys. Spark picks broadcast or shuffle. The cost scales
  with the winner's change and the source size, not with the table size.

6. Scope & interactions

Concurrent shape	Result (+ Case 1 value-exact, 1b)
disjoint keys, different files	commit
disjoint keys, keys inside each other's ranges, any source size	commit (join decides)
same new key / same existing row	abort (genuine)
disjoint keys, same file, both DV-update it	still DELETE_DELETE until stacked with Case 2 (#2)
NOT MATCHED BY SOURCE	unchanged (whole-table read)
value-exact disabled	pre-filters only (file-level / range), never less safe

7. Validation

The MergeSourceKeyReadPredicateSuite (13 tests) runs real interleaved MERGE races. The loser stops
at pre-commit via TransactionExecutionTestMixin, then the winner commits:

• disjoint keys commit (DV and COW), and both MERGEs are applied;
• flag off → ConcurrentAppendException;
• same new key / same row → abort, with no duplicate;
• 20000-key source (range pre-filter): a winner key inside the range but not in the source
  commits, and a winner touching one source key aborts;
• inSetThreshold 1 and 0: disjoint keys still commit;
• multi-column: the exact tuple match commits where per-column IN would conflict;
• insert-only MERGE, NMBS, and event shape.

Mutation check. With the semi-join removed (pre-filter only), exactly the 3 join-dependent tests
fail: the 20000-key, tiny-threshold and multi-column tests. So they really exercise the join.

Results (JDK 17, Spark master build, base conflict-data-skipping-delete-read f3b6a00)

Run	Result
MergeSourceKeyReadPredicateSuite	13 / 13 pass
ConflictDataSkippingSuite + DeleteReadConflictDataSkippingSuite (default config)	32 / 32 pass
spark/scalastyle, spark/Test/scalastyle	clean
23 MERGE suites (Basic*, DVs*, SQL, ExtendedSyntax, Misc, TempViews, NestedData, NMBS, SchemaEvolutionCore, NondeterministicOrder, BasicScala) with the flag forced ON	629 pass, 2 fail, 106 ignored (pre-existing)

With the flag on, the only failures are insert only merge - target data skipping (path-based and
DV). The test asserts exactly 2 source and target scans, and collecting the source keys adds one
more. MERGE results are identical. With the flag off (the default), the test is unaffected.