diff --git a/docs/reference/search.asciidoc b/docs/reference/search.asciidoc index dd7faca60aa92..e99fb6f388d02 100644 --- a/docs/reference/search.asciidoc +++ b/docs/reference/search.asciidoc @@ -154,12 +154,11 @@ configure a soft limit, you can update the `action.search.shard_count.limit` cluster setting in order to reject search requests that hit too many shards. The request parameter `max_concurrent_shard_requests` can be used to control the -maximum number of concurrent shard requests the search API will execute for the -request. This parameter should be used to protect a single request from +maximum number of concurrent shard requests the search API will execute per node +for the request. This parameter should be used to protect a single request from overloading a cluster (e.g., a default request will hit all indices in a cluster which could cause shard request rejections if the number of shards per node is -high). This default is based on the number of data nodes in the cluster but at -most `256`. +high). This default value is `5`. -- diff --git a/docs/reference/search/multi-search.asciidoc b/docs/reference/search/multi-search.asciidoc index 34dc37d794cad..87a87c922b37c 100644 --- a/docs/reference/search/multi-search.asciidoc +++ b/docs/reference/search/multi-search.asciidoc @@ -85,15 +85,16 @@ The msearch's `max_concurrent_searches` request parameter can be used to control the maximum number of concurrent searches the multi search api will execute. This default is based on the number of data nodes and the default search thread pool size. -The request parameter `max_concurrent_shard_requests` can be used to control the -maximum number of concurrent shard requests the each sub search request will execute. -This parameter should be used to protect a single request from overloading a cluster -(e.g., a default request will hit all indices in a cluster which could cause shard request rejections -if the number of shards per node is high). This default is based on the number of -data nodes in the cluster but at most `256`.In certain scenarios parallelism isn't achieved through -concurrent request such that this protection will result in poor performance. For -instance in an environment where only a very low number of concurrent search requests are expected -it might help to increase this value to a higher number. +The request parameter `max_concurrent_shard_requests` can be used to control +the maximum number of concurrent shard requests that each sub search request +will execute per node. This parameter should be used to protect a single +request from overloading a cluster (e.g., a default request will hit all +indices in a cluster which could cause shard request rejections if the number +of shards per node is high). This default value is `5`.In certain scenarios +parallelism isn't achieved through concurrent request such that this protection +will result in poor performance. For instance in an environment where only a +very low number of concurrent search requests are expected it might help to +increase this value to a higher number. [float] [[msearch-security]] diff --git a/docs/reference/search/suggesters/phrase-suggest.asciidoc b/docs/reference/search/suggesters/phrase-suggest.asciidoc index d92c32eddf033..bfbfbc5417977 100644 --- a/docs/reference/search/suggesters/phrase-suggest.asciidoc +++ b/docs/reference/search/suggesters/phrase-suggest.asciidoc @@ -267,7 +267,10 @@ POST _search The `phrase` suggester supports multiple smoothing models to balance weight between infrequent grams (grams (shingles) are not existing in -the index) and frequent grams (appear at least once in the index). +the index) and frequent grams (appear at least once in the index). The +smoothing model can be selected by setting the `smoothing` parameter +to one of the following options. Each smoothing model supports specific +properties that can be configured. [horizontal] `stupid_backoff`:: @@ -288,6 +291,28 @@ the index) and frequent grams (appear at least once in the index). All parameters (`trigram_lambda`, `bigram_lambda`, `unigram_lambda`) must be supplied. +[source,js] +-------------------------------------------------- +POST _search +{ + "suggest": { + "text" : "obel prize", + "simple_phrase" : { + "phrase" : { + "field" : "title.trigram", + "size" : 1, + "smoothing" : { + "laplace" : { + "alpha" : 0.7 + } + } + } + } + } +} +-------------------------------------------------- +// CONSOLE + ==== Candidate Generators The `phrase` suggester uses candidate generators to produce a list of diff --git a/rest-api-spec/src/main/resources/rest-api-spec/api/msearch.json b/rest-api-spec/src/main/resources/rest-api-spec/api/msearch.json index 9c416e841362a..d46f22211a6e7 100644 --- a/rest-api-spec/src/main/resources/rest-api-spec/api/msearch.json +++ b/rest-api-spec/src/main/resources/rest-api-spec/api/msearch.json @@ -43,8 +43,8 @@ }, "max_concurrent_shard_requests" : { "type" : "number", - "description" : "The number of concurrent shard requests each sub search executes concurrently. This value should be used to limit the impact of the search on the cluster in order to limit the number of concurrent shard requests", - "default" : "The default grows with the number of nodes in the cluster but is at most 256." + "description" : "The number of concurrent shard requests each sub search executes concurrently per node. This value should be used to limit the impact of the search on the cluster in order to limit the number of concurrent shard requests", + "default" : 5 }, "rest_total_hits_as_int" : { "type" : "boolean", diff --git a/rest-api-spec/src/main/resources/rest-api-spec/api/search.json b/rest-api-spec/src/main/resources/rest-api-spec/api/search.json index 00ed8d113a00c..6dcc0663847d5 100644 --- a/rest-api-spec/src/main/resources/rest-api-spec/api/search.json +++ b/rest-api-spec/src/main/resources/rest-api-spec/api/search.json @@ -192,7 +192,7 @@ "max_concurrent_shard_requests" : { "type" : "number", "description" : "The number of concurrent shard requests per node this search executes concurrently. This value should be used to limit the impact of the search on the cluster in order to limit the number of concurrent shard requests", - "default" : "The default is 5." + "default" : 5 }, "pre_filter_shard_size" : { "type" : "number", diff --git a/server/src/main/java/org/elasticsearch/action/search/SearchRequest.java b/server/src/main/java/org/elasticsearch/action/search/SearchRequest.java index dea5283a629fb..8dd7414a21b9a 100644 --- a/server/src/main/java/org/elasticsearch/action/search/SearchRequest.java +++ b/server/src/main/java/org/elasticsearch/action/search/SearchRequest.java @@ -135,9 +135,10 @@ public SearchRequest(String[] indices, SearchSourceBuilder source) { } /** - * Creates a new search request by providing the search request to copy all fields from, the indices to search against, the alias of - * the cluster where it will be executed, as well as the start time in milliseconds from the epoch time and whether the reduction - * should be final or not. Used when a {@link SearchRequest} is created and executed as part of a cross-cluster search request + * Creates a new sub-search request starting from the original search request that is provided. + * For internal use only, allows to fork a search request into multiple search requests that will be executed independently. + * Such requests will not be finally reduced, so that their results can be merged together in one response at completion. + * Used when a {@link SearchRequest} is created and executed as part of a cross-cluster search request * performing reduction on each cluster in order to minimize network round-trips between the coordinating node and the remote clusters. * * @param originalSearchRequest the original search request @@ -146,8 +147,8 @@ public SearchRequest(String[] indices, SearchSourceBuilder source) { * @param absoluteStartMillis the absolute start time to be used on the remote clusters to ensure that the same value is used * @param finalReduce whether the reduction should be final or not */ - static SearchRequest crossClusterSearch(SearchRequest originalSearchRequest, String[] indices, - String clusterAlias, long absoluteStartMillis, boolean finalReduce) { + static SearchRequest subSearchRequest(SearchRequest originalSearchRequest, String[] indices, + String clusterAlias, long absoluteStartMillis, boolean finalReduce) { Objects.requireNonNull(originalSearchRequest, "search request must not be null"); validateIndices(indices); Objects.requireNonNull(clusterAlias, "cluster alias must not be null"); @@ -300,7 +301,7 @@ boolean isFinalReduce() { /** * Returns the current time in milliseconds from the time epoch, to be used for the execution of this search request. Used to * ensure that the same value, determined by the coordinating node, is used on all nodes involved in the execution of the search - * request. When created through {@link #crossClusterSearch(SearchRequest, String[], String, long, boolean)}, this method returns + * request. When created through {@link #subSearchRequest(SearchRequest, String[], String, long, boolean)}, this method returns * the provided current time, otherwise it will return {@link System#currentTimeMillis()}. */ long getOrCreateAbsoluteStartMillis() { @@ -308,7 +309,7 @@ long getOrCreateAbsoluteStartMillis() { } /** - * Returns the provided absoluteStartMillis when created through {@link #crossClusterSearch} and + * Returns the provided absoluteStartMillis when created through {@link #subSearchRequest} and * -1 otherwise. */ long getAbsoluteStartMillis() { diff --git a/server/src/main/java/org/elasticsearch/action/search/TransportSearchAction.java b/server/src/main/java/org/elasticsearch/action/search/TransportSearchAction.java index 6eaf53b87c34c..fda3e47cfaab7 100644 --- a/server/src/main/java/org/elasticsearch/action/search/TransportSearchAction.java +++ b/server/src/main/java/org/elasticsearch/action/search/TransportSearchAction.java @@ -29,6 +29,7 @@ import org.elasticsearch.action.support.IndicesOptions; import org.elasticsearch.client.Client; import org.elasticsearch.cluster.ClusterState; +import org.elasticsearch.cluster.block.ClusterBlockException; import org.elasticsearch.cluster.block.ClusterBlockLevel; import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver; import org.elasticsearch.cluster.node.DiscoveryNode; @@ -270,7 +271,7 @@ static void ccsRemoteReduce(SearchRequest searchRequest, OriginalIndices localIn String clusterAlias = entry.getKey(); boolean skipUnavailable = remoteClusterService.isSkipUnavailable(clusterAlias); OriginalIndices indices = entry.getValue(); - SearchRequest ccsSearchRequest = SearchRequest.crossClusterSearch(searchRequest, indices.indices(), + SearchRequest ccsSearchRequest = SearchRequest.subSearchRequest(searchRequest, indices.indices(), clusterAlias, timeProvider.getAbsoluteStartMillis(), true); Client remoteClusterClient = remoteClusterService.getRemoteClusterClient(threadPool, clusterAlias); remoteClusterClient.search(ccsSearchRequest, new ActionListener() { @@ -306,7 +307,7 @@ public void onFailure(Exception e) { String clusterAlias = entry.getKey(); boolean skipUnavailable = remoteClusterService.isSkipUnavailable(clusterAlias); OriginalIndices indices = entry.getValue(); - SearchRequest ccsSearchRequest = SearchRequest.crossClusterSearch(searchRequest, indices.indices(), + SearchRequest ccsSearchRequest = SearchRequest.subSearchRequest(searchRequest, indices.indices(), clusterAlias, timeProvider.getAbsoluteStartMillis(), false); ActionListener ccsListener = createCCSListener(clusterAlias, skipUnavailable, countDown, skippedClusters, exceptions, searchResponseMerger, totalClusters, listener); @@ -316,7 +317,7 @@ public void onFailure(Exception e) { if (localIndices != null) { ActionListener ccsListener = createCCSListener(RemoteClusterAware.LOCAL_CLUSTER_GROUP_KEY, false, countDown, skippedClusters, exceptions, searchResponseMerger, totalClusters, listener); - SearchRequest ccsLocalSearchRequest = SearchRequest.crossClusterSearch(searchRequest, localIndices.indices(), + SearchRequest ccsLocalSearchRequest = SearchRequest.subSearchRequest(searchRequest, localIndices.indices(), RemoteClusterAware.LOCAL_CLUSTER_GROUP_KEY, timeProvider.getAbsoluteStartMillis(), false); localSearchConsumer.accept(ccsLocalSearchRequest, ccsListener); } @@ -472,10 +473,89 @@ private void executeSearch(SearchTask task, SearchTimeProvider timeProvider, Sea Map> routingMap = indexNameExpressionResolver.resolveSearchRouting(clusterState, searchRequest.routing(), searchRequest.indices()); routingMap = routingMap == null ? Collections.emptyMap() : Collections.unmodifiableMap(routingMap); - String[] concreteIndices = new String[indices.length]; - for (int i = 0; i < indices.length; i++) { - concreteIndices[i] = indices[i].getName(); + Map concreteIndexBoosts = resolveIndexBoosts(searchRequest, clusterState); + + if (shouldSplitIndices(searchRequest)) { + //Execute two separate searches when we can, so that indices that are being written to are searched as quickly as possible. + //Otherwise their search context would need to stay open for too long between the query and the fetch phase, due to other + //indices (possibly slower) being searched at the same time. + List writeIndicesList = new ArrayList<>(); + List readOnlyIndicesList = new ArrayList<>(); + splitIndices(indices, clusterState, writeIndicesList, readOnlyIndicesList); + String[] writeIndices = writeIndicesList.toArray(new String[0]); + String[] readOnlyIndices = readOnlyIndicesList.toArray(new String[0]); + + if (readOnlyIndices.length == 0) { + executeSearch(task, timeProvider, searchRequest, localIndices, writeIndices, routingMap, + aliasFilter, concreteIndexBoosts, remoteShardIterators, remoteConnections, clusterState, listener, clusters); + } else if (writeIndices.length == 0 && remoteShardIterators.isEmpty()) { + executeSearch(task, timeProvider, searchRequest, localIndices, readOnlyIndices, routingMap, + aliasFilter, concreteIndexBoosts, remoteShardIterators, remoteConnections, clusterState, listener, clusters); + } else { + //Split the search in two whenever throttled indices are searched together with ordinary indices (local or remote), so + //that we don't keep the search context open for too long between query and fetch for ordinary indices due to slow indices. + CountDown countDown = new CountDown(2); + AtomicReference exceptions = new AtomicReference<>(); + SearchResponseMerger searchResponseMerger = createSearchResponseMerger(searchRequest.source(), timeProvider, + searchService::createReduceContext); + CountDownActionListener countDownActionListener = + new CountDownActionListener(countDown, exceptions, listener) { + @Override + void innerOnResponse(SearchResponse searchResponse) { + searchResponseMerger.add(searchResponse); + } + + @Override + SearchResponse createFinalResponse() { + return searchResponseMerger.getMergedResponse(clusters); + } + }; + + //Note that the indices set to the new SearchRequest won't be retrieved from it, as they have been already resolved and + //will be provided separately to executeSearch. + SearchRequest writeIndicesRequest = SearchRequest.subSearchRequest(searchRequest, writeIndices, + RemoteClusterService.LOCAL_CLUSTER_GROUP_KEY, timeProvider.getAbsoluteStartMillis(), false); + executeSearch(task, timeProvider, writeIndicesRequest, localIndices, writeIndices, routingMap, + aliasFilter, concreteIndexBoosts, remoteShardIterators, remoteConnections, clusterState, countDownActionListener, + SearchResponse.Clusters.EMPTY); + + //Note that the indices set to the new SearchRequest won't be retrieved from it, as they have been already resolved and + //will be provided separately to executeSearch. + SearchRequest readOnlyIndicesRequest = SearchRequest.subSearchRequest(searchRequest, readOnlyIndices, + RemoteClusterService.LOCAL_CLUSTER_GROUP_KEY, timeProvider.getAbsoluteStartMillis(), false); + executeSearch(task, timeProvider, readOnlyIndicesRequest, localIndices, readOnlyIndices, routingMap, + aliasFilter, concreteIndexBoosts, Collections.emptyList(), (alias, id) -> null, clusterState, countDownActionListener, + SearchResponse.Clusters.EMPTY); + } + } else { + String[] concreteIndices = Arrays.stream(indices).map(Index::getName).toArray(String[]::new); + executeSearch(task, timeProvider, searchRequest, localIndices, concreteIndices, routingMap, + aliasFilter, concreteIndexBoosts, remoteShardIterators, remoteConnections, clusterState, listener, clusters); + } + } + + static boolean shouldSplitIndices(SearchRequest searchRequest) { + return searchRequest.scroll() == null && searchRequest.searchType() != DFS_QUERY_THEN_FETCH + && (searchRequest.source() == null || searchRequest.source().size() != 0); + } + + static void splitIndices(Index[] indices, ClusterState clusterState, List writeIndices, List readOnlyIndices) { + for (Index index : indices) { + ClusterBlockException writeBlock = clusterState.blocks().indexBlockedException(ClusterBlockLevel.WRITE, index.getName()); + if (writeBlock == null) { + writeIndices.add(index.getName()); + } else { + readOnlyIndices.add(index.getName()); + } } + } + + private void executeSearch(SearchTask task, SearchTimeProvider timeProvider, SearchRequest searchRequest, + OriginalIndices localIndices, String[] concreteIndices, Map> routingMap, + Map aliasFilter, Map concreteIndexBoosts, + List remoteShardIterators, BiFunction remoteConnections, + ClusterState clusterState, ActionListener listener, SearchResponse.Clusters clusters) { + Map nodeSearchCounts = searchTransportService.getPendingSearchRequests(); GroupShardsIterator localShardsIterator = clusterService.operationRouting().searchShards(clusterState, concreteIndices, routingMap, searchRequest.preference(), searchService.getResponseCollectorService(), nodeSearchCounts); @@ -484,8 +564,6 @@ private void executeSearch(SearchTask task, SearchTimeProvider timeProvider, Sea failIfOverShardCountLimit(clusterService, shardIterators.size()); - Map concreteIndexBoosts = resolveIndexBoosts(searchRequest, clusterState); - // optimize search type for cases where there is only one shard group to search on if (shardIterators.size() == 1) { // if we only have one group, then we always want Q_T_F, no need for DFS, and no need to do THEN since we hit one shard @@ -498,11 +576,9 @@ private void executeSearch(SearchTask task, SearchTimeProvider timeProvider, Sea if (searchRequest.isSuggestOnly()) { // disable request cache if we have only suggest searchRequest.requestCache(false); - switch (searchRequest.searchType()) { - case DFS_QUERY_THEN_FETCH: - // convert to Q_T_F if we have only suggest - searchRequest.searchType(QUERY_THEN_FETCH); - break; + if (searchRequest.searchType() == DFS_QUERY_THEN_FETCH) { + // convert to Q_T_F if we have only suggest + searchRequest.searchType(QUERY_THEN_FETCH); } } @@ -611,22 +687,16 @@ private static void failIfOverShardCountLimit(ClusterService clusterService, int } } - abstract static class CCSActionListener implements ActionListener { - private final String clusterAlias; - private final boolean skipUnavailable; + abstract static class CountDownActionListener implements ActionListener { private final CountDown countDown; - private final AtomicInteger skippedClusters; private final AtomicReference exceptions; - private final ActionListener originalListener; + private final ActionListener delegateListener; - CCSActionListener(String clusterAlias, boolean skipUnavailable, CountDown countDown, AtomicInteger skippedClusters, - AtomicReference exceptions, ActionListener originalListener) { - this.clusterAlias = clusterAlias; - this.skipUnavailable = skipUnavailable; + CountDownActionListener(CountDown countDown, AtomicReference exceptions, + ActionListener delegateListener) { this.countDown = countDown; - this.skippedClusters = skippedClusters; this.exceptions = exceptions; - this.originalListener = originalListener; + this.delegateListener = delegateListener; } @Override @@ -637,26 +707,7 @@ public final void onResponse(Response response) { abstract void innerOnResponse(Response response); - @Override - public final void onFailure(Exception e) { - if (skipUnavailable) { - skippedClusters.incrementAndGet(); - } else { - Exception exception = e; - if (RemoteClusterAware.LOCAL_CLUSTER_GROUP_KEY.equals(clusterAlias) == false) { - exception = wrapRemoteClusterFailure(clusterAlias, e); - } - if (exceptions.compareAndSet(null, exception) == false) { - exceptions.accumulateAndGet(exception, (previous, current) -> { - current.addSuppressed(previous); - return current; - }); - } - } - maybeFinish(); - } - - private void maybeFinish() { + final void maybeFinish() { if (countDown.countDown()) { Exception exception = exceptions.get(); if (exception == null) { @@ -664,17 +715,56 @@ private void maybeFinish() { try { response = createFinalResponse(); } catch(Exception e) { - originalListener.onFailure(e); + delegateListener.onFailure(e); return; } - originalListener.onResponse(response); + delegateListener.onResponse(response); } else { - originalListener.onFailure(exceptions.get()); + delegateListener.onFailure(exceptions.get()); } } } abstract FinalResponse createFinalResponse(); + + @Override + public void onFailure(Exception e) { + if (exceptions.compareAndSet(null, e) == false) { + exceptions.accumulateAndGet(e, (previous, current) -> { + current.addSuppressed(previous); + return current; + }); + } + maybeFinish(); + } + } + + abstract static class CCSActionListener extends CountDownActionListener { + private final String clusterAlias; + private final boolean skipUnavailable; + private final AtomicInteger skippedClusters; + + CCSActionListener(String clusterAlias, boolean skipUnavailable, CountDown countDown, AtomicInteger skippedClusters, + AtomicReference exceptions, ActionListener originalListener) { + super(countDown, exceptions, originalListener); + this.clusterAlias = clusterAlias; + this.skipUnavailable = skipUnavailable; + this.skippedClusters = skippedClusters; + } + + @Override + public final void onFailure(Exception e) { + if (skipUnavailable) { + skippedClusters.incrementAndGet(); + maybeFinish(); + } else { + Exception exception = e; + if (RemoteClusterAware.LOCAL_CLUSTER_GROUP_KEY.equals(clusterAlias) == false) { + exception = wrapRemoteClusterFailure(clusterAlias, e); + } + super.onFailure(exception); + } + } } private static RemoteTransportException wrapRemoteClusterFailure(String clusterAlias, Exception e) { diff --git a/server/src/test/java/org/elasticsearch/action/search/SearchPhaseControllerTests.java b/server/src/test/java/org/elasticsearch/action/search/SearchPhaseControllerTests.java index 084a45267b5c5..3a1adf9748a06 100644 --- a/server/src/test/java/org/elasticsearch/action/search/SearchPhaseControllerTests.java +++ b/server/src/test/java/org/elasticsearch/action/search/SearchPhaseControllerTests.java @@ -330,7 +330,7 @@ private static AtomicArray generateFetchResults(int nShards, } private static SearchRequest randomSearchRequest() { - return randomBoolean() ? new SearchRequest() : SearchRequest.crossClusterSearch(new SearchRequest(), + return randomBoolean() ? new SearchRequest() : SearchRequest.subSearchRequest(new SearchRequest(), Strings.EMPTY_ARRAY, "remote", 0, randomBoolean()); } diff --git a/server/src/test/java/org/elasticsearch/action/search/SearchRequestTests.java b/server/src/test/java/org/elasticsearch/action/search/SearchRequestTests.java index 5aa0d937b981a..c66a9ec93b301 100644 --- a/server/src/test/java/org/elasticsearch/action/search/SearchRequestTests.java +++ b/server/src/test/java/org/elasticsearch/action/search/SearchRequestTests.java @@ -53,21 +53,21 @@ protected SearchRequest createSearchRequest() throws IOException { return request; } //clusterAlias and absoluteStartMillis do not have public getters/setters hence we randomize them only in this test specifically. - return SearchRequest.crossClusterSearch(request, request.indices(), + return SearchRequest.subSearchRequest(request, request.indices(), randomAlphaOfLengthBetween(5, 10), randomNonNegativeLong(), randomBoolean()); } public void testWithLocalReduction() { - expectThrows(NullPointerException.class, () -> SearchRequest.crossClusterSearch(null, Strings.EMPTY_ARRAY, "", 0, randomBoolean())); + expectThrows(NullPointerException.class, () -> SearchRequest.subSearchRequest(null, Strings.EMPTY_ARRAY, "", 0, randomBoolean())); SearchRequest request = new SearchRequest(); - expectThrows(NullPointerException.class, () -> SearchRequest.crossClusterSearch(request, null, "", 0, randomBoolean())); - expectThrows(NullPointerException.class, () -> SearchRequest.crossClusterSearch(request, + expectThrows(NullPointerException.class, () -> SearchRequest.subSearchRequest(request, null, "", 0, randomBoolean())); + expectThrows(NullPointerException.class, () -> SearchRequest.subSearchRequest(request, new String[]{null}, "", 0, randomBoolean())); - expectThrows(NullPointerException.class, () -> SearchRequest.crossClusterSearch(request, + expectThrows(NullPointerException.class, () -> SearchRequest.subSearchRequest(request, Strings.EMPTY_ARRAY, null, 0, randomBoolean())); - expectThrows(IllegalArgumentException.class, () -> SearchRequest.crossClusterSearch(request, + expectThrows(IllegalArgumentException.class, () -> SearchRequest.subSearchRequest(request, Strings.EMPTY_ARRAY, "", -1, randomBoolean())); - SearchRequest searchRequest = SearchRequest.crossClusterSearch(request, Strings.EMPTY_ARRAY, "", 0, randomBoolean()); + SearchRequest searchRequest = SearchRequest.subSearchRequest(request, Strings.EMPTY_ARRAY, "", 0, randomBoolean()); assertNull(searchRequest.validate()); } diff --git a/server/src/test/java/org/elasticsearch/action/search/TransportSearchActionSingleNodeTests.java b/server/src/test/java/org/elasticsearch/action/search/TransportSearchActionSingleNodeTests.java index 82f7c513bf0ce..fa6160839d2a9 100644 --- a/server/src/test/java/org/elasticsearch/action/search/TransportSearchActionSingleNodeTests.java +++ b/server/src/test/java/org/elasticsearch/action/search/TransportSearchActionSingleNodeTests.java @@ -19,11 +19,14 @@ package org.elasticsearch.action.search; +import org.elasticsearch.action.admin.indices.create.CreateIndexResponse; import org.elasticsearch.action.index.IndexRequest; import org.elasticsearch.action.index.IndexResponse; import org.elasticsearch.action.support.IndicesOptions; import org.elasticsearch.action.support.WriteRequest; +import org.elasticsearch.action.support.master.AcknowledgedResponse; import org.elasticsearch.common.Strings; +import org.elasticsearch.common.settings.Settings; import org.elasticsearch.index.query.RangeQueryBuilder; import org.elasticsearch.rest.RestStatus; import org.elasticsearch.search.SearchHit; @@ -46,7 +49,7 @@ public void testLocalClusterAlias() { assertEquals(RestStatus.CREATED, indexResponse.status()); { - SearchRequest searchRequest = SearchRequest.crossClusterSearch(new SearchRequest(), Strings.EMPTY_ARRAY, + SearchRequest searchRequest = SearchRequest.subSearchRequest(new SearchRequest(), Strings.EMPTY_ARRAY, "local", nowInMillis, randomBoolean()); SearchResponse searchResponse = client().search(searchRequest).actionGet(); assertEquals(1, searchResponse.getHits().getTotalHits().value); @@ -58,7 +61,7 @@ public void testLocalClusterAlias() { assertEquals("1", hit.getId()); } { - SearchRequest searchRequest = SearchRequest.crossClusterSearch(new SearchRequest(), Strings.EMPTY_ARRAY, + SearchRequest searchRequest = SearchRequest.subSearchRequest(new SearchRequest(), Strings.EMPTY_ARRAY, "", nowInMillis, randomBoolean()); SearchResponse searchResponse = client().search(searchRequest).actionGet(); assertEquals(1, searchResponse.getHits().getTotalHits().value); @@ -100,13 +103,13 @@ public void testAbsoluteStartMillis() { assertEquals(0, searchResponse.getTotalShards()); } { - SearchRequest searchRequest = SearchRequest.crossClusterSearch(new SearchRequest(), + SearchRequest searchRequest = SearchRequest.subSearchRequest(new SearchRequest(), Strings.EMPTY_ARRAY, "", 0, randomBoolean()); SearchResponse searchResponse = client().search(searchRequest).actionGet(); assertEquals(2, searchResponse.getHits().getTotalHits().value); } { - SearchRequest searchRequest = SearchRequest.crossClusterSearch(new SearchRequest(), + SearchRequest searchRequest = SearchRequest.subSearchRequest(new SearchRequest(), Strings.EMPTY_ARRAY, "", 0, randomBoolean()); searchRequest.indices(""); SearchResponse searchResponse = client().search(searchRequest).actionGet(); @@ -114,7 +117,7 @@ public void testAbsoluteStartMillis() { assertEquals("test-1970.01.01", searchResponse.getHits().getHits()[0].getIndex()); } { - SearchRequest searchRequest = SearchRequest.crossClusterSearch(new SearchRequest(), + SearchRequest searchRequest = SearchRequest.subSearchRequest(new SearchRequest(), Strings.EMPTY_ARRAY, "", 0, randomBoolean()); SearchSourceBuilder sourceBuilder = new SearchSourceBuilder(); RangeQueryBuilder rangeQuery = new RangeQueryBuilder("date"); @@ -156,7 +159,7 @@ public void testFinalReduce() { source.aggregation(terms); { - SearchRequest searchRequest = randomBoolean() ? originalRequest : SearchRequest.crossClusterSearch(originalRequest, + SearchRequest searchRequest = randomBoolean() ? originalRequest : SearchRequest.subSearchRequest(originalRequest, Strings.EMPTY_ARRAY, "remote", nowInMillis, true); SearchResponse searchResponse = client().search(searchRequest).actionGet(); assertEquals(2, searchResponse.getHits().getTotalHits().value); @@ -165,7 +168,7 @@ public void testFinalReduce() { assertEquals(1, longTerms.getBuckets().size()); } { - SearchRequest searchRequest = SearchRequest.crossClusterSearch(originalRequest, + SearchRequest searchRequest = SearchRequest.subSearchRequest(originalRequest, Strings.EMPTY_ARRAY, "remote", nowInMillis, false); SearchResponse searchResponse = client().search(searchRequest).actionGet(); assertEquals(2, searchResponse.getHits().getTotalHits().value); @@ -174,4 +177,62 @@ public void testFinalReduce() { assertEquals(2, longTerms.getBuckets().size()); } } + + public void testSplitIndices() { + { + CreateIndexResponse response = client().admin().indices().prepareCreate("write").get(); + assertTrue(response.isAcknowledged()); + } + { + CreateIndexResponse response = client().admin().indices().prepareCreate("readonly").get(); + assertTrue(response.isAcknowledged()); + } + { + SearchResponse response = client().prepareSearch("readonly").get(); + assertEquals(1, response.getTotalShards()); + assertEquals(1, response.getSuccessfulShards()); + assertEquals(1, response.getNumReducePhases()); + } + { + SearchResponse response = client().prepareSearch("write").get(); + assertEquals(1, response.getTotalShards()); + assertEquals(1, response.getSuccessfulShards()); + assertEquals(1, response.getNumReducePhases()); + } + { + SearchResponse response = client().prepareSearch("readonly", "write").get(); + assertEquals(2, response.getTotalShards()); + assertEquals(2, response.getSuccessfulShards()); + assertEquals(1, response.getNumReducePhases()); + } + { + Settings settings = Settings.builder().put("index.blocks.read_only", "true").build(); + AcknowledgedResponse response = client().admin().indices().prepareUpdateSettings("readonly").setSettings(settings).get(); + assertTrue(response.isAcknowledged()); + } + try { + { + SearchResponse response = client().prepareSearch("readonly").get(); + assertEquals(1, response.getTotalShards()); + assertEquals(1, response.getSuccessfulShards()); + assertEquals(1, response.getNumReducePhases()); + } + { + SearchResponse response = client().prepareSearch("write").get(); + assertEquals(1, response.getTotalShards()); + assertEquals(1, response.getSuccessfulShards()); + assertEquals(1, response.getNumReducePhases()); + } + { + SearchResponse response = client().prepareSearch("readonly", "write").get(); + assertEquals(2, response.getTotalShards()); + assertEquals(2, response.getSuccessfulShards()); + assertEquals(3, response.getNumReducePhases()); + } + } finally { + Settings settings = Settings.builder().put("index.blocks.read_only", "false").build(); + AcknowledgedResponse response = client().admin().indices().prepareUpdateSettings("readonly").setSettings(settings).get(); + assertTrue(response.isAcknowledged()); + } + } } diff --git a/server/src/test/java/org/elasticsearch/action/search/TransportSearchActionTests.java b/server/src/test/java/org/elasticsearch/action/search/TransportSearchActionTests.java index 369d71f05ffb8..60078486335c9 100644 --- a/server/src/test/java/org/elasticsearch/action/search/TransportSearchActionTests.java +++ b/server/src/test/java/org/elasticsearch/action/search/TransportSearchActionTests.java @@ -29,6 +29,10 @@ import org.elasticsearch.action.admin.cluster.shards.ClusterSearchShardsGroup; import org.elasticsearch.action.admin.cluster.shards.ClusterSearchShardsResponse; import org.elasticsearch.action.support.IndicesOptions; +import org.elasticsearch.cluster.ClusterName; +import org.elasticsearch.cluster.ClusterState; +import org.elasticsearch.cluster.block.ClusterBlocks; +import org.elasticsearch.cluster.metadata.IndexMetaData; import org.elasticsearch.cluster.node.DiscoveryNode; import org.elasticsearch.cluster.routing.GroupShardsIterator; import org.elasticsearch.cluster.routing.GroupShardsIteratorTests; @@ -837,4 +841,75 @@ public void testShouldMinimizeRoundtrips() throws Exception { assertFalse(TransportSearchAction.shouldMinimizeRoundtrips(searchRequest)); } } + + public void testShouldSplitIndices() { + { + SearchRequest searchRequest = new SearchRequest(); + assertTrue(TransportSearchAction.shouldSplitIndices(searchRequest)); + } + { + SearchRequest searchRequest = new SearchRequest(); + searchRequest.source(new SearchSourceBuilder()); + assertTrue(TransportSearchAction.shouldSplitIndices(searchRequest)); + } + { + SearchRequest searchRequest = new SearchRequest(); + searchRequest.source(new SearchSourceBuilder().size(randomIntBetween(1, 100))); + assertTrue(TransportSearchAction.shouldSplitIndices(searchRequest)); + } + { + SearchRequest searchRequest = new SearchRequest(); + searchRequest.scroll("5s"); + assertFalse(TransportSearchAction.shouldSplitIndices(searchRequest)); + } + { + SearchRequest searchRequest = new SearchRequest(); + searchRequest.source(new SearchSourceBuilder().size(0)); + assertFalse(TransportSearchAction.shouldSplitIndices(searchRequest)); + } + { + SearchRequest searchRequest = new SearchRequest(); + searchRequest.searchType(SearchType.DFS_QUERY_THEN_FETCH); + assertFalse(TransportSearchAction.shouldSplitIndices(searchRequest)); + } + } + + public void testSplitIndices() { + int numIndices = randomIntBetween(1, 10); + Index[] indices = new Index[numIndices]; + for (int i = 0; i < numIndices; i++) { + String indexName = randomAlphaOfLengthBetween(5, 10); + indices[i] = new Index(indexName, indexName + "-uuid"); + } + { + ClusterState clusterState = ClusterState.builder(ClusterName.DEFAULT).build(); + List writeIndices = new ArrayList<>(); + List readOnlyIndices = new ArrayList<>(); + TransportSearchAction.splitIndices(indices, clusterState, writeIndices, readOnlyIndices); + assertEquals(0, readOnlyIndices.size()); + assertEquals(numIndices, writeIndices.size()); + } + { + List expectedWrite = new ArrayList<>(); + List expectedReadOnly = new ArrayList<>(); + ClusterBlocks.Builder blocksBuilder = ClusterBlocks.builder(); + for (Index index : indices) { + if (randomBoolean()) { + blocksBuilder.addIndexBlock(index.getName(), IndexMetaData.INDEX_WRITE_BLOCK); + expectedReadOnly.add(index.getName()); + } else if(randomBoolean() ){ + blocksBuilder.addIndexBlock(index.getName(), IndexMetaData.INDEX_READ_ONLY_BLOCK); + expectedReadOnly.add(index.getName()); + } else { + expectedWrite.add(index.getName()); + } + } + ClusterState clusterState = ClusterState.builder(ClusterName.DEFAULT).blocks(blocksBuilder).build(); + List writeIndices = new ArrayList<>(); + List readOnlyIndices = new ArrayList<>(); + TransportSearchAction.splitIndices(indices, clusterState, writeIndices, readOnlyIndices); + assertEquals(writeIndices, expectedWrite); + assertEquals(readOnlyIndices, expectedReadOnly); + } + } } diff --git a/x-pack/plugin/src/test/resources/rest-api-spec/test/indices.freeze/10_basic.yml b/x-pack/plugin/src/test/resources/rest-api-spec/test/indices.freeze/10_basic.yml index 16a0aace0e444..f74d6ae900037 100644 --- a/x-pack/plugin/src/test/resources/rest-api-spec/test/indices.freeze/10_basic.yml +++ b/x-pack/plugin/src/test/resources/rest-api-spec/test/indices.freeze/10_basic.yml @@ -82,6 +82,25 @@ - match: {hits.total: 0} +- do: + index: + index: ordinary + id: "1" + body: { "foo": "Hello: 1" } + refresh: wait_for + +- do: + search: + rest_total_hits_as_int: true + index: [test, ordinary] + ignore_throttled: false + body: + query: + match: + foo: hello + +- match: {hits.total: 3} + --- "Test index options":