Distributed Search & Analytics Engines: Inverted Indexes, Columnar Segments & Distributed Aggregations

Full-text search and analytical aggregations represent the two most demanding read paradigms in modern systems engineering. When searching across millions of unstructured documents (e-commerce catalogs, log streams, code repositories) or aggregating metrics across billions of events (financial transactions, infrastructure telemetry), standard relational B-tree indexes collapse under disk I/O bottlenecks and high scan latencies.

This lesson explores the internal architectures of distributed search and analytical engines:

1. Inverted Indexes & Lucene: Term dictionaries, postings lists, skip lists, and Roaring Bitmaps.
2. Columnar Storage & Vectorized Execution: Parquet and ClickHouse physical layouts, dictionary encoding, and SIMD processing.
3. Distributed Scatter-Gather Query Execution: Two-phase query-then-fetch coordination, BM25 scoring, and high-cardinality aggregation handling.

1. Context & Architecture Setting

Consider an enterprise logging and metrics platform ingesting $100,000$ log lines and telemetry events per second ($8.64\text{ billion events/day}$).

+-------------------------------------------------------------------------------+
|                       ROW-STORE VS COLUMN-STORE VS SEARCH                     |
+-------------------------------------------------------------------------------+
| Paradigm                | Physical Layout           | Optimized Operations    |
+-------------------------+---------------------------+-------------------------+
| Row-Oriented OLTP       | [R1: a,b,c] [R2: a,b,c]   | Single-row Point Read,  |
| (PostgreSQL, MySQL)     | Contiguous row blocks     | ACID multi-row Update   |
+-------------------------+---------------------------+-------------------------+
| Inverted Index Search   | Term -> [Doc1, Doc4, ...] | Multi-word boolean AND, |
| (Lucene, Elasticsearch) | Term Dictionary + Postings| Full-text relevance     |
+-------------------------+---------------------------+-------------------------+
| Columnar OLAP           | ColA: [1,2] ColB: [x,y]   | Aggregations (SUM, AVG) |
| (ClickHouse, Parquet)   | Contiguous column arrays  | Bulk column scans       |
+-------------------------+---------------------------+-------------------------+

If an engineer needs to find: WHERE message MATCHES "connection refused" AND latency > 500:

• A traditional row-store must either scan millions of full rows or rely on bloated composite B-trees.
• An inverted index uses the term dictionary to locate the postings lists for "connection" and "refused", performs a SIMD bitset intersection in microseconds, and returns candidate document IDs.
• A columnar engine reads only the contiguous compressed latency and message arrays, skipping 90% of disk bytes.

2. Architectural Mental Model: The Inverted Index

SYSTEM ARCHITECTURE BLUEPRINTDISTRIBUTED SYSTEMS

⤢Expand

ARCHITECTURAL FLOW · STEP 1 OF 3

1. Inverted Index and Postings Lists & Term Dictionary: Finite State Transducer (FST)

Establishes the ingress and structural boundary across Inverted Index and Postings Lists & Term Dictionary: Finite State Transducer (FST): Compact memory prefix/suffix graph (FST); In-memory term lookup in O(term_length); Term: "consensus" → Points to Postings Offset.

Architectural Invariant: Zero lock overhead during concurrent queries.

ARCHITECTURAL FLOW · STEP 2 OF 3

2. Columnar Storage and Vectorization & Row-Store vs Column-Store Layout

Coordinates core data-plane execution and state transitions through Columnar Storage and Vectorization & Row-Store vs Column-Store Layout: Row-Store (Postgres/MySQL); [Row1: id,ts,lat] [Row2: id,ts,lat] [Row3: id,ts,lat]; Column-Store (ClickHouse/Parquet).

Architectural Invariant: Zero virtual method dispatch per row.

ARCHITECTURAL FLOW · STEP 3 OF 3

3. Distributed Scatter-Gather Engine & Two-Phase Query Coordinator

Enforces durability, state convergence, and output verification across Distributed Scatter-Gather Engine & Two-Phase Query Coordinator: 1. Query Phase (Scatter): Broadcasts filter/ranking to all target shards Shards compute local Top-K + BM25 score; 2. Fetch Phase (Gather): Coordinator merges K*N results via Min-Heap Requests full _source docs only for top K; Slashes network bandwidth by 95%.

Architectural Invariant: Adaptive replica selection (lowest queue latency).

Document 1: "Distributed consensus via Raft"
Document 2: "Consensus protocols in distributed systems"
Document 3: "Raft consensus log replication"

