Summary and key takeaways
What every operator is, in one question. For each operator ask: what does it require of its input (order, partitioning, co-location), and what does it provide? An enforcer (a sort, a shuffle, a hash build) is inserted wherever the two do not match. The cost of a plan is almost entirely its enforcers.
Seven articles, one sentence each:
- A query is a specification. Plans differ by enforcers. Some operators are blocking (they must see the last input row before emitting the first output row), memory is the scarce resource, and hash versus sort is the core trade.
- Each relational operator forces something on the engine, and silent wrong answers come from bag semantics, three-valued logic (
NOT INwith NULL), outer-join rewrites, fan-out before an aggregate, and opaque UDFs. - A join needs an "index" that it must build: nested loop, indexed, hash, Grace/hybrid, sort-merge, broadcast, shuffle, co-partitioned. Skew (customer 0, NULL keys) makes one partition the whole runtime. Two-sided skew has no escape.
- Aggregation output is bounded by the number of groups, so it can be reduced before data moves. The ratio N/(P·G) and the function's algebra (distributive, algebraic, holistic) decide everything.
- Sorting computes nothing and supplies a property. Its cost is the only one you can predict (4N I/O, two passes while N ≤ M²/B). The best sort is the one you avoid, by counting how many requirements one order satisfies.
- Dedup is aggregation with an empty accumulator, and
INTERSECT/EXCEPTare semi- and anti-joins. Removing duplicates and counting them are different problems. - This mechanism has two real walls (you cannot know a stream's future; you cannot reorder around NULL-generating semantics), and everything else is a purchase with a price.
Formulas to remember:
| Formula | Meaning | Seen in |
|---|---|---|
| N / (P·G) | rows per group per partition: the payoff of partial aggregation (above about 10 it wins; near 1 it is a tax) | Aggregation algorithms |
| Shuffle volume = min(N, P·G) | what partial aggregation sends over the network | Aggregation algorithms |
| HLL error = 1.04/√m | 12 KB gives 0.81%; 0.1% costs about 68x more memory | Aggregation algorithms |
| Passes = 1 + ⌈log_F(N/M)⌉, F = M/B | external sort passes; one merge pass while N ≤ M²/B | Sorting |
| External sort I/O = 4N | read, write runs, read, write output | Sorting |
| Sample size ≈ P/ε² | splitters for a range-partitioned sort; quadratic in the tolerance | Sorting |
| Collision ≈ 1 − exp(−n²/2^(b+1)) | digest width for dedup: 64 bits gives about 93% at 10¹⁰ rows; use 128 bits | Set and dedup operations |
| Load factor cost ~ 1/(1−ρ) | hash table and queue cost near full | joins and aggregation |
Compressions worth carrying out of this mechanism:
- Blocking is ignorance about completeness. Sorted input, hash partitioning or a watermark each tell the engine when a group is complete. Blocking is also two separate properties: state proportional to input, and inability to emit early.
- The function's algebra determines the engineering. Combiner or not, one shuffle or two, spill partials or raw rows, skew harmless or fatal, incremental maintenance or not. Invertibility separately decides sliding-frame cost and deletions.
- Approximation is safe or unsafe by error direction, not size. A Bloom filter before an exact join costs work. In dedup the same filter loses rows silently.
- Hash and sort are not symmetric. Hash co-locates cheaply and leaves nothing. Sort costs a log factor and leaves an order that composes. Decide by amortization count (how many consumers use the order, including future queries), never per operator.
- Concentration (skew) is expensive for operators that process rows (joins, sorts) and cheap for operators that reduce them (aggregation).
- Costs are conserved, not removed. Broadcast moves a shuffle into memory; bucketing moves it to write time; sketches move cost into error, a currency with no meter.
Patterns that repeat:
- Granularity mismatch creates amplification (fan-out before an aggregate; a 200-byte row against a 16-byte sort key).
- Fixed cost per operation sets a batch size (merge fan-in, flush thresholds, vectorized UDFs).
- A distributed job runs at the speed of its slowest partition.
- Uniformity and independence are assumed and often false (hash partitioning, conjunctive selectivity).
- Randomization replaces coordination (sampled splitters, runtime sampling).
- A bound on what must be answered is a bound on what must be remembered (Top-N, keyset pagination, heavy hitters, sketches). This remains a hypothesis.
What is forbidden and what escape costs:
| Forbidden | Escape | Price |
|---|---|---|
| Output before a blocking operator sees its last input | none; change the question | the answer is not the one asked |
| Knowing data not yet scanned | runtime filters, sampled adaptivity | a barrier and wasted work |
| Parallelism within one key | replicate the other side s ways | s times work; no escape for two-sided skew |
| Exact holistic aggregates in bounded state | sketch, or sort by (group, value) | unmetered error, or a sort |
| Reordering across outer joins and 3VL | none | plans that cannot be expressed |
| Reasoning about a UDF | make it transparent or declared | optimizations stop at its boundary |
| Bounding memory for a final dedup | sort-based dedup | a sort |
Omar's lessons (the story in one list):
- Same answer, different plan: ask what each operator requires and provides, and where the enforcers are.
- Wrong answers with no error: NULLs, outer joins, fan-out, UDFs.
- 199 idle executors: skew (customer 0, NULL keys); split the hot key, filter NULLs, use an as-of join for "price at the time".
COUNT(DISTINCT)kills the cluster: holistic aggregates cannot be reduced early; use the two-shuffle plan or a sketch. Partial aggregation near ratio 1 is a tax.- Five sorts: count requirements, share prefixes, use segmented sorts, and use keyset pagination and heaps instead of offsets and full sorts.
- Lost rows: a filter whose false positive skips a check loses data silently; use 128-bit digests and exact dedup.
- The decision procedure Tarini pinned above his monitor: require/provide, order of cheap operators, NULLs, grain, skew, aggregate class, rows per group per partition, order versus grouping, error direction, UDFs, reproducibility.
Last note. Every decision procedure here takes NDV, key frequency and selectivity as inputs, and those are the quantities an optimizer must estimate instead of measure. The arithmetic here is defensible. Applying it to estimated numbers that are often wrong is the optimizer's real problem.