One pass, and no room

The partition the analysis did not mention

Space-Saving and Misra-Gries are the same structure under a stream, related by subtracting one number. Sharded eight ways and merged, one of them is wrong by 313 where the other is wrong by 927 — and swapping how the arrivals were assigned to machines reverses which is which.

Two structures that are one established something that ought to make this essay impossible. Space-Saving and Misra-Gries, run over the same stream with the same number of counters, hold the same information: subtract the smallest Space-Saving counter from all of them and the result is the Misra-Gries table, at every prefix of the stream and not merely at the end. They are one structure with two readings.

Split the stream across eight machines and merge, and they behave nothing alike.

Space-Saving: the same merge under three partitions, m = 8Worst error over the 8 heaviest keys of 40,000 arrivals of stationary Zipf — the control, where a window is a smaller sample, for three summaries answering the same question. The first holds 32 counters. The second holds as many counters as all the shards hold between them, which is the only honest thing to compare a merge against. The third is the merge itself. The partition is not a detail: round-robin 313, hashed by key 536, consecutive 309. The same stream, the same totals, the same structure, and the error moves by 1.7×.313roundconc 0.14536hashedconc 1.00309blockedconc 0.14how the arrivals were partitionedworst error over the top keysone summary, k = 32one summary, k = 256the merge of 8Space-Saving · stationary Zipf · 40,000 arrivals8 shards
Fig. 1 Space-Saving’s merged error under three ways of assigning arrivals to machines. Round-robin and consecutive blocks are near-identical; hashing by key is nearly twice as bad. The stream, the totals and the structure are the same in all three.

Three ways to split a stream, all of which occur

A partition is an assignment of arrivals to machines, and there are three in ordinary use.

Round-robin. Arrival ii goes to shard imodmi \bmod m. A load balancer with no affinity does this. Every shard sees a sample of the whole distribution, and a key with total frequency ff arrives about f/mf/m times on each machine.

Hashed by key. Key xx goes to shard h(x)modmh(x) \bmod m. Every arrival of a key lands on one machine, which is what any system wanting per-key state must do — a session store, a rate limiter, a per-user counter. It is by far the most common arrangement in practice, because it is the one that makes the state on each machine meaningful on its own.

Consecutive blocks. Shard jj gets the jj-th stretch of the stream. This is what time-partitioned processing does: a summary per minute, merged at the hour.

None of these appears in any bound either structure carries. The analysis is about nn arrivals and kk counters, and how the arrivals were divided is not a parameter of it. It is, however, a parameter of the answer.

The number that predicts it

Before the structures, the partitions themselves can be measured, and the measurement uses no structure at all.

For each of the heaviest keys, take the largest share any one shard holds of that key’s arrivals, and average over the keys. Call it the concentration. A key spread evenly over mm shards scores 1/m1/m; a key confined to one machine scores exactly 1.

On this stream at eight shards, round-robin gives 0.14 and consecutive blocks give 0.14 — both close to the 0.125 an even split would give. Hashing gives 1.00, which is not an approximation but the definition of hashing by key.

Space-Saving merged from 8 shards: worst error over the top 8 keysThe three-way comparison, over 40,000 arrivals of stationary Zipf — the control, where a window is a smaller sample partitioned round-robin: every key on every shard. The merge of 8 summaries of 32 counters is wrong by at most 313 on the heaviest keys. One summary of 32 counters — an eighth of the state — is wrong by 769, which makes the merge look good. One summary of 256 counters, which is exactly the state the shards were holding between them, is wrong by 0. The merge is the price of never having the whole stream in one place, and it is paid against the middle bar rather than the top one.one summary, k = 32769one summary, k = 25608 summaries of 32, merged313worst error over the top keys, in arrivalsconcentration 0.14 — the mean share of a heavy key held by one shard8 shards · round · balancedmerge 313 against matched 0
Fig. 2 Space-Saving under round-robin: worst error 313 against a matched single summary’s zero. The concentration line beneath the bars is 0.14 — every heavy key is present on every shard, and the merge finds it in every table.

For Space-Saving, concentration is very nearly the whole story, and the mechanism is exactly the one the state a merge is standing in for set out. The merge charges a floor for every key absent from a summary. Under round-robin a heavy key is present in all eight tables, so it is charged nothing; under hashing it is present in one and absent from seven, and it pays seven floors.

