Spark Table Statistics

Spark chooses a join strategy before it has read a single row. It chooses by comparing one number — sizeInBytes, the estimated size of each side of the join — against a threshold. That number is almost never the size of your data, and understanding the three rules that produce it explains most of the plans that look insane.

The three rules, all from SizeInBytesOnlyStatsPlanVisitor, the estimator Spark uses when the cost-based optimizer is off — which is the default:

• A file-backed table's size is the compressed bytes on disk, multiplied by spark.sql.sources.fileCompressionFactor, which defaults to 1.0.
• A Filter assumes it removes no rows at all. Its estimated size equals its child's.
• A Join's estimated size is the product of its two children's sizes.

Below is that estimator running on a two-table join. Move the controls and watch the estimate and the truth diverge, then watch the join strategy follow the estimate. The one that matters most is ANALYZE TABLE — because on its own it changes nothing.

Start at the defaults. dim is 180 MB of Parquet, the filter leaves 7.5 MiB of it, and the threshold is 10 MiB — so a broadcast is obviously right, and Spark plans a sort-merge join anyway. Now tick ANALYZE TABLE: still a sort-merge join, because a row count on its own does not reach the filter. Tick FOR COLUMNS + cbo.enabled and the estimate finally moves. Two settings, in that order, and neither is on by default.

Why a WHERE clause changes nothing

This is the surprising one, so here it is as code. With the cost-based optimizer off, Spark estimates a Filter like this:

override def visitFilter(p: Filter): Statistics = visitUnaryNode(p)

