Data Processing Algorithms Mechanics & Internals

The reasoning

Hash aggregation reads each row once, hashes the key, and updates an accumulator. Its memory is the number of *distinct keys*, not the number of rows — five hundred thousand rows over fifty groups needs a table of fifty. That is why it is the default: one pass, and usually a small table.

Sort aggregation orders the rows by the grouping key and then sweeps through, emitting a group whenever the key changes. Running state is a single group, so its memory is tiny — but the sort is `n log n` over everything, and the sort is the thing that spills.

The engine chooses from an estimated distinct count. Low cardinality means hash; very high cardinality, or an input that is already sorted on the key, or an `ORDER BY` on the same key that would need a sort anyway, tips it toward sort. When the estimate is wrong and the hash table does not fit, the engine spills partitions to disk and finishes them separately — which is the same partition-then-conquer idea, applied under duress.

Two consequences worth stating. First, a `GROUP BY` on a near-unique key is the worst case for hashing, because the table approaches the size of the input — grouping by an id column is how an aggregation OOMs. Second, hash aggregation returns groups in no meaningful order, and code that relies on the order it happens to produce breaks when the engine spills and changes it.

The reasoning

Pass one: read as much as fits, sort it in memory, write it out as a sorted run. Repeat. Two terabytes with sixteen gigabytes gives about a hundred and twenty runs. Pass two onward: open `fan_in` runs at once, read their heads into a priority queue, and repeatedly emit the smallest — merging them into longer runs until one remains.

The cost is passes, and passes are `1 + ceil(log_fanin(runs))`. The step worth being explicit about is what a pass costs: it reads every byte and writes every byte, so one pass over two terabytes moves four — two in, two out. Three passes is therefore twelve terabytes of I/O to sort two. That is why the pass count, and not the row count or the CPU, is the number you are optimising.

Which is why memory behaves as a step function. Sixteen gigabytes and four gigabytes both give three passes; sixty-four gigabytes gives two. Between the steps, extra memory buys nothing at all — it makes each run bigger without changing how many merge rounds those runs need. The lever that usually matters more is fan-in, because raising it lowers the log's base.

The practical notes: fan-in is bounded by how much buffer each open run needs, so a very high fan-in means tiny reads and the disk stops being sequential — there is an optimum, not a maximum. And this is exactly what a database means by 'spilling': the hash table that did not fit becomes runs on disk, and the recovery is a merge.

The reasoning

When the table will not fit, the engine partitions the input by a hash of the key — the same key it was grouping on — so that all rows for a given key land in one partition. Each partition is written to disk. Then each is read back and aggregated on its own, because a partition is small enough to fit. The result is identical; the cost is a full write and read of the data, plus the risk of a partition that is itself too big.

That last risk is where spilling turns into a real problem: if one key is enormous, its partition does not shrink no matter how many partitions you make, and the engine recurses or fails. Skew and spilling are the same problem seen from two angles.

It varies run to run because three things vary: how much data actually arrived, how much memory the executor got given other concurrent work, and how good the optimizer's cardinality estimate was. A query at the edge of its budget spills on a busy day and not on a quiet one, which is why 'it worked yesterday' is not evidence.

What to do about it, in order: reduce what is grouped — filter and project before aggregating, so fewer and narrower rows reach the table; check the estimate, since a bad cardinality guess means the engine sized for the wrong thing; give it more memory if the shortfall is small; and if one key dominates, treat it as skew rather than as a memory problem, because more memory will not fix a single enormous group.

The reasoning

**Nested loop** is `O(n·m)` and is chosen when one side is very small or when the join condition is not an equality — a range or inequality join has nothing to hash on, so this is often the only option. With an index on the inner side it becomes an index nested loop, which is the right plan for a small number of outer rows.

**Hash join** builds an in-memory table from the smaller (build) side, then streams the larger (probe) side through it. One pass over each, and the memory is the build side. It is the default for equi-joins and the reason engines care so much about which side is smaller — getting that backwards is a common cause of a job that spills.

