Search Autocomplete System Design

from collections import defaultdict
import heapq

class TrieNode:
    def __init__(self):
        self.children = {}
        self.is_end = False
        self.frequency = 0
        # Cache top-k results at each node to avoid recomputation
        self.top_suggestions = []

class AutocompleteService:
    def __init__(self, top_k=5):
        self.root = TrieNode()
        self.top_k = top_k

    def insert(self, query: str, frequency: int):
        node = self.root
        for char in query.lower():
            if char not in node.children:
                node.children[char] = TrieNode()
            node = node.children[char]
        node.is_end = True
        node.frequency = frequency

    def _collect_suggestions(self, node, prefix, results):
        if node.is_end:
            results.append((node.frequency, prefix))
        for char, child in node.children.items():
            self._collect_suggestions(child, prefix + char, results)

    def build_cache(self):
        """Pre-compute top-k suggestions at every node."""
        self._build_cache_recursive(self.root, "")

    def _build_cache_recursive(self, node, prefix):
        results = []
        self._collect_suggestions(node, prefix, results)
        # Keep top-k by frequency
        node.top_suggestions = [
            query for freq, query
            in heapq.nlargest(self.top_k, results)
        ]
        for char, child in node.children.items():
            self._build_cache_recursive(child, prefix + char)

    def search(self, prefix: str) -> list:
        node = self.root
        for char in prefix.lower():
            if char not in node.children:
                return []
            node = node.children[char]
        # O(1) lookup thanks to pre-computed cache
        return node.top_suggestions

# Build the trie from aggregated search logs
service = AutocompleteService(top_k=5)
queries = [
    ("how to learn python", 50000),
    ("how to learn javascript", 45000),
    ("how to learn go", 30000),
    ("how to cook rice", 28000),
    ("how are you", 90000),
    ("holiday deals", 15000),
]

for query, freq in queries:
    service.insert(query, freq)

service.build_cache()

print(service.search("how to l"))
# ['how to learn python', 'how to learn javascript', 'how to learn go']

# Autocomplete system components

data_pipeline:
  input: search_query_logs (Kafka topic)
  processing: Apache Spark job (runs every 15 minutes)
  steps:
    - aggregate query counts over rolling 7-day window
    - filter offensive or low-quality queries
    - weight recent queries higher (time decay)
    - merge with curated trending topics
  output: ranked_queries.json -> pushed to trie servers

serving_layer:
  edge_cache:
    technology: CloudFront / Fastly
    cache_key: first 3 characters of prefix
    ttl: 5 minutes
    hit_rate: ~60% of all requests

  api_gateway:
    rate_limit: 10 requests/sec per user
    routes_to: trie shard based on first character

  trie_servers:
    instances: 6 shards (partitioned by prefix range)
    memory: 16 GB per shard
    data: in-memory trie with pre-computed top-k
    refresh: hot-swap new trie every 15 minutes

client_side:
  debounce: 200ms between keystrokes
  min_prefix_length: 2 characters
  local_cache: LRU cache of last 50 prefix results
  fallback: show recent user searches if backend is slow