private def visitUnaryNode(p: UnaryNode): Statistics = {
  val childRowSize  = EstimationUtils.getSizePerRow(p.child.output)
  val outputRowSize = EstimationUtils.getSizePerRow(p.output)
  // Assume there will be the same number of rows as child has.
  var sizeInBytes = (p.child.stats.sizeInBytes * outputRowSize) / childRowSize
  ...

The comment says it outright: assume there will be the same number of rows as child has. The only thing that shrinks the estimate is a narrower row — dropping columns. WHERE country = 'US' keeps every column, so outputRowSize equals childRowSize, and the estimate above the filter is exactly the estimate below it. A predicate that eliminates 99.9% of your rows moves this number by zero bytes.

That is not laziness. Without a row count and a distinct-value count for country, there is no defensible way to guess how many rows survive, and guessing low would make Spark broadcast things that are not small. Assuming a filter does nothing is the conservative direction: it produces slow plans rather than failed ones.

There is exactly one filter that does shrink the estimate, and it is not really a filter. If the table is partitioned by dt and you write WHERE dt = '2026-08-07', the matching directories are chosen before the relation is even constructed, so sizeInBytes is the listed bytes of those directories only. That is partition pruning, and it is why a query filtered on a partition column can get a broadcast join while the identical query filtered on a normal column does not.

What ANALYZE TABLE actually gives you, in two steps

ANALYZE TABLE has three forms and they collect different things. Only the third one reaches a WHERE clause.

• ANALYZE TABLE dim COMPUTE STATISTICS NOSCAN — records sizeInBytes from the file listing. For a Parquet table this is roughly what Spark already had.
• ANALYZE TABLE dim COMPUTE STATISTICS — scans the table and records sizeInBytes and rowCount.
• ANALYZE TABLE dim COMPUTE STATISTICS FOR COLUMNS country, id — records, per column, the number of distinct values (usually written NDV, for number of distinct values), the minimum, the maximum, the null count and the average length.

The column statistics are the ones the filter estimator needs, and it needs spark.sql.cbo.enabled = true to be consulted at all — that setting defaults to false. Then, for col = literal with no histogram, FilterEstimation returns a selectivity of 1 / distinctCount. Drag distinct countries in the panel with both boxes ticked: the estimate is the table size divided by that number, exactly.

Two traps sit in the first line of the estimator: if (childStats.rowCount.isEmpty) return None. Column statistics without a row count are useless, so FOR COLUMNS alone does not work — you need the row count too. And on some partitioned-table paths the row count comes back empty even after an ANALYZE, which is why people report the optimizer ignoring statistics they can see in DESCRIBE EXTENDED.

The third trap is time. spark.sql.statistics.size.autoUpdate.enabled defaults to false, so writing to a table does not refresh its statistics. An ANALYZE run once, six months ago, on a table that has since grown 30×, is worse than no statistics at all: the conservative direction is gone and Spark will confidently broadcast something enormous. Tick 'US' is really 40% of dim to see the same failure from a different cause — the uniformity assumption behind 1 / distinctCount breaking on skewed values.

The product rule, and the 8 EiB you see in EXPLAIN COST

For a node with two children and no usable statistics, the default estimate is the product of the children:

Statistics(sizeInBytes = p.children.map(_.stats.sizeInBytes).filter(_ > 0L).product)

A 180 MB table joined to a 40 GB table estimates 1.9 × 10⁸ × 4.3 × 10¹⁰ ≈ 8.1 × 10¹⁸ bytes, which is about 7 exbibytes. That is not a bug in your query; it is what the rule computes. It is also why EXPLAIN COST so often prints sizeInBytes=8.0 EiB partway up a plan: the number has saturated at the largest value a 64-bit signed integer can hold.

The consequence is specific, so be precise about what it does and does not break. Joining a fact table to four dimensions one after another is fine: each dimension is judged on its own sizeInBytes, so each can be broadcast individually no matter how large the running estimate gets. What breaks is broadcasting the result of a join. Build a small lookup by joining three 40 MB reference tables in a common table expression, join that lookup to your fact table, and the planner compares the lookup's estimate — a product — against the threshold. The lookup is 120 MB in reality; on paper it is 40 MiB × 40 MiB × 40 MiB, which saturates at 8 EiB. So it is shuffled, and the 40 GB fact table is shuffled with it.

There is one escape, added to the size-only estimator for equi-joins: if either side's join keys are known to be unique — distinctKeys, which Spark derives from an Aggregate or a DISTINCT, not from any declared primary key — the estimate becomes the sum of the children instead of the product. That is correct, because a join to a unique key cannot produce more rows than the other side has. The panel below is that branch.

Set the count to 2 and each table to 4 MB. Unticked, two 4 MiB tables multiply to a 16 TiB estimate and the lookup is shuffled, along with 40 GiB of fact table, to avoid broadcasting 8 MiB. Tick the box and the estimate becomes the sum, 8 MiB, which is under the threshold — the broadcast comes back and the fact table stays where it is. This is the mechanical reason adding a SELECT DISTINCT or a groupBy inside a subquery sometimes makes a whole query dramatically faster for no visible reason.

Both ways of being wrong, and what each one costs

Overestimate — the estimate says 180 MB, the truth is 3 MB. Spark plans a sort-merge join: both sides are shuffled by the join key, sorted, and merged. You move 40 GB of fact table across the network to avoid copying 3 MB. The job is slow and nothing fails, which is why this one survives in production for years.

Underestimate — the estimate says 8 MB because that is the Parquet file size, and the same rows are 400 MB once decompressed into a hash table. Spark plans a broadcast hash join: the driver collects the whole small side, then ships a copy to every executor. Now the driver is holding 400 MB it did not budget for, and you get either OutOfMemoryError on the driver or Could not execute broadcast in 300 secs. To reach it in the first panel: tick both ANALYZE boxes, set bytes on disk to 400 MB, distinct countries to 2, expansion to 12× and the threshold to 256 MB. The estimate lands at 200 MiB, the truth at 2.3 GiB, and turning the cost-based optimizer on is what walked you into it.

The expansion factor is real and worth a number. Parquet stores columns dictionary-encoded and compressed; the broadcast side is materialised as rows in a hash relation. A 3–10× ratio is ordinary and a 30× ratio happens with heavily repeated string columns. Spark's own configuration documentation for spark.sql.sources.fileCompressionFactor says the default of 1.0 can "lead to a heavily underestimated result". Raising it to a measured value for your data is the single most effective statistics change most teams never make.

Adaptive query execution fixes the overestimate case and only that case. After a shuffle write completes, it knows the true byte size of what was written, and it will convert a sort-merge join into a broadcast join if a side turned out small. It does not help the underestimate: the broadcast is planned before anything runs, so the driver is already collecting 400 MB before there is any runtime measurement to react to. Tick adaptive.enabled in the first panel with the defaults and watch it rescue the first case; then set up the underestimate and watch it not rescue the second.

Reading the numbers in your own cluster

Four commands, in the order you should run them when a join picked the wrong strategy.

• df.explain("cost") prints Statistics(sizeInBytes=…, rowCount=…) on every logical node. This is the ground truth for what the optimizer believes. If rowCount is absent, the cost-based optimizer is inert no matter what you have analyzed. If you see 8.0 EiB, you have found a product-rule saturation.
• DESCRIBE EXTENDED dim shows the table-level Statistics line; DESCRIBE EXTENDED dim country shows that column's distinct count, min, max and null count. Missing means never analyzed; present but stale is worse.
• In the Spark UI's SQL tab, the physical plan node is labelled BroadcastHashJoin or SortMergeJoin, and the BroadcastExchange node carries a data size metric. That metric is the truth. Compare it to what explain("cost") predicted, and the ratio is your table's real expansion factor.
• To confirm a diagnosis in one step, set spark.sql.autoBroadcastJoinThreshold to -1, which disables broadcast selection entirely. If the job stops failing, the estimate was too low. If it gets dramatically slower, the estimate was fine and something else is wrong.

Next: the decision this number drives, in the broadcast join lesson; what happens when the estimate loses and you get a shuffle, in the shuffle; and where the 400 MB you did not budget for has to fit, in executor and driver memory.