### Describe the bug A `top_hits` aggregation whose hits also carry nested `inner_hits` corrupts `_source` reads when concurrent segment search is active. Two slice threads mutate one shared `InnerHitsContext.InnerHitSubContext`, and through it one shared `SourceLookup`, which caches a single Lucene stored-fields merge instance. That instance owns one `IndexInput` and one block state, so interleaved reads decode from a wrong byte offset. The rest of this path is already per-slice by construction: each slice builds its own `SubSearchContext`, and each fetch its own `FetchContext`. The inner-hits contexts are the single piece of per-hit fetch state that escapes that, because `SubSearchContext` inherits `innerHits()` from `FilteredSearchContext`, which forwards to the request-level context. The failure surfaces three ways. Most reads throw a caught `RuntimeException` and the shard fails. Some throw `CorruptIndexException` against files whose checksums are perfect, which reads as index corruption and misdirects diagnosis. A minority throw `AssertionError: Unknown type flag: N`, which is an `Error`, escapes `SourceLookup`'s `catch (Exception)`, reaches `OpenSearchUncaughtExceptionHandler`, and kills the node. ### Related component Search:Aggregations ### To Reproduce Two conditions must both hold, and the second is what makes it show up at all. 1. One request carrying **both** a top-level nested `inner_hits` **and** a `top_hits` aggregation, so `InnerHitsPhase` runs inside `TopHitsAggregator`'s fetch. 2. A **long** per-slice fetch window. The bug needs two slice threads inside `InnerHitsPhase.hitExecute` at the same moment, so the aggregation must produce many buckets each holding hits. With 2 buckets of 3 hits it effectively never fires. With 1000 buckets of 10 hits it fires on almost every request.

Details

Index settings: ``` PUT //_settings { "index.search.concurrent_segment_search.mode": "all", "index.search.concurrent.max_slice_count": 8 } ``` Request shape — a nested field with `inner_hits`, plus a `filter -> terms -> top_hits` aggregation over a high-cardinality keyword field: ```json { "size": 50, "query": { "nested": { "path": "", "score_mode": "max", "query": { "match_all": {} }, "inner_hits": { "size": 5, "name": "innerA" } } }, "aggs": { "f": { "filter": { "match_all": {} }, "aggs": { "t": { "terms": { "field": "", "size": 1000 }, "aggs": { "th": { "top_hits": { "size": 10, "_source": [""], "sort": [ { "": { "order": "desc" } } ] } } } } } } } } ``` Send it about 10 times. Then set `"index.search.concurrent_segment_search.mode": "none"` and send it again. Observed on a single-node cluster, one shard, ~1.25M docs in 10 segments, documents with a multi-kilobyte `_source`. Ten requests per row: | `mode` / `max_slice_count` | HTTP 200 | failed | |---|---|---| | `none` / 1 | 10/10 | 0 | | `all` / 8 | 0/10 | 10 | | `none` / 1, repeated | 10/10 | 0 | | `all` / 2 | 1/10 | 9 | The third row repeats the first *after* the failing arm, so the failures track the setting rather than a warm-up, a cache state, or the order the arms ran in. Small documents make this much harder to hit, because many docs then share one compressed block and no seek occurs between reads. The `filter` wrapper and the `match_all` inner query above come from the workload this was reduced from. We did not test whether a bare `terms` → `top_hits` reproduces without them, so treat both as incidental rather than required.

### Expected behavior Expected: the request succeeds, or fails for a reason unrelated to thread safety. Concurrency settings should not change correctness.

Details

Actual, shard failures and a node exit: ``` index_out_of_bounds_exception Index 8 out of bounds for length 5 index_out_of_bounds_exception Index 17 out of bounds for length 12 array_index_out_of_bounds_exception Index 1 out of bounds for length 1 null_pointer_exception Cannot invoke "SearchHit.score(float)" because ... is null ``` In the fatal case: ``` [ERROR][o.o.b.OpenSearchUncaughtExceptionHandler] fatal error in thread [opensearch[][search][T#12]], exiting java.lang.AssertionError: Unknown type flag: 7 at o.a.l.codecs.lucene90.compressing.Lucene90CompressingStoredFieldsReader.skipField(:320) at o.a.l.codecs.lucene90.compressing.Lucene90CompressingStoredFieldsReader.document(:699) at org.opensearch.search.lookup.SourceLookup.loadSourceIfNeeded(SourceLookup.java:104) at org.opensearch.search.fetch.FetchPhase.prepareNestedHitContext(FetchPhase.java:402) at org.opensearch.search.fetch.FetchPhase.prepareHitContext(FetchPhase.java:320) at org.opensearch.search.fetch.FetchPhase.execute(FetchPhase.java:168) at org.opensearch.search.fetch.subphase.InnerHitsPhase.hitExecute(InnerHitsPhase.java:105) at org.opensearch.search.fetch.subphase.InnerHitsPhase$1.process(InnerHitsPhase.java:81) at org.opensearch.search.fetch.FetchPhase.execute(FetchPhase.java:178) at org.opensearch.search.aggregations.metrics.TopHitsAggregator.buildAggregation(TopHitsAggregator.java:218) ... at org.opensearch.search.aggregations.BucketCollectorProcessor.processPostCollection(BucketCollectorProcessor.java:78) at org.opensearch.search.internal.ContextIndexSearcher.search(ContextIndexSearcher.java:309) at org.apache.lucene.search.TaskExecutor$Task.run(TaskExecutor.java:173) ``` The bottom frames show the fetch running inside a Lucene concurrent-search slice thread.