+-------------------------------------------------------------------------------+
| TERM DICTIONARY (FST in RAM) | POSTINGS LIST (Compressed on Disk)             |
+------------------------------+------------------------------------------------+
| "consensus"                  | [Doc 1, Doc 2, Doc 3]                          |
| "distributed"                | [Doc 1, Doc 2]                                 |
| "log"                        | [Doc 3]                                        |
| "protocols"                  | [Doc 2]                                        |
| "raft"                       | [Doc 1, Doc 3]                                 |
| "replication"                | [Doc 3]                                        |
| "systems"                    | [Doc 2]                                        |
+------------------------------+------------------------------------------------+

Boolean Query: "distributed" AND "raft"
1. Postings("distributed") = [1, 2]
2. Postings("raft")        = [1, 3]
3. Intersection            = [1] -> Returns Document 1

Lucene Segment Lifecycle & Immutability

Apache Lucene represents an index as a collection of immutable segments:

• In-Memory Buffer: Ingested documents are tokenized and written to a RAM buffer (IndexingMemoryBuffer).
• Segment Flush: Periodically or when RAM fills, the buffer flushes to disk as a new immutable segment containing its own FST term dictionary, postings lists, stored fields, and doc values.
• Deletions via Bitsets: Because segments are immutable, deleting a document does not rewrite disk files. Instead, a .del bitset marks the document as deleted.
• Tiered Segment Merge: A background merge thread continuously merges small segments into larger ones, purging deleted tombstones and keeping total segment counts manageable.

3. Columnar Storage Engine Internals: Parquet & ClickHouse

In columnar formats, data is organized physically by column rather than row:

Table Schema: (timestamp UInt32, status UInt16, latency Float32, user_id String)

Disk File Chunk:
[Column: timestamp] -> [1700000000, 1700000001, 1700000002, 1700000003, ...]
[Column: status]    -> [200, 200, 200, 500, 200, 200, 500, 500, ...] (RLE compressed!)
[Column: latency]   -> [12.4, 15.2, 11.9, 850.2, 14.1, 12.8, 920.4, ...]
[Column: user_id]   -> Dictionary Encoded: Dict=[alice, bob] Values=[0, 1, 0, ...]

Compression Techniques

1. Dictionary Encoding: Replaces repetitive high-entropy strings or identifiers with compact 8-bit or 16-bit integers.
2. Run-Length Encoding (RLE): Compresses runs of identical values (e.g., HTTP status 200 repeated 1,000 times) into a single pair: (200, 1000).
3. Frame of Reference (FoR) / Delta Compression: Monotonically increasing numbers (such as timestamps or sorted document IDs) store only differences: [1000, 1005, 1008] -> [1000, +5, +3].
4. SIMD Vectorization: CPUs with AVX-512 or ARM Neon can process a vector of 16 values in a single instruction cycle, evaluating latency > 500 across millions of records in single-digit milliseconds.

4. Distributed Scatter-Gather Coordination

In distributed search engines, tables (indexes) are partitioned into $N$ primary shards distributed across cluster nodes.

                  [Client Search Query]
                           |
                           v
              [Coordinator Node / Router]
                           |
            +--------------+--------------+
            | (Phase 1: Scatter Query)   |
            v                             v
     [Shard 1 Node]                [Shard 2 Node]
     - Evaluates BM25              - Evaluates BM25
     - Sorts local top-10          - Sorts local top-10
     - Returns [DocIDs, Scores]    - Returns [DocIDs, Scores]
            |                             |
            +--------------+--------------+
                           |
                           v
              [Coordinator Node Merges]
              - Min-Heap merges 20 candidates into global top-10
                           |
            +--------------+--------------+
            | (Phase 2: Targeted Fetch)  |
            v                             v
     [Shard 1 Node]                [Shard 2 Node]
     - Fetches _source JSON for    - Fetches _source JSON for
       winning DocIDs only           winning DocIDs only
            |                             |
            +--------------+--------------+
                           |
                           v
              [Returns Final Top-10 to Client]

5. Mathematical Model: BM25 Scoring & Memory Sizing

The Okapi BM25 Scoring Formula

Relevance scoring in modern search engines evaluates the relevance of document $D$ to query $Q=\{q_1,\dots ,q_n\}$:

Where:

