Guessing row counts, picking a plan

Parts 3–5 gave us several ways to do every job: three join algorithms, scan or index, hash or sort aggregation. Which one is right depends on how many rows each operator will see, and the planner has to decide before running anything. This part covers how it estimates those counts, how it prices a plan, and how it searches for the cheapest join order.

Here is a three-table query over the generated dataset:

SELECT c.city, sum(o.qty * p.price)
FROM customers c, orders o, products p
WHERE o.customer_id = c.id AND o.product_id = p.id AND c.city = 'Seattle'
GROUP BY c.city

The SQL says nothing about how to run it. We could join customers to orders first, or orders to products first. Each join could be a hash join (which side builds?), a nested loop, or an index nested-loop join, since orders.customer_id and products.id are indexed. The Seattle filter could narrow customers before the join or be applied after. With three tables there are a dozen reasonable plans. With eight tables there are millions. Their costs can differ by orders of magnitude.

A cost-based planner has three parts:

1. a cardinality estimator that guesses how many rows each operator outputs,
2. a cost model that turns those row counts into a price for each operator,
3. a search over plans that finds the cheapest one without pricing them all.

The search is the elegant part and the cost model is the crude part. Almost all of the trouble comes from the estimator.

ANALYZE: what we know about a column

The estimator can't look at the data while planning. That would mean running the query to plan the query. Instead, an ANALYZE pass reads each table once, ahead of time, and keeps a small summary per column:

function analyze(table)
  for each column c
    values  = column c of every row
    nonNull = values without nulls, sorted
    stats[c] = {
      rows:  len(values),
      nulls: len(values) - len(nonNull),
      ndv:   count distinct nonNull,             -- number of distinct values
      min:   first(nonNull),  max: last(nonNull),
      hist:  equiDepth(nonNull, 8)                -- see below
    }

Real databases sample instead of reading everything. Postgres reads 30,000 rows by default and estimates ndv from the sample, which is a hard problem on its own. Our tables are tiny, so the lesson reads them all.

With only rows and ndv, the natural guess for orders.customer_id = 7 is "every customer has the same number of orders": 5000 / 200 = 25. But the generator is skewed: half of all orders go to customers 1–20. Customer 7 actually has 126 orders, five times the estimate. Customer 150 has 11. A single ndv can't describe a skewed column. A histogram can.

Histograms

A histogram splits the value range into buckets and records how many rows fall in each. Inside a bucket, the estimator assumes values are spread evenly. There are two ways to pick the bucket boundaries:

• Equi-width: equal value ranges (1–25, 26–50, …). Simple, but under skew most rows pile into a few buckets, and within those buckets the evenness assumption is badly wrong.
• Equi-depth: equal row counts. Each bucket holds about 1/8 of the rows, so bucket boundaries crowd together where the data is dense. Hot values end up in narrow buckets, where the evenness assumption is nearly right. Postgres and most others use this, plus a separate list of the most common values.

function equiDepth(sorted, k)
  buckets = []
  for b in 0 .. k-1
    end = index of the ((b+1)/k)-th row, extended so one value never spans two buckets
    buckets.append({ hi: sorted[end], count: rows in this bucket, ndv: distinct in this bucket })
  return buckets

function estimateLessOrEqual(hist, x)              -- col <= x
  sum of count for buckets entirely <= x
  + count * (x - bucket.lo + 1) / (bucket.hi - bucket.lo + 1)   -- linear slice of x's bucket

function estimateEquals(hist, x)                   -- col = x
  b = bucket containing x
  return b.count / b.ndv                           -- values in a bucket are equally common

Slide the predicate around below. The grey bars are the real per-value counts. The teal blocks are what the estimator thinks the distribution looks like: one flat block per bucket. Try orders.customer_id = x with no histogram, then equi-width, then equi-depth, at x = 7 and x = 150. Then try orders.day, which is uniform: every method does well there.

The equi-depth histogram spends four of its eight buckets on customers 1–19, because that's where half the rows are. Its estimate for customer 7 is 143 against an actual 126. The equi-width histogram puts customers 1–25 in a single bucket, mixing the hot range with the start of the cold one, so it's off in both directions.

Selectivity: combining predicates

A selectivity is the fraction of rows a predicate keeps. Estimated output = input rows × selectivity. The lesson code uses these rules, which are close to what every textbook planner starts with:

The AND rule is the independence assumption: knowing one predicate is true tells you nothing about the other. In our data it holds, because city and age were generated independently. In real data it often doesn't. city = 'Austin' AND state = 'TX' has selectivity s(city), not s(city) × s(state), so the product underestimates by a factor of 1/s(state), maybe 50×. Correlated predicates are the main reason estimates go wrong, and the reason Postgres added CREATE STATISTICS for column groups.

