Distributed Query Execution

Overview

Distributed query execution is the process of evaluating a query across multiple independent data partitions (shards) and combining partial results into a single, coherent response. In a search engine like search-service, the index is split across many shards to handle datasets that exceed the memory and storage capacity of a single machine. When a user submits a search query, the system cannot simply look up results from one location—it must coordinate across every shard that might contain relevant documents.

The challenge goes far beyond simple fan-out. Relevance scoring algorithms like BM25 depend on corpus-wide statistics such as document frequency (DF) and average document length. When the corpus is split across shards, each shard only has local statistics, which can produce inconsistent or incorrect rankings if used naively. The query execution engine in index-engine must therefore implement multi-phase protocols that first collect global statistics and then use those statistics to produce globally correct scores during the actual retrieval phase.

Beyond scoring correctness, distributed query execution must also handle practical concerns: merging sorted result lists from multiple shards efficiently, handling shard failures gracefully, enforcing timeouts to maintain tail latency SLAs, and minimizing network round-trips. The scatter-gather pattern—where a coordinator fans out sub-queries and gathers partial results—forms the backbone of this architecture in both search-service and index-engine.

Core Concepts

Scatter-Gather Pattern

The scatter-gather pattern is the fundamental execution model for distributed queries. A coordinator node receives the original query, determines which shards are relevant (shard routing), scatters sub-queries to each selected shard in parallel, and then gathers the partial results. The coordinator is responsible for merging these partial results into the final response. This pattern minimizes latency because all shard queries execute concurrently, meaning total query time is bounded by the slowest shard plus merge overhead.

Distributed TF-IDF and BM25 Statistics

BM25 scoring requires two corpus-wide statistics for each query term: the document frequency (how many documents contain the term) and the average document length across the entire corpus. When the corpus is partitioned, each shard only knows its own local DF and average document length. Using local statistics directly leads to score skew—a term that is rare globally but concentrated in one shard will be underweighted on that shard and overweighted on shards where it is absent. The solution is a two-phase approach: first collect DF and total document counts from all shards, compute global statistics, and then distribute those global statistics back to the shards for the actual scoring phase.

K-Way Merge of Sorted Results

Each shard returns its top-K results sorted by relevance score. The coordinator must merge these N sorted lists into a single globally sorted list. This is a classic k-way merge problem, efficiently solved with a min-heap (or max-heap for descending order). The heap holds one element from each shard at a time, and extraction takes O(log N) per element where N is the number of shards. For a query requesting the top K results from N shards, the coordinator processes at most K*N candidates but only needs O(N) heap space.

Partial Aggregation vs. Final Aggregation

For analytical queries involving aggregations (counts, sums, averages, histograms), each shard computes a partial aggregate over its local data and sends the partial result to the coordinator. The coordinator then combines these partial aggregates into the final result. Some aggregations are trivially composable (sum, count, min, max), while others require careful handling (average requires transmitting both sum and count; percentiles require approximate data structures like t-digests or DDSketches).

Query Planning for Shard Selection

Not every query needs to hit every shard. If the index is partitioned by a known key (e.g., tenant ID, date range, geographic region), the query planner can prune shards that cannot contain matching documents. This shard routing reduces fan-out, lowers network traffic, and improves query latency. In search-service, metadata about each shard’s key range and document count is maintained by the coordinator to enable this optimization.

How It Works

Query Coordinator Receives the Request

The query coordinator is the entry point for all search requests. It parses the query, validates parameters (page size, offset, filters), and determines the execution plan. In search-service, the coordinator is a stateless Go service that can be horizontally scaled behind a load balancer. It maintains a shard map—a routing table that maps shard IDs to their network addresses and metadata (key ranges, replica locations, health status).

Shard Selection and Query Rewriting

The coordinator consults the shard map to determine which shards are relevant. For a full-text search query with no partition key filter, all shards must be queried. For a query filtered by tenant ID in a tenant-partitioned index, only shards containing that tenant’s data are selected. The coordinator may also rewrite the query for each shard—for example, adjusting the requested result count to account for pagination (requesting offset + limit results from each shard to ensure correct global pagination).