### Additional Details ### Root cause Start from what is *not* shared, because it narrows the defect to one object. `AggregationCollectorManager.newCollector()` is called once per slice and builds a fresh aggregator tree, so `TopHitsAggregatorFactory.createInternal` runs `new SubSearchContext(searchContext)` per slice. `FetchPhase.execute` then builds its own context per call, and with it a fresh `SearchLookup` and `SourceLookup`: ```java FetchContext fetchContext = new FetchContext(context); // FetchPhase.execute this.searchLookup = searchContext.getQueryShardContext().newFetchLookup(); // FetchContext ctor ``` One thing escapes that. `SubSearchContext extends FilteredSearchContext`, which owns no `InnerHitsContext` of its own: ```java public InnerHitsContext innerHits() { return in.innerHits(); } // FilteredSearchContext ``` So every slice's sub-context resolves `innerHits()` to the one `InnerHitsContext` built at parse time by `InnerHitContextBuilder.build` → `InnerHitsContext.addInnerHitDefinition`. `InnerHitsPhase.getProcessor` captures that map once, and every slice thread then mutates the same objects once per hit: ```java Map innerHits = searchContext.innerHits().getInnerHits(); // shared across slices ... innerHitsContext.docIdsToLoad(docIdsToLoad, 0, docIdsToLoad.length); innerHitsContext.setId(hit.getId()); innerHitsContext.setRootLookup(rootLookup); fetchPhase.execute(innerHitsContext, "fetch_inner_hits[" + entry.getKey() + "]"); ``` Those three setters are the defect. An `InnerHitSubContext` holds a definition that never changes, plus three fields re-aimed at each hit, and a second thread can re-aim them between another thread's write and the read on the next line. Which field is lost decides the symptom: | Overwritten | The nested fetch then | Surfaces as | |---|---|---| | `docIdsToLoad` | loads another hit's child doc ids under this root | `index_out_of_bounds_exception`, NPE on `SearchHit.score` | | `rootLookup` | reads through another thread's `SourceLookup` | `CorruptIndexException`, `AssertionError: Unknown type flag: N` | `FetchPhase.prepareNestedHitContext` then calls `innerHitsContext.getRootLookup().loadSourceIfNeeded()`. `SourceLookup`'s class javadoc already says "Not thread safe", and it lazily caches a single reader: ```java if (fieldReader == null) { if (reader instanceof SequentialStoredFieldsLeafReader) { SequentialStoredFieldsLeafReader lf = (SequentialStoredFieldsLeafReader) reader; fieldReader = lf.getSequentialStoredFieldsReader()::document; // merge instance } else { fieldReader = reader.storedFields()::document; } } ``` `Lucene90CompressingStoredFieldsReader.serializedDocument` mutates shared state with no lock: ```java if (state.contains(docID) == false) { fieldsStream.seek(indexReader.getStartPointer(docID)); state.reset(docID); } return state.document(docID); ``` One thread seeks and resets for its doc while another does the same, so a thread reads its field header from the other's block position. Valid stored-field type flags are 0 through 5 while `TYPE_MASK` is 7, so 6 and 7 are detectably illegal and produce the `AssertionError`. The other six decode as plausible garbage. Isolating the OpenSearch link on its own: driving one `SourceLookup` from 4 threads failed **42,148 of 80,000** reads, and giving each thread its own failed **0 of 80,000**, with nothing else changed. Note that doc-ID ordering is *not* the problem. `state.document()` indexes a per-doc `offsets[]` array, so out-of-order access inside a loaded block decodes correctly. Sharing is the defect. ### Reaching the fatal throw without OpenSearch The node-killing draw is rare enough that a cluster-level run is a poor way to reach it. Sharing one merge instance across threads reaches it directly, with `lucene-core` alone and no OpenSearch in the loop.

Self-contained harness (lucene-core only)

