Dynamic File Pruning

Dynamic file pruning does not filter your fact table by the rows the join matched. It takes the join-key values found on the small side of the join, tests each one against the minimum and maximum value recorded for every file on the big side, and keeps any file whose range contains at least one of them. How many files that leaves has almost nothing to do with how selective your join was.

The setup is the ordinary star-schema query. A fact table, events, holds 192 million rows in 160 Parquet files. A dimension table, customers, holds 200,000 rows. You filter the dimension, join on customer_id, and you would like Databricks to read only the parts of events that can possibly match.

SELECT e.event_type, count(*)
FROM   events e
JOIN   customers c ON e.customer_id = c.customer_id
WHERE  c.tier = 'enterprise'
GROUP  BY e.event_type

Ordinary file skipping cannot help here. Skipping compares a predicate against the per-file minimum and maximum values stored in the table's _delta_log, and that needs a constant. c.tier = 'enterprise' is a constant, but it is a constant about the wrong table. Which customer_id values it implies is not known until the dimension has actually been read. Dynamic file pruning closes that gap: it runs the small side first, collects the key values it produced, and pushes them into the file filter of the big side before the big side opens a single file.

Two names for the rest of this lesson, because the documentation uses them and they are backwards from how most people say it. The build side is the small table, the one read first to build a hash table — customers here. The probe side is the big table whose rows are then looked up in it, and whose files are the ones being pruned — events.

Below is that pipeline. The events table has been Z-ordered on customer_id, which is the textbook advice, and the dimension filter matches 4,024 rows — two percent of the customers. Read the panel before touching anything.

160 files read out of 160. Nothing was pruned. The table is perfectly Z-ordered on exactly the column being joined, the dimension filter threw away 98% of the customers, and the scan still touched all 21.47 GB.

Now change nothing except the filter on customers, from c.tier to c.signup_date BETWEEN …. Same table, same layout, same 4,024 matching rows. Files read drops to 5, and rows scanned from 192 million to 6 million. The third option, six industry blocks, lands at 10.

The join did not become more selective. What changed is where the matching customer_id values sit relative to each other, and that is the only thing dynamic file pruning can respond to.

Why the count of matching rows tells you nothing

Look at the strip under the map in each case. With c.tier = 'enterprise' the pushed values are spread across the whole customer_id domain, roughly one every 50 ids, because tier has nothing to do with when a customer signed up and ids are issued in signup order. With the date range they are one solid block, because ids are issued in signup order and the predicate is on signup date.

Against that, the map shows what Z-ordering bought. At 100% clustering the median file covers a customer_id range 1,240 wide out of 200,000 — each file holds about 0.6% of the key space, which is as narrow as 160 files can be. That is a good layout by any measure. It is also irrelevant when the pushed values are spread out, because a file is kept if any single one of them falls inside its range, and with 4,024 values spaced 50 apart, every 1,240-wide window contains about 25 of them.

Set the filter back to c.tier and walk rows it matches up from zero. The saturation is abrupt:

• 1 matching row → 1 file read.
• 39 → 39 files. 72 → 68 files. Every value is landing in its own file.
• 132 → 116 files. 243 → 155 files.
• 447 → all 160 files. Every file now contains at least one wanted id.

Under perfect clustering, files read is approximately the smaller of the number of distinct pushed values and the number of files. Once your dimension filter matches more rows than the fact table has files, and those rows are spread across the key domain, pruning is finished — and 447 out of 200,000 customers is a filter almost nobody would describe as unselective. The 4,024 of the default is nine times past the point of no return.

This is the part that inverts the usual intuition. Making the dimension filter tighter helps only until you cross below the file count. Making the values closer together helps without limit. Those are different properties and only the second one is under the control of your table layout.

The contiguous case shows the other half. Switch the filter on customers back to the date range and walk the same slider from the bottom: 1 row reads 1 file, 132 rows still read 1, 447 read 2, 2,790 read 3, 4,024 read 5, and 9,457 read 9. Here the count really does drive the answer, because contiguous values occupy a contiguous stretch of the domain and therefore a contiguous run of files. There is no saturation cliff at 447 the way there was for c.tier.

