diff --git a/modules/parent-join/src/main/java/org/opensearch/join/query/ParentChildInnerHitContextBuilder.java b/modules/parent-join/src/main/java/org/opensearch/join/query/ParentChildInnerHitContextBuilder.java index 089201b6d5b20..78772a5d2dfd0 100644 --- a/modules/parent-join/src/main/java/org/opensearch/join/query/ParentChildInnerHitContextBuilder.java +++ b/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) } } + diff --git a/server/src/main/java/org/opensearch/index/query/NestedQueryBuilder.java b/server/src/main/java/org/opensearch/index/query/NestedQueryBuilder.java index 753cb57c9a8dc..7aa819e53b704 100644 --- a/server/src/main/java/org/opensearch/index/query/NestedQueryBuilder.java +++ b/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) { } } } + diff --git a/server/src/main/java/org/opensearch/search/fetch/subphase/InnerHitsContext.java b/server/src/main/java/org/opensearch/search/fetch/subphase/InnerHitsContext.java index 50dda07263695..cfff7b1ad3ad0 100644 --- a/server/src/main/java/org/opensearch/search/fetch/subphase/InnerHitsContext.java +++ b/server/src/main/java/org/opensearch/search/fetch/subphase/InnerHitsContext.java @@ -79,6 +79,20 @@ public Map 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 copies = new HashMap<>(); + for (Map.Entry 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 } } } + diff --git a/server/src/main/java/org/opensearch/search/fetch/subphase/InnerHitsPhase.java b/server/src/main/java/org/opensearch/search/fetch/subphase/InnerHitsPhase.java index 1c670bbed5b78..f2bf8eff7d992 100644 --- a/server/src/main/java/org/opensearch/search/fetch/subphase/InnerHitsPhase.java +++ b/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 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 innerHits = searchContext.innerHits().copy().getInnerHits(); return new FetchSubPhaseProcessor() { @Override public void setNextReader(LeafReaderContext readerContext) { @@ -117,3 +122,4 @@ private void hitExecute(Map innerHi } } } + diff --git a/server/src/main/java/org/opensearch/search/internal/SubSearchContext.java b/server/src/main/java/org/opensearch/search/internal/SubSearchContext.java index b2c97baf78d91..39030d5a7a3c1 100644 --- a/server/src/main/java/org/opensearch/search/internal/SubSearchContext.java +++ b/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; + } } + diff --git a/server/src/test/java/org/opensearch/search/fetch/subphase/InnerHitsContextTests.java b/server/src/test/java/org/opensearch/search/fetch/subphase/InnerHitsContextTests.java new file mode 100644 index 0000000000000..4a469d4c37efd --- /dev/null +++ b/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()); + } +} \ No newline at end of file