[File-level concurrency] Conflict-time data skipping (min/max stats tier)

[sezruby]

Addresses #4470 (improved concurrency model for liquid / unpartitioned tables) at the file level — avoid conflicts when concurrent transactions touch disjoint files, decided by column statistics without reading row values. Part of the concurrency umbrella (#7360) — Case 1. Full design and rationale: #7356.

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 (#7357), 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, #7356)
same file, disjoint rows	delete/delete, delete/read	#7357 (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 #7357'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 #7357's op-type append-suppression

The #7359 POC (issue #7357) 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. #7357 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

[jiahao-db]

For every readPredicates, we are going to launch a spark job, is that possible we combine those predicates together for one spark job to help improve performance?

[sezruby]

Good catch — done. Replaced the per-read flatMap (which launched one job per read predicate) with a new filterFilesMatchingAnyReadPredicate(files, dataFiltersPerRead) that builds one skipping predicate per read and OR-combines them into a single where, so all reads are evaluated in one Spark job.

I kept them OR-combined rather than flattening every read's filters into a single predicate: independent reads have OR semantics (a file is a candidate if it could match any read), whereas flattening would AND them (read1 AND read2) and incorrectly skip files that one read alone would match. As a bonus, if any read yields no usable skipping predicate (empty/ineligible filters → matches everything), the union matches everything too, so we short-circuit and return all files without launching a job at all.

filterFilesByDataSkipping is now just the single-read case delegating to the new method, so existing behavior/tests are unchanged. Added a test (filterFilesMatchingAnyReadPredicate OR-combines independent reads in one call) that pins the OR semantics.

[jiahao-db]

Should we add fall back mechanism to default conflict behavior if we get failed on evaluation?

[sezruby]

Added. The build + evaluation path is now wrapped in try/catch(NonFatal); on any failure it logs a warning and returns all candidate files, i.e. falls back to the default conflict behavior where every added file remains a conflict candidate (exactly as when the feature is disabled). Since conflict-time data skipping is purely an optimization over the conservative default, failing safe this way means a skipping error can never abort an otherwise-valid commit.

[rahulsmahadev]

high level review:

• changes in conflict checker are well guarded with the spark conf, is it possible to have the code in DataSkippingReader also be either in a separate utility or an isolated method that doesn't touch existing code

will do more deeper reviews soon

[jihoonson]

Thanks @sezruby for the PR! I finally got some time to review your work. Looks good overall, left some comments.

[jihoonson]

filterFilesMatchingAnyReadPredicate applies the data predicates first before the partition pruning happening in the getFirstFileMatchingPartitionPredicatesInternal in the below. Pruning unnecessary partitions first can avoid unnecessary computation in data predicate processing.

[sezruby]

Agreed, done. Data skipping now runs inside the per-read partition probe, on the output of DeltaLog.filterFileList, so both filters are fused into the same head(1) job and the data predicates only apply to files from that read's partitions. Catalyst may combine and reorder the two filters, so this doesn't guarantee that the partition filter is evaluated first for every row, but no separate pass parses stats for every candidate file anymore. Unpartitioned tables (and the readWholeTable path) skip the partition step as before.

[jihoonson]

The predicate evaluation here can have false positives. Suppose you have two queries. Both query has a predicate of (P AND D) where P and D are a partition predicate and a data predicate, respectively. So what we want to evaluate here is (P1 AND D1) OR (P2 AND D2) to find survivors. However, what the code is evaluating here is effectively (P1 OR P2) AND (D1 OR D2). These are not the same, and the latter can have false positives. Consider two reads p=1 AND id<50 and p=2 AND id>=100, and a winning file with usable statistics min(id) = max(id) = 150 in partition p=1. The file will survive data skipping because of read 2, then passes partition filtering because of read 1. This will lead to unnecessary transaction aborts.

[sezruby]

Great catch, you're right. Each step OR'd across reads on its own, so the cross terms P1 AND D2 and P2 AND D1 survived. That never misses a real conflict, but it aborts commits it doesn't need to, as in your p=1 AND id<50 / p=2 AND id>=100 example.

Fixed by pairing each read's data predicates with its own partition predicate. Reads are grouped by canonical partition predicate (non-deterministic predicates are never merged, matching ExpressionSet). Each group runs one probe, filterFileList(P_g).where(OR_{i in g} skip(D_i)), so the result is exactly OR_i (P_i AND D_i). When the feature is off, the probe list is the existing canonicalPredicates with no data filters, so behavior is unchanged.

Added multiple reads: data predicates are paired with their own partition predicate, which uses your exact example. It checks both cross terms (no conflict), a match against each read (conflict), and feature-off (conflict).

[jihoonson]

createTable has a side effect of setting dataSkippingNumIndexedCols to 0, which removes the min/max stats from the stats schema entirely. As a result, no skipping predicate will get built, and the per-file "keep files with missing stats" check will be never reached. Suggest to have these files in the table:

• A file with valid statistics proving it cannot match id < 50.
• Another file with stats = null.

[sezruby]

Good point. To clarify, createTable only sets dataSkippingNumIndexedCols when numIndexedCols is passed, and only the missing stats test passes Some(0). The feature disabled test keeps default stats. But you're right that in the missing stats test no skipping predicate gets built, so we return early and the per-file !verifyStatsForFilter path is never exercised. Added two tests on a table that does have min/max stats:

• files with null stats are kept while files whose stats prove no match are skipped: 10 files with valid stats plus one stats = null file, against id < 50. The 9 files whose stats prove no match are skipped; the [0,100) file and the null-stats file are kept.
• concurrently added file with null stats still conflicts: same check end-to-end through ConflictChecker.

I kept the numIndexedCols = 0 test as well, since it covers the "no stats schema" early return.

[jihoonson]

Would it be nice if we record this fallback as a delta event? Currently this catch block logs the exception and conservatively retains all files, but provides no structured indication that the skipping failed.

[sezruby]

Done. The fallback now lives in ConflictChecker, because the skipping filter is evaluated lazily inside the probe's head(1). On NonFatal it logs and records delta.conflictDetection.dataSkipping.fallback (winning version, exception class and message), then re-runs the probe without skipping. The reader methods no longer catch.

Covered by skipping evaluation failure falls back to treating added files as conflicts (536ae8a): a read predicate id < 'abc' makes the skipping filter fail at evaluation, and the test asserts the commit still conflicts, the fallback event is recorded once, and no noConflict event is recorded.

[jihoonson]

This event is being recorded before the partition pruning. And it will be recorded even when the commit still conflicts. Should we do it only when the commit succeeds after the whole data skipping process is completed?

[sezruby]

Agreed. Removed the filesSkipped pre-pass event. It now works like this:

• When skipping was applied to a probe that found no conflicting file, a flag is set.
• At the end of checkConflicts() the event delta.conflictDetection.dataSkipping.noConflict is recorded. That point is only reached if no check against this winning commit threw.
• So nothing is recorded when the commit conflicts. The new test asserts the event count in both cases.

Two caveats, and I'm happy to change either:

1. The event is per winning commit, so a later winning commit, or the final write, can still fail the transaction. Recording at commit level would mean threading a flag through OptimisticTransaction. I can do that if you prefer.
2. Because of the head(1) fusion it no longer reports a skipped-file count. Counting would need another job, or a check of whether partition pruning alone would have conflicted. It's an extra job, but only on the no-conflict path. Let me know if attribution is worth that.

[sezruby]

Thanks @jihoonson for the detailed review! Addressed everything in 56f12d4. The main change is structural: data skipping is no longer a pre-pass over all reads. Each read's data-skipping filter is now fused into that read's partition probe, so a file conflicts iff it could match P_i AND D_i for some read i. That fixes the false positive you found and also makes partition pruning run first. It adds no extra Spark jobs: the skipping filter rides in the same head(1) job the partition probe already ran.