Design an in-process search index

01The design, in the six-step framework

1. Requirements and use cases

A candidate report describes this as the low-level design of a search engine: not crawling and shards, but the objects inside one index. The aim is a clean pipeline from text to ranked results.

• A caller adds, updates and deletes documents by id.
• The index tokenizes text: splits, lower-cases, drops stop words, stems lightly.
• A caller queries with terms and AND, OR, NOT.
• Results come back ranked by relevance, with the top k.
• A caller asks for prefix completions over indexed terms.

Out of scope: distributed shards, spelling correction, phrase positions beyond a mention, persistence. Non-functional constraints: query time depends on the matched postings, not on the size of the corpus; analysis and ranking are swappable; a reader during an update sees either the old or the new document, not a mix; memory is watched, since the same strings repeat millions of times.

2. Core entities and relationships

• Document: id and text.
• Analyzer: a pipeline of token filters from text to terms.
• TermDictionary: interns each term to one integer id, and keeps a trie for prefixes.
• InvertedIndex: a posting list per term: document id and term frequency, sorted by document id.
• QueryParser: turns a string into a query tree.
• Query: And, Or, Not and Term nodes, each returning a set of document ids.
• Ranker: scores a document for query terms; TF-IDF is the default.
• SearchEngine: the facade.

The engine composes the analyzer and index; a document must be analysed the same way at index and query time, so they belong together. Rankers are injected, because trying a new score should not touch the index.

3. Interfaces and key methods

class Ranker(Protocol):
    def score(self, doc_id: int, term_ids: list[int], index: "InvertedIndex") -> float: ...

class Query(Protocol):
    def run(self, index: "InvertedIndex", universe: set[int]) -> set[int]: ...
    def terms(self) -> list[int]: ...

class InvertedIndex:
    def add(self, doc_id: int, counts: dict[str, int]) -> None: ...
    def remove(self, doc_id: int) -> None: ...
    def docs(self, term_id: int) -> dict[int, int]:
        """Postings for a term: {doc_id: term_frequency}."""

class SearchEngine:
    def add(self, doc_id: int, text: str) -> None:
        """Replaces any earlier version of doc_id."""
    def delete(self, doc_id: int) -> None: ...
    def search(self, query: str, k: int = 10) -> list[tuple[int, float]]:
        """Top k (doc_id, score), best first, ties by doc_id."""
    def complete(self, prefix: str, k: int = 5) -> list[str]: ...

Invariants: a document appears in a term's postings if and only if the term is in its current text; the same analyzer runs at index and query time; every term string is stored once; a term with empty postings is treated as absent.

4. Design patterns and the choices behind them

Analysis as a filter pipeline

Tokenizing is a series of independent steps. A list of filters lets stemming or a Telugu tokenizer be added without touching the index. A single regular expression is the short answer until the second language arrives.

Ranking as a strategy

TF-IDF, BM25 and a learned score all map a document and terms to a number. A strategy makes the experiment a constructor argument. TF-IDF is a sum over query terms:

score(d) = Σ_t (tf(t, d) / len(d)) × log(N / df(t))

A word in every document has a zero weight, and rare words dominate, which is what you want.

Terms as a flyweight

With a billion postings, repeating the string "database" is the memory bill. A flyweight dictionary stores each term once and postings keep small integers. Strings appear only at the edges.

Query as a tree

A parser yields a tree of nodes, each with one run method; adding a phrase node is a new class. The same trie over the dictionary answers prefixes in time proportional to the prefix, and it is kept separate from the postings so that the public API stays four verbs.

5. Core flow and code

1. add removes any old version, analyses the text and counts terms.
2. InvertedIndex.add interns each term and updates its postings and the document's length.
3. search parses the query, runs the tree to a set of candidates, and scores them.
4. A size-k heap picks the best without sorting all matches.
5. complete walks the trie to the prefix and collects terms.

