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

[sezruby]

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:

Concurrent DML shape	Check that fires	Handled by
Different files (disjoint clustering ranges)	append check (winner's added files)	this issue (#4)
Same file, disjoint rows	delete/delete, delete/read (path-keyed)	DV-merge (#5)
Same file, same rows	delete/*	neither — genuine conflict, aborts

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; 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).

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).