Use a copy of the SearchExecutionContext for each Percolator execution

modules/percolator/src/internalClusterTest/java/org/elasticsearch/percolator/PercolatorQuerySearchIT.java

@@ -889,7 +889,11 @@ public void testWithMultiplePercolatorFields() throws Exception {
         assertThat(e.getCause().getMessage(), equalTo("a document can only contain one percolator query"));
     }
 
-    public void testPercolateQueryWithNestedDocuments() throws Exception {
+    /**
+     * Mapping for percolator tests that use nested "employee" documents.
+     * Includes query (percolator), id (keyword), companyname (text), and employee (nested with name).
+     */
+    private XContentBuilder nestedPercolatorMapping() throws IOException {
         XContentBuilder mapping = XContentFactory.jsonBuilder();
         mapping.startObject()
             .startObject("properties")
@@ -912,7 +916,11 @@ public void testPercolateQueryWithNestedDocuments() throws Exception {
             .endObject()
             .endObject()
             .endObject();
-        assertAcked(indicesAdmin().prepareCreate("test").setMapping(mapping));
+        return mapping;
+    }
+
+    public void testPercolateQueryWithNestedDocuments() throws Exception {
+        assertAcked(indicesAdmin().prepareCreate("test").setMapping(nestedPercolatorMapping()));
         prepareIndex("test").setId("q1")
             .setSource(
                 jsonBuilder().startObject()
@@ -1368,4 +1376,119 @@ public void testKnnQueryNotSupportedInPercolator() throws IOException {
         assertThat(exception.getMessage(), containsString("the [knn] query is unsupported inside a percolator"));
     }
 
+    public void testPercolatorBooleanQueriesWithConcurrency() throws Exception {
+        assertAcked(
+            indicesAdmin().prepareCreate("test")
+                .setSettings(Settings.builder().put(indexSettings()).put("index.number_of_shards", 1))
+                .setMapping("field1", "type=long", "query", "type=percolator")
+        );
+
+        prepareIndex("test").setId("1")
+            .setSource(
+                jsonBuilder().startObject()
+                    .field("query", boolQuery().must(rangeQuery("field1").from(10).to(12)).must(rangeQuery("field1").from(12).to(14)))
+                    .endObject()
+            )
+            .get();
+        prepareIndex("test").setId("2")
+            .setSource(
+                jsonBuilder().startObject()
+                    .field("query", boolQuery().must(rangeQuery("field1").from(3).to(4)).must(rangeQuery("field1").from(4).to(6)))
+                    .endObject()
+            )
+            .get();
+        prepareIndex("test").setId("3")
+            .setSource(
+                jsonBuilder().startObject()
+                    .field("query", boolQuery().must(rangeQuery("field1").from(10).to(12)).must(rangeQuery("field1").from(12).to(14)))
+                    .endObject()
+            )
+            .get();
+        prepareIndex("test").setId("4")
+            .setSource(
+                jsonBuilder().startObject()
+                    .field("query", boolQuery().must(rangeQuery("field1").from(3).to(4)).must(rangeQuery("field1").from(4).to(6)))
+                    .endObject()
+            )
+            .get();
+        prepareIndex("test").setId("5")
+            .setSource(
+                jsonBuilder().startObject()
+                    .field("query", boolQuery().must(rangeQuery("field1").from(10).to(12)).must(rangeQuery("field1").from(12).to(14)))
+                    .endObject()
+            )
+            .get();
+        prepareIndex("test").setId("6")
+            .setSource(
+                jsonBuilder().startObject()
+                    .field("query", boolQuery().must(rangeQuery("field1").from(3).to(4)).must(rangeQuery("field1").from(4).to(6)))
+                    .endObject()
+            )
+            .get();
+
+        indicesAdmin().prepareRefresh().get();
+
+        BytesReference source = BytesReference.bytes(jsonBuilder().startObject().field("field1", 12).endObject());
+        assertResponse(prepareSearch().setQuery(new PercolateQueryBuilder("query", source, XContentType.JSON)), response -> {
+            assertHitCount(response, 3);
+        });
+    }
+
+    public void testPercolatorNestedQueriesWithConcurrency() throws Exception {
+        assertAcked(
+            indicesAdmin().prepareCreate("test")
+                .setSettings(Settings.builder().put(indexSettings()).put("index.number_of_shards", 1))
+                .setMapping(nestedPercolatorMapping())
+        );
+
+        QueryBuilder nestedVirginia = QueryBuilders.nestedQuery(
+            "employee",
+            QueryBuilders.matchQuery("employee.name", "virginia potts").operator(Operator.AND),
+            ScoreMode.Avg
+        );
+        QueryBuilder nestedTony = QueryBuilders.nestedQuery(
+            "employee",
+            QueryBuilders.matchQuery("employee.name", "tony stark").operator(Operator.AND),
+            ScoreMode.Avg
+        );
+
+        prepareIndex("test").setId("1").setSource(jsonBuilder().startObject().field("query", nestedVirginia).endObject()).get();
+        prepareIndex("test").setId("2").setSource(jsonBuilder().startObject().field("query", nestedTony).endObject()).get();
+        prepareIndex("test").setId("3").setSource(jsonBuilder().startObject().field("query", nestedVirginia).endObject()).get();
+        prepareIndex("test").setId("4").setSource(jsonBuilder().startObject().field("query", nestedTony).endObject()).get();
+        prepareIndex("test").setId("5").setSource(jsonBuilder().startObject().field("query", nestedVirginia).endObject()).get();
+        prepareIndex("test").setId("6").setSource(jsonBuilder().startObject().field("query", nestedTony).endObject()).get();
+
+        indicesAdmin().prepareRefresh().get();
+
+        BytesReference source = BytesReference.bytes(
+            jsonBuilder().startObject()
+                .startArray("employee")
+                .startObject()
+                .field("name", "virginia potts")
+                .endObject()
+                .startObject()
+                .field("name", "tony stark")
+                .endObject()
+                .endArray()
+                .endObject()
+        );
+        assertResponse(prepareSearch().setQuery(new PercolateQueryBuilder("query", source, XContentType.JSON)), response -> {
+            assertHitCount(response, 6);
+        });
+
+        BytesReference sourceVirginiaOnly = BytesReference.bytes(
+            jsonBuilder().startObject()
+                .startArray("employee")
+                .startObject()
+                .field("name", "virginia potts")
+                .endObject()
+                .endArray()
+                .endObject()
+        );
+        assertResponse(prepareSearch().setQuery(new PercolateQueryBuilder("query", sourceVirginiaOnly, XContentType.JSON)), response -> {
+            assertHitCount(response, 3);
+        });
+    }
+
 }

modules/percolator/src/main/java/org/elasticsearch/percolator/PercolateQueryBuilder.java

@@ -572,8 +572,17 @@ static PercolateQuery.QueryStore createStore(MappedFieldType queryBuilderFieldTy
                         return context.parseDocument(sourceToParse).rootDoc().getBinaryValue(queryBuilderFieldType.name());
                     });
 
-                    queryBuilder = Rewriteable.rewrite(queryBuilder, context);
-                    return queryBuilder.toQuery(context);
+                    // The QueryBuilder.toQuery() function modifies the search context's
+                    // AutoPrefilteringScope in a way that is not thread safe. For other
+                    // queries this is not a problem, but Percolate will call the returned
+                    // PercolateQuery.QueryStore function from multiple threads.
+                    // The context's NestedScope is also vulnerable to concurrent modification.
+                    // Use a cloned SearchExecutionContext for each thread with new instances of
+                    // the mutable fields.
+                    var safeContext = context.copyForConcurrentUse();
+                    queryBuilder = Rewriteable.rewrite(queryBuilder, safeContext);
+                    // toQuery will access localAutoPrefilteringScope
+                    return queryBuilder.toQuery(safeContext);
                 } else {
                     return null;
                 }
@@ -623,7 +632,7 @@ private static QueryBuilder readQueryBuilder(
         }
     }
 