That is the factor of 1.7 between 313 and 536, and it is entirely a property of how many tables the key was in.

And then the other structure does the opposite

Misra-Gries merges differently. Add the tables, keep the kk largest, subtract the (k+1)(k+1)-th largest value from every survivor. One cut, applied to everything, rather than a floor charged per absent key.

Under round-robin, the merged table’s (k+1)(k+1)-th largest value is large. Every shard is holding roughly the same thirty-two keys with roughly the same counts, so the summed table has thirty-two big entries and a crowded shoulder just below them — and the cut is taken from that shoulder and subtracted from all thirty-two.

Under hashing, the eight tables hold nearly disjoint key sets. The summed table has up to two hundred and fifty-six distinct entries, and the (k+1)(k+1)-th largest of that is a key that was heavy on its own machine but is thirty-third overall, which on a Zipf stream is much smaller than the shoulder in the round-robin case.

So Misra-Gries pays a large cut under the partition that suits Space-Saving, and a small one under the partition that hurts it.

Misra-Gries: the same merge under three partitions, m = 8Worst error over the 8 heaviest keys of 40,000 arrivals of stationary Zipf — the control, where a window is a smaller sample, for three summaries answering the same question. The first holds 32 counters. The second holds as many counters as all the shards hold between them, which is the only honest thing to compare a merge against. The third is the merge itself. The partition is not a detail: round-robin 927, hashed by key 152, consecutive 928. The same stream, the same totals, the same structure, and the error moves by 6.1×.927roundconc 0.14152hashedconc 1.00928blockedconc 0.14how the arrivals were partitionedworst error over the top keysone summary, k = 32one summary, k = 256the merge of 8Misra-Gries · stationary Zipf · 40,000 arrivals8 shards
Fig. 3 The same three partitions, the other structure, and the ordering is reversed. Misra-Gries merged under round-robin is wrong by 927 — four arrivals better than a single summary an eighth the size. Merged under hashing it is wrong by 152.

The numbers side by side, on the same stream, at eight shards:

Space-Saving Misra-Gries
round-robin 313 927
consecutive blocks 309 928
hashed by key 536 152

Six times, in opposite directions, for two structures that are the same structure when nothing is merged.

A third structure, to check the story rather than illustrate it

Two structures agreeing with a story is weak evidence; two structures disagreeing with each other, each in the way its merge rule predicts, is stronger. The cardinality estimators are a third case and they settle the shape of the argument, because their answer is no effect at all.

A HyperLogLog merges by taking the larger of each pair of registers. A register holds a maximum over the keys that hashed to it, the maximum over a union is the maximum of the maxima, and nothing in that identity mentions how the union was divided. So a HyperLogLog sharded by hash, by round-robin or by minute produces the same merged registers, bit for bit, and the same estimate. The summaries that add measured exactly that.

The difference is not that one structure is better. It is that a maximum is insensitive to multiplicity and a counter is not. A cardinality estimator does not care how many times it saw a key, so splitting a key’s arrivals across machines costs it nothing; a frequency estimator’s entire content is how many times, and splitting that content is splitting the thing being measured.

Any structure whose merge is idempotent is partition-blind, and any structure whose merge is a sum is not. That is the rule the three cases together support, and it predicts the right answer for a Bloom filter, for a bottom-kk summary, and for the two counter tables here — which is as much generality as three measurements can carry.

Why “the same structure” was never a claim about merging

It is worth being precise about what was established earlier and what was not, because the contradiction here is only apparent.

The claim was that the two structures’ states are related by a single subtraction, at every prefix. That is still true. Run both over any one shard here and the relation holds exactly, as the site’s gate checks on every build.

What does not follow is that operations on those states commute with the relation. Merging is a function of two tables, and the fact that AA and BB carry the same information does not make f(A1,A2)f(A_1, A_2) carry the same information as f(B1,B2)f(B_1, B_2) for an arbitrary ff. The two merge rules are different functions, and the information a merge discards is chosen differently by each.

That is the general shape of it, and it is worth carrying beyond this pair: a bijection between two representations says nothing about how they behave under an operation defined on the representation. Space-Saving’s merge is defined in terms of its floors, which the Misra-Gries reading has subtracted away; Misra-Gries’s merge is defined in terms of a cut, which the Space-Saving reading has no name for.