Parallel Fan-Out of Sub-Queries

The coordinator sends sub-queries to all selected shards concurrently. In search-service, this is implemented using goroutines with a shared context for cancellation. Each goroutine opens a gRPC stream to its target shard, sends the sub-query, and reads the response. A timeout context ensures that slow shards do not block the entire query indefinitely. In index-engine, the fan-out uses async I/O with an event loop, sending all sub-queries without blocking and processing responses as they arrive.

Shard-Local Execution

Each shard executes the sub-query against its local index. For a BM25 text query, this involves looking up the posting lists for each query term, intersecting or unioning them according to the boolean logic, scoring each candidate document, and returning the top-K results sorted by score. The shard also returns local statistics (document count, term frequencies) if the coordinator is performing a two-phase query for global scoring.

K-Way Merge with a Heap

As shard responses arrive, the coordinator feeds them into a k-way merge. A max-heap is initialized with the highest-scoring result from each shard. The coordinator repeatedly extracts the maximum element from the heap, adds it to the final result list, and pushes the next element from the same shard into the heap. This continues until the desired number of results is collected or all shards are exhausted. The merge is O(K log N) where K is the number of results requested and N is the number of shards.

Global Score Computation

In a two-phase query, the coordinator first collects term statistics from all shards, computes global DF and average document length, and then issues a second round of sub-queries with the global statistics attached. Each shard re-scores its local candidates using the global statistics, producing scores that are directly comparable across shards. The coordinator then performs the k-way merge on these globally-scored results. Both search-service and index-engine support this two-phase mode for queries where ranking precision is critical.

Practical Applications

Distributed Full-Text Search

The primary application of distributed query execution is full-text search across a sharded inverted index. A user types a query, and the system must search billions of documents spread across hundreds of shards, rank them by relevance, and return the top results in under 100 milliseconds. The scatter-gather pattern with k-way merge is the standard architecture used by Elasticsearch, Solr, and custom search engines like search-service.

Distributed Vector Similarity Search

Vector search (nearest-neighbor queries) follows the same distributed pattern. Each shard holds a partition of the vector index (HNSW graph or IVF structure) and returns its local top-K nearest neighbors. The coordinator merges these partial results using a heap, selecting the globally closest K vectors. Unlike text search, vector similarity scores are absolute distances, so no global statistics correction is needed—but the challenge of slow shards and tail latency remains.

Cross-Shard Analytics

Analytical queries that compute aggregations (facet counts, histograms, time-series rollups) over the full dataset require distributing the aggregation computation. Each shard computes partial aggregates, and the coordinator merges them. For faceted search, each shard returns its local term counts for each facet field, and the coordinator sums them to produce global facet counts. This pattern enables real-time analytics over datasets too large for any single node.

Federated Queries Across Heterogeneous Indexes

In advanced architectures, a single user query may need to search across multiple types of indexes: a text index for keyword matching, a vector index for semantic similarity, and a structured index for metadata filters. The coordinator acts as a federation layer, translating the query into sub-queries appropriate for each index type, executing them in parallel, and fusing the results using a hybrid scoring function (e.g., reciprocal rank fusion or weighted linear combination).

Implementation Patterns

Scatter-Gather with Goroutines

In Go, the scatter-gather pattern maps naturally to goroutines and channels. The coordinator launches one goroutine per shard, each sending a sub-query over gRPC and writing the result to a shared channel. A collector goroutine reads from the channel, feeding results into the merge heap. A context.WithTimeout ensures the entire operation is bounded. If a shard fails to respond in time, its partial results are simply excluded from the merge, and the coordinator returns a degraded but timely response.

Streaming Merge with Priority Queue

