Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Cleanup search sub transport actions and collapse o.e.action.search.t…
…ype package into o.e.action.search TransportSearchTypeAction and subclasses are not actually transport actions, but just support classes useful for their inner async actions that can easily be extracted out so that we get rid of one too many level of abstraction. Same pattern can be applied to TransportSearchScrollQueryAndFetchAction & TransportSearchScrollQueryThenFetchAction which we could remove in favour of keeping only their inner classes named SearchScrollQueryAndFetchAsyncAction and SearchScrollQueryThenFetchAsyncAction. Remove org.elasticsearch.action.search.type package, collapsed remaining classes into existing org.elasticsearch.action.search package Make also ParsedScrollId ScrollIdForNode and TransportSearchHelper classes and their methods package private. Closes #11710
- Loading branch information
Showing
29 changed files
with
1,836 additions
and
2,054 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
380 changes: 380 additions & 0 deletions
380
core/src/main/java/org/elasticsearch/action/search/AbstractSearchAsyncAction.java
Large diffs are not rendered by default.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
67 changes: 67 additions & 0 deletions
67
core/src/main/java/org/elasticsearch/action/search/SearchCountAsyncAction.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,67 @@ | ||
| /* | ||
| * Licensed to Elasticsearch under one or more contributor | ||
| * license agreements. See the NOTICE file distributed with | ||
| * this work for additional information regarding copyright | ||
| * ownership. Elasticsearch licenses this file to you under | ||
| * the Apache License, Version 2.0 (the "License"); you may | ||
| * not use this file except in compliance with the License. | ||
| * You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, | ||
| * software distributed under the License is distributed on an | ||
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY | ||
| * KIND, either express or implied. See the License for the | ||
| * specific language governing permissions and limitations | ||
| * under the License. | ||
| */ | ||
|
|
||
| package org.elasticsearch.action.search; | ||
|
|
||
| import org.elasticsearch.action.ActionListener; | ||
| import org.elasticsearch.cluster.ClusterService; | ||
| import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver; | ||
| import org.elasticsearch.cluster.node.DiscoveryNode; | ||
| import org.elasticsearch.common.logging.ESLogger; | ||
| import org.elasticsearch.common.util.concurrent.AtomicArray; | ||
| import org.elasticsearch.search.action.SearchServiceTransportAction; | ||
| import org.elasticsearch.search.controller.SearchPhaseController; | ||
| import org.elasticsearch.search.fetch.FetchSearchResultProvider; | ||
| import org.elasticsearch.search.internal.InternalSearchResponse; | ||
| import org.elasticsearch.search.internal.ShardSearchTransportRequest; | ||
| import org.elasticsearch.search.query.QuerySearchResultProvider; | ||
| import org.elasticsearch.threadpool.ThreadPool; | ||
|
|
||
| class SearchCountAsyncAction extends AbstractSearchAsyncAction<QuerySearchResultProvider> { | ||
|
|
||
| SearchCountAsyncAction(ESLogger logger, SearchServiceTransportAction searchService, ClusterService clusterService, | ||
| IndexNameExpressionResolver indexNameExpressionResolver, | ||
| SearchPhaseController searchPhaseController, ThreadPool threadPool, SearchRequest request, | ||
| ActionListener<SearchResponse> listener) { | ||
| super(logger, searchService, clusterService, indexNameExpressionResolver, searchPhaseController, threadPool, | ||
| request, listener); | ||
| } | ||
|
|
||
| @Override | ||
| protected String firstPhaseName() { | ||
| return "query"; | ||
| } | ||
|
|
||
| @Override | ||
| protected void sendExecuteFirstPhase(DiscoveryNode node, ShardSearchTransportRequest request, ActionListener<QuerySearchResultProvider> listener) { | ||
| searchService.sendExecuteQuery(node, request, listener); | ||
| } | ||
|
|
||
| @Override | ||
| protected void moveToSecondPhase() throws Exception { | ||
| // no need to sort, since we know we have no hits back | ||
| final InternalSearchResponse internalResponse = searchPhaseController.merge(SearchPhaseController.EMPTY_DOCS, firstResults, | ||
| (AtomicArray<? extends FetchSearchResultProvider>) AtomicArray.empty(), request); | ||
| String scrollId = null; | ||
| if (request.scroll() != null) { | ||
| scrollId = TransportSearchHelper.buildScrollId(request.searchType(), firstResults, null); | ||
| } | ||
| listener.onResponse(new SearchResponse(internalResponse, scrollId, expectedSuccessfulOps, successfulOps.get(), buildTookInMillis(), buildShardFailures())); | ||
| } | ||
| } |
142 changes: 142 additions & 0 deletions
142
core/src/main/java/org/elasticsearch/action/search/SearchDfsQueryAndFetchAsyncAction.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,142 @@ | ||
| /* | ||
| * Licensed to Elasticsearch under one or more contributor | ||
| * license agreements. See the NOTICE file distributed with | ||
| * this work for additional information regarding copyright | ||
| * ownership. Elasticsearch licenses this file to you under | ||
| * the Apache License, Version 2.0 (the "License"); you may | ||
| * not use this file except in compliance with the License. | ||
| * You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, | ||
| * software distributed under the License is distributed on an | ||
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY | ||
| * KIND, either express or implied. See the License for the | ||
| * specific language governing permissions and limitations | ||
| * under the License. | ||
| */ | ||
|
|
||
| package org.elasticsearch.action.search; | ||
|
|
||
| import org.elasticsearch.action.ActionListener; | ||
| import org.elasticsearch.action.ActionRunnable; | ||
| import org.elasticsearch.cluster.ClusterService; | ||
| import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver; | ||
| import org.elasticsearch.cluster.node.DiscoveryNode; | ||
| import org.elasticsearch.common.logging.ESLogger; | ||
| import org.elasticsearch.common.util.concurrent.AtomicArray; | ||
| import org.elasticsearch.search.action.SearchServiceTransportAction; | ||
| import org.elasticsearch.search.controller.SearchPhaseController; | ||
| import org.elasticsearch.search.dfs.AggregatedDfs; | ||
| import org.elasticsearch.search.dfs.DfsSearchResult; | ||
| import org.elasticsearch.search.fetch.QueryFetchSearchResult; | ||
| import org.elasticsearch.search.internal.InternalSearchResponse; | ||
| import org.elasticsearch.search.internal.ShardSearchTransportRequest; | ||
| import org.elasticsearch.search.query.QuerySearchRequest; | ||
| import org.elasticsearch.threadpool.ThreadPool; | ||
|
|
||
| import java.io.IOException; | ||
| import java.util.concurrent.atomic.AtomicInteger; | ||
|
|
||
| class SearchDfsQueryAndFetchAsyncAction extends AbstractSearchAsyncAction<DfsSearchResult> { | ||
|
|
||
| private final AtomicArray<QueryFetchSearchResult> queryFetchResults; | ||
|
|
||
| SearchDfsQueryAndFetchAsyncAction(ESLogger logger, SearchServiceTransportAction searchService, | ||
| ClusterService clusterService, IndexNameExpressionResolver indexNameExpressionResolver, | ||
| SearchPhaseController searchPhaseController, ThreadPool threadPool, | ||
| SearchRequest request, ActionListener<SearchResponse> listener) { | ||
| super(logger, searchService, clusterService, indexNameExpressionResolver, searchPhaseController, threadPool, request, listener); | ||
| queryFetchResults = new AtomicArray<>(firstResults.length()); | ||
| } | ||
|
|
||
| @Override | ||
| protected String firstPhaseName() { | ||
| return "dfs"; | ||
| } | ||
|
|
||
| @Override | ||
| protected void sendExecuteFirstPhase(DiscoveryNode node, ShardSearchTransportRequest request, | ||
| ActionListener<DfsSearchResult> listener) { | ||
| searchService.sendExecuteDfs(node, request, listener); | ||
| } | ||
|
|
||
| @Override | ||
| protected void moveToSecondPhase() { | ||
| final AggregatedDfs dfs = searchPhaseController.aggregateDfs(firstResults); | ||
| final AtomicInteger counter = new AtomicInteger(firstResults.asList().size()); | ||
|
|
||
| for (final AtomicArray.Entry<DfsSearchResult> entry : firstResults.asList()) { | ||
| DfsSearchResult dfsResult = entry.value; | ||
| DiscoveryNode node = nodes.get(dfsResult.shardTarget().nodeId()); | ||
| QuerySearchRequest querySearchRequest = new QuerySearchRequest(request, dfsResult.id(), dfs); | ||
| executeSecondPhase(entry.index, dfsResult, counter, node, querySearchRequest); | ||
| } | ||
| } | ||
|
|
||
| void executeSecondPhase(final int shardIndex, final DfsSearchResult dfsResult, final AtomicInteger counter, | ||
| final DiscoveryNode node, final QuerySearchRequest querySearchRequest) { | ||
| searchService.sendExecuteFetch(node, querySearchRequest, new ActionListener<QueryFetchSearchResult>() { | ||
| @Override | ||
| public void onResponse(QueryFetchSearchResult result) { | ||
| result.shardTarget(dfsResult.shardTarget()); | ||
| queryFetchResults.set(shardIndex, result); | ||
| if (counter.decrementAndGet() == 0) { | ||
| finishHim(); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public void onFailure(Throwable t) { | ||
| try { | ||
| onSecondPhaseFailure(t, querySearchRequest, shardIndex, dfsResult, counter); | ||
| } finally { | ||
| // the query might not have been executed at all (for example because thread pool rejected execution) | ||
| // and the search context that was created in dfs phase might not be released. | ||
| // release it again to be in the safe side | ||
| sendReleaseSearchContext(querySearchRequest.id(), node); | ||
| } | ||
| } | ||
| }); | ||
| } | ||
|
|
||
| void onSecondPhaseFailure(Throwable t, QuerySearchRequest querySearchRequest, int shardIndex, DfsSearchResult dfsResult, | ||
| AtomicInteger counter) { | ||
| if (logger.isDebugEnabled()) { | ||
| logger.debug("[{}] Failed to execute query phase", t, querySearchRequest.id()); | ||
| } | ||
| this.addShardFailure(shardIndex, dfsResult.shardTarget(), t); | ||
| successfulOps.decrementAndGet(); | ||
| if (counter.decrementAndGet() == 0) { | ||
| finishHim(); | ||
| } | ||
| } | ||
|
|
||
| private void finishHim() { | ||
| threadPool.executor(ThreadPool.Names.SEARCH).execute(new ActionRunnable<SearchResponse>(listener) { | ||
| @Override | ||
| public void doRun() throws IOException { | ||
| sortedShardList = searchPhaseController.sortDocs(true, queryFetchResults); | ||
| final InternalSearchResponse internalResponse = searchPhaseController.merge(sortedShardList, queryFetchResults, | ||
| queryFetchResults, request); | ||
| String scrollId = null; | ||
| if (request.scroll() != null) { | ||
| scrollId = TransportSearchHelper.buildScrollId(request.searchType(), firstResults, null); | ||
| } | ||
| listener.onResponse(new SearchResponse(internalResponse, scrollId, expectedSuccessfulOps, successfulOps.get(), | ||
| buildTookInMillis(), buildShardFailures())); | ||
| } | ||
|
|
||
| @Override | ||
| public void onFailure(Throwable t) { | ||
| ReduceSearchPhaseException failure = new ReduceSearchPhaseException("query_fetch", "", t, buildShardFailures()); | ||
| if (logger.isDebugEnabled()) { | ||
| logger.debug("failed to reduce search", failure); | ||
| } | ||
| super.onFailure(t); | ||
| } | ||
| }); | ||
|
|
||
| } | ||
| } |
Oops, something went wrong.