[RLC] Case 1 — conflict-time data skipping (disjoint data ranges) #4

Part of the row-level-concurrency umbrella (#3, Case 1). Implemented in #8.

Apply the reader's data skipping at conflict time so writers touching disjoint data ranges
don't falsely conflict. The biggest win on unpartitioned / liquid-clustered tables, the lowest
incremental cost (the plumbing already exists), and independent of deletion vectors and row tracking.
Scope: spark/ module (ConflictChecker, DataSkippingReaderBase).

Problem

OSS conflict detection is correct-by-abort: it cancels conservatively so a bad interleaving never
commits, at the cost of cancelling more than necessary. One large source of unnecessary
cancellations is that the conflict checker filters candidate files by partition only — it does
no data skipping.

getFirstFileMatchingPartitionPredicates (in ConflictChecker) — the heart of the added-files
(append) check — does:

if (currentTransactionInfo.readWholeTable ||
    currentTransactionInfo.readSnapshot.metadata.partitionColumns.isEmpty) {
  return files.headOption      // unpartitioned -> ANY winner-added file conflicts
}
// else: DeltaLog.filterFileList(partitionSchema, files, partitionFilters)  // partition filter only

So on an unpartitioned table (which includes every liquid-clustered table), any winner
changed-data file conflicts with a concurrent writer that has any read predicate — even when the two
operations touch entirely disjoint data ranges.

Half the plumbing already exists: DeltaTableReadPredicate carries dataPredicates
(populated by trackReadPredicates); the checker just ignores the data half — it references only
partitionPredicate(s), and DeltaLog.filterFileList filters by partition only.

Goal

Skip winner-added files whose stats prove they cannot match the current transaction's data
predicates — i.e. apply the same data skipping the reader uses, at conflict time. Removes false
conflicts on unpartitioned/LC tables (and tightens partitioned tables) without weakening correctness.

Design (as implemented)

• DataSkippingReaderBase.buildDataSkippingPredicate(dataFilters) — mirrors filesForScan's
  eligibility filtering + DataFiltersBuilder + conjunction fold, returning a DataSkippingPredicate
  (skip expression + referenced stats), or None.
• DataSkippingReaderBase.filterFilesByDataSkipping(files, dataFilters) — parses each file's stats
  against this snapshot's statsSchema (shared parseAndDecodeStats, so it matches the scan path)
  and keeps files where expr || !verifyStatsForFilter(referencedStats). Returns the surviving
  original AddFiles (matched by path); wrapped in recordFrameProfile.
• ConflictChecker.getFirstFileMatchingPartitionPredicates consumes the current txn's
  dataPredicates, applied per read predicate with the survivors unioned (see semantics below),
  then the existing partition filter / headOption runs on the narrowed set. Emits a
  delta.conflictDetection.dataSkipping.filesSkipped event.
• Gated by spark.databricks.delta.conflictDetection.dataSkipping.enabled (internal, default off).
  No protocol / reader / action changes.

Which operations benefit — not just append

The check improved is the append (added-files) check, which fires whenever the winner adds data
files. That is not only INSERT — UPDATE and MERGE add new files (rewritten/inserted row
images), and because those are non-blind changed-data files they are conflict-checked under the
default WriteSerializable isolation (a blind INSERT is only checked under Serializable). So the
headline beneficiaries are concurrent UPDATE / DELETE / MERGE, under the default isolation.

The two levers partition the DML-concurrency space by which conflict check fires:

Because clustering separates key ranges into different files, the common LC concurrent-DML case is
"different files" → squarely this issue's domain. (delete/read and delete/delete are path-keyed,
so this change does not affect them; the same-file case is #5's job.)

Roles: the loser (whose commit is being checked) supplies the read predicate (WHERE /
MERGE ON) that drives the skip — DELETE/UPDATE/MERGE all have one. The winner supplies the
added files whose stats are pruned — INSERT, UPDATE, MERGE, or clustering OPTIMIZE.

The one DML case this does not cover: a DV-DELETE winner re-adds the same file (same
path, tightBounds=false) rather than a new one — that is the same-file case with wide stats, owned
by #5; here it is conservatively kept.

Worked example (liquid clustering)

LC table orders CLUSTER BY (customer_id) (unpartitioned), default WriteSerializable. Clustering
gives every file a tight, well-separated customer_id min/max.

• Job A (loser): UPDATE orders SET tier='gold' WHERE customer_id BETWEEN 0 AND 1000
  → rewrites low-id files, read predicate customer_id ∈ [0,1000].
• Job B (winner): UPDATE orders SET tier='gold' WHERE customer_id BETWEEN 900000 AND 901000
  → adds new-image files with stats [900000, 901000].

They touch different files, so delete/read + delete/delete don't fire. The append check does:

without #4:  unpartitioned → headOption → B's added file conflicts → ConcurrentAppendException → A aborts
with #4:     A's predicate [0,1000] vs B's file stats [900000,901000] → no overlap → SKIP → both commit

This works under the default isolation because B's UPDATE files are non-blind changed data. The
same shape holds for MERGE (ON keys → predicate when the ON has target-only conjuncts — see Gap: key-only MERGE below; images → added files), non-DV DELETE, and blind
INSERT (the last only under Serializable). Clustering is what makes it pay off: the tight,
separated stats make the skip decisive — on a non-clustered unpartitioned table files often span the
whole range and little could be skipped. It restores the concurrency partitioning used to
provide before the shift to LC (delta-io#7057's motivation).

Gap: key-only MERGE registers no usable read predicate

ClassicMergeExecutor.findTouchedFiles records only the target-only conjuncts of the ON
condition (getTargetOnlyPredicates). For the common upsert shape ON t.id = s.id there are none, so
the loser's read predicate is true: every winner-added file matches, neither tier can skip anything,
and concurrent upserts on disjoint keys still abort with ConcurrentAppendException on
unpartitioned / LC tables.

Follow-up — prototyped in #20: have MERGE register a source-derived read predicate on the
equi-join keys: the loser read exactly the target rows whose key tuple appears in its source. The
existing tiers then decide per key:

• winner added a row whose key is in the loser's source → match → abort. This includes two
  concurrent MERGEs inserting the same new key: the loser's NOT MATCHED decision is stale.
• disjoint keys → no match → both commit.

Why it is sound: a MERGE's MATCHED / NOT MATCHED outcome depends only on target rows whose join
key equals some source key, so a predicate covering the source keys is a superset of what the loser
actually read. Limits: only for equi-joins on key columns (other ON shapes keep true);
NOT MATCHED BY SOURCE still reads the whole target (umbrella Case 6); a scoped ON (e.g.
... AND t.p = 'A') is checked only within its scope, so the same key inserted outside that scope is a
uniqueness concern, not a conflict-detection one.

Which keys. The predicate must cover all source keys, not only the NOT MATCHED ones. A conditional matched clause (e.g. WHEN MATCHED AND s.ts > t.ts THEN UPDATE, typical for CDC upserts) reads row k and may leave it untouched; if a concurrent writer changes k, that no-op decision is stale, and only the read predicate catches it. Rows the loser actually writes are already protected by the deletion-vector overlap checks, so restricting to NOT MATCHED keys is a valid optimization only when every MATCHED clause is unconditional. Case 1b's removed-side check must use the same key predicate.

Representation (#20) — exact semi-join, no size threshold. A literal IN list needs a size cap,
and a min/max fallback above it degrades to range-level conflicts for the large upserts that need this
most; the reference implementation reconciles per key with no source-size cap (black-box tested to 4M
source rows). So #20 registers two predicates:

• a pre-filter for file pruning and the min/max tier — IN (<keys>) up to
  merge.sourceKeyReadPredicate.inSetThreshold (default 10000) distinct keys, else the source key
  [min, max] range. It never decides an outcome.
• an exact marker, MergeSourceKeyMatch, carrying the loser's source keys: true in generic paths
  (sound over-approximation), and evaluated by this issue's value-exact tier (and [Spark] Conflict-time data skipping (delete/read tier) #15's removed side)
  as a left-semi join of the winner's surviving rows with the source keys.

Opt-in via merge.sourceKeyReadPredicate.enabled (default off).

Benefit depends on key overlap. Concurrent upserts that genuinely share a key still abort, as they must. For large batches of keys drawn from a large, actively updated key space, two concurrent batches sharing at least one key is likely, so this reconciles mainly when concurrent writers' key sets are actually disjoint (e.g. writers partitioned by key, or small batches).

Correctness — the one-way safety rule

Data skipping in conflict detection is sound in one direction only: it may skip a file only when
stats prove no overlap. A wrongly-skipped file = a missed conflict = the corruption the abort
was preventing. Guaranteed by:

• Reusing the reader's machinery (buildDataSkippingPredicate + verifyStatsForFilter), not a
  hand-rolled min/max comparison — so semantics match the scan path exactly.
• expr || !verifyStatsForFilter(referencedStats) keeps any file whose referenced stats are
  missing/NULL.
• Per-read OR, never cross-read AND. Data predicates from independent reads have OR semantics (a
  file conflicts if it could match any read); combining them with AND could drop a file that
  matches one read → missed conflict. So skipping is applied per read predicate and the survivors
  unioned; a file is dropped only if it fails every read. Whole-table reads disable skipping
  entirely.
• Non-deterministic / subquery / metadata-column filters are dropped up front (mirrors
  filesForScan), so they can't produce an unsafe skip.

Deletion-vector (non-tight) stats are safe. DV/OPTIMIZE-touched files carry tightBounds=false;
updateStatsToWideBounds keeps min/max as valid outer bounds (a superset of live values) and only
rewrites nullCount to a tri-state. A superset can only cause fewer skips, never an unsafe one —
the exact property the reader relies on to skip DV'd files at scan time, inherited here for free via
verifyStatsForFilter. The consequence is effectiveness, not correctness: on heavily-DV'd files
the wide bounds rarely prove non-overlap, so they are kept (conservative). Most effective on fresh,
tight-stat files (appends, clustering output, update images) and on predicates over clustering
columns.

Multi-cluster / multi-JVM

Correct across clusters/JVMs by construction: conflict detection is log-based. The loser reads the
winner's committed actions — including their persisted stats — from the shared _delta_log into
its own JVM, and evaluates skipping against its own local read predicates. No shared memory or
cross-JVM RPC; it reads the same log the existing partition check reads, and the skip decision is
deterministic. Other-engine / no-stats writers are handled safely (missing stats → keep). The
standing prerequisite (true of any multi-writer Delta) is an atomic commit primitive — atomic
put-if-absent on the log or a commit coordinator — which guarantees a single winner per version.

Tests (ConflictDataSkippingSuite)

• disjoint ranges (DELETE id<50 vs append [1000,1100)) → added file skipped, both commit.
• overlapping ranges → still conflicts (ConcurrentAppendException).
• feature disabled → disjoint ranges still conflict.
• missing stats (dataSkippingNumIndexedCols=0) → disjoint ranges still conflict (one-way-safety
  invariant).
• partitioned table → data skipping on a non-partition column avoids the conflict.
• empty read predicates on a non-blind txn → all added files kept (no unsafe skip).

Relationship to #5 — supersedes its op-type append-suppression

The #2 POC (issue #5) took a shortcut for the concurrent-UPDATE case: a blanket append-suppression
that skips all of a rewrite-only DML winner's new image files in the append check, keyed only
on the winner being DELETE/UPDATE. That is unsound — an UPDATE can rewrite a value so a row
moves into the loser's predicate (winner SET x=15, loser DELETE WHERE x>10, row was x=5), a
genuine write-skew the blanket suppression hides: the loser commits without deleting the now-matching
row, strictly weaker than base WriteSerializable. Those image files belong here: they are ordinary
non-blind changed-data files (a different path from the DV'd source), so this stats-based skip is
the correct arbiter — abort on a real flip (stats overlap the loser's predicate), reconcile only on
proven no-overlap, keep (abort) on missing stats. #5 drops the suppression and depends on this for
UPDATE image-file reconcile; the two stack (this data-skipping check under Case 2's DV merge).