[BUG] Shared InnerHitSubContext corrupts `_source` reads under concurrent segment search

[bowenlan-amzn]

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 /<index>/_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:

{
  "size": 50,
  "query": {
    "nested": {
      "path": "<nested_path>",
      "score_mode": "max",
      "query": { "match_all": {} },
      "inner_hits": { "size": 5, "name": "innerA" }
    }
  },
  "aggs": {
    "f": {
      "filter": { "match_all": {} },
      "aggs": {
        "t": {
          "terms": { "field": "<high_cardinality_keyword>", "size": 1000 },
          "aggs": {
            "th": {
              "top_hits": {
                "size": 10,
                "_source": ["<some_field>"],
                "sort": [ { "<sort_field>": { "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[<node-id>][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:

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:

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:

Map<String, InnerHitsContext.InnerHitSubContext> 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:

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:

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)

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<String, AtomicLong> 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.

• [Feature Request] Concurrent executing query in InnerHitsPhase #16878 asked for concurrent inner_hits execution to improve fetch latency, which is the same code from the performance direction.
• Revert "Removing unused fetch sub phase processor initialization during fetch… (#12503)" #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]

I have implemented a fix in PR #22872. The root cause is exactly as analyzed in the issue description: InnerHitsPhase.getProcessor captures the shared request-level inner-hits contexts, so every concurrent slice thread mutates the same InnerHitSubContext (docIdsToLoad / id / rootLookup) — and the shared SourceLookup is not thread safe. The fix deep-copies the inner-hits contexts per fetch so each slice works on its own instance.