```java import org.apache.lucene.codecs.StoredFieldsReader; import org.apache.lucene.document.*; import org.apache.lucene.index.*; import org.apache.lucene.store.Directory; import org.apache.lucene.store.MMapDirectory; import java.nio.file.*; import java.util.*; import java.util.concurrent.*; import java.util.concurrent.atomic.*; /** * Shares ONE stored-fields merge instance across threads, as a shared SourceLookup * does inside InnerHitsPhase, and counts the type tags that come back. * * javac -cp lucene-core.jar F7.java * java -Xmx4g -DrunMs=150000 -cp lucene-core.jar:. F7 // assertions OFF, like production */ public class F7 { static final int NDOCS = 20000, PAYLOAD = 1500, EXTRA_FIELDS = 12, THREADS = 10; public static void main(String[] args) throws Exception { long runMs = Long.getLong("runMs", 60_000L); Path dir = Files.createTempDirectory("f7"); try (Directory d = new MMapDirectory(dir)) { index(d); try (DirectoryReader dr = DirectoryReader.open(d)) { LeafReader leaf = dr.leaves().get(0).reader(); StoredFieldsReader base = ((CodecReader) leaf).getFieldsReader(); // The defect in one line: one merge instance, many threads. hammer(base.getMergeInstance(), runMs); } } } static void index(Directory d) throws Exception { char[] filler = new char[PAYLOAD]; Arrays.fill(filler, 'x'); try (IndexWriter w = new IndexWriter(d, new IndexWriterConfig().setUseCompoundFile(false))) { for (int i = 0; i < NDOCS; i++) { Document doc = new Document(); doc.add(new StringField("id", Integer.toString(i), Field.Store.NO)); doc.add(new StoredField("_source", "{\"n\":" + i + ",\"pad\":\"" + new String(filler) + "\"}")); for (int f = 0; f < EXTRA_FIELDS; f++) doc.add(new StoredField("f" + f, i * 31L + f)); w.addDocument(doc); } w.forceMerge(1); } } /** Illegal type tags stay verbatim; everything else is normalised so the histogram stays small. */ static String classify(Throwable e) { String m = String.valueOf(e.getMessage()); if (m.startsWith("Unknown type flag: ")) return m; return e.getClass().getSimpleName() + ": " + m.replaceAll("-?\\d+", "N").replaceAll("\$resource=[^)]*\$", ""); } static void hammer(StoredFieldsReader shared, long runMs) throws Exception { AtomicLong reads = new AtomicLong(), fails = new AtomicLong(); Map byMessage = new ConcurrentHashMap<>(); long deadline = System.nanoTime() + runMs * 1_000_000L; ExecutorService pool = Executors.newFixedThreadPool(THREADS); for (int t = 0; t < THREADS; t++) { pool.submit(() -> { Random rnd = new Random(); StoredFieldVisitor v = new StoredFieldVisitor() { @Override public Status needsField(FieldInfo fi) { // A corrupted read yields a garbage field number, and thus a null // FieldInfo. Tolerating it lets the read travel on to the tag switch. if (fi == null) return Status.NO; return "_source".equals(fi.name) ? Status.YES : Status.NO; } }; while (System.nanoTime() < deadline) { reads.incrementAndGet(); try { shared.document(rnd.nextInt(NDOCS), v); } catch (Throwable e) { fails.incrementAndGet(); byMessage.computeIfAbsent(classify(e), k -> new AtomicLong()).incrementAndGet(); } } }); } pool.shutdown(); pool.awaitTermination(runMs * 3, TimeUnit.MILLISECONDS); System.out.println("reads=" + reads + " failures=" + fails); byMessage.entrySet().stream() .sorted((a, b) -> Long.compare(b.getValue().get(), a.getValue().get())) .forEach(e -> System.out.println(" " + e.getValue() + " " + e.getKey())); } } ``` Two notes on running it. Give it several GB of heap, because a corrupted read can decode a huge length and try to allocate it. And the visitor must tolerate a `null` `FieldInfo`: a garbage field number produces one, and failing there is what stops most corrupted reads from ever reaching the tag switch.

Three runs on `lucene-core` 10.2.1, each 10 threads on one shared merge instance: | Reads | Failing reads | flag 6 | flag 7 | Illegal tag rate | |---|---|---|---|---| | 5,211,343 | 5,210,158 | 189 | 103 | 1 in 17,800 | | 6,098,851 | 6,073,384 | 203 | 73 | 1 in 22,000 | | — | 5,856,445 | 570 | 547 | 1 in 5,200 | No value other than 6 and 7 ever appeared. The rate is not stable — the last row is an earlier run on a different document shape — so read it as order of magnitude only, and as an upper bound besides, since the null-tolerant visitor lets reads travel further than they would in OpenSearch. The point the numbers carry is narrower than a rate: an impossible tag is rare per corrupted read, and a busy shard produces corrupted reads in the millions. Assertions change which error you get, so a regression test should pin this down. With `-ea`, the `assert bits <= NUMERIC_DOUBLE` in `document()` fires first and reports `bits=7`. With assertions off, the read falls through to the switch and `skipField` throws `Unknown type flag: 7`, which is what production traces show. ### Related issues Neither of these covers this defect, and both touch the same path. - #16878 asked for concurrent `inner_hits` execution to improve fetch latency, which is the same code from the performance direction. - #13486 reverted an earlier regression in `InnerHitsPhase`, `InnerHitsContext` and `SearchContext`, where reviewers asked for wider coverage of the uncovered fetch flows. The regression test proposed below is that coverage.

waterWang analysis page

water