[Bug] copy inner-hits contexts per fetch to fix concurrent segment search corruption

modules/parent-join/src/main/java/org/opensearch/join/query/ParentChildInnerHitContextBuilder.java

@@ -91,7 +91,7 @@ protected void doBuild(SearchContext context, InnerHitsContext innerHitsContext)
         if (joinFieldMapper != null) {
             String name = innerHitBuilder.getName() != null ? innerHitBuilder.getName() : typeName;
             JoinFieldInnerHitSubContext joinFieldInnerHits = new JoinFieldInnerHitSubContext(
-                name,
+                getName(),
                 context,
                 typeName,
                 fetchChildInnerHits,
@@ -124,6 +124,19 @@ static final class JoinFieldInnerHitSubContext extends InnerHitsContext.InnerHit
             this.joinFieldMapper = joinFieldMapper;
         }
 
+        @Override
+        public InnerHitsContext.InnerHitSubContext copy() {
+            JoinFieldInnerHitSubContext copy = new JoinFieldInnerHitSubContext(
+                getName(),
+                context,
+                typeName,
+                fetchChildInnerHits,
+                joinFieldMapper
+            );
+            copyTo(copy);
+            return copy;
+        }
+
         @Override
         public TopDocsAndMaxScore topDocs(SearchHit hit) throws IOException {
             Weight innerHitQueryWeight = getInnerHitQueryWeight();
@@ -210,3 +223,4 @@ private String getSortedDocValue(String field, SearchContext context, int docId)
     }
 
 }
+

server/src/main/java/org/opensearch/index/query/NestedQueryBuilder.java

@@ -511,6 +511,13 @@ public TopDocsAndMaxScore topDocs(SearchHit hit) throws IOException {
         public ObjectMapper getChildObjectMapper() {
             return childObjectMapper;
         }
+
+        @Override
+        public InnerHitsContext.InnerHitSubContext copy() {
+            NestedInnerHitSubContext copy = new NestedInnerHitSubContext(getName(), context, parentObjectMapper, childObjectMapper);
+            copyTo(copy);
+            return copy;
+        }
     }
 
     @Override
@@ -524,3 +531,4 @@ public void visit(QueryBuilderVisitor visitor) {
         }
     }
 }
+

server/src/main/java/org/opensearch/search/fetch/subphase/InnerHitsContext.java

@@ -79,6 +79,20 @@ public Map<String, InnerHitSubContext> getInnerHits() {
         return innerHits;
     }
 
