Interview questions

Elapsed time is work divided by the capacity you kept busy, not by the capacity you have. Machines sit idle for three reasons: waiting for a producer (dependency), waiting for a slow peer at a barrier (imbalance), and waiting for a shared resource (exclusion). Find which one dominates.

Read moreThree idlenesses

A barrier waits for the slowest task: about 1 + 0.3 × √(2 ln 1000) ≈ 2.1× the mean, from variance alone. With heavy-tailed task times the maximum grows faster, so this is the optimistic case.

Read moreThe barrier and the maximum

With pull, the consumer asks for each unit, so it can always say "not yet": backpressure is free, at the cost of a call per unit. With push, the producer runs a tight loop, so a bounded queue or credit system must be added to let the consumer refuse. Examples: Flink's credit-based flow control, actor mailboxes that grow without bound, and TCP's receive window.

Read morePush-based execution

No. Its cost is one virtual call and generic attribute access per row, which also prevents keeping values in registers and inlining. The direction of flow and the unit of work are independent: Vectorwise was pull-based and vectorized, and was the fastest analytical engine for years.

Read moreThe cube, and a break in scaleVolcano: iterator pull

Flow control. The producer decides when the next row moves, so the consumer cannot refuse. Across a thread or machine boundary you must add a bounded queue or a credit protocol, and LIMIT becomes a stop signal sent upstream.

Read morePush-based execution

Keep a batch's working set in the L1 cache: about 32 KB ÷ (live columns × 8 bytes), around 1,024 values. Smaller batches bring back the fixed cost per call; larger ones push intermediate results out to main memory.

Read moreVectorized execution

Batches spread the call overhead over many values and keep many independent cache misses in flight at once (Little's Law). Compiled code wins on computation-heavy queries, vectorized code on memory-bound ones, and vectorization is also easier to debug and port.

Read moreCompiled query execution

When the rows processed exceed the compile cost divided by the time saved per row. With 50 ms to compile and 20 ns saved per row, the break-even is 2.5 million rows. That threshold is why engines tier: interpret first, compile in the background, then switch.

Read moreCompiled query execution

One permanent thread per core repeatedly takes a small chunk of input, a morsel, from a queue and runs the whole pipeline on it. Because threads, not the plan, own the parallelism, the degree of parallelism can change mid-query, and imbalance is bounded by one morsel rather than one partition. A morsel should take about 100 µs to 1 ms (10⁴–10⁵ rows), so dispatch costs under 1%.

Read moreMorsel-driven parallelism

Batching spreads the per-call cost across a block. A shared columnar in-memory format removes the cost of crossing the boundary altogether.

Read moreVectorized UDFs and batch inference

Long enough that dispatching it is a small overhead: at least the fixed dispatch cost divided by the overhead you accept. With 10 ms of dispatch and a 1% target, about 1 second. Hence the usual guidance of 100 ms to a few seconds per task.

Read moreThe scale break

Compute n × T × λ: tasks × task duration × failure rate per unit time. Well below 1, stream and accept restarting the whole query on a rare failure. Near or above 1, materialize, so that a single task is the unit of failure and retry.

Read moreThe scale breakDataflow / DAG execution

The straggler penalty, the slowest task over the mean, is paid in every round and does not average out, because the barrier forbids starting the next round early. A 2× penalty over 30 rounds is paid thirty times.

Read moreBSP supersteps

Running out of memory in a mailbox. Mailboxes are unbounded by default, the sender decides when to send, and the receiver cannot refuse. Every mature actor system has had to add explicit backpressure.

Read moreActor-model execution

All data exchange goes through object storage, writing M × R objects for M producers and R consumers. At about $5 per million requests, 100 KB objects cost $0.05 per GB in requests and 10 KB objects $0.50 per GB. Past some width, the requests cost more than the compute.

Read moreServerless execution

T_p ≥ max(T₁ / p, T∞): no schedule on p workers beats the total work divided by p, and none beats the longest chain of steps that must run in sequence. Amdahl's Law is the special case where that chain is one serial block.

Read moreThe one bound, in three costumes

Data-parallel speedup is capped at 1 / f, where f is the largest partition's share of the work. A key holding 5% of the rows caps a thousand-core cluster at 20×.

Read moreData parallelism

Those that can be written as partition, compute locally, merge: sum, count, min, max, and average as a (sum, count) pair. Exact median and exact distinct count cannot; they need a sort, two passes or a sketch.

Read moreThe algebraic gate

The far end holds the oldest, largest task, the one least likely to be in the victim's cache, so each steal moves the most work and fewer steals are needed. Randomized stealing gets within T₁ / p + O(T∞) with no coordinator. It cannot split a task, so it does nothing for skew.

Read moreWork stealing

When the cluster is busy: duplicates take capacity from other queries, and response time grows as 1 / (1 − ρ) with utilization ρ. It also fails when the detector compares elapsed time only, because it then duplicates the biggest task, and the copy takes just as long.

Read moreSpeculative execution

Plot each task's duration against its input bytes. Random spread at equal bytes is imbalance; one slow task at equal bytes is a straggler (a bad machine); duration proportional to unequal bytes is skew.

Read moreClassify the long tail

A key with L rows on the left and R on the right produces L × R output rows, so skew multiplies: 10× on both sides is 100× the work.

Read moreWhy join skew is worse than it looks

Each of the M × R blocks must stay above the medium's minimum efficient read, so M = R ≤ √(S / b_min) for S bytes shuffled. Ten times more data allows only about √10 ≈ 3.2 times the width: 1 TB with a 1 MB floor gives 1,000 × 1,000.

Read moreThe width ceiling

Hash shuffle writes M × R files, needs R open file handles per task, and R buffers per core: 16 cores × 1,000 × 32 KB = 512 MB. Sort shuffle writes one file per task, plus an index of offsets.

Read moreHash shuffle versus sort shuffle

It merges blocks from many producers, so each consumer does one large read instead of M small ones. It costs a second network hop, a shared stateful service and a fallback path. Its biggest benefit is elasticity: executors can be released, because their shuffle output no longer lives on them.

Read morePush-based and remote shuffle servicesWhy remote shuffle really won

When the codec's throughput exceeds BW × r / (r − 1), for link bandwidth BW and ratio r. On a 100 Gbps network, nothing wins: shuffle data is read once, so a slow, strong codec is never amortized.

Read moreShuffle compression