Concurrent Segment Search

A shard is a Lucene index, and a Lucene index is a pile of immutable segments. Classically, the query phase walked those segments one thread, one segment at a time: a single search thread iterated every LeafReaderContext sequentially, collecting hits. On a shard with many segments and idle CPU cores, that is a waste — the segments are independent and could be searched in parallel.

Concurrent segment search does exactly that: it splits a shard's segments into slices and searches the slices in parallel on the search threadpool, then reduces the per-slice results into one shard-local answer. It is one of the cleanest examples in this curriculum of an RFC that soaked behind a setting and then became a default at the cluster level in 3.0. It is also a clean lesson in Lucene's CollectorManager contract and the latency-vs-CPU trade-off that every parallelism feature must confront.

After this chapter you should be able to: explain what a slice is and how slices map to threads; describe the CollectorManager/Collector contract and why concurrency requires it; find the settings that turn it on and bound slice count; reason about when it helps and when it hurts; and grep your way to the code that owns the slicing decision.

Note: This sits squarely on the OpenSearch/Lucene boundary. The slicing and the IndexSearcher executor are Lucene mechanisms; OpenSearch chooses the executor, the slice strategy, the settings, and the integration with the search execution path.

The problem, precisely

A search request fans out to one copy of each shard (the Hitchhiker's Guide "return trip"). On each shard, the query phase runs the Lucene query and collects the top-K. Without concurrency, that phase is single-threaded per shard, so a shard's query latency is bounded by all its segments processed in series — even when the node has spare cores.

The win is highest when:

• A shard has many sizable segments (the parallelism has something to chew).
• The query is CPU-bound per segment (aggregations, expensive scoring).
• The node has idle cores to spend.

The win evaporates — or reverses — when segments are tiny (coordination overhead dominates), when the node is already CPU-saturated (you just added contention), or when the query is cheap (latency was never the bottleneck). Concurrency is not free, and this feature's whole design is about spending CPU to buy latency only when that trade is good.

Slices: the unit of parallelism

A slice is a group of one or more segments (LeafReaderContexts) assigned to a single task. Lucene's IndexSearcher, when given an Executor, partitions its leaves into slices and submits one task per slice. OpenSearch controls how many slices there are and how leaves are grouped.

flowchart TD
    Shard["shard query phase"] --> Slice["compute leaf slices"]
    Slice --> S0["slice 0: seg _0,_1"]
    Slice --> S1["slice 1: seg _2,_3"]
    Slice --> S2["slice 2: seg _4"]
    S0 --> C0["leaf collector 0"]
    S1 --> C1["leaf collector 1"]
    S2 --> C2["leaf collector 2"]
    C0 --> R["CollectorManager.reduce(...)"]
    C1 --> R
    C2 --> R
    R --> Top["shard-local top-K"]

Slice count is the central knob. Too few slices and you leave parallelism on the table; too many and you pay task-scheduling and reduce overhead for nothing. OpenSearch bounds it with a max-slice-count setting and a default strategy that balances segment sizes across slices rather than blindly one-segment-per-slice (which would be terrible for many tiny segments).

The CollectorManager contract (why this needs a new abstraction)

A Lucene Collector is not thread-safe — it accumulates state (the heap of top hits, the aggregation buckets) as it visits docs. You cannot share one collector across slices running on different threads. The solution is the CollectorManager:

CollectorManager<C extends Collector, T> {
    C newCollector();           // one fresh collector PER slice/thread
    T reduce(Collection<C> cs); // merge the per-slice collectors into one result
}

This is the heart of the feature. Each slice gets its own collector via newCollector(); the slices run in parallel, each collecting independently; then reduce(...) merges them into the shard-local answer. Converting an aggregation or a top-docs collector to be concurrency-safe means expressing it as a CollectorManager with a correct, associative reduce — and that is exactly the work that made many aggregations concurrent-search-compatible (and exactly where the subtle bugs live).

cd ~/src/OpenSearch
# OpenSearch's collector-manager wiring for the query phase.
grep -rln "CollectorManager" server/src/main/java/org/opensearch/search/ | head
grep -rn "newCollector\|public .* reduce" \
  server/src/main/java/org/opensearch/search/query/ | head

Warning: A reduce that is not associative/commutative is a latent correctness bug. Single-threaded, the segments are visited in a fixed order and the bug hides; concurrent, the order changes and results wobble. This is the single most common class of concurrent-search defect — see the bugs table.

The executor: which threadpool runs the slices

The per-slice tasks run on a dedicated executor so they do not starve other work. OpenSearch wires a search executor into the Lucene IndexSearcher for concurrent mode. Find it and the threadpool it draws from:

# Where the concurrent searcher / executor is constructed.
grep -rn "IndexSearcher\|setExecutor\|sliceExecutor\|SliceExecutor\|index_searcher" \
  server/src/main/java/org/opensearch/search/ | head
# The threadpool registry (look for an index_searcher / search pool).
grep -rn "index_searcher\|ThreadPool.Names" \
  server/src/main/java/org/opensearch/threadpool/ThreadPool.java | head

The relationship to the broader thread model is in threadpools and concurrency. The key point: concurrent segment search consumes a finite pool, which is why turning it on globally changes the CPU economics of the whole node — the trade-off is cluster-wide, not per-query.

The reduce after slices

After all slices finish, CollectorManager.reduce produces one shard-local result. That is distinct from the coordinating-node reduce you already know from aggregations: there are now two reduce levels.

flowchart TD
    subgraph Shard A
      sa0["slice 0"] --> sared["slice reduce"]
      sa1["slice 1"] --> sared
    end
    subgraph Shard B
      sb0["slice 0"] --> sbred["slice reduce"]
      sb1["slice 1"] --> sbred
    end
    sared --> coord["coordinating-node reduce (SearchPhaseController)"]
    sbred --> coord
    coord --> answer["global answer"]

This two-level reduce is why an aggregation must be correct both at the slice level and at the shard level. An aggregation that was only ever tested with the single-level (coordinating) reduce can be subtly wrong once a second reduce level appears beneath it.

Settings and defaults

# Find the real setting keys and defaults in your checkout (do not trust docs blindly).
grep -rn "concurrent_segment_search\|concurrent.max_slice_count\|CONCURRENT_SEGMENT_SEARCH" \
  server/src/main/java/org/opensearch/ | head -20

# Turn it on/off and bound slices at runtime.
curl -s -XPUT 'localhost:9200/_cluster/settings' -H 'Content-Type: application/json' -d '{
  "persistent": {
    "search.concurrent_segment_search.enabled": true,
    "search.concurrent.max_slice_count": 4
  }
}'