Misra-Gries: error against the number of shards, k = 32Worst error over the 8 heaviest keys of 40,000 arrivals of stationary Zipf — the control, where a window is a smaller sample, for three summaries answering the same question. The first holds 32 counters. The second holds as many counters as all the shards hold between them, which is the only honest thing to compare a merge against. The third is the merge itself. From 2 shards to 32, the merge's worst error goes 931 to 917 while the state the shards hold between them rises 16-fold. More machines is more state and a worse answer.931m = 2conc 0.51930m = 4conc 0.27927m = 8conc 0.14924m = 16conc 0.08917m = 32conc 0.04shardsworst error over the top keysone summary, k = 32one summary, k = k·mthe merge of mMisra-Gries · stationary Zipf · 40,000 arrivalspartition: round
Fig. 4 Misra-Gries under round-robin, against the shard count, and the merge is barely better than one small summary at any point on the sweep. Every shard holds the same keys, so the summed table’s shoulder is crowded, so the cut is large, so almost everything the shards learned is subtracted away.

What the shoulder is, and why it decides everything

The word shoulder has been doing a lot of work above, and it is worth making concrete, because the whole reversal turns on it.

Take the eight tables, add them key by key, and sort the result descending. Misra-Gries’s merge keeps the first thirty-two and subtracts the thirty-third’s value from each of them. So the quantity that decides the damage is a single number: how large the thirty-third entry of the summed table is.

Under round-robin, each shard’s table holds the same thirty-two keys — the globally heavy ones — because each shard sees a scaled copy of the global distribution. The summed table therefore has about thirty-two entries and a thirty-third that is a genuine near-miss: a key that was thirty-third on every shard and is thirty-third overall. On a Zipf stream with exponent near one, the thirty-third key is not much smaller than the thirty-second, so the cut is large and it is taken off everything.

Under hashing, each shard’s table holds the thirty-two keys heavy on that shard, and those sets barely overlap. The summed table has closer to two hundred and fifty-six entries, and its thirty-third is the thirty-third globally — much further down the Zipf curve than the thirty-third-on-a-shard, and therefore much smaller.

That is the entire mechanism, and it explains a number that looks strange in the table above: Misra-Gries merged under round-robin is wrong by 927 where a single summary of a thirty-second of the state is wrong by 932. Eight machines bought four arrivals of accuracy. Everything the eight shards learned was subtracted away by one cut, because they had all learned the same thing.

The same 8 summaries, combined three waysEach row is one of the 10 heaviest keys; each column is one order of combining the same 8 per-shard summaries. Where the three columns differ, the count a query returns depends on nothing but the shape of the merge tree — 27 of the keys on this plate. The truth column is the exact count over the union and is what all three are estimating.keytruthfoldedtreereversed16,3625,4355,4355,43423,0742,1482,1482,14731,9631,0411,0411,04041,37344944944851,098195195194693569696877691212118656686956611051012shaded rows are keys whose count depends on the order aloneMisra-Gries, k = 32 · 8 shards · round27 keys move with the order
Fig. 5 The top ten keys of the union, with the counts three different merge orders produce and the exact count beside them. The shortfall against the truth is the cut, and under this partition it is nearly the whole of each count.

Which partition a system actually has

The practical reading is uncomfortable, because the partition is almost never chosen with the merge in mind.

Hashing by key is chosen for locality. It is what makes a per-key structure possible at all: a rate limiter needs every request for a user on the machine holding that user’s counter. Nobody picks it to make a merged heavy-hitter estimate more or less accurate, and yet it decides that estimate by a factor of six.

Round-robin is chosen for load. It is what a stateless balancer does, and it spreads a hot key across the fleet, which is exactly what a system wants when the hot key is a thundering herd.

Consecutive blocks are chosen by the clock. A summary per minute is a partition by time, and it behaves here like round-robin because this stream is stationary — the first minute looks like the ninth. On a stream whose distribution moves, it would not, and the measurement would need a different corpus to say anything about it.

So the partition is set by requirements that have nothing to do with summarisation, and the structure is then chosen from a literature that does not mention it. The pairing is arbitrary, and it is worth a factor of six.

What a system can do about it

Three responses are available, and they cost different things.

Change nothing and quote the right structure. The cheapest, and on the evidence above it is worth a factor of six. A system already hashing by key and already summarising heavy hitters is running the wrong one of the two if it is running Space-Saving, and the change is a different merge function over the same counters.

