Co-Partitioned Datasets and Shuffle-Free Joins

Joining two large tables on a key requires the matching rows to meet somewhere. If neither table is small enough to copy in full, the shuffle that brings them together is usually the most expensive part of the query — every row of both sides written to a shuffle file, fetched across the network, and merged before a single join comparison happens. Co-partitioning is the technique of paying that cost once, at write time, and storing the result in the table layout so later queries can skip it.

The shuffle cost a join normally pays

A shuffled sort-merge join writes and fetches both inputs and sorts both sides by the join key. Its shuffle volume is roughly the sum of the two tables’ sizes after projection, and both sides must be fully materialized and merged before the join begins. A shuffled hash join does the same for the build side, which must be partitioned and then held in a hash table per partition.

The alternative is a broadcast, which avoids shuffling the large side at all by copying the small side to every task. A broadcast costs the size of the small side multiplied by the number of tasks that must receive it: a ten-mebibyte dimension table distributed to a thousand tasks moves on the order of ten gigabytes across the network, which is a fine trade when the small side really is small — engines commonly default the broadcast threshold to about ten mebibytes — and a bad one when the “small” side is a hundred megabytes and the task count is in the thousands.

Both approaches leave the large side’s data resident in shuffled files that are discarded when the query ends. A table scanned by the same join every night pays the same cost every night, which is the observation that motivates making the partitioning persistent.

Bucketing as a shuffle written into the table layout

A bucketed table stores its data already divided by the join key. On write, each row is assigned to bucket $b = h(k) \bmod B$ for a fixed bucket count $B$ and a fixed hash function $h$, and each bucket is written as its own file or file group. If both tables are bucketed by the same key with the same $B$ and the same $h$, then a join on that key pairs bucket $b$ of the first table with bucket $b$ of the second, and the join becomes $B$ independent tasks that each read one bucket from each side. No shuffle is created at query time.

The write itself is a shuffle, so nothing is saved on the first join; the saving accumulates on repeats. That makes bucketing worthwhile for tables joined on a stable key by many queries, and pointless for a table that is scanned once or joined on varying columns. It also raises the cost of updates: rewriting rows in a bucket means rewriting whole bucket files, and appending new writers must use identical boundary or hash definitions, which is why the open table formats store the bucketing specification as table metadata rather than as a convention among writers.

Range bucketing achieves the same co-partitioning with ordered buckets chosen from a sample of the key distribution. Ordered buckets add two capabilities hash buckets do not have. A sorted merge join can stream through a bucket from each side in key order without holding a hash table, so the per-task memory requirement drops to the size of one key group plus fetch buffers, which makes it usable for buckets far larger than the available memory. And a predicate on the join key eliminates entire buckets from the scan before any bytes are read, because the stored boundary values make the elimination exact.

The preconditions, and how they break quietly

Co-partitioning rests on three conditions being satisfied for both inputs: the same partition function, the same partition count, and the same key values as seen by that function. Each has a distinct failure mode.

A differing bucket count, 64 against 128, means bucket $b$ of one table contains keys that live in two buckets of the other. Query planners detect this from the table metadata and fall back to a shuffle, so the cost reappears but the answer stays correct — a performance surprise rather than a correctness one, and the reason a bucketed join should be confirmed by inspecting the plan rather than assumed from the table definition.

Agreement on the hash function is subtler, because it is applied to values and the values must be encoded identically. An integer key stored as 32 bits in one table and 64 bits in the other hashes differently if the function is sensitive to the byte width. A string key compared case-insensitively but hashed on the raw bytes sends Alpha and alpha to different buckets, which changes where rows land without changing what the join considers equal; if a subsequent bucketed step assumes key equality implies bucket equality, rows that should match end up in different buckets and the join drops them. This is the mismatch that planners cannot always detect, because the metadata records a function name and a column type rather than the semantics of the comparison.

Skew survives bucketing untouched. A key with a large share of the rows still hashes to exactly one bucket, so bucketing does not remove the sixty-times-the-mean partition described in hot keys and hash partitioning. It can make skew worse to detect, since a bucketed table’s bucket sizes are fixed by the written data and no longer visible as a shuffle-time statistic. Null join keys are a specific instance: every null hashes the same way on every row, so all null-keyed rows collect in one bucket, where they do no join work at all, since null never equals null in an equi-join.

File counts, reuse, and where the trade turns

Bucketing multiplies file count. A table written in $P$ output partitions with $B$ buckets produces on the order of $P \times B$ files, so a 128-bucket table written by a 200-partition job emits tens of thousands of small files, and those files become the input splits for the next reader, with the small-file penalty described in input splits and task granularity. Keeping buckets reasonably large — tens of megabytes at least — avoids trading one shuffle problem for a metadata problem.

Reuse also depends on the bucket count matching the query’s natural parallelism. A hundred-bucket table gives a join at most a hundred concurrent tasks; if the cluster has more slots than that, the job runs under-parallelized, and the bucket count chosen for a nightly join of two huge tables may be far too small for a join that now reads a fraction of the data. Because the count is baked into the files, it cannot be adjusted for one query without a rewrite.

The trade that decides whether to bucket at all is the number of future joins against the write cost plus the file-count tax. Two tables joined nightly for a year on the same key justify the write-time shuffle many times over. A one-off join gains nothing, and a join whose key changes monthly gains nothing either, since the layout would have to be rewritten each time.

A last interaction is with the bucket count itself, which is chosen at write time and cannot be revisited per query. A statically bucketed table fixes its parallelism, while a query reading a fraction of the data would prefer more, smaller buckets; because the count is baked into the file layout, one query cannot reconcile the two without a rewrite. Engines that resize partitions between stages — the runtime repartitioning covered in adaptive partitioning at runtime — regain that freedom inside a single query, but they cannot make a stored bucket layout denser, since those bytes were written long ago. Persisted co-partitioning therefore pays off in exchange for committing to a parallelism level and a partition function in advance, and the commitment is only worth making for keys and workloads that are themselves stable.

The comparison against a shuffle written at query time is also instructive, because the two produce the same physical arrangement by different means. A query-time shuffle rebuilds the partition assignment every run and discards it, while bucketing writes the assignment into the table and keeps it. The first adapts to the day’s data and costs the same every day; the second costs nothing on repeat and cannot react to a change in key distribution, which is exactly the stability assumption that makes it worthwhile.