Step-by-Step Teaching

Step 1: Understanding the Problem Domain

What is a Non-Rollup Metric?

First, let's establish why this problem exists at all.

Metrics in data pipelines fall into two categories:

The core challenge:

Daily Active Users:
  Monday:    User A, User B, User C  → 3
  Wednesday: User A, User D          → 2

Naive weekly sum = 3 + 2 = 5  ← WRONG
Correct weekly count = 4 (A, B, C, D)  ← REQUIRES tracking all identities simultaneously

Key insight: Non-rollup metrics require holding every unique identifier in memory across the entire aggregation window simultaneously. At quarterly scale, that means 3.6 billion UUIDs at once.

---

Step 2: Why Standard Solutions Fail

Before understanding the solution, you must understand why each existing approach breaks.

Approach 1: RoaringBitmap (32-bit)

What it does: Encodes set membership as compressed bits. Uses bitwise OR for fast set unions.

Why it fails at scale:

RoaringBitmap works on 32-bit integers
UUIDs are 128-bit strings
→ Requires a Global Dictionary: UUID → unique integer rank

Problem cascade:
  ├── Dictionary = single point of failure (corruption = all jobs invalid)
  ├── Concurrent jobs generate overlapping rank assignments
  └── Backfill jobs MUST run sequentially
      → 2-year backfill takes 20+ days

Approach 2: Roaring64Bitmap (64-bit)

What it does: Hashes UUIDs directly to 64-bit integers. No dictionary needed.

Why it fails at scale:

JVM Rule: No single array (byte[]) can exceed ~2 GB

Hive/Spark UDAF contract:
  → Aggregation state MUST be returned as a single byte[]
  → At ~179 million unique UUIDs, the serialized bitmap hits 2 GB
  → At 3.6 billion UUIDs: DETERMINISTIC OutOfMemoryError on every run

This is a structural JVM constraint. No configuration, no memory tuning, no framework upgrade fixes it.

Approach 3: HyperLogLog (Approximate)

What it does: Estimates distinct counts using probabilistic data structures.

Why it fails here:

Error rate: 1–5%
MAU underpins financial reporting
→ Approximate answers are unacceptable

Summary of Failures

Bitmap-32    → Sequential execution, dictionary corruption risk
Bitmap-64    → Hard OOM at 179M identifiers (JVM 2 GB limit)
HyperLogLog  → Inexact (1–5% error)

The requirement: Exact + Dictionary-free + Never materializes into a single byte[]

---

Step 3: The Core Insight — "More Boxes"

"Sometimes the solution isn't found by thinking outside the box — it's found by utilizing more boxes."

The JVM refuses to allocate one array larger than 2 GB.

The solution: Never create one large array. Create many small ones.

Instead of:
  ONE Roaring64Bitmap → ONE byte[] → CRASHES at 2 GB

Use:
  Map<Integer, Roaring64Bitmap>
  ├── Chunk 0    → small byte[]  ✅
  ├── Chunk 1    → small byte[]  ✅
  ├── Chunk 2    → small byte[]  ✅
  └── ... 65,536 chunks total

Because chunks cover non-overlapping (disjoint) segments of the value space:

Total distinct count = sum of each chunk's cardinality

No data is double-counted. No data is lost.

---

Step 4: The Chunked Aggregation Buffer Strategy — Mechanics

How Routing Works

Every incoming UUID goes through this pipeline:

UUID (128-bit string)
    ↓
xxHash64()  →  64-bit integer hash
    ↓
Top 16 bits extracted:  chunkId = (int)(hash >>> 48)
    ↓
Stored in:  map.get(chunkId).add(fullHash)

Visual example:

hash = 1010101100110101 | 0101010101010101010101010101010101010101010101010
        ←  top 16 bits →   ←           remaining 48 bits                →
        chunkId = 43829         stored inside chunk 43829's bitmap

Why xxHash64?

Two specific reasons:

1. Built into Hive — no external dependencies
2. Avalanche property — a small change in input causes completely different output, guaranteeing uniform distribution across all 65,536 chunks

---

Step 5: Why 16 Bits Specifically?

This is a deliberate engineering tradeoff. Let's examine the math:

16 bits → 2^16 = 65,536 chunks

At 3.6 billion identifiers:
  Average per chunk = 3,600,000,000 / 65,536 ≈ 55,000 values

OOM threshold per chunk = ~179 million values
Safety margin = 179,000,000 / 55,000 ≈ 3,254×

Why not fewer bits?

8 bits → 256 chunks
  Average per chunk = 3.6B / 256 ≈ 14 million values
  Worst-case skew could approach OOM threshold
  → Insufficient safety margin

Why not more bits?

24 bits → 16,777,216 chunks
  Most chunks nearly empty
  → Map overhead and garbage collection pressure become the bottleneck
  → Diminishing returns