Rather than waiting for all shards to respond before merging, the coordinator can begin merging as soon as the first shard responds. A priority queue (min-heap or max-heap) is initialized lazily: as each shard’s response arrives, its top result is pushed into the heap. Results can be streamed to the client as they are extracted from the heap, reducing time-to-first-byte. This streaming approach is especially beneficial for large result sets or when the client consumes results incrementally.

Two-Phase Query for Statistics Collection

The two-phase query protocol separates statistics collection from document retrieval. In phase one, the coordinator sends a lightweight “stats-only” request to each shard, which returns term document frequencies and total document counts without fetching any actual documents. The coordinator aggregates these into global statistics. In phase two, the coordinator sends the full query with global statistics attached, and each shard scores documents using the corrected IDF values. This adds one network round-trip but guarantees globally correct BM25 rankings.

Circuit Breakers for Slow Shards

In production systems, individual shards can become slow due to GC pauses, disk I/O spikes, or network congestion. A circuit breaker monitors each shard’s response time and error rate. If a shard exceeds the latency threshold or error budget, the circuit breaker “opens” and the coordinator stops sending queries to that shard for a cooldown period, falling back to replicas or returning partial results. This prevents one degraded shard from causing cascading latency across all queries.

Code Examples

Scatter-Gather Fan-Out in Go

This example demonstrates the core scatter-gather pattern used in search-service to fan out a search query across multiple shards concurrently and collect partial results.

package search

import (
	"context"
	"fmt"
	"sync"
	"time"
)

// SearchResult represents a single document match from a shard.
type SearchResult struct {
	DocID   uint64
	Score   float64
	ShardID int
}

// ShardResponse holds the results from a single shard query.
type ShardResponse struct {
	ShardID int
	Results []SearchResult
	Stats   ShardStats
	Err     error
}

// ShardStats holds per-shard statistics for global scoring.
type ShardStats struct {
	TotalDocs int64
	TermDF    map[string]int64 // term -> document frequency
	AvgDocLen float64
}

// ShardClient is the interface for querying individual shards.
type ShardClient interface {
	Query(ctx context.Context, query string, topK int) (*ShardResponse, error)
}

// ScatterGather fans out a query to all shards and collects results.
// It respects the context deadline and returns partial results if
// some shards time out.
func ScatterGather(
	ctx context.Context,
	shards map[int]ShardClient,
	query string,
	topK int,
) ([]ShardResponse, error) {
	ctx, cancel := context.WithTimeout(ctx, 500*time.Millisecond)
	defer cancel()

	var (
		mu        sync.Mutex
		responses []ShardResponse
		wg        sync.WaitGroup
	)

	for shardID, client := range shards {
		wg.Add(1)
		go func(id int, c ShardClient) {
			defer wg.Done()

			resp, err := c.Query(ctx, query, topK)
			if err != nil {
				// Log the error but don’t fail the entire query.
				// The coordinator will return partial results.
				resp = &ShardResponse{
					ShardID: id,
					Err:     fmt.Errorf("shard %d: %w", id, err),
				}
			} else {
				resp.ShardID = id
			}

			mu.Lock()
			responses = append(responses, *resp)
			mu.Unlock()
		}(shardID, client)
	}

	wg.Wait()
	return responses, nil
}

// ComputeGlobalStats aggregates per-shard statistics into global
// corpus-wide statistics for BM25 scoring.
func ComputeGlobalStats(responses []ShardResponse) ShardStats {
	global := ShardStats{
		TermDF: make(map[string]int64),
	}

	var totalDocLen float64
	var shardCount int

	for _, resp := range responses {
		if resp.Err != nil {
			continue
		}
		global.TotalDocs += resp.Stats.TotalDocs
		totalDocLen += resp.Stats.AvgDocLen * float64(resp.Stats.TotalDocs)
		shardCount++

		for term, df := range resp.Stats.TermDF {
			global.TermDF[term] += df
		}
	}

	if global.TotalDocs > 0 {
		global.AvgDocLen = totalDocLen / float64(global.TotalDocs)
	}

	return global
}

K-Way Merge of Sorted Results in Go

