The order nobody fixed

The summaries that add said something about the exact structures that is worth quoting back before it is taken away:

It makes the operation associative and commutative, which is what allows a tree. A hundred summaries can be combined pairwise in any arrangement, in parallel, and the answer does not depend on the arrangement.

That is true of HyperLogLog, of Count-Min and of bottom-$k$, and it is the reason those structures are pleasant to deploy. It is not true of the counter tables, and this essay is about how untrue.

What associativity was buying #

Associativity is not an aesthetic property. In a distributed aggregation it is doing four jobs at once, and it is worth separating them, because losing it loses all four and they fail differently.

It permits a tree, which is what makes the depth logarithmic in the number of shards rather than linear. It permits partial results to be combined in any grouping, so a rack can pre-aggregate before sending. It permits re-ordering, so a slow shard does not stall the ones behind it. And it makes the result a function of the set of summaries rather than of the sequence, which is what lets two runs be compared.

The first three are performance and the fourth is correctness-adjacent, and the fourth is the one that goes unnoticed. A system can lose associativity, keep building trees, keep pre-aggregating and keep re-ordering — everything still runs, every answer is inside the guarantee — and only discover the loss when two numbers that should match do not.

Three shapes, all of which get built #

There is no such thing as the merge of eight summaries. There is a binary operation and a way of applying it to eight things, and three ways occur in practice.

Folded, one at a time. A coordinator holds a running summary and combines each shard’s report into it as the report arrives: $(((s_1 \oplus s_2) \oplus s_3) \oplus s_4)\dots$ This is what a receiving loop does, and it is the shape with the smallest working state — never more than two summaries at once.

Combined pairwise, in a tree. $((s_1 \oplus s_2) \oplus (s_3 \oplus s_4))\dots$ This is what any parallel reduction builds, and what a map-reduce framework will do without being asked.

Folded, last first. The same as the first shape with the shards in the other order — which is what happens when a slow machine reports late, or a retry re-orders the arrivals at the coordinator.

For an associative operation these three are the same. For these structures they are not, and the differences are not small.

What the disagreement looks like #

At eight shards, hashed by key, on a stationary Zipf stream of forty thousand arrivals: the three shapes produce tables that differ on thirty-seven keys, the largest disagreement being 705 arrivals.

Both numbers grow with the shard count and neither goes away:

The row at four shards is worth a second look. Thirty-seven keys move at eight shards and twenty-seven at four, but the largest gap at four shards is one arrival and at eight it is seven hundred and five. So the number of affected keys and the size of the effect are separate quantities that grow at different rates, and quoting either alone gives the wrong impression.

Why an ordering can matter at all #

The merge rule is a function of two tables. It is symmetric — $A \oplus B = B \oplus A$ — so commutativity is not the problem. What fails is associativity, and it fails for a reason that is easy to state once the operation is written out.

Both merge rules truncate. Space-Saving keeps the $k$ largest of the union and discards the rest; Misra-Gries does the same and additionally subtracts the cut. Truncation is where the information goes, and which information goes depends on what else is in the table at the moment of truncation.

Consider a key that is thirty-fifth in $s_1 \oplus s_2$ and would have been twentieth in $s_1 \oplus s_2 \oplus s_3$. Merging left to right, it is discarded at the first step and its arrivals never reach the third; merging as a tree, $s_3$ and $s_4$ combine first, the key survives in the other half, and the final merge finds it in both. The order decides which keys are still present to be counted, and a discarded key cannot be recovered by any later step.

That is why the exact structures escape. A HyperLogLog register never discards anything — a maximum keeps its argument. A Count-Min cell never discards anything — a sum keeps both. The structures that merge exactly are exactly the structures whose merge has no truncation in it, and truncation is what a bounded-space summary is.

The jump between four shards and eight #

The largest gap goes from one arrival at four shards to seven hundred and five at eight, and a hundredfold jump in one doubling deserves an explanation rather than a shrug.

At four shards, hashed by key, each shard holds thirty-two counters and the union of four such tables is at most one hundred and twenty-eight keys. The truncation to thirty-two throws away keys that are genuinely small — the fourth quartile of a union that is itself mostly the head of the distribution — and whichever shape does the throwing, roughly the same small keys go. The tables differ, because different small keys go, but the counts of the surviving heavy keys barely move: a key at rank ten is at rank ten in every intermediate, in every shape.