16 bits is the Goldilocks choice: enough chunks to guarantee safety, few enough to avoid overhead.

The Chernoff bound (a statistical tool for bounding worst-case deviation) confirms: the probability of any single chunk reaching 179 million values is effectively zero at production scale.

---

Step 6: The Four Engineering Mechanisms

The strategy has four interlocking components:

Mechanism 1: Streaming Serialization

Traditional approach:
  Serialize entire bitmap → one giant byte[] → OOM

Chunked approach:
  For each chunk:
    serialize chunk → small byte[]
    write to output stream
    release memory
  → Peak memory = largest single chunk, NOT total

Mechanism 2: Self-Describing Magic Number

Problem: During migration, some pipeline data is old format (Bitmap-64), some is new (chunked).

Solution: Prefix every chunked partial result with 0xDEADBEEF

Read first 4 bytes of partial result:
  ├── == 0xDEADBEEF  →  Route to CHUNKED merge path
  └── != 0xDEADBEEF  →  Route to LEGACY deserialization path

Why 0xDEADBEEF specifically?

Roaring64Bitmap's internal cookie header can NEVER equal 0xDEADBEEF
→ False positives are structurally impossible
→ No historical recompute needed during migration

Mechanism 3: Cardinality-Only Final Output

At the final reduce stage:
  DON'T serialize bitmaps
  DO sum cardinalities across all chunks
  Return: single 8-byte long

Result: Final output is ALWAYS 8 bytes
  → Regardless of whether you counted 1 user or 3.6 billion
  → Zero OOM risk at reduce stage

Mechanism 4: Concurrent Execution (Bonus Unlock)

Bitmap-32 (dictionary-based):
  Job A assigns rank 1 to User X
  Job B also assigns rank 1 to User Y  ← COLLISION
  → Jobs must run sequentially

Chunked Bitmap-64 (hash-based):
  No shared state between jobs
  → Multiple backfill windows run in parallel
  → Mean 3.8× speedup from 3 concurrent windows

---

Step 7: How It Fits Into the Hive UDAF Lifecycle

A UDAF (User-Defined Aggregate Function) has four phases:

PARTIAL1  (per-mapper)
  Each mapper processes its local data
  → Serialize only local chunks (fraction of full value space)
  → Small, safe byte[]

PARTIAL2  (combiner/merge)
  Combine partial results from multiple mappers
  → Magic number routes to correct merge path
  → Stream-merge one chunk at a time, release after each

FINAL  (reducer)
  Merge all partial results
  → Same streaming approach

terminate()
  Return final answer
  → Sum chunk cardinalities
  → Return 8-byte long

---

Step 8: Results and Validation

Correctness Validation

Test: Compare chunked UDAF vs raw COUNT(DISTINCT identifier)
Scope: 10 metric families × 3 time windows (daily, monthly, quarterly) = 30 runs
Result: EXACT match across all 30 runs, zero deviation

Performance Results

Scalability Improvement

Old safe threshold:  ~179 million unique identifiers
New safe threshold:  ~11.7 trillion unique identifiers

Improvement: 65,000×

---

Step 9: Generalizing the Pattern

This solution is not Uber-specific. Apply this mental model whenever you see:

Condition 1: JVM-based distributed pipeline (Hive, Spark, etc.)
Condition 2: Aggregation state must be returned as single byte[]
Condition 3: High-cardinality exact distinct counting required
→ You are one cardinality threshold away from this exact failure

The generalizable pattern:

When a single object exceeds a hard size constraint:
  1. Partition the value space into disjoint chunks (use top N bits of hash)
  2. Process each chunk independently
  3. Combine results by summing (works because chunks are non-overlapping)
  4. Use a sentinel value for backward-compatible migration
  5. Return only the final scalar, not the intermediate structure

This applies beyond UUIDs to: device IDs, session tokens, geohash-encoded locations — any 64-bit hashed domain.

---

Concept Map Summary

PROBLEM
  Non-rollup metrics require exact distinct counts
  → At scale: 3.6 billion UUIDs simultaneously
  → JVM 2 GB array limit causes deterministic OOM

FAILED APPROACHES
  Bitmap-32    → Dictionary = sequential jobs, corruption risk
  Bitmap-64    → Monolithic serialization hits 2 GB at 179M IDs
  HyperLogLog  → Approximate (unacceptable for financial metrics)

SOLUTION: Chunked Aggregation Buffer
  ├── Hash UUID → 64-bit integer (xxHash64)
  ├── Route by top 16 bits → 65,536 disjoint chunks
  ├── Serialize each chunk independently (streaming)
  ├── Magic number 0xDEADBEEF → backward-compatible routing
  └── terminate() returns 8-byte long (sum of chunk cardinalities)

RESULTS
  ├── 65,000× scalability improvement
  ├── 65% faster backfills (up to 94%)
  ├── Zero OOM failures
  └── Exact accuracy maintained