This example implements the k-way merge using a max-heap to combine sorted result lists from multiple shards into a single globally-ranked list. This pattern is central to result aggregation in both search-service and index-engine.

package search

import "container/heap"

// heapEntry wraps a search result with its source shard index and
// position within that shard’s result list.
type heapEntry struct {
	result   SearchResult
	shardIdx int // index into the responses slice
	pos      int // position within the shard’s result list
}

// maxHeap implements heap.Interface for heapEntry, ordered by
// descending score (highest score = highest priority).
type maxHeap []heapEntry

func (h maxHeap) Len() int            { return len(h) }
func (h maxHeap) Less(i, j int) bool  { return h[i].result.Score > h[j].result.Score }
func (h maxHeap) Swap(i, j int)       { h[i], h[j] = h[j], h[i] }

func (h *maxHeap) Push(x interface{}) {
	*h = append(*h, x.(heapEntry))
}

func (h *maxHeap) Pop() interface{} {
	old := *h
	n := len(old)
	entry := old[n-1]
	*h = old[:n-1]
	return entry
}

// KWayMerge merges sorted result lists from multiple shards into
// a single globally-sorted list of the top-K results.
//
// Each shard’s results must be pre-sorted in descending score order.
// The merge runs in O(K log N) time where K is the requested result
// count and N is the number of shards.
func KWayMerge(responses []ShardResponse, topK int) []SearchResult {
	h := &maxHeap{}
	heap.Init(h)

	// Seed the heap with the top result from each shard.
	for i, resp := range responses {
		if resp.Err != nil || len(resp.Results) == 0 {
			continue
		}
		heap.Push(h, heapEntry{
			result:   resp.Results[0],
			shardIdx: i,
			pos:      0,
		})
	}

	results := make([]SearchResult, 0, topK)

	for h.Len() > 0 && len(results) < topK {
		// Extract the highest-scoring result across all shards.
		entry := heap.Pop(h).(heapEntry)
		results = append(results, entry.result)

		// Push the next result from the same shard, if available.
		nextPos := entry.pos + 1
		shardResults := responses[entry.shardIdx].Results
		if nextPos < len(shardResults) {
			heap.Push(h, heapEntry{
				result:   shardResults[nextPos],
				shardIdx: entry.shardIdx,
				pos:      nextPos,
			})
		}
	}

	return results
}

Common Pitfalls

1. Using shard-local statistics for BM25 scoring. When each shard computes IDF using only its local document frequency, terms that are unevenly distributed across shards receive inconsistent scores. This causes documents from shards with fewer occurrences of the term to be over-ranked relative to documents from shards where the term is more common. Always use global statistics or accept the ranking accuracy trade-off explicitly.
2. Requesting only top-K from each shard when paginating. If the client requests page 5 (results 41–50), the coordinator must request at least the top 50 results from each shard, not just 10. Otherwise, a shard might have relevant results ranked 41st–50th locally that rank higher globally but are never retrieved. This means deep pagination is inherently expensive in distributed search: the cost grows linearly with the page offset.
3. Ignoring tail latency from the slowest shard. The scatter-gather pattern means the query is as slow as the slowest shard. Without timeouts, a single shard experiencing a GC pause or disk stall can cause the entire query to hang. Always set aggressive timeouts, return partial results when shards fail, and use hedged requests or replica routing to mitigate stragglers.
4. Merging non-comparable scores from heterogeneous indexes. When federating across different index types (text BM25 scores, vector cosine distances, custom ranking signals), the raw scores are on different scales and cannot be directly compared. Use score normalization (min-max scaling, z-score normalization) or rank-based fusion (reciprocal rank fusion) to combine results from heterogeneous sources.
5. Failing to account for shard imbalance in query planning. If shards are unevenly sized (some hold many more documents than others), the larger shards will consistently be slower, becoming a bottleneck. Monitor per-shard query latency distributions and rebalance shards proactively. Consider replicating hot shards to spread the query load.

Key Takeaways