Note: "Default-on in 3.0" is a cluster-level default; it can still be disabled per-index or per-cluster. When you benchmark, always record the effective setting, because a regression report that does not state the slice count is unfalsifiable.

The trade-offs, as a decision table

The honest summary: concurrent segment search trades CPU and throughput for single-query latency. On a lightly loaded cluster with fat segments it is a clear win; on a saturated cluster it can make things worse. This is why the optimizing-slicing capstone is about the slicing heuristic, not the feature's existence.

Trace it yourself

# 1. Create an index, force several segments by indexing in batches with refresh.
curl -s -XPUT 'localhost:9200/csstest' -H 'Content-Type: application/json' -d '
{ "settings": { "number_of_shards": 1, "number_of_replicas": 0 } }'
for b in 1 2 3 4 5; do
  for i in $(seq 1 2000); do printf '{"index":{}}\n{"v":%s,"g":"k%s"}\n' "$((RANDOM))" "$((RANDOM%50))"; done \
    | curl -s -H 'Content-Type: application/x-ndjson' \
        -XPOST 'localhost:9200/csstest/_bulk?refresh=true' --data-binary @- >/dev/null
done
curl -s 'localhost:9200/_cat/segments/csstest?v'   # confirm multiple segments

# 2. Run a CPU-bound agg with concurrency OFF, then ON; compare took.
curl -s -XPUT 'localhost:9200/_cluster/settings' -H 'Content-Type: application/json' \
  -d '{"transient":{"search.concurrent_segment_search.enabled":false}}'
curl -s 'localhost:9200/csstest/_search?pretty' -H 'Content-Type: application/json' \
  -d '{"size":0,"aggs":{"g":{"terms":{"field":"g","size":50}}}}' | grep '"took"'

curl -s -XPUT 'localhost:9200/_cluster/settings' -H 'Content-Type: application/json' \
  -d '{"transient":{"search.concurrent_segment_search.enabled":true}}'
curl -s 'localhost:9200/csstest/_search?pretty' -H 'Content-Type: application/json' \
  -d '{"size":0,"aggs":{"g":{"terms":{"field":"g","size":50}}}}' | grep '"took"'

# 3. profile=true shows per-slice work in the breakdown.
curl -s 'localhost:9200/csstest/_search?pretty' -H 'Content-Type: application/json' \
  -d '{"profile":true,"size":0,"aggs":{"g":{"terms":{"field":"g"}}}}' | grep -i "slice\|collector" | head

# 4. Read the owning code.
grep -rln "ContextIndexSearcher\|slice\|CollectorManager\|reduce" \
  server/src/main/java/org/opensearch/search/internal/ | head
./gradlew :server:test --tests "*ConcurrentSegmentSearch*" 2>&1 | tail -15
./gradlew :server:internalClusterTest --tests "*ConcurrentSearch*" 2>&1 | tail -15

Common bugs and symptoms

Validation: prove you understand this

1. Define a slice and explain how slices map to threads and to segments. Why is one-segment-per-slice a bad default for many tiny segments?
2. Write the CollectorManager contract from memory and explain why concurrent search requires it rather than a single shared Collector.
3. Draw the two reduce levels (slice reduce and coordinating-node reduce) and say which classes own each.
4. Name the setting that enables the feature and the one that bounds slice count, and state what changed in 3.0.
5. Give two concrete situations where concurrent segment search hurts, and the metric you would watch to detect each.
6. Explain why a non-associative reduce is invisible single-threaded but breaks under concurrency.

Next: Backpressure and Admission Control, or the optimizing-slicing capstone.