Concurrent Segment Search — Intensive

This masterclass extends the Concurrent Segment Search engineering chapter. That chapter told you what concurrent segment search is — slices, the CollectorManager contract, the two-level reduce, the trade-offs. Read it first; this intensive assumes it. Here we go down to the algorithm and the code: the exact slicing heuristic and its constants, the Lucene IndexSearcher machinery that builds LeafSlices, how OpenSearch's ContextIndexSearcher drives them on a real threadpool, the precise CollectorManager contract with associativity proofs, and the byte-for-byte reduce chain from per-slice → per-shard → coordinator. Then three labs make you profile it, build a CollectorManager from scratch, and benchmark the crossover.

By the end you can: derive the default slice count for a given segment topology by hand; explain why a non-associative reduce is invisible single-threaded and broken concurrent; locate every class in the path with grep; and state, with numbers, exactly when concurrency wins and when it loses.

Note: 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. Throughout, "OpenSearch class" vs "Lucene class" matters — when you file a bug you need to know which project owns the code.

First principles: why parallelize across segments at all

A shard is a Lucene index. A Lucene index is a pile of immutable segments. Each segment is a fully self-contained mini-index: its own terms dictionary, postings, doc-values, points. Critically, segments are independent — to count how many documents match a query, you can count per segment and sum, with no cross-segment coordination during the scan.

That independence is the whole opportunity. Classically the query phase walked the segments one thread, one LeafReaderContext at a time:

for (LeafReaderContext leaf : reader.leaves()) {
    LeafCollector lc = collector.getLeafCollector(leaf);
    BulkScorer scorer = weight.bulkScorer(leaf);
    scorer.score(lc, leaf.reader().getLiveDocs());   // scan this segment
}

A shard with eight fat segments and eight idle cores spends seven cores doing nothing while one core grinds through all eight segments in series. The latency of that query is the sum of the per-segment costs. If you could run each segment on its own core, the latency would approach the max of the per-segment costs instead. That is the entire pitch: turn a sum into a max by spending idle CPU.

The reason this is Amdahl-friendly is that the per-segment scan is the dominant, embarrassingly parallel part of an expensive query. The serial part — building the query weight, the final reduce — is small relative to scanning millions of docs through an aggregation. So the speedup ceiling is high when the per-segment work is large. When the per-segment work is tiny (a few-term match on a small segment), the serial overhead dominates and parallelism is a net loss. Hold that thought; it's the crux of the trade-off and of Lab CS3.

flowchart LR
    subgraph Sequential["Sequential: latency = sum of segment costs"]
        seq0["seg _0 (12ms)"] --> seq1["seg _1 (10ms)"] --> seq2["seg _2 (11ms)"] --> seq3["seg _3 (9ms)<br/>total 42ms"]
    end
    subgraph Concurrent["Concurrent: latency ≈ max + reduce"]
        c0["seg _0 (12ms)"]
        c1["seg _1 (10ms)"]
        c2["seg _2 (11ms)"]
        c3["seg _3 (9ms)"]
        c0 --> red["reduce (~1ms)<br/>total ≈ 13ms"]
        c1 --> red
        c2 --> red
        c3 --> red
    end

The slice model: LeafSlice, IndexSearcher#slices, and the heuristic

