Scalable distributed Inverted Index Design
Problem Statement
Design a scalable distributed inverted index system that enables fast full-text search across
a large corpus of documents (100 Million Documents). The system should support document
ingestion, provide sub-second query responses, and scale to handle enterprise-grade
workloads.
As discussed during the discussions :-
Core Features
● Document Ingestion: Add/update/delete documents with metadata
● Full-Text Search: Query documents using natural language terms
● Relevance Ranking: Return results ordered by relevance score (Based on scoring
Strategy)
● Real-time Updates: Near real-time availability of indexed documents
Scale Parameters
● Document Count: 100 million documents
● Document Size: 10KB average
● Total Storage: ~1TB raw data + 3-5TB index data
Performance Parameters
● Query Throughput: 50,000 QPS
● Indexing Throughput: 10,000 documents/second
● Query Latency: P95 < 100ms, P99 < 200ms
Reliability
● Replication Factor: 3x across availability zones
● Recovery Time: In case of failure - RTO < 15 minutes (max 15 mins downtime), RPO <
5 minutes (max 5 mins data loss)
Here is a quick summary of Design :-
● Scale Target: Handle 100M documents (~1TB raw data, 3–5TB indexed).
● Performance Goals: 50K QPS, 10K docs/sec ingestion, P99 latency under 200ms.
● Core Components: API Gateway, Query Workers, Indexing Workers, Index Nodes,
Document Store, Redis Cache.
● Document Ingestion: Real-time add/update/delete with metadata via indexing pipeline
handled by Indexing workers
● Full-Text Search: Natural language search with term parsing, sharding, and ranking.
● API Gateway: Routes indexing and search requests to appropriate worker nodes.
● Sharding Strategy: Hash-based partitioning across 1000+ shards (~100K docs each).
● Query Execution: Fan-out to relevant shards, parallel search, and global result
ranking.
● (Optionally) For greater performance and async processing. Kafka priority queues with
scalable workers ensure SLA-based ingestion.
● Replication: 3x replication per shard with auto failover and quorum-based consistency.
● Caching: Multi-level cache (L1 in-memory, L2 Redis, optional L3 shard-level) for hot
queries.
● Document Store: Sharded NoSQL storage for raw documents and metadata, separate
from index.
● Horizontal Scalability: All services are stateless and each component can be scaled
independently. Worker nodes can be autoscaled. Shards can be scaled and more
replicas can be added.
● Disaster Recovery: Multi-tier backup, cross-region replication, and auto health
monitoring.
High Level Design
Components Description
API Gateway
To route the requests to different services. Primarily acts as a request router along with other
functionalities.
● Search Related Queries -> Routed to the Query Worker Nodes
● Indexing Related Queries -> Routed to Indexing Worker Nodes
Query Worker Nodes ( Powers all search queries )
● Query parsing and validation
● Shard routing based on term distribution
● Parallel query execution across shards
● Result aggregation and ranking
● Caching for hot queries
Indexing Worker Nodes ( Powers all Indexing requests )
● Parser: Processes incoming documents for indexing
● Router: Determines target shard based on document ID hash
● Tokenizes document content and generates the data to be indexed
● Updates relevant shards with new index entries
Index Nodes ( Stores Indices )
● Shard 1, 2, 3...N: Each stores subset of documents (~100K docs each)
● Maintains inverted index (term → document mappings)
● Handles search queries on local document subset
● Provides 3x replication for fault tolerance
Document Store ( Stores raw documents and metadata )
● Persistent storage for original document content and metadata
● Provides document retrieval for search results
● Separate from index storage for scalability
● This can also be sharded for scalability.
● NoSQL would be preferable in our use case where horizontal scalability is required and
data is unstructured.
Cache (Redis Cluster)
● Stores hot query results for sub-millisecond response times
● Shared cache accessible by all Query Worker nodes
● Reduces load on Index Nodes for popular searches
Assumptions :-
** I have skipped load balancer etc for the sake of simplicity in design. I have assumed that we
have handled that at infra level.
** Assuming Shards can recover from failures using Leader election if primary shard fails.
Data Flow
Query Flow
1. User submits a search query via API Gateway with authentication and rate limiting.
2. Cache Check: Redis cache checked for existing results to avoid redundant computation.
3. Query Parsing: Parser validates syntax, tokenizes terms, normalizes text, and extracts
filters/operators.
4. Route Planning: Router identifies relevant shards based on query terms and metadata.
5. Fan-out Execution: Query is distributed in parallel to relevant index shards.
6. Shard Processing: Each shard searches local inverted indices and returns ranked
results.
7. Result Aggregation: Responses from shards are merged, including posting lists.
8. Global Ranking: Results are re-ranked using global relevance metrics (e.g., TF-IDF,
BM25).
9. Response Formatting: Results formatted with metadata, pagination, and query stats.
10.Cache Update: Final results cached in Redis with TTL for faster future retrieval.
Indexing Flow
1. Document submitted via API Gateway with metadata and content.
2. API Gateway Routing: API Gateway validates, authenticates, and routes to indexing
worker nodes. [ Or to indexing queue based on priority if queues are used ]
3. [ Optional if queue is used ] Queue Buffering: Kafka queues (Hot/Warm/Cold) buffer
documents based on SLA and criticality.
4. Worker Processing: Indexing workers consume messages, parse content, tokenize
text, and extract terms.
5. Shard Distribution: Hash-based routing determines target shards using document ID.
6. Store Index & Index Updates: Workers update inverted indices, posting lists, and term
stats in parallel across shards.
7. Document Storage: Raw content and metadata stored in document store for retrieval.
8. Replication Sync: Primary shard syncs updates to replicas using quorum (2/3 replicas).
9. Commit Confirmation: Success acknowledged once a write quorum is achieved.
Sharding
We can use hash-based document sharding to horizontally partition 100 million documents
across 1000+ shards.
Each shard contains:
● Term Dictionary (B+ Tree): Fast term lookup with O(log n) complexity
● {term : [ doc list ]} : Document IDs containing each term with positional data.
Algorithm: shard_id = hash(document_id) % num_shards
Shard Size: ~100K documents per shard.
Query Processing with Sharding with Fan-out Strategy :
● Query parser determines affected shards based on term distribution
● Parallel execution across relevant shards (typically 20-50% of total shards)
● Results merged and re-ranked at the Query Worker nodes.
Adding New Shards:
● Provision new empty shards with initialized indices
● Background rebalancing using hashing migration
● Update routing tables atomically across all required nodes
● Gradual traffic migration with zero downtime
Other pointers :-
1. In order to handle the hot shard problem, we can do Replica read distribution + caching
frequently accessed data
2. Right now the design uses simple modulo based hashing. We can further increase the
scalability and maintainability of the system by using consistent hashing.
Replication
Replication: 3x factor (1 primary + 2 replicas) for fault tolerance
Node Distribution: Multiple shards per physical node to maximize resource utilization.
Node 1: [ Shard 1 Primary, Shard 5 Replica, Shard 9 Replica]
Node 2: [ Shard 2 Primary, Shard 1 Replica, Shard 10 Replica]
Node 3: [ Shard 3 Primary, Shard 2 Replica, Shard 1 Replica]
Failure Handling
● If a primary replica fails, We can do leader election among replicas. Promote replica to
new primary + Reroute writes to new primary + Spawn new replica on healthy node.
● In case of Replica Failure - We can continue serving from primary + remaining replica .
Then Spawn replacement replica on healthy node and Sync data from primary to new
replica
Performance Optimisation
Multi-Level Caching
1. L1 Cache: Query Worker-level in-memory LRU cache
2. L2 Cache: Redis cluster for frequent queries (shared across all query workers)
3. ( Optionally ) L3 Cache: Shard-level caching
Optimised Query Planning
1. Selective Shard Querying: Only query shards containing relevant terms (typically
20-50% of total shards based on the category etc.)
2. Early Termination: Stop processing when sufficient high-quality results found
3. We can analyze document frequency (DF) for each query term and start with rarest
terms to minimize intermediate result sets.
Queue with Priority
Kafka Queue :- We can use the queue to receive the requests and process the requests
gracefully.
Moreover, we can have priority queues to index documents in a greedy manner.
● High Priority Queue: Critical updates indexed within 1 seconds
● Medium Priority Queue: Bulk updates processed within 30 seconds
● Low Priority Queue: Historical data reindexed periodically
Indexing Scalability & Lag
1. We can have Multi-Tier Indexing Pipeline (10K Requests/sec)
a. High Priority Queue: Critical updates (<1s SLA) – e.g., breaking news, premium
users, real-time alerts.
b. Medium Priority Queue: Standard/bulk updates (<30s SLA) – e.g., Regular API
calls, user-generated content etc
c. Low Priority Queue: Historical/archive data (<5min SLA) – e.g.,Bulk imports,
historical data, archive processing
d. Kafka Queues: Segmented by priority with dedicated consumer groups per tier.
2. Scalable Worker Architecture
a. Auto-Scaling: 50–100 workers scale based on priority queue depth.
b. Parallelism: Each worker processes 100–200 docs/sec with concurrent shard
writes.
c. Micro-Batching: Used in Medium/Low Priority paths (10–50 docs/batch).
Disaster Recovery
1. Multi-Level Failure Handling with Automated Recovery
a. Node Failures: Automatic failover within 30s using Raft leader election; 3× shard
replication.
b. Cluster Failures: Manual intervention with RTO < 15 min; auto traffic redirection
to healthy clusters.
c. Region Failures: Cross-region DR with RTO < 60 min, RPO < 30 min via async
replication.
2. Multi-Tier Backup Strategy with Geographic Distribution
a. Hot Backup: Real-time sync replication across AZs, RPO < 5s, supports auto
failover.
b. Warm Backup: Hourly async snapshots stored cross-region for manual restore.
c. Cold Backup: Daily batch archives for long-term compliance and
point-in-time recovery.
3. Automated Health Monitoring & Failure Detection
a. Continuous Monitoring: Per-second checks on network, resources, index
integrity, and query performance.
b. Alerts Setup: Automated alerts trigger for