One more notch, 10,685 rows, and the readout goes to all 160. That is not saturation either — the pruning stopped being attempted, and the section after next is about why. Note what it means for the ladder above: on this table the contiguous case never gets to demonstrate its own ceiling, because a different limit arrives at 10,240 matched rows and takes the feature away entirely.

What is actually pushed down

Spark builds this in the optimizer as a DynamicPruningSubquery attached to the probe-side scan. At planning time PlanDynamicPruningFilters looks for a broadcast hash join on the same keys whose broadcast exchange it can reuse. If it finds one, the subquery becomes:

DynamicPruningExpression(InSubqueryExec(customer_id, SubqueryBroadcastExec))

SubqueryBroadcastExec waits on the broadcast the join was going to build anyway, pulls the join-key column out of the hash relation, and hands the scan a list of values. That reuse is the whole trick: the build side is read exactly once and serves both the join and the pruning. The subquery is named dynamicpruning#<exprId>, which is the string to look for in a physical plan.

Databricks' contribution is what happens next. In open-source Spark the value list prunes partition directories — it can only eliminate data whose key is encoded in a path. Databricks pushes the same list into the Delta scan instead, where it is tested against the minValues and maxValues that every add action in the log carries for the first 32 columns. The 2020 announcement puts it in one sentence: "a dynamic filter is created from the build side of the join and passed into the SCAN operation." That is why the feature works on a non-partitioned column, and it is the reason it is worth understanding separately from partition pruning.

The test each file gets is the ordinary interval test. Given pushed values V and a file with recorded bounds [min, max], the file can be skipped only if no member of V lies in that interval. The decision log in the panel prints one worked example of each outcome on every change, with the actual boundary values and the actual nearest pushed value on either side, so you can check the arithmetic rather than take it.

Two consequences follow directly and neither is obvious from the docs. The filter is a set of values, so its cost grows with the size of the build side — this is why the feature is tied to a broadcast in the first place. And the test is per-file and independent, so pruning has no way to exploit the fact that your 4,024 ids came from one contiguous predicate. If the layout does not put them in few files, nothing downstream can.

The two ways it silently does nothing

Both of these leave a dynamicpruning expression sitting in the physical plan. Neither logs a warning.

No statistics for the join key. Put rows it matches back to 4,024, where the date-range filter was reading 5 files, then switch the join key in the schema to column 41 of 60. Files read jumps from 5 to 160, the verdict readout says no bounds, and the map goes flat — every bar becomes full width, because unknown bounds are unbounded bounds. Delta records min/max only for columns whose position is below delta.dataSkippingNumIndexedCols, and the default in DataSkippingReader.scala is DATA_SKIPPING_NUM_INDEXED_COLS_DEFAULT_VALUE = 32. A column at position 41 has no minValues entry, a file whose bounds are unknown could contain anything, and a pruning rule that must never drop a matching row has to keep it.

Three ways to land here without noticing. Someone adds columns to the middle of the schema and pushes your join key past 32. Someone sets delta.dataSkippingStatsColumns to an explicit list — the Delta source is blunt that "the column stats not mentioned by this config will be ignored even if they exist", so the statistics can be physically present in the log and still unused. Or the key is a long string: the third option in that dropdown is a 44-character order_ref whose first 32 characters are a constant prefix, and spark.databricks.io.skipping.stringPrefixLength defaults to 32, so substring(min(c), 0, 32) records the identical value for every file in the table. The bounds exist, they are correct, and they separate nothing.

The broadcast went away. Put the join key in the schema back to column 3 of 60 first — the paragraph above left it at column 41, and with no statistics every reading below is 160 whatever else you do. Then set the filter to the date range and rows it matches to 9,457. Nine files read. Move it one notch, to 10,685 rows. All 160.