from __future__ import annotations
import heapq, math, re, threading
from collections import Counter
from dataclasses import dataclass, field
from typing import Callable, Optional, Protocol

STOP_WORDS = frozenset({"a", "an", "the", "of", "to", "in", "and", "or", "not"})
TokenFilter = Callable[[list[str]], list[str]]


def lowercase(tokens: list[str]) -> list[str]: return [t.lower() for t in tokens]
def drop_stop_words(tokens: list[str]) -> list[str]: return [t for t in tokens if t not in STOP_WORDS]
def light_stem(tokens: list[str]) -> list[str]:
    return [t[:-1] if t.endswith("s") and len(t) > 3 else t for t in tokens]


class Analyzer:
    def __init__(self, filters: list[TokenFilter]) -> None:
        self.filters = filters
    def terms(self, text: str) -> list[str]:
        tokens = re.findall(r"\w+", text)
        for f in self.filters:
            tokens = f(tokens)
        return tokens


@dataclass
class TrieNode:
    children: dict[str, "TrieNode"] = field(default_factory=dict)
    is_term: bool = False


class TermDictionary:
    def __init__(self) -> None:
        self.ids: dict[str, int] = {}
        self.names: list[str] = []
        self.trie = TrieNode()

    def intern(self, term: str) -> int:
        if term in self.ids:
            return self.ids[term]
        self.ids[term] = len(self.names)
        self.names.append(term)
        node = self.trie
        for ch in term:
            node = node.children.setdefault(ch, TrieNode())
        node.is_term = True
        return self.ids[term]

    def lookup(self, term: str) -> Optional[int]:
        return self.ids.get(term)

    def complete(self, prefix: str, k: int) -> list[str]:
        node = self.trie
        for ch in prefix:
            node = node.children.get(ch)
            if node is None:
                return []
        out: list[str] = []
        stack = [(node, prefix)]
        while stack and len(out) < k:
            cur, word = stack.pop()
            if cur.is_term:
                out.append(word)
            for ch in sorted(cur.children, reverse=True):
                stack.append((cur.children[ch], word + ch))
        return out


class InvertedIndex:
    def __init__(self) -> None:
        self.dictionary = TermDictionary()
        self.postings: dict[int, dict[int, int]] = {}
        self.doc_len: dict[int, int] = {}
        self.doc_terms: dict[int, list[int]] = {}

    def add(self, doc_id: int, counts: dict[str, int]) -> None:
        ids = []
        for term, n in counts.items():
            tid = self.dictionary.intern(term)
            self.postings.setdefault(tid, {})[doc_id] = n
            ids.append(tid)
        self.doc_terms[doc_id] = ids
        self.doc_len[doc_id] = sum(counts.values())

    def remove(self, doc_id: int) -> None:
        for tid in self.doc_terms.pop(doc_id, []):
            self.postings[tid].pop(doc_id, None)
        self.doc_len.pop(doc_id, None)

    def docs(self, term_id: int) -> dict[int, int]:
        return self.postings.get(term_id, {})


class Query(Protocol):
    def run(self, index: InvertedIndex, universe: set[int]) -> set[int]: ...
    def terms(self) -> list[int]: ...


@dataclass
class Term:
    term_id: Optional[int]
    def run(self, index: InvertedIndex, universe: set[int]) -> set[int]:
        return set() if self.term_id is None else set(index.docs(self.term_id))
    def terms(self) -> list[int]:
        return [] if self.term_id is None else [self.term_id]


@dataclass
class Bool:
    op: str
    left: Query
    right: Query
    def run(self, index: InvertedIndex, universe: set[int]) -> set[int]:
        a, b = self.left.run(index, universe), self.right.run(index, universe)
        return a & b if self.op == "AND" else a | b
    def terms(self) -> list[int]:
        return self.left.terms() + self.right.terms()


@dataclass
class Negate:
    inner: Query
    def run(self, index: InvertedIndex, universe: set[int]) -> set[int]:
        return universe - self.inner.run(index, universe)
    def terms(self) -> list[int]:
        return []