**Sort-merge** sorts both inputs on the join key and walks them together. It costs two sorts but needs almost no memory during the merge, it handles inputs that are already sorted for free, and it degrades gracefully to disk. It is the fallback when the build side will not fit and the choice when both sides are enormous.

The distributed layer adds a second decision on top: broadcast versus shuffle. A broadcast hash join ships the small side to every node and avoids moving the large side at all; a shuffle hash or sort-merge join repartitions both sides by the key. Which one is chosen comes from an estimate of the small side's size, and the classic failure is a broadcast of something the optimizer thought was small — which is an out-of-memory error on the driver rather than a slow query.

The reasoning

The engine builds the hash table for the build side and, while doing so, inserts each join key into a small Bloom filter. That filter is then sent to the probe side's scan. Rows whose key the filter rejects cannot possibly join, so they are dropped at the scan — before decompression, before the shuffle, before the hash probe.

The saving is proportional to how selective the join is. Joining a billion-row fact table to a thousand-customer dimension means almost every fact row has no match, and a runtime filter can eliminate most of them at the storage layer. That is a much bigger win than making the join itself faster, because the cheapest row is the one never read.

What it cannot do is be exact. A Bloom filter has false positives, so some non-matching rows survive the filter and are correctly discarded later by the actual join. It has no false negatives, which is the property that makes this safe: a row it rejects is guaranteed not to match, so no result is ever lost. Using a structure with false negatives here would silently drop rows.

The cost side, since interviewers push on it: the filter has to be built before the probe side can be scanned, which serialises two things that might otherwise overlap, and it is wasted work when the join is not selective. Engines therefore apply it adaptively — build the filter, check how selective it turned out to be, and skip it if it is rejecting almost nothing.

The reasoning

The structure: a bit array of `m` bits and `k` hash functions. To add a key, set the `k` bits it hashes to. To test, check them all — if any is zero the key was definitely never added, and if all are one it was *probably* added. The asymmetry is the whole point: no false negatives, some false positives.

The sizing: the false-positive rate is `(1 - e^(-kn/m))^k`, and the optimal `k` for a given `m/n` is `(m/n)·ln2`. Ten bits per key with seven hashes lands near 1%; the table beside this shows measured rates tracking the formula closely at four, eight, twelve and sixteen bits per key. Knowing the shape matters more than the exact constant: the rate falls exponentially in bits per key, so 1% costs about ten bits and 0.1% about fifteen — cheap to improve.

What it buys is that the expensive lookup only happens for keys the filter admits. Checking a hundred million keys against a remote store becomes a bit-array probe that rejects most of them locally, and only the maybes go over the network. The false-positive rate is therefore a *cost* parameter, not a correctness one — a 1% rate means 1% wasted lookups, and the answer is still exact because the real store confirms.

The limits to state before being asked: you cannot delete from a standard Bloom filter (clearing bits would create false negatives) — a counting Bloom filter or a cuckoo filter can. You cannot enumerate what is in it. And the rate degrades as you exceed the `n` you sized for, so a filter built for a hundred million keys and fed two hundred million is far worse than its stated rate, silently.

The reasoning

Row layout stores each record's fields together, so reading one column means reading — and decompressing — every other column with it. Column layout stores each column contiguously, so a query touching two of forty columns reads a twentieth of the bytes. That is projection pushdown, and on wide tables it is usually the largest single win.

Compression is the second effect and often the bigger one. A column is values of one type with real redundancy, so dictionary, run-length and delta encodings work extremely well — where a row-wise page interleaves types and defeats them. Better compression is fewer bytes read, which compounds with reading fewer columns.

The third is execution shape. A contiguous run of same-typed values is what a vectorised engine wants: process a batch of ten thousand values in a tight loop, no per-row interpretation, SIMD where available. Row-at-a-time execution spends most of its time on overhead rather than on arithmetic.