• $f(q_i,D)$ = Term frequency of word $q_i$ in document $D$.
• $|D|$ and $\text{avgdl}$ = Length of document $D$ and average document length across the entire index.
• $k_1$ (typically $1.2$) = Calibrates term frequency saturation (preventing a document with 100 repetitions from dominating).
• $b$ (typically $0.75$) = Length normalization factor.
• $\text{IDF}(q_i)=\ln \left(\frac{N-n(q_i)+0.5}{n(q_i)+0.5}+1\right)$, penalizing common stop-words.

Deep Pagination Heap Footprint

If a client requests page 10,000 with page size 10 across $S$ shards: $${\text{Memory}}_{coord}=S\times (\text{from}+\text{size})\times {\text{Size}}_{\text{entry}}$$ For $S=50$, $\text{from}=100,000$, $\text{size}=10$, ${\text{Size}}_{\text{entry}}\approx 64\text{ bytes}$: $${\text{Memory}}_{coord}=50\times 100,010\times 64\text{ bytes}\approx 320\text{ MB JVM Heap per Query!}$$

With 20 concurrent deep pagination queries, coordinator nodes immediately crash with java.lang.OutOfMemoryError: Java heap space.

6. Practical Engineering Simulation: Inverted Index & Scatter-Gather Coordinator

This self-contained TypeScript simulation implements an inverted index with tokenization, postings lists, BM25 scoring, and a distributed two-phase scatter-gather coordinator.

/**
 * Systems Notebook - Distributed Inverted Index & Scatter-Gather Coordinator
 */

interface DocumentRecord {
  id: string;
  title: string;
  content: string;
}

interface ScoredDoc {
  docId: string;
  score: number;
  shardId: string;
}

class InvertedIndexShard {
  public readonly shardId: string;
  private documents: Map<string, DocumentRecord> = new Map();
  private postings: Map<string, string[]> = new Map(); // term -> docIds
  private docLengths: Map<string, number> = new Map();
  private totalLength: number = 0;

  constructor(shardId: string) {
    this.shardId = shardId;
  }

  private tokenize(text: string): string[] {
    return text
      .toLowerCase()
      .replace(/[^a-z0-9\s]/g, '')
      .split(/\s+/)
      .filter((w) => w.length > 2);
  }

  public indexDocument(doc: DocumentRecord): void {
    this.documents.set(doc.id, doc);
    const tokens = this.tokenize(doc.title + ' ' + doc.content);
    this.docLengths.set(doc.id, tokens.length);
    this.totalLength += tokens.length;

    const uniqueTokens = new Set(tokens);
    for (const token of uniqueTokens) {
      if (!this.postings.has(token)) {
        this.postings.set(token, []);
      }
      this.postings.get(token)!.push(doc.id);
    }
  }

  /**
   * Phase 1: Query Phase (Returns DocID + BM25 Score)
   */
  public queryPhase(queryTerms: string[], topK: number): ScoredDoc[] {
    const candidateDocScores: Map<string, number> = new Map();
    const avgdl = this.documents.size > 0 ? this.totalLength / this.documents.size : 1;
    const N = this.documents.size;

    for (const term of queryTerms) {
      const docIds = this.postings.get(term) ?? [];
      const n_q = docIds.length;
      if (n_q === 0) continue;

      // Simplified IDF
      const idf = Math.log((N - n_q + 0.5) / (n_q + 0.5) + 1);

      for (const docId of docIds) {
        const doc = this.documents.get(docId)!;
        const tokens = this.tokenize(doc.title + ' ' + doc.content);
        const tf = tokens.filter((t) => t === term).length;
        const dl = this.docLengths.get(docId) ?? avgdl;

        // BM25 term calculation
        const k1 = 1.2;
        const b = 0.75;
        const score = idf * ((tf * (k1 + 1)) / (tf + k1 * (1 - b + b * (dl / avgdl))));

        const current = candidateDocScores.get(docId) ?? 0;
        candidateDocScores.set(docId, current + score);
      }
    }

    const scoredList: ScoredDoc[] = Array.from(candidateDocScores.entries()).map(
      ([docId, score]) => ({ docId, score, shardId: this.shardId })
    );

    // Sort descending by score and return topK
    scoredList.sort((a, b) => b.score - a.score);
    return scoredList.slice(0, topK);
  }

  /**
   * Phase 2: Fetch Phase (Fetches full document source)
   */
  public fetchDocument(docId: string): DocumentRecord | undefined {
    return this.documents.get(docId);
  }
}

class DistributedSearchCoordinator {
  private shards: InvertedIndexShard[] = [];

  constructor(shards: InvertedIndexShard[]) {
    this.shards = shards;
  }