+    /**
+     * Returns a deep copy of this context. Each inner-hit definition is copied so that a
+     * fetch running on a concurrent segment search slice thread never shares the mutable
+     * {@link InnerHitSubContext} (doc IDs to load, root id, root {@link SourceLookup})
+     * with fetches running on other slices.
+     */
+    public InnerHitsContext copy() {
+        Map<String, InnerHitSubContext> copies = new HashMap<>();
+        for (Map.Entry<String, InnerHitSubContext> entry : innerHits.entrySet()) {
+            copies.put(entry.getKey(), entry.getValue().copy());
+        }
+        return new InnerHitsContext(copies);
+    }
+
     public void addInnerHitDefinition(InnerHitSubContext innerHit) {
         if (innerHits.containsKey(innerHit.getName())) {
             throw new IllegalArgumentException(
@@ -116,6 +130,26 @@ protected InnerHitSubContext(String name, SearchContext context) {
 
         public abstract TopDocsAndMaxScore topDocs(SearchHit hit) throws IOException;
 
+        /**
+         * Returns a deep copy of this inner-hit definition. The copy keeps the immutable
+         * definition (query, sort, from/size, highlight, fetch configuration) but gets
+         * fresh result holders and per-fetch state, so that concurrent segment search
+         * slice threads never share the mutable fields mutated by
+         * {@link org.opensearch.search.fetch.subphase.InnerHitsPhase#hitExecute}.
+         */
+        public abstract InnerHitSubContext copy();
+
+        /**
+         * Copies the fetch configuration of this context onto a freshly constructed
+         * {@code target} of the same concrete type. Child inner hits are copied recursively.
+         */
+        protected void copyTo(InnerHitSubContext target) {
+            super.copyFetchStateTo(target);
+            if (childInnerHits != null) {
+                target.setChildInnerHits(childInnerHits.copy().getInnerHits());
+            }
+        }
+
         public String getName() {
             return name;
         }
@@ -213,3 +247,4 @@ public static void intersect(Weight weight, Weight innerHitQueryWeight, Collecto
         }
     }
 }
+

server/src/main/java/org/opensearch/search/fetch/subphase/InnerHitsPhase.java

@@ -67,7 +67,12 @@ public FetchSubPhaseProcessor getProcessor(FetchContext searchContext) {
         if (searchContext.innerHits() == null || searchContext.innerHits().getInnerHits().isEmpty()) {
             return null;
         }
-        Map<String, InnerHitsContext.InnerHitSubContext> innerHits = searchContext.innerHits().getInnerHits();
+        // Each fetch (and therefore each concurrent segment search slice thread) gets its
+        // own copy of the inner-hits contexts. The request-level InnerHitsContext and its
+        // InnerHitSubContext hold mutable per-hit state (docIdsToLoad, root id, root
+        // SourceLookup) that is re-aimed for every hit; sharing it across slice threads
+        // corrupts _source reads and can kill the node (see GH-22868 for root cause).
+        Map<String, InnerHitsContext.InnerHitSubContext> innerHits = searchContext.innerHits().copy().getInnerHits();
         return new FetchSubPhaseProcessor() {
             @Override
             public void setNextReader(LeafReaderContext readerContext) {
@@ -117,3 +122,4 @@ private void hitExecute(Map<String, InnerHitsContext.InnerHitSubContext> innerHi
         }
     }
 }
+

server/src/main/java/org/opensearch/search/internal/SubSearchContext.java

@@ -377,4 +377,35 @@ public FetchSearchResult fetchResult() {
     public long getRelativeTimeInMillis() {
         throw new UnsupportedOperationException("Not supported");
     }
+
+    /**
+     * Copies the fetch configuration of this context onto {@code target}, which must be a
+     * freshly constructed {@link SubSearchContext} of the same concrete type. Result holders
+     * ({@link #fetchSearchResult}, {@link #querySearchResult}) and per-fetch state
+     * (doc IDs to load) are intentionally not copied: every fetch gets its own fresh
+     * instances so that concurrent segment search slice threads never share mutable state.
+     */
+    protected void copyFetchStateTo(SubSearchContext target) {
+        target.from = from;
+        target.size = size;
+        target.sort = sort;
+        target.parsedQuery = parsedQuery;
+        target.query = query;
+        target.storedFields = storedFields;
+        if (scriptFields != null) {
+            for (ScriptFieldsContext.ScriptField field : scriptFields.fields()) {
+                target.scriptFields().add(field);
+            }
+        }
+        target.fetchSourceContext = fetchSourceContext;
+        target.docValuesContext = docValuesContext;
+        target.fetchFieldsContext = fetchFieldsContext;
+        target.highlight = highlight;
+        target.explain = explain;
+        target.trackScores = trackScores;
+        target.includeNamedQueriesScore = includeNamedQueriesScore;
+        target.version = version;
+        target.seqNoAndPrimaryTerm = seqNoAndPrimaryTerm;
+    }
 }
+

server/src/test/java/org/opensearch/search/fetch/subphase/InnerHitsContextTests.java

@@ -0,0 +1,57 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.search.fetch.subphase;
+
+import org.opensearch.test.OpenSearchTestCase;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+public class InnerHitsContextTests extends OpenSearchTestCase {
+
+    public void testCopyProducesIndependentDefinitionInstances() {
+        InnerHitsContext.InnerHitSubContext original = mock(InnerHitsContext.InnerHitSubContext.class);
+        when(original.getName()).thenReturn("inner");
+        InnerHitsContext.InnerHitSubContext copiedValue = mock(InnerHitsContext.InnerHitSubContext.class);
+        when(original.copy()).thenReturn(copiedValue);
+
+        InnerHitsContext context = new InnerHitsContext();
+        context.addInnerHitDefinition(original);
+
+        InnerHitsContext copy = context.copy();
+
+        assertNotSame(context, copy);
+        assertNotSame(context.getInnerHits(), copy.getInnerHits());
+        assertSame(copiedValue, copy.getInnerHits().get("inner"));
+        verify(original).copy();
+    }
+
+    public void testCopyKeepsDefinitionNames() {
+        InnerHitsContext.InnerHitSubContext original = mock(InnerHitsContext.InnerHitSubContext.class);
+        when(original.getName()).thenReturn("innerA");
+        InnerHitsContext.InnerHitSubContext copiedValue = mock(InnerHitsContext.InnerHitSubContext.class);
+        when(original.copy()).thenReturn(copiedValue);
+
+        InnerHitsContext context = new InnerHitsContext();
+        context.addInnerHitDefinition(original);
+
+        InnerHitsContext copy = context.copy();
+
+        assertTrue(copy.getInnerHits().containsKey("innerA"));
+        assertEquals(1, copy.getInnerHits().size());
+    }
+
+    public void testCopyOfEmptyContextIsEmpty() {
+        InnerHitsContext context = new InnerHitsContext();
+        InnerHitsContext copy = context.copy();
+        assertNotSame(context, copy);
+        assertTrue(copy.getInnerHits().isEmpty());
+    }
+}