Hive Cost-Based Optimizer — Join Ordering and Plan Selection

What the CBO decides — and what it does not

The CBO's authority is narrower than its reputation suggests, and knowing the boundary saves a lot of misdirected tuning. It owns join order, the physical join algorithm for each join, the placement of aggregations relative to joins, and a family of algebraic rewrites: predicate pushdown, projection pruning, constant folding, join-condition inference, limit pushdown and semi-join introduction. Those decisions are all made at compile time, before a single byte is read.

It does not own anything downstream of the plan. Container sizing, vectorization, the number of Tez reducers, file formats and compression codecs are separate concerns configured separately, and no amount of statistics will make the optimizer fix a table stored as uncompressed text in nine million small files. Nor does it choose partitioning or bucketing for you — it only exploits the physical design you already committed to. When a query is slow because the layout is wrong, the CBO is not the lever.

The practical consequence is a triage order. Ask first whether the plan shape is wrong — a join order that materialises an enormous intermediate result, or a shuffle where a broadcast was possible. That is the optimizer's department, and the fix is usually statistics rather than configuration. Only if the plan shape is defensible should you move on to execution-layer tuning, which is where most people start and where the returns are far smaller. A plan that joins two billion-row tables in the wrong order cannot be rescued by a bigger heap. The full compile-to-DAG pipeline this sits inside is covered in Hive query execution.

Turning it on, and proving it actually ran

CBO is controlled by hive.cbo.enable, on by default in every modern Hive release. It is rarely off, but it is frequently ineffective, and the two failure modes look identical from the outside.

The first is missing inputs. The optimizer can only cost what it can measure, and column-level statistics are fetched only when hive.stats.fetch.column.stats is enabled. With column stats unavailable, Calcite still runs, still reorders, and still emits a plan — it just prices every alternative from row counts and raw sizes alone, with no selectivity information at all. You get cost-based planning on a starvation diet, which frequently produces the same plan a rule-based optimizer would have.

The second is the silent bail-out. Certain query constructs fall outside what Hive's Calcite integration can represent, and rather than fail, the compiler abandons CBO for that statement and falls back to the older rule-based path. The query returns correct results; it is simply planned by a component that cannot reorder joins. Nothing in the result set tells you this happened.

The observable test costs nothing. Run EXPLAIN CBO on the statement: if a Calcite relational-algebra tree comes back, the optimizer planned the query; if it does not, the plan you are staring at was produced by the fallback path and no amount of running ANALYZE will change it. Make that check the first step of any Hive plan investigation, before reading a single operator. hive.cbo.show.warnings is the complementary signal for the first failure mode — it surfaces the optimizer's warnings about tables it had to plan without column statistics.

SET hive.cbo.enable=true;
SET hive.stats.fetch.column.stats=true;
SET hive.cbo.show.warnings=true;
SET hive.compute.query.using.stats=true;

EXPLAIN CBO
SELECT c.region, sum(o.amount)
  FROM orders o JOIN customers c ON o.cust_id = c.cust_id
 WHERE c.country = 'NL'
 GROUP BY c.region;

The statistics the cost model consumes

Two families of numbers feed the cost model, and they behave very differently. Table statistics — row count, raw data size, total size, held per partition — anchor the estimate for every scan. Column statistics — number of distinct values, null count, min/max, average and maximum column width — are what make selectivity estimation possible at all. Without NDV the optimizer cannot say how many rows survive country = 'NL', and without that it cannot say which join order is cheaper.

Table statistics largely maintain themselves: hive.stats.autogather updates row counts and sizes as a side effect of writes through Hive. Column statistics are the ones that go missing, though hive.stats.column.autogather extends the same treatment to them for inserts that Hive performs. Anything that lands data outside Hive's write path — a Spark job writing Parquet into the partition directory, a distcp, an external tool registering a partition — leaves the statistics untouched and quietly stale.

The explicit refresh is ANALYZE TABLE ... COMPUTE STATISTICS FOR COLUMNS, restricted where possible to the columns queries actually join, filter and group on. Scanning and sketching two hundred columns when eight of them carry every predicate in the workload is wasted cluster time. NDV is computed as a sketch rather than an exact distinct count, and hive.stats.ndv.error governs the accuracy-versus-cost trade; hive.stats.fetch.bitvector lets the optimizer retrieve the underlying sketches so that partition-level NDVs can be merged without double-counting values that appear in several partitions.

ANALYZE TABLE customers COMPUTE STATISTICS;
ANALYZE TABLE customers COMPUTE STATISTICS FOR COLUMNS cust_id, country, region;