Where it loses: fetching a whole row means touching every column's file and stitching the record back together, so point lookups by primary key are much worse than a row store's. Writes are worse — inserting one row touches every column — which is why columnar formats prefer large batch appends and why OLTP stays row-oriented. And a query selecting `*` on a narrow table gives up the projection advantage entirely.

The reasoning

**Dictionary** replaces each value with an index into a list of distinct values. It is dramatic on low-cardinality strings — a country column becomes a few bits per row — and it is worse than raw when nearly every value is distinct, because you store the values *and* the indexes. The measured table shows exactly that: dictionary on an ascending id column is larger than the raw encoding.

**Run-length** stores value-and-count pairs. It collapses a sorted column almost to nothing, and on an interleaved column with no runs it is worse than raw — every 'run' is length one, so you have doubled the data. The same column, sorted or not, differs by orders of magnitude.

**Delta** stores each value's difference from the previous one, which is small and fixed-width for a monotonically increasing column like a timestamp or an id. On unordered numbers the deltas are as large as the values and it buys nothing.

Which leads to the point that matters operationally: **sort order is an encoding decision.** Sorting a table by a low-cardinality column before writing turns run-length from useless into free, and improves everything downstream of it. And general-purpose compression on top (gzip, zstd) is not an alternative to encodings — it works with them, and it costs CPU on every read, which is the trade a query engine is making when it chooses a lighter codec than the smallest one available.

The reasoning

**Partition pruning** is first and cheapest. If the data is laid out as `region=EU/...`, the planner never lists the other directories. No file is opened. This is why partition columns should be the ones you filter on most, and why partitioning by a high-cardinality column is a mistake — it creates millions of tiny files and the listing itself becomes the bottleneck.

**File and row-group statistics** come next. Columnar formats store min/max per column per row group, so a group whose `region` range excludes 'EU' is skipped without decompressing it. This only works if the data is clustered — if every row group contains every region, every min/max spans everything and nothing is skipped. Sorting on the filter column is what makes zone maps effective, which is the same lesson as the encoding one.

**Scan-level evaluation** applies the predicate during decoding, often against dictionary codes rather than decoded values — so a dictionary-encoded column can be filtered by comparing integers and never materialising the strings. Bloom filters attached to a row group serve the same purpose for equality on high-cardinality columns, and a runtime filter from a join arrives at this level too.

**Engine-level filtering** is the fallback: rows are fully read and then discarded. It is correct and it is the level where you have already paid for everything. So when a query is slow, the diagnostic question is which of the higher levels failed to engage — usually because the predicate is on a column the layout does not support, or is wrapped in a function the engine cannot push through, such as `WHERE upper(region) = 'EU'`.

The reasoning

In classic row-at-a-time execution, each row is pulled through a tree of operators, and every operator does a virtual call and a type check per row. For a simple sum, the interpretation overhead dwarfs the addition — the CPU spends its time deciding what to do rather than doing it.

A vectorised engine passes batches — typically a thousand to ten thousand values of one column — between operators. The inner loop is then a simple typed loop over contiguous memory: no branching per value, good cache locality, and a shape the compiler can auto-vectorise into SIMD instructions that do several values per cycle.

The batch size is a real tuning parameter, and the reason it is not simply 'as big as possible' is cache. A batch should fit comfortably in L1 or L2 so that a chain of operators works on hot data; too large and each operator evicts the next one's working set, too small and the per-batch overhead returns.

The alternative approach worth naming is compilation — generating machine code for the whole query so there are no operator boundaries at all. Both attack the same problem, and both need columnar input to work well. That is the deeper point: columnar storage, compression, and vectorised execution are one design, not three features, and an engine that reads column batches from disk can hand them to the execution layer without transformation.

The reasoning

