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

[sezruby]

Part of the row-level-concurrency umbrella (#3) — Case 1. Full design and rationale: #4.

Summary

Reader-side conflict-time data skipping. Conflict detection currently filters candidate files by partition only. On an unpartitioned table — which includes every liquid-clustered table — getFirstFileMatchingPartitionPredicates returns files.headOption, so the append check conflicts on any concurrently-added file, even when the two operations touch entirely disjoint data ranges. This is the dominant source of false conflicts for liquid clustering, where incremental clustering and DML usually touch different clustering-key regions.

This teaches conflict detection to apply the reader's column-statistics data skipping: an added file whose stats prove it cannot match the current transaction's data predicates is excluded from the conflict check.

What it does

• DataSkippingReaderBase.filterFilesByDataSkipping(files, dataFilters) — reuses the reader's DataFiltersBuilder + verifyStatsForFilter to build the skipping predicate, parses each file's stats against the snapshot's statsSchema, and applies expr || !verifyStatsForFilter(...).
• ConflictChecker.getFirstFileMatchingPartitionPredicates consumes the already-captured DeltaTableReadPredicate.dataPredicates (previously ignored). Skipping is applied per read predicate with the survivors unioned (OR across independent reads; a read's own predicates are ANDed), and is not applied for a whole-table read — so it can only ever narrow, never miss, a conflict. 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 improved check is the added-files (append) check, which fires whenever the winner adds data files — so it is not only INSERT. UPDATE/MERGE add new row-image files, and because those are non-blind changed-data files they are checked under the default WriteSerializable (a blind INSERT is only checked under Serializable). So the headline beneficiaries are concurrent UPDATE / DELETE / MERGE on disjoint clustering ranges (different files), under the default isolation.

This complements the same-file DV merge (#5), splitting the space by which check fires:

Concurrent DML shape	Check that fires	Handled by
different files (disjoint ranges)	append check	this PR (Case 1, #4)
same file, disjoint rows	delete/delete, delete/read	#5 (Case 2, DV merge)
same file, same rows	delete/*	neither (genuine conflict)

Roles: the loser supplies the read predicate (WHERE / MERGE ON); the winner supplies the added files whose stats are pruned. (A DV-DELETE winner re-adds the same file, so that's #5's same-file case, not this PR's.)

One-way safety

Data skipping here is sound in one direction only — a file is skipped only when its stats prove no overlap; a wrongly-skipped file would be a missed conflict. Guaranteed by reusing the reader's machinery:

• expr || !verifyStatsForFilter(referencedStats) keeps any file whose referenced stats are missing/NULL.
• Deletion-vector (tightBounds=false) stats are safe: updateStatsToWideBounds keeps min/max as valid outer bounds (a superset of live values), so wide bounds only cause fewer skips, never an unsafe one — the same property the reader uses to skip DV'd files. The effect on DV-heavy files is reduced benefit (conservative keep), not incorrectness.
• Subquery / non-deterministic / metadata-column filters are dropped up front (mirrors filesForScan).
• Data predicates from independent reads are OR-combined, never AND-combined; whole-table reads disable skipping entirely; an empty read-predicate set on a non-blind txn keeps all files.

Correct across clusters/JVMs by construction — conflict detection is log-based: the loser reads the winner's committed stats from _delta_log and evaluates skipping against its own local predicates (no shared state; deterministic; other-engine/no-stats writers are kept).

Supersedes #5's op-type append-suppression

The #2 POC (issue #5) shortcut the concurrent-UPDATE case with a blanket append-suppression for rewrite-only DML winners — skipping all their new image files in the append check, keyed only on the winner's op type. That is unsound: an UPDATE can flip a row into the loser's predicate (winner SET x=15, loser DELETE WHERE x>10, row was x=5), a genuine write-skew the suppression hides. Those image files are ordinary non-blind changed-data files and are correctly arbitrated by this PR's stats-based skip — abort on stats overlap (real flip), reconcile only on proven no-overlap, keep on missing stats. #5 drops the suppression and depends on this PR; they stack (this data skipping under the DV merge).

Testing

ConflictDataSkippingSuite (new), phase-locking harness on a Serializable table (10 files, id in [0,1000)):

• disjoint (DELETE id<50 vs append [1000,1100)) → added file skipped, both commit.
• overlapping (DELETE id>=950 vs append [1000,1100)) → 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).
• plus unit coverage for the per-read AND-combine and the whole-table-read guard.

With the flag off, getFirstFileMatchingPartitionPredicates is equivalent to before (candidateFiles = files), so no default-path regression.

🤖 Generated with Claude Code