ANALYZE TABLE orders PARTITION (dt='2026-08-01')
  COMPUTE STATISTICS FOR COLUMNS cust_id, product_id, amount;

DESCRIBE FORMATTED customers country;

The collection side — what the gathering job costs, how incremental collection accumulates metadata, and how to audit for tables the pipeline forgot — is developed from the Impala angle in Impala table statistics, and the mechanics carry over because both engines store these numbers in the same metastore.

The join-order search space: left-deep, bushy, and bounded

Join reordering is where the CBO earns its keep, and it is a search problem with brutal arithmetic. For n tables joined together, the number of distinct join trees grows factorially; past a handful of relations, exhaustive enumeration is not merely slow, it is impossible within a compile-time budget measured in hundreds of milliseconds. Every real optimizer therefore bounds the search, and the bounds are what determine plan quality on wide queries.

The first bound is shape. A left-deep tree joins one base table at a time into a growing accumulator: each join's build side is a fresh table and its probe side is the previous result. A bushy tree allows two intermediate results to be joined to each other — joining the two fact tables' filtered subsets separately and then combining them. Left-deep trees pipeline naturally and there are far fewer of them, so they are the traditional default; bushy trees win precisely when two independent branches each shrink dramatically under their own filters, because a left-deep tree is forced to carry one branch's full width through the other branch's joins.

The second bound is the algorithm. Hive collapses a chain of inner joins into a single multi-join operator, then re-derives an order over that operator's inputs. For a small number of inputs the planner can afford near-exhaustive exploration; beyond that it switches to a greedy heuristic that repeatedly picks the cheapest next join. Greedy is not optimal and does not pretend to be — it is a bet that a locally cheapest join is usually globally reasonable, which holds for star schemas and holds much less well for snowflakes and self-join chains.

What this means in practice: a five-table star query is almost always well optimised, and a fifteen-table report with several independent join clusters is where you should expect to inspect the plan rather than trust it. Outer joins constrain the search further, because they are not freely reassociable — the optimizer may not reorder across them without changing semantics, so a chain of left outer joins leaves it with very little freedom regardless of how good the statistics are.

How a candidate plan is priced

Every candidate plan is priced bottom-up, and each operator's cost depends on its estimated output cardinality — which then becomes the next operator's input. This is why estimation errors compound: a mistake at the leaf does not stay at the leaf, it multiplies up the tree.

Scan cardinality comes from table statistics, adjusted by any predicates pushed into the scan. Filter selectivity comes from column statistics: an equality predicate on a column with NDV d is assumed to pass roughly 1/d of the rows, a range predicate is interpolated against min/max, and IN lists are scaled by list length under hive.stats.filter.in.factor. Join output is estimated from the containment assumption — for an equi-join on a single key, roughly |A| × |B| divided by the larger of the two NDVs, so the join's cardinality is dominated by the side with more distinct key values. For a join on several key columns at once, hive.stats.correlated.multi.key.joins selects whether those columns are assumed independent or correlated. The distinction is not academic: assuming independence multiplies their NDVs together, producing a far larger divisor and therefore a much smaller estimated result — and underestimates are the direction that gets you a broadcast you cannot afford.

Those cardinalities are converted to cost by a model that prices CPU, local I/O and network transfer. hive.cbo.costmodel.extended selects a richer model that accounts for the execution engine's characteristics rather than a simple row-count proxy; the practical effect is better discrimination between plans whose row counts are similar but whose shuffle volumes are not. Ancillary factors round out the estimate: hive.stats.deserialization.factor relates on-disk bytes to in-memory row footprint, which matters because the map-join decision is made against memory, not disk, and hive.stats.join.factor applies a correction to join output sizes.

One optimisation short-circuits the whole pipeline. With hive.compute.query.using.stats enabled, a query that asks only for count(*), or a min or max over a column with statistics, is answered from the metastore without launching a job at all. It is exact only when statistics are current, which is a good reminder that a stale row count is not merely a planning risk — it can become a wrong answer.

Map-join conversion: two thresholds and an in-memory estimate

The single highest-impact decision the optimizer makes is whether a join can avoid the shuffle entirely by broadcasting one side into every task's memory. Getting this right turns a network-bound job into a scan; getting it wrong turns a fast query into an out-of-memory failure.

Conversion is gated by hive.auto.convert.join, and the threshold that matters on Tez is hive.auto.convert.join.noconditionaltask.size, used when hive.auto.convert.join.noconditionaltask is enabled. The near-universal misreading is to treat that number as a file size. It is not: it bounds the estimated in-memory footprint of the hash table built from the small side. Highly compressed columnar data expands substantially once decoded into objects, so a 20MB Parquet file can easily exceed a 25MB threshold in memory, and conversely a wide table filtered down to three columns occupies far less than its file size suggests. The default is conservative, and raising it is one of the few Hive settings where a considered increase reliably pays off — provided the task heap has room, because this budget is spent inside the same container that runs everything else.

