Sorting a billion integers from a stream

August 29, 2026 · personal · algorithms · dotnet · elixir · distributed-systems

the short version

At a billion keys, radix sort is the only sort that keeps up with the network. A billion 32-bit integers is 4 GB. Over 10 GbE that takes about three seconds to arrive; Array.Sort then needs 66 s to sort them on one thread, and the sixteen-thread radix sort from the previous post needs 0.7. Once the sort is faster than the ingest, the design question changes: it's no longer how fast you can sort, it's how much of the sort you can do while the data is still arriving. The answer is about a third of it. Partition each chunk by its top bits the moment it lands, count as you go, and after the last byte only the two private scatters remain: half a second to a fully sorted billion, down from 0.9.

The shape is the one production engines use. An analytics node ingesting a billion rows an hour doesn't sort a billion rows; it sorts each arriving block as it lands, writes it as a sorted run, and merges runs in the background. Radix sort is the tool inside that shape for numeric keys, and this post is the inside of one such block, at a size where the constants are the whole story.

the case: a billion rows an hour, one node

In 2018 Cloudflare wrote up their HTTP analytics pipeline: 6M requests a second at the edge, 11M rows a second into a 36-node ClickHouse cluster, 47 Gbps of inserts, Kafka in between with 106 partitions and a Go consumer per partition batching rows into inserts. Per node that's about 300,000 rows a second, a billion rows every hour, arriving from a dozen or so streams. That's the shape and the scale of this post, and it hasn't got smaller since.

What the node does with each insert is the interesting part. A MergeTree table writes every insert as a separate part, and "each of them is lexicographically sorted by primary key"; a background process then merges parts together. The sort on insert is where the row order comes from, the merge keeps the number of parts bounded, and for numeric key columns the sort ClickHouse reaches for is its own radix sort (LSD, stable, in RadixSort.h). So the unit of work is: rows arrive from N streams, a block of them is sorted by key, the sorted block is written. That is exactly what follows, with the block scaled up to a billion 32-bit account ids so that the constants have nowhere to hide. Real rows carry a hundred other columns; those move by the sorted permutation afterwards, and the key sort is still the part that decides whether the node keeps up.

what can happen while the data arrives

The previous post ended with the fastest in-memory version: one MSD pass partitions the input into 1,024 buckets by the top 10 bits, then each bucket is sorted privately on its low 22 bits with two 11-bit passes. Look at that from the point of view of a chunk landing off the network.

The MSD partition doesn't need anything global. A key's bucket is its top 10 bits, and if every stream keeps its own 1,024 buckets as growable lists of pages, a key can go into its page the moment it arrives. No prefix sum, no knowledge of n, nothing shared between streams. The histograms of the low digits don't need anything global either: two increments per key into the arriving bucket's counters.

What can't happen on arrival is the scatter. A stable scatter into a bucket's final range needs the bucket's complete histogram, and that isn't known until the last key has landed. So the two private passes are the floor: after the last byte you owe two sweeps over the data, and nothing else.

Here's the partitioner, one per input stream, called on the stream's own thread with each chunk. Three things happen per key: it's appended to its bucket's open page, the bucket's count goes up, and its two lower digits are counted for the passes that come later.

// A 32-bit key is sorted as three digits. The top one picks the bucket on
// arrival; the other two are sorted privately inside the bucket afterwards.
public static int TopDigit(int key) => key >> 22;            // bits 22..31, 1,024 buckets
public static int MidDigit(int key) => (key >> 11) & 0x7FF;  // bits 11..21, second private pass
public static int LowDigit(int key) => key & 0x7FF;          // bits 0..10,  first private pass

