diff --git a/server/src/main/java/org/opensearch/search/DefaultSearchContext.java b/server/src/main/java/org/opensearch/search/DefaultSearchContext.java index f377a5e315e1b..d10173184f1c6 100644 --- a/server/src/main/java/org/opensearch/search/DefaultSearchContext.java +++ b/server/src/main/java/org/opensearch/search/DefaultSearchContext.java @@ -183,6 +183,7 @@ final class DefaultSearchContext extends SearchContext { private final QueryShardContext queryShardContext; private final FetchPhase fetchPhase; private final Function requestToAggReduceContextBuilder; + private final boolean useConcurrentSearch; DefaultSearchContext( ReaderContext readerContext, @@ -213,13 +214,14 @@ final class DefaultSearchContext extends SearchContext { this.indexShard = readerContext.indexShard(); this.clusterService = clusterService; this.engineSearcher = readerContext.acquireSearcher("search"); + this.useConcurrentSearch = useConcurrentSearch(executor); this.searcher = new ContextIndexSearcher( engineSearcher.getIndexReader(), engineSearcher.getSimilarity(), engineSearcher.getQueryCache(), engineSearcher.getQueryCachingPolicy(), lowLevelCancellation, - executor, + useConcurrentSearch ? executor : null, this ); this.relativeTimeSupplier = relativeTimeSupplier; @@ -878,18 +880,7 @@ public Profilers getProfilers() { */ @Override public boolean isConcurrentSegmentSearchEnabled() { - if (FeatureFlags.isEnabled(FeatureFlags.CONCURRENT_SEGMENT_SEARCH) - && (clusterService != null) - && (searcher().getExecutor() != null)) { - return indexService.getIndexSettings() - .getSettings() - .getAsBoolean( - IndexSettings.INDEX_CONCURRENT_SEGMENT_SEARCH_SETTING.getKey(), - clusterService.getClusterSettings().get(CLUSTER_CONCURRENT_SEGMENT_SEARCH_SETTING) - ); - } else { - return false; - } + return useConcurrentSearch; } public void setProfilers(Profilers profilers) { @@ -932,4 +923,24 @@ public void setBucketCollectorProcessor(BucketCollectorProcessor bucketCollector public BucketCollectorProcessor bucketCollectorProcessor() { return bucketCollectorProcessor; } + + /** + * Evaluate based on cluster and index settings if concurrent segment search should be used for this request context + * @return true: use concurrent search + * false: otherwise + */ + private boolean useConcurrentSearch(Executor concurrentSearchExecutor) { + if (FeatureFlags.isEnabled(FeatureFlags.CONCURRENT_SEGMENT_SEARCH) + && (clusterService != null) + && (concurrentSearchExecutor != null)) { + return indexService.getIndexSettings() + .getSettings() + .getAsBoolean( + IndexSettings.INDEX_CONCURRENT_SEGMENT_SEARCH_SETTING.getKey(), + clusterService.getClusterSettings().get(CLUSTER_CONCURRENT_SEGMENT_SEARCH_SETTING) + ); + } else { + return false; + } + } } diff --git a/server/src/main/java/org/opensearch/search/internal/FilteredSearchContext.java b/server/src/main/java/org/opensearch/search/internal/FilteredSearchContext.java index bb990e69e7722..02e6568369e16 100644 --- a/server/src/main/java/org/opensearch/search/internal/FilteredSearchContext.java +++ b/server/src/main/java/org/opensearch/search/internal/FilteredSearchContext.java @@ -559,4 +559,9 @@ public void setBucketCollectorProcessor(BucketCollectorProcessor bucketCollector public BucketCollectorProcessor bucketCollectorProcessor() { return in.bucketCollectorProcessor(); } + + @Override + public boolean isConcurrentSegmentSearchEnabled() { + return in.isConcurrentSegmentSearchEnabled(); + } } diff --git a/server/src/test/java/org/opensearch/search/SearchServiceTests.java b/server/src/test/java/org/opensearch/search/SearchServiceTests.java index 876a2d15cad7e..8643941970fe0 100644 --- a/server/src/test/java/org/opensearch/search/SearchServiceTests.java +++ b/server/src/test/java/org/opensearch/search/SearchServiceTests.java @@ -1275,8 +1275,77 @@ public void testConcurrentSegmentSearchSearchContext() throws IOException { .getSetting(index, IndexSettings.INDEX_CONCURRENT_SEGMENT_SEARCH_SETTING.getKey()) ); assertEquals(concurrentSearchEnabled, searchContext.isConcurrentSegmentSearchEnabled()); + // verify executor nullability with concurrent search enabled/disabled + if (concurrentSearchEnabled) { + assertNotNull(searchContext.searcher().getExecutor()); + } else { + assertNull(searchContext.searcher().getExecutor()); + } + } + } + // Cleanup + client().admin() + .cluster() + .prepareUpdateSettings() + .setTransientSettings(Settings.builder().putNull(SearchService.CLUSTER_CONCURRENT_SEGMENT_SEARCH_SETTING.getKey())) + .get(); + } + + /** + * Test that the Search Context for concurrent segment search enabled is set correctly at the time of construction. + * The same is used throughout the context object lifetime even if cluster setting changes before the request completion. + */ + public void testConcurrentSegmentSearchIsSetOnceDuringContextCreation() throws IOException { + String index = randomAlphaOfLengthBetween(5, 10).toLowerCase(Locale.ROOT); + IndexService indexService = createIndex(index); + final SearchService service = getInstanceFromNode(SearchService.class); + ShardId shardId = new ShardId(indexService.index(), 0); + long nowInMillis = System.currentTimeMillis(); + String clusterAlias = randomBoolean() ? null : randomAlphaOfLengthBetween(3, 10); + SearchRequest searchRequest = new SearchRequest(); + searchRequest.allowPartialSearchResults(randomBoolean()); + ShardSearchRequest request = new ShardSearchRequest( + OriginalIndices.NONE, + searchRequest, + shardId, + indexService.numberOfShards(), + AliasFilter.EMPTY, + 1f, + nowInMillis, + clusterAlias, + Strings.EMPTY_ARRAY + ); + + Boolean[] concurrentSearchStates = new Boolean[] { true, false }; + for (Boolean concurrentSearchSetting : concurrentSearchStates) { + // update concurrent search cluster setting and create search context + client().admin() + .cluster() + .prepareUpdateSettings() + .setTransientSettings( + Settings.builder().put(SearchService.CLUSTER_CONCURRENT_SEGMENT_SEARCH_SETTING.getKey(), concurrentSearchSetting) + ) + .get(); + try (DefaultSearchContext searchContext = service.createSearchContext(request, new TimeValue(System.currentTimeMillis()))) { + // verify concurrent search state in context + assertEquals(concurrentSearchSetting, searchContext.isConcurrentSegmentSearchEnabled()); + // verify executor state in searcher + assertEquals(concurrentSearchSetting, (searchContext.searcher().getExecutor() != null)); + + // update cluster setting to flip the concurrent segment search state + client().admin() + .cluster() + .prepareUpdateSettings() + .setTransientSettings( + Settings.builder().put(SearchService.CLUSTER_CONCURRENT_SEGMENT_SEARCH_SETTING.getKey(), !concurrentSearchSetting) + ) + .get(); + + // verify that concurrent segment search is still set to same expected value for the context + assertEquals(concurrentSearchSetting, searchContext.isConcurrentSegmentSearchEnabled()); } } + // Cleanup client().admin() .cluster()