At eight shards the union is up to two hundred and fifty-six keys and each shard’s thirty-two are the keys heavy on that shard. Now a key can be rank five on shard three and absent from the other seven, and whether it survives to the final table depends on which other shards it was merged with first. When it does not survive, its several hundred arrivals are simply gone, and the shape that kept it and the shape that did not differ by that much.

The threshold is where the union of the tables outgrows the truncation by enough that genuinely heavy keys are in the discarded part. Below it the shapes disagree about which small keys to lose; above it they disagree about which large ones. The count of affected keys rises smoothly through the transition and the magnitude does not, which is why both numbers have to be quoted.

Neither answer is wrong #

Every one of these tables satisfies the structure’s guarantee. Space-Saving’s counters are never below the truth in any shape; Misra-Gries’s are never above it. The site’s gate asserts the first on every build, over every key of the merged table, in the direction where a bug would make the numbers look better.

So a system running any of the three shapes is running a correct implementation, and a system that changes shape between releases has changed nothing it promised. What it has changed is the answer.

This is a guarantee is not a result again, and the merge is the sharpest case in the collection so far, because here the two answers being compared are both inside the bound and both produced by the same code from the same data. The bound was never tight enough to pin the answer down, and nothing else was ever pinning it.

What is actually lost #

The accuracy question is the smaller half. Order-dependence costs three properties that a system may have been relying on without noticing.

A summary stops being a value. The summaries that add made the point that a bit-identical merge lets a summary be compared for equality, cached by content, checkpointed and deduplicated. None of that survives here: two runs of the same pipeline over the same data can produce different bytes, so a cache keyed on the summary’s hash is a cache that never hits, and a checkpoint compared against a recomputation reports a difference that means nothing.

A retry stops being free. If a shard’s report is delivered twice, an idempotent merge absorbs it. Neither of these merges is idempotent — merging a Space-Saving summary with itself doubles the counts, correctly — so a duplicate is a correctness problem the merge cannot detect, and the protection has to come from somewhere else. That much is true of Count-Min too. What is new here is that a duplicate also changes the shape, so the failure is not confined to the duplicated shard’s contribution.

And a comparison across time stops meaning anything. A dashboard showing this hour’s merged table against last hour’s is showing two numbers whose difference includes whatever the two hours’ merge orders happened to be. On this stream that is worth up to 950 arrivals on a single key, which is comfortably larger than most of the changes anybody would be watching for.

A taxonomy that falls out of three phases #

Put the structures this collection has merged into one list and a rule appears that is worth more than any of the individual measurements.

Read down the columns and the structure of the field is visible. Exactness and order-blindness travel together, and both are properties of merges that do not discard: a maximum, an or, a sum, and a re-truncation of a set that was never lossy to begin with. Idempotence is a separate axis and depends on whether the combining operation is a maximum or a sum.

The bottom-$k$ row is the interesting one, because it truncates and is nevertheless exact. Taking the $k$ smallest hashes of a union is order-blind because the $k$ smallest of a union are among the $k$ smallest of each part — the discarded elements provably could not have mattered. Neither counter table has that property: a key discarded from one intermediate might have been the largest in the final table, and there is no way to know at the time of discarding.

So the dividing line is not truncation as such. It is whether the truncation can be shown, at the moment it happens, not to matter. That is a sharper statement than “these two structures behave badly”, and it predicts the column for a structure this site has not measured yet.

The repair, and what it costs #

There is an obvious fix and it is worth pricing rather than dismissing.

Fix the shape. Declare that merges are always performed in shard-index order, folded, and the answer becomes a function of the data alone. It costs the parallelism: a fold is a serialisation point, and the whole reason to build a tree is that the tree is $\log m$ deep and the fold is $m$.

Or fix the truncation. Merge without discarding — keep the union of all the tables, up to $mk$ entries, and truncate once at the end. That is associative, because a single truncation at the end cannot depend on the order of the sums that preceded it. It costs the space: the merged intermediate is $m$ times a summary, which is exactly the state the state a merge is standing in for measured the deployment as already holding.

The second is the more interesting trade, because it says the order-dependence is bought rather than inherent: a merge is order-dependent precisely because it truncates early, and truncating early is what keeps the intermediate small. A coordinator with room for $mk$ counters can have a reproducible answer; one with room for $k$ cannot.

