Scenarios

Multiply row counts along the plan and find where they peak. The usual cause is a join or an unnest that runs before a filter or an aggregation.

Filters should sit below joins, and aggregates should be reduced to the join key's grain before joining: a join that fans out rows is a granularity mismatch.

How many distinct sort orders and shuffles does the plan add?

Reduce rows early, then join; verify with EXPLAIN ANALYZE.

The optimizer's estimates may be wrong because correlated predicates were assumed independent.

Broadcast moves 500 MB × 200 = 100 GB; shuffling moves about 400 GB. The size ratio, 800, exceeds P = 200, so the network favours broadcast.

Broadcast means 100 GB of copies across the cluster and 200 hash-table builds, and 500 MB is far above typical broadcast thresholds of tens of MB.

Broadcasting keeps the orders table's existing partitioning, which can save a later shuffle.

Broadcast if each executor has room in its 16 GB and no driver becomes a bottleneck; otherwise shuffle with a runtime filter.

A central driver collecting the whole table before broadcasting it.

Check the frequency distribution of join keys, and the fraction of NULL keys. With Zipf-distributed keys, the top key carries about 7% of rows, which caps the speedup near 14× however many workers you add.

1. Filter NULL keys out before the shuffle: free.
2. Split the hot key: let the engine split skewed partitions at run time, or salt it with s copies of the other side if the hot keys are stable.
3. Use sampled range partitioning.

If the hot key is hot on both sides, the output itself is huge; pre-aggregate one side.

A NULL entered the subquery's results. Under three-valued logic, x NOT IN (…, NULL) is UNKNOWN for every row, so nothing passes.

Rewrite with NOT EXISTS, or filter NULLs out of the subquery.

Add a data-quality check on the column, and consider a lint rule against NOT IN on nullable columns. While there, also check for right-side conditions in WHERE on LEFT JOINs, and joins that fan out rows before a SUM.

A hash join: the condition is a range, not an equality.

If it is "the latest row at or before t", use an as-of join operator.

A band merge join, or range partitioning with replication. Estimate the replication factor (average interval length ÷ bucket width), and check the distribution of interval lengths for a heavy tail. A merge needs memory only for the longest run of duplicates.

There is no simple anchor for the output size, so test it on a sample.

A hybrid hash join spills only (1.1M − M) / 1.1M ≈ 9%: a small penalty.

A Grace hash join splits it into 3 TB ÷ 16 GB ≈ 188 partitions. The possible fan-out is 16 GB ÷ 8 MB = 2,048, so one level of partitioning is enough (the threshold for a second is about 32 TB), at the cost of 2 × (|R| + |S|) extra I/O.

Neither case is a cliff. A single huge key would be, since it cannot be partitioned. A sort-merge join is the constant-memory alternative.

About (row size ÷ roughly 1.2 bytes per key in the filter) × (1 − the fraction of rows that match). For 200-byte rows with 5% matching: about 167 × 0.95 ≈ 158×. For 8-byte rows with 90% matching: about 7 × 0.1 = 0.7, so no gain.

A barrier: the probe side must wait for the build side to finish the filter.

A min/max range when the data is clustered, an IN-list for a few keys, a Bloom filter otherwise.

Each distinct (PARTITION BY, ORDER BY) specification is a full sort.

Reorder or unify the specifications so they share prefixes: one sort on (a, b) also provides (a). An equality predicate makes a column constant, so it drops out of the required order. Where input is already sorted on (a) and the requirement is (a, b), a segmented sort needs memory only for the largest group, and streams.

A window that only needs each partition's rows together does not need a global order.

Default RANGE frames with ties cost more than ROWS. Sliding MAX or MIN costs O(log w) per row, and holistic functions over a frame cost O(w).

A clustered or bucketed layout pays for the sort once, for every query, at write time, at the cost of constraining other queries on that table. And whenever specifications change, verify the results, since semantics may change too.

The function is holistic, so partial aggregation cannot shrink anything, and the group key is skewed. If the home page has 10% of events, that is 5 billion rows; at 8 bytes per user ID, 40 GB sent to one executor with 16 GB.

HyperLogLog: 12 KB per page, mergeable, one shuffle, immune to skew, and its merges are exactly reproducible.