The map side: each task computes `hash(key) % partitions` for every row, buffers rows per partition, sorts and spills them, and finally writes a single shuffle file plus an index of where each partition starts. That write is to *local disk*, which is the part people miss — a shuffle is not a network operation with a bit of buffering, it is a full materialisation.

The reduce side: reducer `i` must fetch partition `i` from every one of the map tasks. With `m` mappers and `r` reducers that is `m × r` fetches, which is why shuffles scale badly in the number of tasks and why very high partition counts hurt even when each partition is small.

Then the reducer merges what it fetched — typically a sorted merge, which is external merge sort again if the partition does not fit in memory. So one shuffle can involve two spills: one on the map side and one on the reduce side.

The consequences that make this worth knowing. Shuffles are where jobs fail, because they touch disk, network and memory at once. Skew is fatal here specifically: one reducer fetching a partition ten times the size of the others takes ten times as long and the stage waits for it. And the way to make a job faster is almost always to shuffle less — filter and aggregate before the shuffle, use a broadcast join to avoid one entirely, or partition the data on disk so the shuffle is unnecessary.

The reasoning

The mechanics: hash the key, index into an array of buckets, and resolve collisions — open addressing probes nearby slots, chaining follows a linked list. Open addressing is what analytical engines use, because probing adjacent slots is cache-friendly where chasing pointers is not. The load factor is watched, and exceeding it triggers a resize, which rehashes everything.

The first cliff is cache. While the table fits in L2 or L3, a probe is a few nanoseconds. Once it exceeds the last-level cache, most probes become main-memory accesses at roughly a hundred nanoseconds, and since the access pattern is random by construction, prefetching cannot help. The same code gets an order of magnitude slower with no change in complexity — which is why measured performance disagrees with the `O(1)` on the whiteboard.

The second cliff is memory: exceed the budget and the engine spills, converting an in-memory operation into a partitioned, disk-based one.

This is why engines partition proactively. Radix-partitioning the input into pieces whose hash tables each fit in cache turns one cache-missing table into many cache-resident ones, and the extra pass over the data is repaid several times over. It is the same partition-then-conquer structure as spilling and as the bounded-memory dedup — the recurring answer whenever a structure does not fit a level of the hierarchy.

The reasoning

A shuffle join repartitions *both* sides by the join key so matching keys meet on the same node. That moves the billion-row side across the network, which is the dominant cost. A broadcast join instead sends the whole small side to every node; the large side stays where it is and is joined locally. When the small side really is small, this is enormously cheaper — no shuffle at all.

The decision comes from an estimated size compared against a threshold. Estimates come from statistics, and statistics go stale or were never collected, so the engine can believe a table is ten megabytes when it is ten gigabytes. It then collects that side into a single process to build the broadcast — and that is where it fails, with an out-of-memory error that names the driver or coordinator rather than the join.

The signature is worth memorising because it is unambiguous: a job that fails during collection, before any real work, with memory pressure in the coordinator. That is a broadcast of something too large, and the fix is either to refresh statistics, lower the threshold, or hint the plan away from it.

The subtler failure is broadcasting something that is small but joined against a *filtered* large side that would have been cheap to shuffle anyway — you pay to ship the small side to a hundred nodes that each use a fraction of it. And the reverse: a shuffle chosen because the estimate said 'too big', where the small side was actually tiny, gives you a full shuffle of a billion rows for nothing. Both are estimate failures, which is why the practical advice is that keeping statistics current is worth more than any individual hint.

The reasoning

Sort-merge join costs two sorts plus a merge. If the inputs arrive sorted, the sorts are free and only the merge remains: walk both cursors forward, emit matches. Memory is a handful of rows, so it never spills, and it scales to inputs far larger than any hash table could hold.

That is why bucketed or clustered tables exist. Writing both tables partitioned and sorted on `customer_id` means the engine can join them without shuffling *or* sorting — matching buckets are already colocated and ordered. The cost has been paid once at write time and is reused by every query.