  public search(queryString: string, topK: number): DocumentRecord[] {
    const terms = queryString.toLowerCase().split(/\s+/).filter((t) => t.length > 2);

    console.log(`[Coordinator] Phase 1: Scattering query [${terms.join(', ')}] across ${this.shards.length} shards...`);

    // Phase 1: Scatter query
    const allCandidates: ScoredDoc[] = [];
    for (const shard of this.shards) {
      const shardResults = shard.queryPhase(terms, topK);
      allCandidates.push(...shardResults);
    }

    // Merge and extract top-K global winners
    allCandidates.sort((a, b) => b.score - a.score);
    const winningCandidates = allCandidates.slice(0, topK);

    console.log(`[Coordinator] Merged ${allCandidates.length} candidate scores. Selected top ${winningCandidates.length} winners.`);

    // Phase 2: Targeted Fetch
    const finalDocuments: DocumentRecord[] = [];
    for (const winner of winningCandidates) {
      const shard = this.shards.find((s) => s.shardId === winner.shardId);
      const fullDoc = shard?.fetchDocument(winner.docId);
      if (fullDoc) {
        finalDocuments.push(fullDoc);
      }
    }

    return finalDocuments;
  }
}

// -------------------------------------------------------------
// Demonstration Execution
// -------------------------------------------------------------
function runSearchDemo(): void {
  const shard1 = new InvertedIndexShard('shard-us-east-1');
  const shard2 = new InvertedIndexShard('shard-us-west-1');

  shard1.indexDocument({
    id: 'doc-1',
    title: 'Raft Consensus Protocol Internals',
    content: 'Raft ensures distributed state machine consistency across cluster quorums.'
  });
  shard1.indexDocument({
    id: 'doc-2',
    title: 'PostgreSQL Relational Storage Engine',
    content: 'B-tree index and write ahead logging in relational systems.'
  });

  shard2.indexDocument({
    id: 'doc-3',
    title: 'Distributed Systems & Paxos Algorithms',
    content: 'Distributed consensus algorithms like Paxos and Raft prevent split-brain states.'
  });

  const coordinator = new DistributedSearchCoordinator([shard1, shard2]);
  const results = coordinator.search('distributed consensus raft', 2);

  console.log('\n=== Final Search Results ===');
  results.forEach((doc, idx) => {
    console.log(`${idx + 1}. [${doc.id}] ${doc.title}`);
  });
}

runSearchDemo();

7. Observability & SRE Metrics

+-------------------------------------------------------------------------------+
|                       DISTRIBUTED SEARCH SRE DASHBOARD                        |
+-------------------------------------------------------------------------------+
| Metric Indicator       | Prometheus / JMX Metric| SRE Alert Threshold         |
+------------------------+------------------------+-----------------------------+
| Query Phase Latency    | search_query_time_ms   | Alert if p99 > 150ms;       |
|                        |                        | slow shard dragging cluster |
| Fetch Phase Latency    | search_fetch_time_ms   | Alert if p99 > 50ms;        |
|                        |                        | disk seek bottleneck        |
| JVM Garbage Collection | jvm_gc_pause_seconds   | Alert if pause > 1000ms;    |
|                        |                        | deep pagination heap leak   |
| Segment Merge Throttle | segment_merge_throttle_| Alert if throttled > 10s;   |
| Time                   | seconds                | disk write IOPS saturated   |
+------------------------+------------------------+-----------------------------+

8. Comprehensive Architectural Summary

Distributed search and analytical engines succeed by aligning data representations with physical computer hardware:

• Inverted Indexes (Lucene) accelerate full-text and multi-word boolean searches by pre-sorting postings lists and compressing document arrays via Roaring Bitmaps.
• Columnar Engines (ClickHouse, Parquet) accelerate massive analytical queries by loading only query-relevant column bytes, compressing redundant runs, and executing computations over vectorized SIMD CPU lanes.
• Two-Phase Scatter-Gather Coordination minimizes network transit by separating lightweight ranking candidates (docIDs + scores) from heavyweight document payloads.

9. Design Alternatives & Search Analytics Trade-Off Matrix

10. Failure Modes, Query Storms & Recovery

• Deep Pagination Outage (from=100000, size=50): In scatter-gather search, a deep pagination query requires every shard to sort and return 100,050 documents to the coordinator node; merging 100k items across 50 shards exhausts heap memory. Mitigate using cursor-based search (search_after in Elasticsearch).
• Wildcard Prefix Explosion (query): Leading wildcards force the engine to scan the entire Lucene term dictionary rather than using FST prefix lookups. Mitigate with N-gram tokenizers or edge-ngram indexing.
• Circuit Breaking Heap Protection: Configure parent heap memory circuit breakers (indices.breaker.total.use_real_memory) to reject expensive queries before JVM OutOfMemoryErrors occur.

