Principle:Neuml Txtai Distributed Search
| Knowledge Sources | |
|---|---|
| Domains | Distributed_Systems, Scalability |
| Last Updated | 2026-02-09 17:00 GMT |
Overview
Distributed Search is txtai's architecture for sharding embeddings indexes across multiple HTTP microservices, enabling horizontal scaling of vector search beyond single-machine memory and compute limits.
Description
As document collections grow beyond what a single machine can efficiently index and serve, txtai provides a Cluster class that distributes the embeddings index across multiple independent txtai API instances. Each instance (shard) holds a partition of the full document collection and runs as a standalone HTTP microservice. The Cluster class acts as a coordinator: it accepts queries from the application, fans them out to all shards in parallel, collects the partial results, and merges them into a single globally ranked result set. This scatter-gather architecture enables linear horizontal scaling -- adding more shards increases both the total index capacity and the aggregate query throughput.
The sharding strategy is straightforward: documents are assigned to shards at indexing time, typically by round-robin distribution or hash-based partitioning on the document id. Each shard independently maintains its own ANN index, optional content database, and optional graph layer, functioning as a complete txtai Embeddings instance. The Cluster class maintains a registry of shard URLs and health status, routing queries only to healthy shards. If a shard becomes unavailable, the Cluster can either return partial results (best-effort mode) or raise an error (strict mode), depending on the application's consistency requirements.
Query execution in distributed mode follows a three-phase pattern:
- Scatter phase -- The Cluster sends the query (either a simple similarity query or a full SQL query) to each shard via HTTP POST
- Gather phase -- Each shard executes the query against its local index and returns its top-k results with scores
- Merge phase -- The Cluster's Aggregate module combines all shard results, sorts them globally by score, removes duplicates, enforces the requested LIMIT, and returns the final result set
The application code uses the same Embeddings API regardless of whether the index is local or distributed.
The Cluster class communicates with shards using txtai's REST API protocol, which means each shard is a standard txtai API server that can also be queried independently. This design simplifies deployment: shards can be deployed as containers (Docker, Kubernetes pods), serverless functions, or traditional server processes. Load balancers can be placed in front of shard groups for additional fault tolerance. The coordinator itself is stateless, holding only the shard registry, so it can be replicated for high availability.
Usage
Use Distributed Search when the document collection exceeds the memory capacity of a single machine, when query latency requirements demand parallel execution across multiple nodes, or when you need fault tolerance through shard replication. A typical deployment starts with a single-node index during development and transitions to a distributed cluster as the dataset grows. The Cluster class is configured with a list of shard URLs and can be used as a drop-in replacement for a local Embeddings instance, requiring no changes to query code.
Theoretical Basis
1. Index Sharding Strategies: The two primary sharding strategies are:
- Hash-based partitioning -- Each document is assigned to shard hash(id) mod S where S is the number of shards, ensuring uniform distribution regardless of insertion order
- Range-based partitioning -- Documents are assigned to shards based on id ranges or timestamp ranges, enabling time-based data management (e.g., archiving old shards)
Hash-based partitioning provides better load balancing; range-based partitioning enables targeted queries when a filter predicate aligns with the partition key.
2. Scatter-Gather Query Pattern: The Cluster dispatches each query to all S shards in parallel using concurrent HTTP requests. Each shard returns its local top-k results. The total network payload is O(S * k * result_size). The merge step sorts S * k results in O(S * k * log(S * k)) time. The overall query latency is max(shard_latencies) + merge_time, dominated by the slowest shard (the straggler problem). This pattern is the standard distributed search architecture used by systems like Elasticsearch and Apache Solr.
3. Result Merging and Deduplication: During the merge phase, results from all shards are combined into a single list, sorted by similarity score in descending order. If documents are replicated across shards (for fault tolerance), deduplication removes entries with identical document ids, keeping only the highest-scoring instance. The final LIMIT is applied after deduplication to ensure the correct number of results. For SQL queries with GROUP BY or aggregate functions, the Aggregate module performs the aggregation across shard-level partial results.
4. Shard Health Monitoring: The Cluster periodically sends health check requests (HTTP GET to a status endpoint) to each shard. Shards that fail to respond within a configurable timeout are marked as unhealthy and excluded from subsequent queries until they recover. When a shard transitions from unhealthy back to healthy, it is automatically re-added to the active set. This health monitoring ensures that transient failures do not block query execution, at the cost of potentially incomplete results when shards are down.
5. Scaling Properties: With S shards, the system exhibits the following scaling characteristics:
- Aggregate index capacity: S * C where C is the per-shard capacity (determined by available RAM and disk)
- Query throughput: scales as S * T where T is the per-shard throughput, assuming the merge phase and network are not bottlenecks
- Query latency: does not improve with more shards (bounded by the slowest shard plus merge overhead) but remains constant as the total dataset grows, since each shard's local index size stays fixed at N / S documents
This makes the architecture suitable for datasets ranging from millions to billions of documents.