Distributed Search Design

Motivation and Goals

There is a need to support homogeneous indicies that are larger than will fit on a single machine and still provide acceptable latency. Aggregate throughput can also increase if a larger percent of the index can be held in memory.


Nice to haves:

Simple Distribution

The normal responses from the standard and dismax query handlers provide almost enough information in order to merge multiple responses from sub-searchers into a single client response.

Document Lists:

Highlighting info:

Faceted Browsing info:

Debug info: currently not meant for machine parsing... we could simply include debug info from all sub-searchers



Merge XML responses w/o Schema

Create an external service to make HTTP requests to sub-searchers and simply combine the current XML responses. All operations operate on the XML alone, w/o reliance on the Solr schema.




Merge responses with Schema awareness

Have a query handler that makes HTTP requests to sub-searchers and merges responses.



Stateless request handlers

Optional: Have request handlers and APIs that don't use docids, and don't require query consistency. There may not be as much value in this option... If we need custom query handlers, complex distribution that allows index consistency and docids would probably be a better choice.

Complex Distribution

The big distinction here is index consistency... a single request handler would be guaranteed that the view of the index does not change during a single request (just like non-distributed Solr).

If the index doesn't change then:

Follow the basic Lucene design for MultiSearcher/RemoteSearcher as a template.

Areas that will need change:


Transport Syntax:

If custom query handlers are to be allowed with Distributed Search, there is a problem of how to represent Queries. The Lucene MultiSearcher distributes Weights to sub-searchers... this was done for global-idf reasons... the weights are calculated in the super-searcher and can thus use doc counts from all sub-searchers.

Network Transports:

Need new style APIs geared toward faceted browsing for a distributed solution (avoid instantiating DocSets... pass around symbolic sets and set operations)

Consistency via Retry

Maintaining a consistent view of the index can be difficult. A different method would return an index version with any request to a sub-index. If that version ever changed during the course of a request, the entire request could be restarted.



Consistency via Specifying Index Version

Every request returns an index version that created it. On any request, one may specify the version of the index they want to serve the request. An older IndexSearcher may be made available for some time after a new IndexSearcher is in place to service requests that started with it. Every request to a particular version of a searcher could extend the "lease" for another 5 seconds (configurable, or even per-request).


Q: how would request handler get a particular version of a SolrIndexSearcher? One is already bound to a SolrQueryRequest (the newest), but if an older one is requested should that logic be built into SolrCore? Handling it in SolrCore would be both cleaner for request handlers, but it would complicate SolrCore too.

Multi-phased approach, allowing for inconsistency

Do a mulit-phased approach (separate query phase from stored field retrievial and document highlighting), but communicate using the uniqueKey fields rather than internal docids.

This is simpler, but opens a window of inconsistency because the index could always change between phases. This level of inconsistency may be acceptable given that there is already inconsistency caused by clients paging through results.

Downsides of using uniqueKeys instead of lucene docids:

High Availability

How can High Availability be obtained on the query side?


How should the collection be updated? It would be complex for the client to partition the data themselves, since they would have to ensure that a particular document always went to the same server. Although user partitioning should be possible, there should be an easier default.

Single Master

A single master could partition the data into multiple local indicies and subsearchers would only pull the local index they are configured to have.

Directory structure for indicies: Current: solr/data/index OptionA: solr/data/index0, solr/data/index1, solr/data/index2, OptionB: solr/data/index, solr/data2/index, solr/data3/index, OptionC: solr/data/index, another_solr_home/data/index, yet_another_solr_home/data/index


Multiple Masters

There could be a master for each slice of the index. An external module could provide the update interface and forward the request to the correct master based on the unique key field.

The directory structure for indicies would not have to change since each master would have it's own solr home and hence it's own index directory.




How to synchronize commits across subsearchers and top-level-searchers? This is probably only needed if one is trying to present a SolrSearcher java interface with a consistent index view.



What won't work

Current approach

"Multi-phased approach, allowing for inconsistency" is what is being first used for the query side of https://issues.apache.org/jira/browse/SOLR-303 . Distributing the indexing will be up to users via a Multiple Master approach. In the future, we may want to migrate to "Consistency via Specifying Index Version" and lucene internal docids.

The query is executed in phases. In each phase a request is sent to relevant shards in a separate thread. After all the responses are received for all requests the next phase is executed.


Each shard is requested for the top matching document's unique keys and sort fields with facets for the given query. The number of keys requested in this phase is 'N' (start=0&rows=N) regardless of the start specified, so that the results can be correctly merged together.

The response gets the unique keys for each document and their scores. If GET_FACETS is requested it returns the top 'N' facets. n=facet.count. After the responses are obtained they are merged and sorted by the rank. From the sorted list the documents to be returned are identified on the basis of 'start' and 'rows' parameter.

Phase 2

Request are sent to fetch fields, highlighting and MoreLikeThis information only for the documents identified in Phase 1. The request contains the document unique keys and is sent to only the relevant shard which has the document.

Phase 3: REFINE_FACETS (only for faceted search)

The original returned facets may have insufficient information. So more requests are sent to shards for refining facets. Note that the approach applied here gives accurate counts but theoretically, it is possible to miss some facet terms.

After the document fields and facets are obtained the response is constructed and sent back to client.

It is possible that during the small window of time (from phase 1-3) the index may change. In that case the responses may have incorrect data. That is ignored for the time-being.

DistributedSearchDesign (last edited 2009-09-20 22:05:03 by localhost)