The estimated build side crossed spark.sql.autoBroadcastJoinThreshold — 9.24 MB to 10.43 MB against a 10 MB default — so the join is planned as a sort-merge join instead. There is now no broadcast exchange to reuse, and PlanDynamicPruningFilters takes its other branch: DynamicPruningExpression(Literal.TrueLiteral). A filter that is the literal true is still a filter in the plan. It removes nothing. Tick force the broadcast with a hint and the same 10,685-row query reads 10 files again.

Making a dimension filter slightly less selective can therefore cost you an order of magnitude, which is not a shape anyone expects from a query optimizer. It also means an out-of-date sizeInBytes on the dimension can switch dynamic file pruning off by mistake: the planner only has an estimate, and if it guesses the build side is 12 MB when it is really 2 MB you lose both the broadcast and the pruning. When adaptive query execution converts a sort-merge join back to a broadcast at runtime, it does so after the shuffle has already been written — too late to prune the scan that fed it.

The join type is the third condition and it is a hard one. Switch join type to LEFT OUTER JOIN: 160 files. A left outer join with events on the left must emit every fact row whether or not it matched, so no fact file can be excluded on the strength of the dimension. The 2020 announcement lists the condition as "The join type is INNER or LEFT-SEMI"; the current documentation page dropped it, which is worth knowing before you go looking for it. Spark's own rule is the authority underneath: canPruneLeft in joins.scala returns true only for Inner, LeftSemi and RightOuter, and the fact table is on the left here. LEFT SEMI in the dropdown behaves identically to INNER, because both discard non-matching fact rows.

The thresholds, and the cliff nobody expects

Put the join type back to INNER, untick the broadcast hint and put rows it matches back to 4,024 — the last two paragraphs moved all three. Leave the filter on the date range and drag files in events down to 74. The table is 9.93 GB and all 74 files are read. Now put it up by one, to 75 files and 10.07 GB. Two files are read.

Adding 128 MB to a table reduced what the query scans from 9.93 GB to 0.27 GB, because spark.databricks.optimizer.deltaTableSizeThreshold is "the minimum size (in bytes) of the Delta table on the probe side of the join required to trigger dynamic file pruning", and its default is 10,000,000,000 bytes. Below it the optimizer does not attempt pruning at all — the reasoning being that on a small table the cost of computing the filter outweighs the scan it would save, which is defensible on average and surprising in the specific case.

deltaTableFilesThreshold is the second gate, defaulting to 10 files. Which of the two binds first is decided by your file size, and this is worth a moment because the defaults encode an assumption. 10,000,000,000 bytes divided by 10 files is 1 GB per file, and optimize.maxFileSize defaults to 1 GiB — so on a table written with default-sized files the two gates open within a few percent of each other. On a table of 128 MiB files, the file gate opens at 11 files and the size gate not until 75, so the file gate never binds and the size gate binds 65 files later than you would guess from reading about it. On a table of thousands of tiny files the size gate is the only one that ever matters.

Both defaults are session settings you can lower. The honest reason to do so is a table just under the line whose queries are latency-sensitive; the dishonest one is to make a benchmark look better on a table small enough that it would have been cached anyway.

Raising the ceiling

Everything above says pruning is bounded by the layout. So change the layout. Put files in events back to 160 — the threshold section left it at 11, where pruning is not attempted at all — keep the date-range filter and the match count at 4,024, and sweep clustering on the join key from 0 to 100:

• 0% — arrival order. 160 files read. The median file spans 197,894 of the 200,000 possible ids, so every file's range contains the whole block. This is what an un-optimized event table looks like and it is why a point lookup on a single customer, at the bottom of the match slider, also reads all 160 files.
• 50% — 113 files. Median range 140,383.
• 80% — 44 files. Median range 49,467.
• 100% — 5 files. Median range 1,240.

The curve is steep at the top, which matters practically: partial clustering buys you very little. Going from 0% to 50% removes 29% of the work; going from 90% to 100% removes another 11% of the original but cuts the remaining work by more than three quarters. A table that is "roughly sorted" by the join key is not most of the way there.