That is a clean statement of the kind this collection prefers — two resources, traded against each other, with the exchange rate measured — and it is the reason to prefer it over the advice to standardise the shape.

Where the shape is decided, in practice #

It is worth naming the places a shape gets chosen, because none of them looks like a choice at the time.

A reduce in a dataflow framework builds a tree and the arity is a configuration value — often literally a tuning parameter for memory, unconnected to correctness in the framework’s own model, because the framework assumes the operation is associative and says so in the interface it asks the caller to implement.

A gRPC fan-in with a completion handler folds, in the order the responses arrive, which is the order the network chose.

A retry re-orders. A shard that times out and is re-requested arrives after the others, so its contribution is folded in last on one run and in the middle on the next. Nothing has failed; the answer has moved.

And a backfill combines a different set. Re-running yesterday’s aggregation over a repaired shard produces a merge of the same summaries in a different arrangement, and the reconciliation report shows a difference on forty keys that nobody can explain, because both numbers are right.

The last of these is the one worth guarding against, and the guard is cheap: if a system compares merged summaries across runs, it has to fix the shape, and fixing the shape is a line of code and a serialisation point. That is a real cost and it is smaller than the cost of an unexplainable diff.

The partition decides whether the order matters #

The last plate on this page carries the largest qualification in it, and a caption is not where it belongs. Under two of its three partitions the fold orders agree to a fifth of a per cent. Every number in the body above is measured under the third.

That is not a footnote to the result; it is the condition on it, and the mechanism is the one this essay has already established.

The disagreement comes from truncation, and truncation only discards something that mattered when the union of the shards’ tables is much larger than $k$. So the controlling quantity is how much the shards’ tables overlap, and the partition sets it directly.

Hashed by key, every key lives on exactly one shard. Eight shards of thirty-two counters hold up to 256 distinct keys between them and the merge keeps 32, so seven eighths of the union is discarded and which eighth survives depends on the grouping. That is the setting of the tables above.

Evenly loaded, every key lands on every shard. Each shard’s table is a scaled-down copy of the same global distribution, so all eight agree on nearly the same thirty-two keys, the union is barely larger than $k$, and there is almost nothing for the truncation to decide. The orders agree because they are choosing between identical candidates.

So the rule is the inverse of the obvious one. Order-dependence is worst when the shards’ tables are nearly disjoint and vanishes when they are nearly identical — and spreading a key across every machine, which looks like the wasteful arrangement, is the reproducible one.

Which makes the practical warning both narrower and more pointed than nothing in a deployment fixes the shape. A deployment does not fix its merge shape, and it does fix its partition, and the partition is written down: it is a routing rule in a configuration file. The test is one question — can the same key arrive at two shards? If it cannot, the shape matters and everything above applies. If it can, the shape is very likely irrelevant and no serialisation point is needed.

The tension that leaves is a real one and nothing names it. Hashing by key is chosen because it lets a shard pre-aggregate its own keys completely, which is most of why sharding is worth doing; and it is exactly the choice that makes the merge order-dependent. Round-robin gives up the pre-aggregation and gets a reproducible answer. Two goods, in direct opposition, decided by a parameter selected for neither of them — which is the partition the analysis did not mention’s subject arriving one level up.

It also adds a third repair to the two priced above, and it is the one a deployment usually cannot take. Fixing the shape costs the parallelism; fixing the truncation costs the coordinator the state it was already holding; changing the partition costs the shard locality the whole architecture was built for. The first two are engineering decisions and the third is not, which is why the honest recommendation stays where it was — fix the shape, and pay the serialisation point — and why it should be given to the systems that hash rather than to all of them.

What this does not measure #

Nothing here is a bound on the disagreement. The gaps quoted are what these shapes produced on this stream at these settings. No claim is made about a worst case, and the worst case over all shapes of a merge tree is a quantity this site has not computed.

Three shapes out of many. A tree of eight has more arrangements than the three drawn; balanced, folded and reverse-folded were chosen because they are the three that occur, not because they bracket the range.

And the stream is stationary. Every number here is from a Zipf draw whose distribution does not move. On a stream where a key is heavy early and absent later, truncation order interacts with arrival order as well, and that is a second mechanism this measurement cannot separate from the first.

What is established is that the operation is not associative, that the disagreement grows with the shard count, and that both the accuracy and the reproducibility of a merged summary depend on a parameter no deployment writes down.