The older hive.mapjoin.smalltable.filesize governs the conditional-task path, where Hive generates a plan containing both a map-join and a shuffle-join branch and resolves between them at runtime based on the observed input size. That mechanism is a safety net for exactly the case where the estimate is untrustworthy, and it is why some plans contain branches that never execute.

Crucially, the size the optimizer compares against the threshold is the post-filter estimated size, not the table's size on disk. A dimension table of 50 million rows filtered to one country is a broadcast candidate if and only if the optimizer believes the filter is selective — which it only does with fresh column statistics on the filtered column. This is the mechanism behind the most common Hive regression: the table grew or its distribution shifted, nobody re-analyzed, the stale estimate keeps the broadcast decision alive, and the build side no longer fits. Related decisions — bucket map joins and sort-merge bucket joins, enabled by hive.optimize.bucketmapjoin and hive.auto.convert.sortmerge.join — depend on physical layout rather than on estimates; see Hive bucketing for how that co-location is established.

Reading EXPLAIN CBO and its costed variants

Hive exposes the plan at several altitudes, and choosing the wrong one wastes a lot of time. EXPLAIN CBO shows the Calcite relational-algebra tree the optimizer settled on — HiveJoin, HiveFilter, HiveProject, HiveAggregate, HiveTableScan nodes with their join conditions and algorithms. This is the right view for one specific question: what order did it pick, and which joins are broadcast? Plain EXPLAIN, by contrast, shows the physical Tez plan — vertices, edges and operator trees — which answers how the work is distributed but buries the join order under mechanical detail.

The costed variant is what turns plan reading from guesswork into diagnosis. EXPLAIN CBO COST annotates each node with the estimated row count and the cost the optimizer assigned it, so the join order and the arithmetic that justified it appear together. Reading a costed tree bottom-up, the diagnostic move is always the same: find the lowest node whose estimated row count you know to be wrong. Everything above it is downstream of that error, so there is no point analysing the join order until the leaf estimates are credible.

