Distributed Randomized Multiset (insert, remove, getRandom)

0) Problem Restatement

First the classic single-machine problem (LeetCode 381): a multiset (duplicates allowed) with:

• insert(val), remove(val) (one occurrence), and
• getRandom(): return an element where each occurrence is equally likely (so a value present 3 times is 3x as likely),
• all in O(1) average time.

Then LinkedIn's follow-up: scale it across multiple servers with the same operations and the same randomness guarantee.

1) Single Machine: O(1) Design

• An array items holding every occurrence (so random choice = pick a random index).
• A map val → set of indexes where that value sits in the array.
• Remove in O(1): take any index i of val, move the last array element into position i, update that element's index set, and pop the last slot.

import random
from collections import defaultdict

class RandomizedMultiset:
    def __init__(self):
        self.items, self.pos = [], defaultdict(set)
    def insert(self, val):
        self.pos[val].add(len(self.items)); self.items.append(val)
    def remove(self, val):
        if not self.pos[val]: return False
        i = self.pos[val].pop(); last = self.items[-1]
        self.items[i] = last
        self.pos[last].add(i); self.pos[last].discard(len(self.items) - 1)
        self.items.pop()
        if not self.pos[val]: del self.pos[val]
        return True
    def get_random(self):
        return random.choice(self.items)
    def __len__(self):
        return len(self.items)

2) Distributing It

• P(item) = (size_s / total) × (1 / size_s) = 1 / total. Uniform.

flowchart LR
    C["Client"] --> R["Router"]
    R -->|"insert/remove: hash(val)"| S1["Shard 1 - count 40M"]
    R --> S2["Shard 2 - count 25M"]
    R --> S3["Shard 3 - count 35M"]
    R -->|"getRandom: pick shard by size, then local random"| S2
    S1 -->|"size updates"| SZ[("Shard size table")]
    S2 --> SZ
    S3 --> SZ
    SZ --> R

2.1 Knowing shard sizes

• Each shard keeps its count and reports it to the router (or a small size table).
• getRandom: draw r uniformly in [0, total), then walk the prefix sums of shard sizes to find the shard (with a Fenwick tree when there are many shards), and call shard.get_random().

2.2 Consistency trade-off

• Sizes change constantly. Exact uniformity would need a consistent snapshot of all sizes during getRandom, which is expensive.
• Practical choice: refresh sizes every few milliseconds, or send deltas. The small staleness causes a tiny, bounded bias.
• If exactness is required: rejection sampling. Pick a shard using an upper bound on each shard's size (say, its max possible), then have the shard accept with probability actual_size / bound (otherwise retry). This is exactly uniform with stale bounds, at the cost of some retries.
• An empty shard chosen because of stale info simply triggers a retry.

3) Other Details

• Hot values: a single value with a huge count stays on one shard. That's fine for correctness. For load, the shard can be replicated for reads (getRandom).
• Rebalancing: consistent hashing with virtual nodes. Moving a range moves its occurrences, and sizes are updated.
• Replication for fault tolerance: each shard has replicas, and writes go to the leader.

4) Wrap-Up

On one machine, use an array of all occurrences plus a map from value to index set, removing by swapping with the last element, so all three operations are O(1). Across servers, hash-partition values so insert and remove hit one shard, and make getRandom uniform by choosing a shard with probability proportional to its size (prefix sums over reported counts), then sampling inside it. Use rejection sampling with size upper bounds when stale counts must not bias the result.