The same property helps beyond joins: a `GROUP BY` on the sort key becomes a streaming sweep rather than a hash table, range predicates prune effectively because min/max statistics are tight, and run-length encoding becomes useful. One decision improves four things, which is what makes it worth making deliberately.

The costs to be honest about: maintaining sort order on write is expensive, especially for incremental appends, and you can only be sorted on one thing. Choosing the sort key is choosing which query shape to privilege. And it decays — a table sorted at creation and appended to for six months is no longer sorted, which is what compaction and clustering maintenance exist to fix, and why 'we sorted it once' is not a durable answer.

The reasoning

The base is collected by analysing the table: total rows, per-column distinct counts (often themselves estimated with a sketch), null fractions, min and max, and sometimes a histogram of the value distribution. Everything above the scan is *derived* from those by rules.

The rules encode assumptions, and each one is a place error enters. Uniformity: a filter on one of ten distinct values is assumed to keep a tenth of the rows, which is wildly wrong when the distribution is skewed. Independence: two predicates are assumed to multiply, so `country = 'FR' AND city = 'Paris'` is estimated far smaller than it is, because the columns are correlated. Containment for joins, which assumes the smaller side's keys all appear in the larger.

Errors then compound multiplicatively up the plan. A join whose inputs are each estimated three times too small produces an output estimate that is off by an order of magnitude, and the plan for the *next* join is chosen from that. This is why bad plans usually appear in deep queries rather than simple ones, and why the fix is often to break a query up so the engine can see real intermediate sizes.

What to do in practice: keep statistics fresh, since stale stats on a growing table are the single most common cause; add multi-column statistics where the engine supports them, which directly addresses the independence assumption; materialise a problem intermediate so its true size is known; and only then reach for hints. Adaptive execution — re-planning mid-query once real row counts are known, as Spark's AQE does — is the systemic answer, and knowing it exists is a good signal.

The reasoning

Listing comes first. Object stores paginate listings at a thousand keys, so enumerating two million files is two thousand sequential API calls before planning can even start. On a partitioned table with many directories this is often the largest single component, and it happens every query.

Then per-file open cost. A columnar file has a footer with its schema and statistics that must be read before any data — so each file costs at least one extra round trip. At tens of milliseconds each, two million files is hours of latency that no amount of parallelism inside a file can help.

Then scheduling. Engines assign work per file or per row group, so two million files becomes two million tasks. Task scheduling overhead is milliseconds each, and tasks that do a few kilobytes of work are almost entirely overhead.

Then compression and encoding quality, which is the one people forget: a file too small to fill a row group compresses badly and its statistics are useless for pruning, because a min/max over a handful of rows rarely excludes anything. So small files also defeat the pushdown that would have saved you.

The fixes follow from which cost dominates: compaction to rewrite many small files into few large ones, which is the general answer; larger write batches or fewer output partitions so they are not created in the first place; and coarser partitioning, since partitioning by a high-cardinality column is the usual root cause. The target is normally a few hundred megabytes per file — big enough to amortise the overhead, small enough to parallelise.

The reasoning

Early materialisation reads every requested column for every row in the block, assembles the rows, then filters. Late materialisation reads only the predicate column, evaluates it to produce a selection vector — a list of positions that survived — and then reads the other columns at just those positions.

The saving scales with selectivity. A predicate keeping one percent of rows means the other nineteen columns are read for one percent of the data, which is a hundred-fold reduction in decompression work. When the predicate keeps most rows the technique buys nothing and the positional access can be slower than a sequential scan, which is why engines choose adaptively rather than always doing it.

It composes with the encodings. A dictionary-encoded column can be filtered by comparing dictionary codes rather than decoded strings, so the predicate is evaluated on integers and the strings are never materialised for rejected rows at all. That is why the technique and the encoding are designed together.

The limit is random access cost. Fetching scattered positions from a compressed column means decompressing the blocks that contain them, so if the survivors are spread evenly across every block you decompress everything anyway. Clustering the data on the filter column concentrates survivors into fewer blocks — which is the same reason sort order helps zone maps, arriving from a different direction.