HiveAggregate(group=[{0}], agg#0=[sum($1)])
  HiveProject(region=[$3], amount=[$1])
    HiveJoin(condition=[=($0, $2)], joinType=[inner], algorithm=[MapJoin],
             cost=[{3.1E6 rows, ...}])
      HiveTableScan(table=[[db, orders]], rowcount=[1.7E8])
      HiveProject(cust_id=[$0], region=[$4])
        HiveFilter(condition=[=($2, 'NL')], rowcount=[9.0E5])
          HiveTableScan(table=[[db, customers]], rowcount=[5.0E7])

Two supporting forms are worth knowing. EXPLAIN EXTENDED adds the file and partition lists actually selected, which is how you confirm partition pruning really happened rather than assuming it. EXPLAIN FORMATTED emits JSON, which is what you want when diffing plans programmatically across a Hive upgrade — optimizers change their minds between versions, and a plan-diff over a benchmark suite catches regressions before users do. hive.explain.user toggles the simplified user-facing rendering; turn it off when you need the full operator detail.

Where cardinality estimation goes wrong

Cardinality estimation rests on assumptions that data routinely violates, and the failures are systematic rather than random. Knowing the four common shapes lets you predict where a plan will be wrong before you measure it.

Correlated predicates. The model assumes predicates are independent, so two filters each passing 10% are assumed to pass 1% together. When the columns are correlated — city = 'Amsterdam' AND country = 'NL', or a status column that only takes one value for a given product type — the true combined selectivity is close to the more selective predicate alone, and the estimate is low by an order of magnitude. Underestimates are the dangerous direction: they make the optimizer think an input is small enough to broadcast.

Uniformity versus skew. The 1/NDV selectivity rule assumes distinct values occur equally often. Real key distributions are heavy-tailed, so a filter on a rare value is overestimated and a filter on a hot value is badly underestimated. The same assumption makes join output estimates optimistic on skewed keys, which is why a plan can look perfectly reasonable and still produce a single reducer holding forty percent of the rows — see Hive skew join optimization for what to do about that once it happens.

Opaque expressions. Statistics describe columns, not expressions. A predicate on upper(country), a date arithmetic expression, a CASE expression, or any user-defined function is something the optimizer has no distribution for, so it falls back to a fixed default selectivity that is unrelated to reality. Rewriting a filter so the predicate lands on the bare column — the same rewrite that lets predicate pushdown reach the scan — restores estimation as well as pushdown.

Joins on non-key columns and post-filter NDV. The containment assumption behaves well for foreign-key joins and poorly for joins on attributes with no referential relationship. Related to this, when a filter reduces a table, the optimizer must guess what happened to that table's other columns' NDVs. The statistics describe the whole table, but the join downstream sees only the surviving rows, and distinct values do not shrink in proportion to row count — filtering a customer table to one country removes 98% of the rows but leaves the customer-id NDV essentially equal to the surviving row count, while barely touching the NDV of a low-cardinality status column. Whatever heuristic bridges that gap is an approximation applied at every filter in the tree. Deep in a multi-join tree, the compounding of these guesses is why the estimate at the top can be off by several orders of magnitude even when every leaf statistic is fresh.

EXPLAIN ANALYZE and the runtime corrections

Since estimation is imperfect by construction, the useful question is how quickly you can find out. EXPLAIN ANALYZE answers it directly: it executes the query and returns the plan annotated with both the estimated and the actual row count at each operator. That side-by-side comparison converts plan debugging from inference into measurement, and it is the single most valuable command in this whole area. A large divergence at one operator names the statistic to refresh; agreement everywhere means the plan is a fair reading of the data and the problem lies in execution.

EXPLAIN ANALYZE
SELECT c.region, sum(o.amount)
  FROM orders o JOIN customers c ON o.cust_id = c.cust_id
 WHERE c.country = 'NL'
 GROUP BY c.region;

Hive can also correct itself after the fact. With hive.query.reexecution.enabled, a query whose execution fails can be recompiled and retried using row counts observed during the failed attempt rather than the metastore's stale numbers, so the second compilation prices the plan against what the data turned out to be; hive.query.reexecution.strategies selects which mechanisms participate. This is genuinely useful for the canonical failure — a map-join whose build side did not fit — but note what it is: a recovery path triggered by a failure, not a substitute for statistics. You pay for the failed attempt every time, and a query that is merely slow rather than fatal never triggers it at all.

Complementing that, two runtime mechanisms narrow what the plan committed to. Dynamic partition pruning, enabled by hive.tez.dynamic.partition.pruning, lets the actual join keys produced by a dimension scan restrict which fact-table partitions are read — a pruning decision the compiler could not have made because the key values were unknown at compile time. Dynamic semi-join reduction, enabled by hive.tez.dynamic.semijoin.reduction, builds a bloom filter from one side's join keys and applies it to the other side's scan, discarding rows that cannot possibly join before they are shuffled.

It is worth being clear about the boundary. Both of these narrow the data flowing through a plan whose shape was fixed at compile time; neither re-orders joins mid-flight. That contrast is the main structural difference from Spark, whose adaptive execution re-plans between stages using measured statistics — see Spark SQL optimizer architecture for how that changes the trade-off.

Intervening when the optimizer is wrong

When a plan is wrong, the remedies form a strict preference order, and skipping down the list is the most common self-inflicted wound in Hive operations.

First, fix the statistics. Run ANALYZE TABLE ... COMPUTE STATISTICS FOR COLUMNS on the tables whose estimates diverged, then re-read the plan. This is the only remedy that improves every query against those tables rather than the one in front of you, and it is correct more often than any hint. Make it structural: the pipeline that writes a table should own analyzing it, so statistics are refreshed in the same job that changed the data and fail visibly when they do not.

Second, adjust thresholds — deliberately, at the right scope. Raising hive.auto.convert.join.noconditionaltask.size to match a task heap that has grown is a legitimate cluster-level change. Setting it at session scope for one report is a workaround with a memory footprint; setting it globally to a number nobody has reconciled against the container size is how clusters acquire mysterious intermittent out-of-memory failures.

Third, hint — and record why. A MAPJOIN hint forces the broadcast the optimizer declined; STREAMTABLE tells it which relation to stream rather than buffer. Hints are appropriate when a report is due in ten minutes and the analyze job takes an hour. They are also permanent in the worst way: a hint pins one query's decision to the data as it looked on the day it was written, and it goes on being obeyed long after the table it describes has changed shape. Every hint should carry a comment naming the estimate it is overriding, so a future reader can test whether it is still true.

Last, consider whether the query should run at all. If the same expensive join and aggregation runs hundreds of times a day, the leveraged fix is not a better plan but a smaller input — a materialized view that the optimizer rewrites queries onto automatically, governed by hive.materializedview.rewriting. That mechanism is covered in Hive materialized views, and it is the case where optimizer work compounds instead of accumulating as per-query debt.