Rewrite as SELECT page, COUNT(*) FROM (SELECT DISTINCT page, user_id …) GROUP BY page. The first shuffle is by the (page, user) pair, so the hot page spreads over all executors. It costs two shuffles but runs in parallel. Plain random salting would double-count users.

Several distinct aggregates in one query mean several copies of the data on the wire.

Sketches for dashboards, where the error has no meter and is paid silently by the reader; the exact two-shuffle plan for billing numbers.

Each executor sees 10 million rows. With 40 million customers, it can expect about 40M × (1 − e^(−0.25)) ≈ 8.85 million distinct customers, so partial aggregation shrinks its data only about 1.13×. It pays a hash and a memory write per row, plus an 8.85-million-entry table, for a 12% smaller shuffle.

Turn partial aggregation off for this query, or enable an adaptive mode that samples the first tens of thousands of rows and abandons it when few rows hit existing groups. A plan-time estimate of distinct values is the least reliable statistic, so prefer sampling at run time.

For GROUP BY country, the ratio is in the tens of thousands, and partial aggregation is a huge win.

200 GB per node. With 32 GB of memory for the sort and 8 MB buffers, M² / B = 32 GB × 4,000 = 128 TB, so one merge pass suffices, and I/O is about 4N = 800 GB per node. At 3 GB/s that is under 5 minutes, so the node is likely CPU-bound if keys are strings with collation: use normalized keys and sort compact (prefix, pointer) pairs.

Range-partition on boundaries chosen from a random sample (a few hundred samples per partition), sort locally, and concatenate.

• Biased samples from time-ordered files: sample at random, never the first rows.
• A single key with more rows than one partition should hold defeats any boundaries.
• Sampling is a barrier before any rows can be routed.
• On object storage, larger buffers bring the third pass closer.

The filter has no false negatives but does have false positives. In deduplication, a false positive says "seen before" about a new event, so a real row is dropped, and nothing downstream checks. At 10 billion rows, 0.1% is 10 million lost rows, invisible to monitoring.

Use the filter only as a negative oracle: "definitely new" emits immediately, "maybe seen" consults an exact store or an exact sort- or hash-based dedup. Or deduplicate exactly on a 128-bit digest (with 64 bits, a collision is about 93% likely at 10¹⁰ rows), using a sort-based dedup if the final stage does not fit in memory.

Best-effort telemetry, consumers that are idempotent anyway, and bounded-window filters that reset.

Reconcile row counts against the source after every run.

A merge join needs order on the key, sort-based aggregation needs order (or grouping) on the key, and ORDER BY needs order on the key: three consumers of one order.

Sort once, merge-join, aggregate in one streaming pass with constant memory, and the output is already ordered. A hash plan would build one hash table for the join, another for the aggregation, then sort for ORDER BY. With only one consumer of the order, hashing would win.

If the table is clustered or bucketed on the key, the sort is free.

A hot key breaks range partitioning and hashing alike, and the sort needs N ≤ M² / B per node to finish in one merge pass.

Floating-point results depend on the order of additions. Use decimal or fixed-point types for money, compensated (Kahan or Neumaier) summation, or sort before summing.

Choose the sketch by how it merges: t-digest merges are lossy and depend on the merge tree, so they are not reproducible; KLL has a bounded error; exact quantiles come from sorting by (group, value) and reading positions.

Exactness costs a sort; a sketch costs an error with no meter. t-digest is good for accuracy at p99 and p99.9, but not for reproducibility.

The output is 40 million rows. Partial aggregation reduces data only about 1.13×, so the shuffle moves nearly all 2 billion rows × about 16 bytes (key and value) ≈ 32 GB. Each executor then owns 200,000 customers, which fits easily in memory (about 5 MB).

Make the shuffle a range partitioning on customer_id, and aggregate with a sort on each executor: the output is grouped and ordered at once, so ORDER BY costs nothing extra.

customer_id = 0 is 140 million rows. SUM is distributive, so wherever partial aggregation applies it collapses that key to one row per executor and the skew is harmless; an adaptive engine keeps partial aggregation on when hot keys give a high hit rate.

The 2% of NULL keys form one more group; decide whether the report should include it.