11. Guided Design Exercises

Exercise 1: Lucene BM25 Relevance Calculation & Distributed IDF Skew

Consider an index shard containing $N=10,000$ documents with average document length $\text{avgdl}=200\text{ tokens}$, BM25 parameters $k_1=1.2$, and $b=0.75$. A user searches for the two-term query $Q=\{\text{"raft"},\text{"consensus"}\}$:

• Term "raft" appears in $n(\text{"raft"})=100$ documents; Term "consensus" appears in $n(\text{"consensus"})=1,000$ documents.
• Document A has length $|D_A|=100\text{ tokens}$ with $f(\text{"raft"},D_A)=3$ and $f(\text{"consensus"},D_A)=1$.
• Document B has length $|D_B|=600\text{ tokens}$ with $f(\text{"raft"},D_B)=6$ and $f(\text{"consensus"},D_B)=2$.

1. Compute $\text{IDF}(\text{"raft"})$, $\text{IDF}(\text{"consensus"})$, and the exact BM25 scores $\text{Score}(D_A,Q)$ and $\text{Score}(D_B,Q)$ to determine which document ranks higher.
2. Explain Distributed IDF Skew: how can an unevenly hashed 20-shard cluster produce inaccurate BM25 rankings using local shard statistics (search_type=query_then_fetch), and when is dfs_query_then_fetch required?

Exercise 2: Row-Store vs. ClickHouse Columnar Scan & SIMD Bandwidth Derivation

An observability table stores $1\text{ billion}$ ($10^9$) HTTP request rows. Each row is $512\text{ bytes}$ wide across 28 columns (including URL and headers), while status (UInt16, $2\text{ bytes}$ raw, compressed $8\times$ via RLE/LZ4 to $0.25\text{ B/row}$) and latency_ms (Float32, $4\text{ bytes}$ raw, compressed $2\times$ via Gorilla/ZSTD to $2.0\text{ B/row}$) are queried by: SELECT avg(latency_ms), quantile(0.99)(latency_ms) FROM http_logs WHERE status = 500;

1. Calculate the total disk bytes read and minimum scan time on a single server with $4\text{ GB/s}$ NVMe read bandwidth for a row-oriented store (PostgreSQL sequential heap scan) versus a columnar store (ClickHouse reading only status and latency_ms).
2. Calculate how many AVX-512 SIMD vector instructions (512-bit = 64-byte registers processing sixteen Float32 lanes per instruction) are needed to filter and accumulate latency_ms if $1\%$ of rows ($10^7$ rows) match status = 500.

Exercise 3: Deep Pagination Heap Exhaustion & Roaring Bitmap Filter Cache Sizing

An OpenSearch cluster with $S=40$ primary shards experiences recurring coordinator OutOfMemoryError crashes when a scraper bot issues concurrent queries with from = 250000, size = 50 ($64\text{ bytes}$ per priority-queue score entry on the coordinator).

1. Calculate the coordinator JVM heap memory consumed by a single from = 250000, size = 50 request and by $30$ concurrent requests.
2. Redesign the pagination API using search_after with a deterministic tie-breaker sort ([timestamp: desc, _id: asc]) and compute the new coordinator priority-queue footprint for $30$ concurrent queries.
3. For a boolean filter cache storing a posting list of $200,000$ matching document IDs inside a segment of $5,000,000$ documents, compare the memory footprint of an uncompressed bitset, a 32-bit integer array, and a Roaring Bitmap (which divides the 32-bit range into $2^{16}=65,536$-integer chunks using either a 16-bit sorted array for $<4,096$ docs or a fixed $8\text{ KB}$ bitmap container for $\ge 4,096$ docs).

12. Authoritative References & Version Context

• Stephen Robertson and Hugo Zaragoza (2009), “The Probabilistic Relevance Framework: BM25 and Beyond”, Foundations and Trends in Information Retrieval 3(4) — doi.org/10.1561/1500000019
• Apache Lucene Official Documentation (v9.x/10.x) — lucene.apache.org/core/
• OpenSearch Project Documentation (v2.x LTS) — Query Phase, Fetch Phase, and Postings Compression — opensearch.org/docs/latest/
• Version Context: Search internals reflect Apache Lucene 9.x/10.x, OpenSearch 2.12+ LTS, and Elasticsearch 8.x architectures.