The reasoning

`SELECT DISTINCT x` is exactly `GROUP BY x` with no aggregate function, and it is executed the same way — a hash table keyed on the distinct values, or a sort followed by a sweep that emits on change. Memory is the number of distinct values, so distinct over a near-unique column is the expensive case.

Distributed, it is a two-stage operation: deduplicate locally on each node to shrink what moves, shuffle by a hash of the value so that equal values meet, then deduplicate again. The local pre-pass is what makes it affordable, and it works because distinctness is decomposable — the union of per-node distinct sets contains the global distinct set.

`COUNT(DISTINCT x)` is different because counts are not decomposable the same way. You cannot add per-node distinct counts — the same customer appearing on two nodes would be counted twice — so you must ship the values, not the counts. That is the fundamental reason a `COUNT(DISTINCT)` across a large cluster is so much more expensive than a `COUNT(*)`, and why multiple `COUNT(DISTINCT)`s in one query are worse still, since each needs its own shuffle.

Which is exactly the opening a sketch fills: HyperLogLog registers *are* mergeable, so each node builds a sketch, the sketches are combined by taking the maximum per register, and the count comes out of the merged sketch. You give up exactness and gain a computation that distributes in constant space. That trade — mergeability bought with approximation — is the recurring shape behind every sketch in this subject.

The reasoning

The `PARTITION BY` becomes a shuffle: rows are repartitioned by a hash of `customer_id` so every row for a customer is colocated. The `ORDER BY` becomes a sort within each partition. Only then can the function be evaluated, sweeping the sorted rows and assigning numbers.

So the cost is a shuffle plus a per-partition sort, and both can spill. A window over a partition key with few distinct values is the pathological case: a thousand customers over a billion rows means a thousand partitions, one node gets an enormous one, and that partition must be sorted in its entirety. Window functions are therefore acutely sensitive to skew, more so than a plain aggregation, because a partition cannot be split across nodes at all.

Frames matter for the cost of the sweep. `ROW_NUMBER` and running aggregates over an unbounded-preceding frame are a single forward pass with constant state. A sliding frame like `ROWS BETWEEN 100 PRECEDING AND CURRENT ROW` needs a buffer of that width. `RANGE` frames on a value rather than a row count need a search per row, and are considerably more expensive than the `ROWS` equivalent people usually mean.

The optimisation that follows: consecutive window functions sharing the same `PARTITION BY` and `ORDER BY` are evaluated in one shuffle-and-sort, so writing several windows with an identical `OVER` clause is nearly free compared to writing them with slightly different ones. And if the table is already partitioned and sorted on those columns on disk, the engine can skip both steps — which is the sorted-input argument again.

The reasoning

State is `keys × open windows`. Tumbling windows keep that small because each key has one open window at a time; sliding windows multiply it by the overlap factor — a five-minute window sliding every minute means five open windows per key — and session windows are the worst, because a session's end is not known until a gap has elapsed and two sessions can merge when a late event bridges them.

Release is governed by the watermark, which is the job's assertion that no event older than time T will arrive. Once the watermark passes a window's end plus `allowedLateness`, that window can be emitted and its state dropped. That is the entire purpose of a watermark: it is the mechanism that makes unbounded state bounded.

Which is why a watermark that cannot advance is a memory leak, not just a latency problem. If one partition stops producing events, the watermark — usually the minimum across partitions — freezes, no window closes, and state grows until the job dies. The symptom is a streaming job with rising memory and no visible errors, and the cause is an idle partition rather than anything about the data volume.

The trade-off to state explicitly: longer allowed lateness means more correct results and proportionally more state retained. Shorter means less memory and more events arriving after their window has closed, which then need a side output or a batch reconciliation to recover. That decision is a data-correctness decision with a memory price, and it belongs to whoever owns the number, not to whoever tunes the cluster.