This is where Databricks' own advice to use liquid clustering to maximise dynamic file pruning comes from, and it is the right advice for a reason that is easy to miss: clustering has to survive your writes. A Z-ordered table degrades towards arrival order with every append, so the 100% row above is the state of the table immediately after OPTIMIZE and not the state your 3 a.m. query runs against. Liquid clustering re-clusters incrementally as data lands, which keeps you near the top of that curve instead of sliding down it between maintenance windows.

One interaction to be aware of if you have deletion vectors on. The Delta protocol lets a file carry tightBounds: false, meaning the recorded minimum is merely "less than or equal to all valid values" rather than equal to the smallest one. Deleting the only enterprise customer in a file does not narrow that file's range, so it keeps qualifying for pruning tests it can no longer satisfy. The bounds stay correct — they are still safe to skip on — they just stop being tight, and only a rewrite makes them tight again.

Checking it on a real system

The diagnosis is always the same three questions in the same order: was pruning attempted, did it have statistics to work with, and was the layout capable of using them. Answer them separately or you will fix the wrong one.

Was it attempted? Run DESCRIBE DETAIL events and read sizeInBytes and numFiles. If sizeInBytes is under 10,000,000,000 or numFiles is 10 or fewer, dynamic file pruning was never in play and no amount of clustering will change that. Then check the join: EXPLAIN FORMATTED and look for BroadcastHashJoin rather than SortMergeJoin, and for a scan filter containing dynamicpruning#. A dynamicpruning expression that reads simply true means the broadcast was not reusable and the filter was planned away — that is the Literal.TrueLiteral branch, and the fix is a BROADCAST hint or a corrected sizeInBytes, not a table rewrite.

Did it have statistics? Query the log directly. This returns one row per live file with the recorded bounds for your join key:

SELECT count(*)                                  AS files,
       count(get_json_object(add.stats, '$.minValues.customer_id')) AS with_stats
FROM   json.`/path/to/events/_delta_log/*.json`
WHERE  add IS NOT NULL

If with_stats is 0 the column is outside the statistics set: check its ordinal position against SHOW TBLPROPERTIES events for delta.dataSkippingNumIndexedCols and delta.dataSkippingStatsColumns. Changing either property does not recompute anything for files already written; on Databricks Runtime 14.3 LTS and above ANALYZE TABLE events COMPUTE DELTA STATISTICS backfills them, and otherwise the files have to be rewritten. If the values are present but identical everywhere, you have the truncated-string case or a layout problem, not a statistics problem.

Was the layout capable? Take the same query and compute the two numbers the panel shows. From the log, the median width of maxValues.customer_id - minValues.customer_id across files. From the dimension, count(DISTINCT customer_id) under your filter and the span between its smallest and largest value. If the distinct count exceeds numFiles and the values span most of the domain, the join is at its pruning ceiling and clustering the fact table harder will not move it. Cluster on something the predicate is actually contiguous in instead — often the date column the dimension filter is really about.

The number that confirms it. In the Databricks query profile, open the scan node for the fact table and compare files read against the table's file count; the filter icon on the metric shows the percent pruned. Do this on the scan node, not the whole query, because a query that spends its time in the aggregation will look fine in wall-clock terms right up until the fact table doubles. A useful habit is to record files read alongside the query, so that the day pruning silently stops — a schema change, a dimension that grew past the broadcast threshold, an OPTIMIZE that stopped running — you see a step change in a counter rather than a gradual complaint about latency.

Two things this lesson does not cover but which will bite in the same area. Dynamic file pruning in MERGE, UPDATE and DELETE requires Photon-enabled compute; without it those statements do not get it at all, which is the usual explanation for a MERGE with a 200-row source rewriting a whole target. And dynamic file pruning is a Databricks Runtime feature, not part of the Delta Lake protocol — an open-source Spark job against the same table gets Spark's partition-level pruning only, which is why advice that works in a notebook can fail in a self-managed cluster reading the identical files.