class QueryParser:
    """Left-to-right; a bare space means AND; NOT binds to the next term."""
    def __init__(self, analyzer: Analyzer, dictionary: TermDictionary) -> None:
        self.analyzer, self.dictionary = analyzer, dictionary

    def _term(self, word: str) -> Query:
        terms = self.analyzer.terms(word)
        return Term(self.dictionary.lookup(terms[0]) if terms else None)

    def parse(self, text: str) -> Optional[Query]:
        tree, op, negate = None, "AND", False
        for word in text.split():
            if word in ("AND", "OR"):
                op = word; continue
            if word == "NOT":
                negate = True; continue
            node = self._term(word)
            node = Negate(node) if negate else node
            tree = node if tree is None else Bool(op, tree, node)
            op, negate = "AND", False
        return tree


class Ranker(Protocol):
    def score(self, doc_id: int, term_ids: list[int], index: InvertedIndex) -> float: ...


class TfIdfRanker:
    def score(self, doc_id: int, term_ids: list[int], index: InvertedIndex) -> float:
        n = len(index.doc_len)
        total = 0.0
        for tid in term_ids:
            docs = index.docs(tid)
            if doc_id in docs:
                total += docs[doc_id] / index.doc_len[doc_id] * math.log(n / len(docs))
        return total


class SearchEngine:
    def __init__(self, analyzer: Analyzer, ranker: Ranker) -> None:
        self.analyzer, self.ranker = analyzer, ranker
        self.index = InvertedIndex()
        self.lock = threading.RLock()

    def add(self, doc_id: int, text: str) -> None:
        counts = Counter(self.analyzer.terms(text))   # slow work outside the lock
        with self.lock:
            self.index.remove(doc_id)
            self.index.add(doc_id, counts)

    def delete(self, doc_id: int) -> None:
        with self.lock:
            self.index.remove(doc_id)

    def search(self, query: str, k: int = 10) -> list[tuple[int, float]]:
        with self.lock:
            tree = QueryParser(self.analyzer, self.index.dictionary).parse(query)
            if tree is None:
                return []
            hits = tree.run(self.index, set(self.index.doc_len))
            terms = tree.terms()
            scored = [(d, self.ranker.score(d, terms, self.index)) for d in hits]
        return heapq.nsmallest(k, scored, key=lambda x: (-x[1], x[0]))

    def complete(self, prefix: str, k: int = 5) -> list[str]:
        with self.lock:
            return self.index.dictionary.complete(prefix.lower(), k)

6. Extensions, concurrency and what interviewers probe

• Same analysis both ways: indexing "Databases" and querying "database" must meet. The query parser runs the same Analyzer; a mismatch here is the classic silent zero-results bug.
• Fast intersection: Python sets are short to write. Real postings are sorted arrays, intersected smallest-first with skip pointers, and compressed by storing gaps between ids.
• Phrases: store positions in postings, then check adjacency; add a Phrase node. Memory roughly doubles, so name the cost.
• Updates at scale: rewriting postings is costly, so real indexes buffer new documents in a small segment, merge in the background, and mark deletes with tombstones; that is the inverted index concept in full.
• Concurrency: one lock gives atomic updates. A read-heavy index prefers immutable segments swapped by reference, so readers take no lock. Analysis runs outside the lock because it is the slow part.
• BM25 is a new Ranker that adds length normalisation and saturation.
• Test first: update removes old terms; NOT excludes; a term in every document scores zero; unknown terms return nothing rather than raising; top k orders ties by id; prefix completion honours k.
• Common mistake: storing a list of words per document and scanning them. Say "inverted" in the first minute.

02Where this question was reported

• LLD Mastery · Google-tagged (Hard); inverted index, relevance ranking
• Glassdoor · Glassdoor candidate report: 'low level design of a search engine' at Google (page blocks bots; via search snippet)