Part of the row-level-concurrency umbrella (#7360, Case 1). Implemented in #7358.
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 onlySo 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 (#7356) |
| Same file, disjoint rows |
delete/delete, delete/read (path-keyed) |
DV-merge (#7357) |
| 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 #7357'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 #7357; 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 #7356: unpartitioned → headOption → B's added file conflicts → ConcurrentAppendException → A aborts
with #7356: 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 (#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 #7357 — supersedes its op-type append-suppression
The #7359 POC (issue #7357) 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. #7357 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).
Part of the row-level-concurrency umbrella (#7360, Case 1). Implemented in #7358.
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(inConflictChecker) — the heart of the added-files (append) check — does: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:
DeltaTableReadPredicatecarriesdataPredicates(populated bytrackReadPredicates); the checker just ignores the data half — it references onlypartitionPredicate(s), andDeltaLog.filterFileListfilters 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)— mirrorsfilesForScan's eligibility filtering +DataFiltersBuilder+ conjunction fold, returning aDataSkippingPredicate(skip expression + referenced stats), or None.DataSkippingReaderBase.filterFilesByDataSkipping(files, dataFilters)— parses each file's stats against this snapshot'sstatsSchema(sharedparseAndDecodeStats, so it matches the scan path) and keeps files whereexpr || !verifyStatsForFilter(referencedStats). Returns the surviving originalAddFiles (matched by path); wrapped inrecordFrameProfile.ConflictChecker.getFirstFileMatchingPartitionPredicatesconsumes the current txn'sdataPredicates, applied per read predicate with the survivors unioned (see semantics below), then the existing partition filter /headOptionruns on the narrowed set. Emits adelta.conflictDetection.dataSkipping.filesSkippedevent.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 —
UPDATEandMERGEadd 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/readanddelete/deleteare path-keyed, so this change does not affect them; the same-file case is #7357's job.)Roles: the loser (whose commit is being checked) supplies the read predicate (
WHERE/ MERGEON) that drives the skip — DELETE/UPDATE/MERGE all have one. The winner supplies the added files whose stats are pruned — INSERT, UPDATE, MERGE, or clusteringOPTIMIZE.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 #7357; 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-separatedcustomer_idmin/max.UPDATE orders SET tier='gold' WHERE customer_id BETWEEN 0 AND 1000→ rewrites low-id files, read predicatecustomer_id ∈ [0,1000].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:
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 (#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:
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.filesForScan), so they can't produce an unsafe skip.Deletion-vector (non-tight) stats are safe. DV/OPTIMIZE-touched files carry
tightBounds=false;updateStatsToWideBoundskeeps min/max as valid outer bounds (a superset of live values) and only rewritesnullCountto 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 viaverifyStatsForFilter. 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_loginto 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)DELETE id<50vs append[1000,1100)) → added file skipped, both commit.ConcurrentAppendException).dataSkippingNumIndexedCols=0) → disjoint ranges still conflict (one-way-safety invariant).Relationship to #7357 — supersedes its op-type append-suppression
The #7359 POC (issue #7357) took a shortcut for the concurrent-
UPDATEcase: 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 beingDELETE/UPDATE. That is unsound — anUPDATEcan rewrite a value so a row moves into the loser's predicate (winnerSET x=15, loserDELETE WHERE x>10, row wasx=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. #7357 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).