A slice is a group of one or more segments assigned to a single task. It is not one segment per thread — that would be a disaster for a shard with hundreds of tiny segments (you'd spawn hundreds of tasks, each scanning a handful of docs). Lucene groups segments into slices with a heuristic that balances two budgets:

• maxDocsPerSlice — a slice should hold roughly this many documents before a new slice is started. The Lucene default is about 250,000.
• maxSegmentsPerSlice — a slice should hold at most this many segments regardless of doc count. The Lucene default is 5.

These constants live in Lucene's IndexSearcher. Do not trust this page — grep your checkout to confirm the exact values and signature for your version:

cd ~/src/lucene   # an apache/lucene checkout
grep -rn "maxDocsPerSlice\|maxSegmentsPerSlice\|MAX_DOCS_PER_SLICE\|MAX_SEGMENTS_PER_SLICE" \
  lucene/core/src/java/org/apache/lucene/search/IndexSearcher.java
# The slicing method itself:
grep -n "protected LeafSlice\[\] slices\|static LeafSlice\[\] slices\|class LeafSlice" \
  lucene/core/src/java/org/apache/lucene/search/IndexSearcher.java

LeafSlice is a tiny value class — essentially an array of LeafReaderContextPartition (or, in older Lucene, LeafReaderContext):

// org.apache.lucene.search.IndexSearcher.LeafSlice  (shape; grep to confirm fields)
public static class LeafSlice {
    public final LeafReaderContext[] leaves;     // the segments in this slice
    // (newer Lucene wraps these in LeafReaderContextPartition to allow
    //  splitting a single huge segment across slices)
}

The grouping algorithm, step by step

The default slices(...) algorithm is deterministic and worth knowing cold, because the labs make you reproduce its output by hand:

1. Sort the leaves by doc count, descending (biggest segments first).
2. Walk the sorted leaves, accumulating them into the current slice.
3. Start a new slice when adding the next leaf would push the current slice over maxDocsPerSlice and the current slice already has at least one leaf — unless the current leaf itself is bigger than maxDocsPerSlice (a giant segment gets its own slice).
4. Also start a new slice once the current slice reaches maxSegmentsPerSlice leaves.
5. The result is an array of LeafSlice, each a balanced bundle of segments.

Sorting biggest-first matters: it puts a giant segment in its own slice early and packs the long tail of small segments together, so you don't end up with one slice holding all the big work and another holding dust.

flowchart TD
    Start["leaves sorted by docCount desc"] --> Loop{"more leaves?"}
    Loop -- yes --> Check{"current slice<br/>over maxDocsPerSlice<br/>OR == maxSegmentsPerSlice?"}
    Check -- "yes (and slice non-empty)" --> New["close slice,<br/>start new slice"]
    Check -- no --> Add["add leaf to current slice"]
    New --> Add
    Add --> Loop
    Loop -- no --> Done["emit LeafSlice[]"]

Worked example — derive the slice count by hand

Suppose a shard has these segments (doc counts), the defaults maxDocsPerSlice = 250_000, maxSegmentsPerSlice = 5:

_0: 600,000   _1: 240,000   _2: 200,000   _3: 50,000
_4: 40,000    _5: 30,000    _6: 20,000    _7: 10,000

Walk the algorithm (already sorted desc):

Result: 4 slices. Slice A is one giant segment; B and C are ~200–250k each; D bundles the four small segments into one task. That's exactly the balancing you want — four tasks of comparable cost, no task scanning dust alone. Reproduce this in Lab CS1 and watch profile=true report the same slice count.

Note: OpenSearch can override the slice count. With search.concurrent.max_slice_count = 0 you get Lucene's heuristic above. With max_slice_count = N > 0 you get at most N slices, packed differently — a fixed cap rather than the doc-budget heuristic. Grep ContextIndexSearcher (next section) to see which path your branch takes.

ContextIndexSearcher: OpenSearch's driver

OpenSearch doesn't use a bare Lucene IndexSearcher. It subclasses it as ContextIndexSearcher, which (a) carries the per-shard SearchContext, (b) decides whether to run concurrently at all, (c) supplies the executor — the search threadpool — and (d) can override slices(...) to honor max_slice_count. Find it:

cd ~/src/OpenSearch
find server -name ContextIndexSearcher.java
grep -n "class ContextIndexSearcher\|slices\|getExecutor\|setExecutor\|getSlices\|sliceCount\|searchContext" \
  server/src/main/java/org/opensearch/search/internal/ContextIndexSearcher.java | head -30

The executor is the key wiring. A Lucene IndexSearcher runs concurrently only if it was constructed with an Executor. OpenSearch passes the search executor when concurrent mode is on, and null (or a same-thread executor) when it's off — that single choice flips the whole feature:

# Where the searcher gets its executor, and the threadpool it draws from.
grep -rn "ContextIndexSearcher(" \
  server/src/main/java/org/opensearch/search/ | head
grep -n "index_searcher\|INDEX_SEARCHER\|ThreadPool.Names" \
  server/src/main/java/org/opensearch/threadpool/ThreadPool.java | head

The dedicated pool (commonly index_searcher) exists so that per-slice tasks don't starve — or get starved by — the main search pool. The relationship to the broader thread model is in threadpools and concurrency. The load-bearing consequence: concurrent segment search consumes a finite pool, so turning it on globally changes the CPU economics of the whole node, not just one query. A node serving 200 concurrent searches, each now wanting 4 slices, suddenly wants 800 slice-tasks of search-pool work.

flowchart TD
    QP["QueryPhase.execute(searchContext)"] --> CIS["ContextIndexSearcher.search(query, collectorManager)"]
    CIS --> Decide{"executor != null<br/>(concurrent mode)?"}
    Decide -- no --> Seq["single-thread: iterate leaves,<br/>one Collector"]
    Decide -- yes --> Slices["slices(leaves) -> LeafSlice[]"]
    Slices --> Submit["submit one task per slice<br/>to index_searcher pool"]
    Submit --> NC["collectorManager.newCollector() per slice"]
    NC --> Scan["each task scans its segments"]
    Scan --> Reduce["collectorManager.reduce(collectors)"]
    Seq --> Result["shard-local result"]
    Reduce --> Result

The CollectorManager contract (this is the heart of it)

A Lucene Collector accumulates mutable state as it visits docs — the heap of top hits, the aggregation buckets, a running count. It is not thread-safe, and you cannot share one across slices running on different threads. The abstraction that makes concurrency safe is the CollectorManager:

// org.apache.lucene.search.CollectorManager<C extends Collector, T>
public interface CollectorManager<C extends Collector, T> {
    C newCollector() throws IOException;          // a FRESH collector per slice
    T reduce(Collection<C> collectors) throws IOException;  // merge them
}

Two methods, one invariant, and the whole feature lives or dies on the invariant:

• newCollector() is called once per slice. Each slice gets its own collector with its own mutable state. No two slices ever touch the same collector instance. This is what makes the parallel scan correct without locks.
• reduce(collectors) merges the per-slice collectors into one shard-local result T. It runs after all slices finish, on a single thread.

Contrast with a plain Collector, which the sequential path uses directly: one collector, visited segment-by-segment, no reduce. The CollectorManager is the concurrent generalization, and Lucene's IndexSearcher.search(Query, CollectorManager) overload is the entry point that uses the executor.

cd ~/src/OpenSearch
grep -rln "implements CollectorManager\|CollectorManager<" \
  server/src/main/java/org/opensearch/search/ | head
grep -rn "newCollector()\|public .* reduce(" \
  server/src/main/java/org/opensearch/search/query/ | head

Why reduce MUST be associative and commutative

Here is the subtlety that produces the nastiest concurrent-search bugs. Slices run in nondeterministic order, and reduce receives the collectors in an order that can vary run to run. If reduce is associative and commutative, the merged result is identical regardless of order. If it isn't, the result wobbles — and, fatally, the wobble is invisible single-threaded, because the sequential path visits segments in a fixed order and never calls reduce at all.

A correct example — counting matched docs:

// COUNT collector manager: reduce is sum, which is associative + commutative.
static final class CountCM implements CollectorManager<CountCollector, Long> {
    public CountCollector newCollector() { return new CountCollector(); }
    public Long reduce(Collection<CountCollector> cs) {
        long total = 0;
        for (CountCollector c : cs) total += c.count;   // order-independent
        return total;
    }
}

A broken example — "first non-empty wins":

// BROKEN: reduce depends on iteration order. Single-threaded this never runs;
// concurrent, the answer changes when slice order changes.
public String reduce(Collection<LabelCollector> cs) {
    for (LabelCollector c : cs) if (c.label != null) return c.label;  // WRONG
    return null;
}

Top-K is the canonical correct non-trivial case: each slice keeps its own priority queue of the top size hits; reduce does a merge of the queues (take all entries, keep the global top size), which is associative and commutative under the sort comparator. You build exactly this in Lab CS2.

Warning: A reduce that isn't associative/commutative is a latent correctness bug, not a crash. It passes single-threaded tests forever and only manifests once concurrency is on and a shard has more than one slice. The fix in the test suite is to force multi-slice execution and assert identical results against the sequential path — see the bugs table and the labs.

Why aggregations had to become slice-safe

Aggregations were the hard part of shipping concurrent segment search. An aggregation is a tree of collectors (AggregatorBase subclasses), and the tree had been written assuming one collector visited all segments in order. Concurrency breaks that assumption in two places:

1. Each slice needs its own aggregator tree. You can't share bucket ordinals, hash tables, or big-array state across threads. So the aggregation path was reworked to produce a CollectorManager whose newCollector() builds a fresh aggregator (or a fresh MultiBucketCollector wrapping several) per slice.
2. A new reduce level appears beneath the coordinator reduce. Previously aggregations reduced once, on the coordinating node, via InternalAggregation.reduce (see Aggregations). Now there is a per-slice reduce on the shard first, then the per-shard result goes to the coordinator for the second reduce. An aggregation that was only ever correct for the single (coordinator) reduce can be subtly wrong once a second reduce level is introduced beneath it.

grep -rn "class AggregatorBase\|MultiBucketCollector\|CollectorManager\|getLeafCollector" \
  server/src/main/java/org/opensearch/search/aggregations/ | head -20
# The wrapper that lets several bucket sub-aggregators be collected together:
find server -name MultiBucketCollector.java

MultiBucketCollector is the multiplexer: a bucket aggregation has many sub-aggregators, and MultiBucketCollector wraps them so they're collected together and reduced together, per slice. The combination — AggregatorBase producing per-slice trees, MultiBucketCollector multiplexing sub-aggregators, and a slice-level reduce — is what made aggregations concurrent-safe.

The full reduce chain: per-slice → per-shard → coordinator

This is the single most important diagram in the masterclass. There are now three levels where partial results combine, and confusing them is the source of half the bugs:

flowchart TD
    subgraph ShardA["Shard A (data node)"]
      a0["slice 0 collector"] --> ar["CollectorManager.reduce<br/>(slice reduce)"]
      a1["slice 1 collector"] --> ar
      a2["slice 2 collector"] --> ar
      ar --> aqr["QuerySearchResult (shard-local top-K + agg partials)"]
    end
    subgraph ShardB["Shard B (data node)"]
      b0["slice 0 collector"] --> br["CollectorManager.reduce<br/>(slice reduce)"]
      b1["slice 1 collector"] --> br
      br --> bqr["QuerySearchResult"]
    end
    aqr --> coord["SearchPhaseController.reducedQueryPhase<br/>(coordinator reduce)"]
    bqr --> coord
    coord --> ans["global top-K + reduced aggs"]

The slice reduce is new with concurrent search; the coordinator reduce you already knew from search execution and aggregations. The discipline a contributor must internalize: an aggregation has to be correct at both the slice reduce and the coordinator reduce. A test that only exercises one level can pass while the other is broken.

# The two reduce sites, side by side:
grep -n "reduce" server/src/main/java/org/opensearch/search/query/QueryPhase.java | head
grep -n "reducedQueryPhase\|InternalAggregation\|reduce" \
  server/src/main/java/org/opensearch/action/search/SearchPhaseController.java | head

Settings and defaults

# Confirm the real keys, defaults, and types in YOUR checkout — don't trust docs.
grep -rn "concurrent_segment_search\|concurrent.max_slice_count\|CONCURRENT_SEGMENT_SEARCH\|MAX_SLICE_COUNT" \
  server/src/main/java/org/opensearch/ | head -20

# Toggle at runtime and bound slices.
curl -s -XPUT 'localhost:9200/_cluster/settings' -H 'Content-Type: application/json' -d '{
  "persistent": {
    "search.concurrent_segment_search.mode": "all",
    "search.concurrent.max_slice_count": 4
  }
}'