-    static SearchExecutionContext wrap(SearchExecutionContext delegate) {
+    private static SearchExecutionContext wrap(SearchExecutionContext delegate) {
         return new SearchExecutionContext(delegate) {
 
             @Override

modules/percolator/src/main/java/org/elasticsearch/percolator/PercolatorFieldMapper.java

@@ -578,7 +578,7 @@ static byte[] encodeRange(String rangeFieldName, byte[] minEncoded, byte[] maxEn
     // This is sane behavior for typical usage. But for percolator, the fields for the may not have any terms
     // Consequently, we may erroneously skip expanding those term fields.
     // This override allows mapped field values to expand via wildcard input, even if the field is empty in the shard.
-    static SearchExecutionContext wrapAllEmptyTextFields(SearchExecutionContext searchExecutionContext) {
+    private static SearchExecutionContext wrapAllEmptyTextFields(SearchExecutionContext searchExecutionContext) {
         return new FilteredSearchExecutionContext(searchExecutionContext) {
             @Override
             public boolean fieldExistsInIndex(String fieldname) {

modules/percolator/src/test/java/org/elasticsearch/percolator/QueryBuilderStoreTests.java

@@ -19,31 +19,38 @@
 import org.apache.lucene.search.TermQuery;
 import org.apache.lucene.store.Directory;
 import org.elasticsearch.TransportVersion;
+import org.elasticsearch.cluster.metadata.IndexMetadata;
 import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.core.CheckedFunction;
+import org.elasticsearch.index.IndexSettings;
 import org.elasticsearch.index.IndexVersion;
+import org.elasticsearch.index.fielddata.FieldDataContext;
+import org.elasticsearch.index.fielddata.IndexFieldData;
 import org.elasticsearch.index.fielddata.plain.BytesBinaryIndexFieldData;
 import org.elasticsearch.index.mapper.BinaryFieldMapper;
 import org.elasticsearch.index.mapper.DocumentParserContext;
 import org.elasticsearch.index.mapper.KeywordFieldMapper;
 import org.elasticsearch.index.mapper.MappedFieldType;
 import org.elasticsearch.index.mapper.MapperBuilderContext;
+import org.elasticsearch.index.mapper.MapperMetrics;
+import org.elasticsearch.index.mapper.MappingLookup;
 import org.elasticsearch.index.mapper.TestDocumentParserContext;
+import org.elasticsearch.index.query.FilteredSearchExecutionContext;
 import org.elasticsearch.index.query.SearchExecutionContext;
 import org.elasticsearch.index.query.TermQueryBuilder;
 import org.elasticsearch.script.field.BinaryDocValuesField;
 import org.elasticsearch.search.SearchModule;
 import org.elasticsearch.search.aggregations.support.CoreValuesSourceType;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.xcontent.NamedXContentRegistry;
-import org.mockito.Mockito;
+import org.elasticsearch.xcontent.XContentParserConfiguration;
 
 import java.io.IOException;
 import java.util.Collections;
+import java.util.function.BiFunction;
 
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
+import static org.elasticsearch.index.query.SearchExecutionContextHelper.SHARD_SEARCH_STATS;
 
 public class QueryBuilderStoreTests extends ESTestCase {
 
@@ -67,7 +74,6 @@ public void testStoringQueryBuilders() throws IOException {
             BinaryFieldMapper fieldMapper = PercolatorFieldMapper.Builder.createQueryBuilderFieldBuilder(
                 MapperBuilderContext.root(false, false)
             );
-            MappedFieldType.FielddataOperation fielddataOperation = MappedFieldType.FielddataOperation.SEARCH;
 
             try (IndexWriter indexWriter = new IndexWriter(directory, config)) {
                 for (int i = 0; i < queryBuilders.length; i++) {
@@ -84,18 +90,53 @@ public void testStoringQueryBuilders() throws IOException {
                 }
             }
 
-            SearchExecutionContext searchExecutionContext = mock(SearchExecutionContext.class);
-            when(searchExecutionContext.indexVersionCreated()).thenReturn(IndexVersion.current());
-            when(searchExecutionContext.getWriteableRegistry()).thenReturn(writableRegistry());
-            when(searchExecutionContext.getParserConfig()).thenReturn(parserConfig());
-            when(searchExecutionContext.getForField(fieldMapper.fieldType(), fielddataOperation)).thenReturn(
-                new BytesBinaryIndexFieldData(fieldMapper.fullPath(), CoreValuesSourceType.KEYWORD, BinaryDocValuesField::new)
+            IndexVersion indexVersion = IndexVersion.current();
+            NamedWriteableRegistry writeableRegistry = writableRegistry();
+            XContentParserConfiguration parserConfig = parserConfig();
+            BytesBinaryIndexFieldData fieldData = new BytesBinaryIndexFieldData(
+                fieldMapper.fullPath(),
+                CoreValuesSourceType.KEYWORD,
+                BinaryDocValuesField::new
             );
-            when(searchExecutionContext.getFieldType(Mockito.anyString())).thenAnswer(invocation -> {
-                final String fieldName = (String) invocation.getArguments()[0];
-                return new KeywordFieldMapper.KeywordFieldType(fieldName);
-            });
-            PercolateQuery.QueryStore queryStore = PercolateQueryBuilder.createStore(fieldMapper.fieldType(), searchExecutionContext);
+            BiFunction<MappedFieldType, FieldDataContext, IndexFieldData<?>> indexFieldDataLookup = (mft, fdc) -> fieldData;
+            Settings indexSettingsSettings = indexSettings(indexVersion, 1, 1).build();
+            IndexSettings indexSettings = new IndexSettings(
+                IndexMetadata.builder("test").settings(indexSettingsSettings).build(),
+                Settings.EMPTY
+            );
+            SearchExecutionContext searchExecutionContext = new SearchExecutionContext(
+                0,
+                0,
+                indexSettings,
+                null,
+                indexFieldDataLookup,
+                null,
+                MappingLookup.EMPTY,
+                null,
+                null,
+                parserConfig,
+                writeableRegistry,
+                null,
+                null,
+                System::currentTimeMillis,
+                null,
+                null,
+                () -> true,
+                null,
+                Collections.emptyMap(),
+                null,
+                MapperMetrics.NOOP,
+                SHARD_SEARCH_STATS
+            );
+
+            var filteredSEC = new FilteredSearchExecutionContext(searchExecutionContext) {
+                @Override
+                public MappedFieldType getFieldType(String name) {
+                    return new KeywordFieldMapper.KeywordFieldType(name);
+                }
+            };
+
+            PercolateQuery.QueryStore queryStore = PercolateQueryBuilder.createStore(fieldMapper.fieldType(), filteredSEC);
 
             try (IndexReader indexReader = DirectoryReader.open(directory)) {
                 LeafReaderContext leafContext = indexReader.leaves().get(0);

server/src/main/java/org/elasticsearch/index/query/FilteredSearchExecutionContext.java

@@ -33,6 +33,7 @@
 import org.elasticsearch.index.mapper.ParsedDocument;
 import org.elasticsearch.index.mapper.SourceLoader;
 import org.elasticsearch.index.mapper.SourceToParse;
+import org.elasticsearch.index.query.support.AutoPrefilteringScope;
 import org.elasticsearch.index.query.support.NestedScope;
 import org.elasticsearch.script.Script;
 import org.elasticsearch.script.ScriptContext;
@@ -390,4 +391,17 @@ public void setRewriteToNamedQueries() {
     public boolean rewriteToNamedQuery() {
         return in.rewriteToNamedQuery();
     }
+
+    @Override
+    public AutoPrefilteringScope autoPrefilteringScope() {
+        return in.autoPrefilteringScope();
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public FilteredSearchExecutionContext copyForConcurrentUse() {
+        return new FilteredSearchExecutionContext(this);
+    }
 }

server/src/main/java/org/elasticsearch/index/query/SearchExecutionContext.java

@@ -264,6 +264,19 @@ private void reset() {
         this.autoPrefilteringScope = new AutoPrefilteringScope();
     }
 
+    /**
+     * Creates a shallow copy of the context with new instances of the fields
+     * that are mutable - notably {@code AutoPrefilteringScope} and {@code NestedScope}.
+     * The mutable elements cannot be shared safely between threads, this function
+     * returns a copy that is safe to use in a concurrent environment without the
+     * overhead of a full copy.
+     *
+     * @return The shallow copy of the context.
+     */
+    public SearchExecutionContext copyForConcurrentUse() {
+        return new SearchExecutionContext(this);
+    }
+
     // Set alias filter, so it can be applied for queries that need it (e.g. knn query)
     public void setAliasFilter(QueryBuilder aliasFilter) {
         this.aliasFilter = aliasFilter;