Join cardinality

For an equi-join L.a = R.b, the standard estimate is:

The reasoning: assume that every value on the side with fewer distinct values appears on the other side too (the containment assumption), and that values are evenly spread. Then each of the |L| rows matches |R| / ndv(R.b) rows on average, and the max picks the side whose values are the subset. For a foreign-key join like orders.customer_id = customers.id, this gives exactly |orders|, which is right.

Now add the filter. Seattle is 1 of 8 cities, so the estimator guesses 200/8 = 25 Seattle customers. The generator makes Seattle rare, and there are only 10. The join estimate is 5000 · 25 / 200 = 625 orders, and the actual count is 622. The estimate is nearly perfect, by accident: the customer count is 2.5× too high, but those 10 customers include some of the hot ones from 1–20, so they have more orders than average. The two errors cancel. In general they don't, and they multiply instead.

The cost model

Costs are in made-up units, roughly "rows touched". The only thing that matters is that they rank plans in the right order:

cost(Scan t)               = rows(t)
cost(Filter)               = rows(input)
cost(HashJoin)             = 2 · rows(build) + rows(probe)       -- build side = right input
cost(NLJoin)               = rows(outer) · rows(inner)
cost(IndexNLJoin)          = rows(outer) · (log2(rows(inner)) + matches / rows(outer))
cost(Sort)                 = n · log2(n)
cost(HashAggregate)        = rows(input)
total(plan)                = sum over every operator

Real cost models separate CPU from I/O, sequential from random reads, and account for memory (a hash join that spills costs far more). They are still rough guesses, and Leis et al. found that once the cardinalities are right, even a very simple cost model picks good plans. The cardinalities are what matter.

Searching: dynamic programming over subsets

With n tables there are n! left-deep join orders (each join's right input is a base table), and far more if you allow bushy trees. Pricing each one separately doesn't scale. In 1979, System R observed that the problem has optimal substructure: the best plan joining {c, o, p} must contain the best plan for whichever two-table subset it joins first. So compute the best plan for every subset, smallest first, and reuse those results:

function bestPlan(relations, predicates)
  best = {}
  for each r in relations                        -- level 1: access paths
    best[{r}] = cheapest of (Scan r + filters, IndexScan r if an index fits a filter)

  for size in 2 .. n
    for each subset S of relations with |S| = size
      for each r in S                            -- r joins last (left-deep)
        left = best[S - {r}]
        if left is null then continue
        if no predicate connects S - {r} and r then continue   -- skip cross products
        for alg in [HashJoin, NLJoin, IndexNLJoin if r has an index on its join column]
          candidate = Join(alg, left, accessPath(r))
          if cost(candidate) < cost(best[S]) then best[S] = candidate
  return best[relations]

There are 2ⁿ subsets with up to n choices each, which is exponential but far better than n!. The usual practical limit is about 10–12 tables; past that, planners switch to greedy or genetic search. (Postgres's geqo_threshold defaults to 12.) Step through it below, then switch scenarios: the filters change the estimates, and the estimates change which plan wins.

Some things to notice:

• {c, p} is never built. No predicate connects customers and products, so joining them first would produce a 10,000-row cross product. Selinger's heuristic refuses to consider it. That's usually right, though not always: a cross product of two tiny filtered tables can be the best first step.
• The algorithm changes with the estimate. With a small filtered outer side, an index nested-loop join beats building a hash table. That's the trade-off from part 5, now decided automatically.
• Compare est and actual for p.category = 'games'. The estimator assumes 50/5 = 10 games products; the generator produced 4. Everything above that filter is 2.5× too big. Here the mistake is harmless, because the plan is the same either way. When a 2.5× error flips a hash join to a nested loop over a million rows, it isn't.

EXPLAIN: estimates next to reality

The integrated engine prints its chosen physical plan with an estimated row count on every node. The lesson demo runs the plan and prints the actual count next to it, like Postgres's EXPLAIN ANALYZE. The last step of the figure above shows the same thing for the chosen plan. Reading these side by side is the most useful debugging habit for slow queries: find the lowest node where estimate and actual diverge, because every operator above it was chosen on bad information.

Errors compound. If each estimate is off by 2× and a query has five joins, the top of the plan can be off by 2⁵ = 32×, in either direction. Leis et al. measured this on a real dataset (IMDB) with the planners of several major databases. The errors routinely reached multiple orders of magnitude, they grew exponentially with the number of joins, and they caused nearly all the bad plans. Planners are only as good as their row counts.

cd go && go test ./exercises/lesson06/
cd clojure && clojure -M:ex -d ex -n exercises.lesson06.planner-test