// Called on the stream's own thread with each chunk as it lands. Nothing in here needs
// the total count or any other stream: the bucket is the key's top digit, and the
// histograms are per bucket, per stream. The totals are summed once, after the last chunk.
public void Ingest(ReadOnlySpan<int> keys)
{
    if (countOnArrival)
    {
        // Refs to the first element of each histogram: Unsafe.Add skips the bounds check on
        // the two random increments per key, which is where this loop spends its time.
        ref var lowCounts = ref MemoryMarshal.GetArrayDataReference(LowDigitCounts);
        ref var midCounts = ref MemoryMarshal.GetArrayDataReference(MidDigitCounts);
        foreach (var key in keys)
        {
            var bucket = key >> Radix.TopShift;                 // top 10 bits pick the bucket
            var n = used[bucket];
            openPage[bucket][n] = key;                          // append to the bucket's open page
            if (++n == PageSize)                                // page full: file it, open a fresh one
            {
                fullPages[bucket].Add(openPage[bucket]);
                openPage[bucket] = new int[PageSize];
                n = 0;
            }
            used[bucket] = n;
            KeysPerBucket[bucket]++;
            // count the two lower digits now, so the private passes later need no counting read
            Unsafe.Add(ref lowCounts, bucket * Radix.LowRadix + (key & Radix.LowMask))++;
            Unsafe.Add(ref midCounts, bucket * Radix.LowRadix + ((key >> Radix.LowBits) & Radix.LowMask))++;
        }
        return;
    }
    // ... the same loop without the two counting lines
}

PageSize is 4,096 ints, 16 KB, so each stream has 16 MB of open pages at any time and appends are sequential within a page. LowDigitCounts and MidDigitCounts are the histograms for every bucket, flattened as [bucket * 2048 + digit], 8 MB each per stream; the Unsafe.Add form is there because those two increments are the loop's cost, and a version with plain jagged arrays partitions 13% slower when nothing is pacing the input (and identically at 10 GbE).

what's left after the last byte

Once every stream has stopped, the bucket sizes are the sum of each stream's counts, which gives every bucket its final range in the output. Then, per bucket, in parallel across workers: sum the streams' histograms for the bucket (2,048 adds each), scatter straight from the pages into a scratch buffer on the low 11 bits, and scatter the scratch into the bucket's final range on the next 11. Both scatters use the write-combining staging from the previous post.

// Pass A: low digit. Read the pages in arrival order and scatter into scratch.
{
    ToOffsets(l.LowHist);                                              // histogram -> where each digit's run starts
    ref var offsets = ref MemoryMarshal.GetArrayDataReference(l.LowHist);
    ref var dest = ref MemoryMarshal.GetReference(tmp);
    ref var lines = ref MemoryMarshal.GetArrayDataReference(l.WcLines);
    ref var lineFill = ref MemoryMarshal.GetArrayDataReference(l.WcFill);
    foreach (var s in streams) foreach (var page in s.Pages(b))
        foreach (var key in page.Span) Radix.Put(ref lines, ref lineFill, ref offsets, ref dest, Radix.LowDigit(key), key);
    Radix.Drain(l.WcLines, l.WcFill, l.LowHist, tmp, Radix.LowRadix);   // flush the partial lines
}
// Pass B: middle digit. Scatter scratch into the bucket's final range. Stable, so
// keys with the same middle digit keep the low-digit order pass A gave them.
{
    ToOffsets(l.MidHist);
    ref var offsets = ref MemoryMarshal.GetArrayDataReference(l.MidHist);
    ref var dest = ref MemoryMarshal.GetReference(bucket);
    ref var lines = ref MemoryMarshal.GetArrayDataReference(l.WcLines);
    ref var lineFill = ref MemoryMarshal.GetArrayDataReference(l.WcFill);
    foreach (var key in tmp) Radix.Put(ref lines, ref lineFill, ref offsets, ref dest, Radix.MidDigit(key), key);
    Radix.Drain(l.WcLines, l.WcFill, l.MidHist, bucket, Radix.LowRadix);
}

l is the worker's reusable scratch state (one bucket's worth of temp space, the two histograms, the write-combining lines), Radix.Put and Radix.Drain are the write-combining staging from the previous post, and ToOffsets turns a histogram into start positions in place.

Pass A reads the pages in arrival order, so the sort is stable with respect to arrival within a stream, and streams are visited in a fixed order, so it's deterministic across them too. Peak memory is the pages (4 GB), the output (4 GB) and one scratch buffer per worker the size of its largest bucket (about 4 MB), the same two buffers the batch version needs. The difference is that a bucket's pages can be dropped the moment it's sorted, which the benchmark doesn't bother to do and a long-running node would.

the numbers