Note: "Default-on in 3.0" is a cluster-level default — still overridable per index and per cluster. The historical arc is a clean example of RFC → feature flag → opt-in setting → default: the original request-for-comments lived at https://github.com/opensearch-project/OpenSearch/issues/10024 (search "concurrent segment search" in the OpenSearch issues if that number has moved). Whenever you report a regression, record the effective mode and slice count — a report that omits them is unfalsifiable.

Trade-offs: when it wins, when it loses

The honest one-liner: concurrent segment search trades CPU and throughput for single-query latency. On a lightly loaded cluster with fat segments it's a clear win; on a saturated cluster it can make things worse.

The force-merge interaction (a subtle one)

Force-merge to a single segment is the classic "make this read-only index fast" move. But a shard with one segment has one slice — there is nothing to parallelize. So force-merging an index to 1 segment disables concurrent segment search for that index, by construction. For a read-heavy index you often want force-merge to a small number of segments (say, one per core you're willing to spend) rather than to 1, so concurrency still has something to chew. This is a real tuning decision, not a footnote — quantify it in Lab CS3.

# One segment after this -> no concurrency:
curl -s -XPOST 'localhost:9200/idx/_forcemerge?max_num_segments=1'
# A few segments -> concurrency still applies:
curl -s -XPOST 'localhost:9200/idx/_forcemerge?max_num_segments=4'

Common bugs and symptoms

Validation: prove you understand this

1. Derive, by hand, the number of slices for a shard with segments of {500k, 300k, 250k, 100k, 60k, 40k, 30k} docs under maxDocsPerSlice=250k, maxSegmentsPerSlice=5. Show your slices.
2. Write the CollectorManager contract from memory. Explain why concurrent search needs it instead of a single shared Collector, in terms of mutable state and locks.
3. Draw the three reduce levels (slice → shard result → coordinator) and name the class that owns each.
4. Give a concrete reduce that is correct and one that is broken, and explain why the broken one passes single-threaded tests forever.
5. A read-only index is force-merged to 1 segment and concurrent search "stopped helping." Explain why and give the fix.
6. Name the setting that enables the feature, the one that selects the mode, and the one that bounds slice count; state what changed in 3.0.
7. Why must an aggregation be correct at both reduce levels? Give a failure mode that only shows up at the slice level.

Next: Lab CS1 — Enable and Profile. Then Lab CS2 — CollectorManager and Lab CS3 — Benchmark and Tune. This intensive feeds Capstone Project 3.