Re-partition before merging. A shuffle that gives every shard a sample of the whole distribution turns the hashed case into the round-robin case. It costs a full redistribution of the arrivals, which is the thing the partition existed to avoid, so it is almost never worth it for a summary — but it is worth knowing that the accuracy difference has a price in bandwidth and that the price is enormous.

Merge hierarchically along the partition. If shards are grouped so that a group holds all of a key’s arrivals, the within-group merges are exact and only the between-group merge pays. This is what a rack-aware or region-aware aggregation does for other reasons, and it happens to be the arrangement that suits a counter merge.

None of these is a recommendation, because the right answer depends on what the summary is for. What all three share is that they are decisions somebody has to make, and the literature the structure came from does not signal that there is a decision.

Space-Saving: error against the number of shards, k = 32Worst error over the 8 heaviest keys of 40,000 arrivals of stationary Zipf — the control, where a window is a smaller sample, for three summaries answering the same question. The first holds 32 counters. The second holds as many counters as all the shards hold between them, which is the only honest thing to compare a merge against. The third is the merge itself. From 2 shards to 32, the merge's worst error goes 656 to 318 while the state the shards hold between them rises 16-fold. More machines is more state and a worse answer.656m = 2conc 0.51314m = 4conc 0.27313m = 8conc 0.14313m = 16conc 0.08318m = 32conc 0.04shardsworst error over the top keysone summary, k = 32one summary, k = k·mthe merge of mSpace-Saving · stationary Zipf · 40,000 arrivalspartition: round
Fig. 6 Space-Saving under the partition that suits it, across the shard count. Even here the merge does not converge on the matched summary as machines are added — the gap is a floor charged per absent key, and adding machines adds keys to be absent from.

The rule that falls out

There is a usable statement here, and it is small enough to remember.

If every heavy key is on every shard, prefer the structure whose merge charges per absent key — Space-Saving. There are no absent keys to charge for, and the merge is nearly free.

If every heavy key is on one shard, prefer the structure whose merge takes one cut — Misra-Gries. The summed table is wide and shallow below the top, so the cut is small.

Both halves are measured rather than argued, and both are measured on one stream, which is the limit worth stating plainly. A flatter distribution than this Zipf would crowd both shoulders and narrow the gap; a steeper one would make the heavy keys survive any partition and narrow it from the other side. What the exponent does not change is the direction: the mechanisms are structural, and each structure’s merge is sensitive to precisely the thing the other’s is not.

Three sizes, and the merge is two of themA merge of 8 summaries of 32 counters is 16,384 bits while it is being computed and 2,048 bits afterwards, and neither number is the one usually quoted. The first is what the deployment costs — every shard holds its summary at the same time — and it is exactly what a single summary of 256 counters would cost. The second is what the answer costs, and it is 8× smaller. A structure described as "32 counters" is 8 times that during the operation it exists for.held by the 8 shards, all at once16,384 bitsone summary of k·m = 25616,384 bitsthe merged answer2,048 bitsone summary of k = 322,048 bitsstate, in bitsMisra-Gries · 8 shards of 32 · hashed8× between the deployment and the answer
Fig. 7 And the accounting is unchanged by any of it. Whichever partition and whichever structure, the deployment holds what a matched single summary would hold and the answer holds a fraction of it — the choice of partition moves the accuracy and not the arithmetic.

The measurement that would settle it

There is an experiment this essay does not run, and naming it is more useful than pretending the question is closed.

The concentration measured above is a property of the partition. What decides the Misra-Gries merge is a property of the summed table, and the two are related but not the same: a partition can concentrate keys without widening the summed table, if the shards happen to hold overlapping sets for another reason. A structure-free predictor for the second quantity — something like the expected number of distinct keys across mm tables of kk, as a function of the distribution and the partition — would let both structures’ merged errors be predicted before either is run.

It is computable from the stream and the partition alone, it would take one figure, and it is not here. It is written down in the site’s own plan as work rather than left implied, because a phase that measures two mechanisms and unifies neither has found a pattern rather than a rule.