Sixteen simulated streams, each producing its share of the keys in 1 MB chunks from a seeded generator, either as fast as the partitioner will take them or paced to an aggregate 1.25 GB/s, which is 10 GbE. Every result was checked for order and against the sum and xor of the generated keys. .NET 8 in a Linux arm64 container on an M3 Max, sixteen cores.

At 100M the same five rows run in 0.03 to 0.32 s of ingest and 0.05 to 0.12 s after the last chunk (Array.Sort: 6.1 s), with the same ordering.

Three things to read off it.

The batch sort is already faster than the network. At a billion keys the in-memory sort takes about 0.7 s, and a single-threaded Array.Sort takes 65.6 s. Once you're at radix, the ingest is the bottleneck by a factor of four, and any further work on the sort only shortens the tail after the last byte.

Partitioning on arrival is free at network speed and not free otherwise. Unpaced, the partitioner ingests slower than a memcpy because the scatter into 1,024 open pages is real work; paced to 10 GbE, the producers have time to spare and the partition rides along at no cost to the ingest.

The floor is two-thirds of the sort. Counting on arrival gets the tail to 0.46 s unpaced and 0.50 s paced, against 0.62 and 0.89 for the batch sort. The two scatters that need the totals are two of the three passes, so at most a third of the work can hide behind the ingest, and that's what hid. Counting has a price when the wire is faster than 10 GbE: the histograms are 16 MB per stream and every key does two random increments into them, which unpaced turns a 0.4 s ingest into 2.0. Scattering straight from the pages didn't beat the plain copy, because it reads the pages twice (once to count, once to scatter) through an enumerator, and that costs what the copy cost. The version to ship is the one whose extra work fits in the idle time you actually have.

the same billion on the BEAM

For the record, the :atomics version from the previous post scales to a billion keys without changing: two off-heap arrays of a billion 64-bit slots (16 GB), sixteen processes each owning a range.

Both runs verified sorted with the input's sum. Ten seconds for a billion against 0.6 in .NET is the same fifteen-fold BIF-call-per-element gap the previous post measured at ten million, unchanged by scale. It's also an order of magnitude inside what Enum.sort would take (a couple of minutes, extrapolated from 10M, on a list that would itself be 16 GB), and it's the only pure-BEAM sort that gets a billion integers into order in a time you'd wait for.

The BEAM can't do the streaming variant as written, because there's no way to hand pages between processes without copying them, but the same partition-on-arrival idea maps onto per-process :atomics buckets sized from a first pass over the stream's own counts. That's a post of its own.

if it doesn't fit, or must stream out

Everything above assumes the billion fits in memory and the output is wanted all at once. Drop either assumption and radix moves from being the frame to being the tool inside a different frame: sort each arriving block (with radix, since the keys are numeric), write it as a sorted run, and k-way merge the runs. That's MergeTree's parts and background merges, and DuckDB's per-thread radix sort followed by Merge Path. The merge is bandwidth-bound and parallel, and its first output row comes out immediately, which a full-pass radix sort can never offer. On disk the MSD partition survives too: 1,024 buckets are 1,024 sequential appends, which is what an external distribution sort has always been.

what to take from it

At a billion keys of a fixed-width type, the sort is a solved problem and the network isn't: a sixteen-core radix sort beats the wire by a factor of four and the library sort by a hundred. Design for the tail after the last byte, and the way to shorten it is to do the parts of the algorithm that need no global knowledge (partition by the top digits, count the lower ones) as the data lands, leaving only the scatters that need the totals.

The production shape is sort-per-block then merge, and radix is the sort per block. That's why an analytics node can absorb a billion rows an hour with its keys in order.

where to look

• The previous post: Radix sort on sixteen cores, in .NET and Elixir, with the in-memory version this builds on and the write-combining scatter it reuses.
• Cloudflare, HTTP Analytics for 6M requests per second using ClickHouse, 2018.
• ClickHouse, MergeTree: parts sorted by primary key on insert, merged in the background; and RadixSort.h.
• Laurens Kuiper, Fastest table sort in the West, 2021: per-thread radix sort, then Merge Path.
• Wassenberg and Sanders, Faster Radix Sort via Virtual Memory and Write-Combining, 2010.