Misra-Gries: keys whose count depends on the order the shards were combinedThree merge orders over the same 40,000 arrivals: folded in one at a time, combined pairwise in a tree, and folded in with the shards reversed. All three are the same associative-looking operation applied to the same summaries, and the tables they produce are not the same table. At 32 shards, 52 keys carry a different count depending only on the shape, the largest difference being 298 arrivals. Nothing in a deployment fixes the shape: a coordinator folds, a parallel reduction trees, and a retry changes the order.0231436845165232shards mergedkeys whose count movedthree orders:folded in one at a timecombined pairwise, in a treefolded in, last shard firstMisra-Gries, k = 32 · hashedworst gap 298 arrivals
Fig. 8 And a reminder that the partition is not the only unstated parameter. These counts also depend on the order the shards were combined in, which no deployment fixes and which the next essay measures.

Half of the missing predictor is a closed form

The experiment named above wants a structure-free way to predict the summed table’s width, and one of the three partitions gives it up without any experiment at all.

Under hashing the shards’ key sets are disjoint by construction. Every arrival of a key goes to one machine, so no key can appear in two tables, and the summed table has exactly mkmk entries when the shards fill — 256 at eight shards of thirty-two, with no overlap to estimate. The global top thirty-two are spread across the eight machines by the same hash, so each of them is retained by its own shard, and the summed table’s ranking below them is the global ranking.

That makes the cut computable: it is f(k+1)f(k+1), the global count of the thirty-third heaviest key, and nothing about the structure enters.

Evaluate it on this stream. The heaviest key holds 16% of the arrivals, which for a Zipf exponent of 1.1 fixes the normaliser, so key ii holds 0.16i1.10.16\,i^{-1.1} of them. At i=33i = 33 that is 0.34%, and 0.34% of forty thousand arrivals is about 136.

The measured Misra-Gries error under hashing is 152. A prediction from three numbers — the exponent, the head’s share, and k+1k+1 — landing within a tenth of a measurement taken from eight merged tables.

So half the predictor exists and it is the half a deployment is most likely to need, since hashing by key is what the essay names as by far the most common arrangement. A system that hashes can compute its merged Misra-Gries error from its own traffic profile before allocating a counter, and the same three numbers say what a wider table would buy: raising kk to 64 moves the cut to f(65)f(65), which on this exponent is 65 rather than 136, so doubling the counters roughly halves the error.

The other half genuinely needs the measurement, and it is worth being clear about why rather than filing it as undone. Under round-robin the shards hold overlapping sets, so the summed table’s width is somewhere between kk and mkmk and depends on how far down each shard’s own top-kk the global heavy keys sit. That is a question about mm independent samples of the same distribution agreeing on their top kk — answerable in principle, and not by the same one-line argument, because the disjointness that made the hashed case trivial is exactly what is absent.

Which sharpens the rule this page ends on into something with an arithmetic behind it. If every heavy key is on one shard, the merged cut is a global frequency and can be looked up. If every heavy key is on every shard, the cut is a property of how the samples overlap and has to be measured — and the measurement above says it is six times larger, which is two structures that are one’s bijection failing at exactly the point where the two merge rules stop discarding the same information.

It also explains why the concentration statistic orders the Space-Saving numbers and not the Misra-Gries ones, as this page reports without accounting for it. Concentration is about where a key’s mass sits, which is what the state a merge is standing in for’s per-absent-key floor charges for. The cut is about the summed table’s thirty-third entry, which concentration says nothing about — two quantities, one measured and one not, and the closed form above is the second one in the case where it has one.

What is not established

Nothing here is a bound. Both structures’ guarantees hold in every cell of that table — the merged Space-Saving counters are never below the truth, and the site’s gate checks it on every build. What moves by a factor of six is the realised error, which no bound in the field is about.

The concentration is a predictor and not a formula. It orders the partitions correctly for Space-Saving and it does not, on its own, predict the Misra-Gries numbers, which depend on the shape of the summed table below the cut rather than on where a key’s mass sits. Two quantities are doing the work and only one of them has been isolated.

And a partition is not only an accuracy decision. Hashing by key gives every shard a self-contained answer about the keys it holds, which is often the thing actually wanted — and a merged estimate is then a secondary product rather than the point. This essay measures one axis of a choice that has several.

What this makes readable

Essays that name this one as a prerequisite.

Named alongside this one

Essays reaching for the same objects. Nobody chose these; they are what the concept index makes visible.

What links here

Every essay whose body links to this one.

The objects this essay names

Each one links to every other essay that touches it.

GuaranteeHeavy hitterMergeable summaryMisra–GriesPartitionShardSpace-savingState bitsStreaming modelZipf distribution