Skip to content

Commit

Permalink
Merge pull request #21393 from masaruh/alias_boost
Browse files Browse the repository at this point in the history 
Resolve index names in indices_boost
  • Loading branch information
@masaruh
masaruh authored Dec 16, 2016
2 parents 41ffb00 + 3df2a08 commit a0185c8
Show file tree
Hide file tree
Showing 25 changed files with 555 additions and 233 deletions.
13 changes: 9 additions & 4 deletions core/src/main/java/org/elasticsearch/action/search/AbstractSearchAsyncAction.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -50,6 +50,7 @@


abstract class AbstractSearchAsyncAction<FirstResult extends SearchPhaseResult> extends AbstractAsyncAction {
private static final float DEFAULT_INDEX_BOOST = 1.0f;

protected final Logger logger;
protected final SearchTransportService searchTransportService;
Expand All @@ -66,16 +67,17 @@ abstract class AbstractSearchAsyncAction<FirstResult extends SearchPhaseResult>
private final AtomicInteger totalOps = new AtomicInteger();
protected final AtomicArray<FirstResult> firstResults;
private final Map<String, AliasFilter> aliasFilter;
private final Map<String, Float> concreteIndexBoosts;
private final long clusterStateVersion;
private volatile AtomicArray<ShardSearchFailure> shardFailures;
private final Object shardFailuresMutex = new Object();
protected volatile ScoreDoc[] sortedShardDocs;

protected AbstractSearchAsyncAction(Logger logger, SearchTransportService searchTransportService,
Function<String, DiscoveryNode> nodeIdToDiscoveryNode,
Map<String, AliasFilter> aliasFilter, Executor executor, SearchRequest request,
ActionListener<SearchResponse> listener, GroupShardsIterator shardsIts, long startTime,
long clusterStateVersion, SearchTask task) {
Map<String, AliasFilter> aliasFilter, Map<String, Float> concreteIndexBoosts,
Executor executor, SearchRequest request, ActionListener<SearchResponse> listener,
GroupShardsIterator shardsIts, long startTime, long clusterStateVersion, SearchTask task) {
super(startTime);
this.logger = logger;
this.searchTransportService = searchTransportService;
Expand All @@ -91,6 +93,7 @@ protected AbstractSearchAsyncAction(Logger logger, SearchTransportService search
expectedTotalOps = shardsIts.totalSizeWith1ForEmpty();
firstResults = new AtomicArray<>(shardsIts.size());
this.aliasFilter = aliasFilter;
this.concreteIndexBoosts = concreteIndexBoosts;
}

public void start() {
Expand Down Expand Up @@ -125,8 +128,10 @@ void performFirstPhase(final int shardIndex, final ShardIterator shardIt, final
} else {
AliasFilter filter = this.aliasFilter.get(shard.index().getUUID());
assert filter != null;

float indexBoost = concreteIndexBoosts.getOrDefault(shard.index().getUUID(), DEFAULT_INDEX_BOOST);
ShardSearchTransportRequest transportRequest = new ShardSearchTransportRequest(request, shardIt.shardId(), shardsIts.size(),
filter, startTime());
filter, indexBoost, startTime());
sendExecuteFirstPhase(node, transportRequest , new ActionListener<FirstResult>() {
@Override
public void onResponse(FirstResult result) {
Expand Down
9 changes: 5 additions & 4 deletions core/src/main/java/org/elasticsearch/action/search/SearchDfsQueryAndFetchAsyncAction.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -47,10 +47,11 @@ class SearchDfsQueryAndFetchAsyncAction extends AbstractSearchAsyncAction<DfsSea
private final SearchPhaseController searchPhaseController;
SearchDfsQueryAndFetchAsyncAction(Logger logger, SearchTransportService searchTransportService,
Function<String, DiscoveryNode> nodeIdToDiscoveryNode,
Map<String, AliasFilter> aliasFilter, SearchPhaseController searchPhaseController,
Executor executor, SearchRequest request, ActionListener<SearchResponse> listener,
GroupShardsIterator shardsIts, long startTime, long clusterStateVersion, SearchTask task) {
super(logger, searchTransportService, nodeIdToDiscoveryNode, aliasFilter, executor,
Map<String, AliasFilter> aliasFilter, Map<String, Float> concreteIndexBoosts,
SearchPhaseController searchPhaseController, Executor executor, SearchRequest request,
ActionListener<SearchResponse> listener, GroupShardsIterator shardsIts,
long startTime, long clusterStateVersion, SearchTask task) {
super(logger, searchTransportService, nodeIdToDiscoveryNode, aliasFilter, concreteIndexBoosts, executor,
request, listener, shardsIts, startTime, clusterStateVersion, task);
this.searchPhaseController = searchPhaseController;
queryFetchResults = new AtomicArray<>(firstResults.length());
Expand Down
10 changes: 5 additions & 5 deletions core/src/main/java/org/elasticsearch/action/search/SearchDfsQueryThenFetchAsyncAction.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -55,11 +55,11 @@ class SearchDfsQueryThenFetchAsyncAction extends AbstractSearchAsyncAction<DfsSe

SearchDfsQueryThenFetchAsyncAction(Logger logger, SearchTransportService searchTransportService,
Function<String, DiscoveryNode> nodeIdToDiscoveryNode,
Map<String, AliasFilter> aliasFilter, SearchPhaseController searchPhaseController,
Executor executor, SearchRequest request, ActionListener<SearchResponse> listener,
GroupShardsIterator shardsIts, long startTime, long clusterStateVersion,
SearchTask task) {
super(logger, searchTransportService, nodeIdToDiscoveryNode, aliasFilter, executor,
Map<String, AliasFilter> aliasFilter, Map<String, Float> concreteIndexBoosts,
SearchPhaseController searchPhaseController, Executor executor, SearchRequest request,
ActionListener<SearchResponse> listener, GroupShardsIterator shardsIts, long startTime,
long clusterStateVersion, SearchTask task) {
super(logger, searchTransportService, nodeIdToDiscoveryNode, aliasFilter, concreteIndexBoosts, executor,
request, listener, shardsIts, startTime, clusterStateVersion, task);
this.searchPhaseController = searchPhaseController;
queryResults = new AtomicArray<>(firstResults.length());
Expand Down
4 changes: 2 additions & 2 deletions core/src/main/java/org/elasticsearch/action/search/SearchQueryAndFetchAsyncAction.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -40,12 +40,12 @@ class SearchQueryAndFetchAsyncAction extends AbstractSearchAsyncAction<QueryFetc

SearchQueryAndFetchAsyncAction(Logger logger, SearchTransportService searchTransportService,
Function<String, DiscoveryNode> nodeIdToDiscoveryNode,
Map<String, AliasFilter> aliasFilter,
Map<String, AliasFilter> aliasFilter, Map<String, Float> concreteIndexBoosts,
SearchPhaseController searchPhaseController, Executor executor,
SearchRequest request, ActionListener<SearchResponse> listener,
GroupShardsIterator shardsIts, long startTime, long clusterStateVersion,
SearchTask task) {
super(logger, searchTransportService, nodeIdToDiscoveryNode, aliasFilter, executor,
super(logger, searchTransportService, nodeIdToDiscoveryNode, aliasFilter, concreteIndexBoosts, executor,
request, listener, shardsIts, startTime, clusterStateVersion, task);
this.searchPhaseController = searchPhaseController;

Expand Down
6 changes: 3 additions & 3 deletions core/src/main/java/org/elasticsearch/action/search/SearchQueryThenFetchAsyncAction.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -50,13 +50,13 @@ class SearchQueryThenFetchAsyncAction extends AbstractSearchAsyncAction<QuerySea
private final SearchPhaseController searchPhaseController;

SearchQueryThenFetchAsyncAction(Logger logger, SearchTransportService searchTransportService,
Function<String, DiscoveryNode> nodeIdToDiscoveryNode, Map<String,
AliasFilter> aliasFilter,
Function<String, DiscoveryNode> nodeIdToDiscoveryNode,
Map<String, AliasFilter> aliasFilter, Map<String, Float> concreteIndexBoosts,
SearchPhaseController searchPhaseController, Executor executor,
SearchRequest request, ActionListener<SearchResponse> listener,
GroupShardsIterator shardsIts, long startTime, long clusterStateVersion,
SearchTask task) {
super(logger, searchTransportService, nodeIdToDiscoveryNode, aliasFilter, executor, request, listener,
super(logger, searchTransportService, nodeIdToDiscoveryNode, aliasFilter, concreteIndexBoosts, executor, request, listener,
shardsIts, startTime, clusterStateVersion, task);
this.searchPhaseController = searchPhaseController;
fetchResults = new AtomicArray<>(firstResults.length());
Expand Down
38 changes: 32 additions & 6 deletions core/src/main/java/org/elasticsearch/action/search/TransportSearchAction.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,7 @@
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.index.Index;
import org.elasticsearch.search.SearchService;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.internal.AliasFilter;
import org.elasticsearch.tasks.Task;
import org.elasticsearch.threadpool.ThreadPool;
Expand Down Expand Up @@ -84,6 +85,29 @@ private Map<String, AliasFilter> buildPerIndexAliasFilter(SearchRequest request,
return aliasFilterMap;
}

private Map<String, Float> resolveIndexBoosts(SearchRequest searchRequest, ClusterState clusterState) {
if (searchRequest.source() == null) {
return Collections.emptyMap();
}

SearchSourceBuilder source = searchRequest.source();
if (source.indexBoosts() == null) {
return Collections.emptyMap();
}

Map<String, Float> concreteIndexBoosts = new HashMap<>();
for (SearchSourceBuilder.IndexBoost ib : source.indexBoosts()) {
Index[] concreteIndices =
indexNameExpressionResolver.concreteIndices(clusterState, searchRequest.indicesOptions(), ib.getIndex());

for (Index concreteIndex : concreteIndices) {
concreteIndexBoosts.putIfAbsent(concreteIndex.getUUID(), ib.getBoost());
}
}

return Collections.unmodifiableMap(concreteIndexBoosts);
}

@Override
protected void doExecute(Task task, SearchRequest searchRequest, ActionListener<SearchResponse> listener) {
// pure paranoia if time goes backwards we are at least positive
Expand All @@ -107,6 +131,8 @@ protected void doExecute(Task task, SearchRequest searchRequest, ActionListener<
searchRequest.preference());
failIfOverShardCountLimit(clusterService, shardIterators.size());

Map<String, Float> 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_A_F, no need for DFS, and no need to do THEN since we hit one shard
Expand All @@ -125,7 +151,7 @@ protected void doExecute(Task task, SearchRequest searchRequest, ActionListener<
}

searchAsyncAction((SearchTask)task, searchRequest, shardIterators, startTimeInMillis, clusterState,
Collections.unmodifiableMap(aliasFilter), listener).start();
Collections.unmodifiableMap(aliasFilter), concreteIndexBoosts, listener).start();
}

@Override
Expand All @@ -135,6 +161,7 @@ protected final void doExecute(SearchRequest searchRequest, ActionListener<Searc

private AbstractSearchAsyncAction searchAsyncAction(SearchTask task, SearchRequest searchRequest, GroupShardsIterator shardIterators,
long startTime, ClusterState state, Map<String, AliasFilter> aliasFilter,
Map<String, Float> concreteIndexBoosts,
ActionListener<SearchResponse> listener) {
final Function<String, DiscoveryNode> nodesLookup = state.nodes()::get;
final long clusterStateVersion = state.version();
Expand All @@ -143,22 +170,22 @@ private AbstractSearchAsyncAction searchAsyncAction(SearchTask task, SearchReque
switch(searchRequest.searchType()) {
case DFS_QUERY_THEN_FETCH:
searchAsyncAction = new SearchDfsQueryThenFetchAsyncAction(logger, searchTransportService, nodesLookup,
aliasFilter, searchPhaseController, executor, searchRequest, listener, shardIterators, startTime,
aliasFilter, concreteIndexBoosts, searchPhaseController, executor, searchRequest, listener, shardIterators, startTime,
clusterStateVersion, task);
break;
case QUERY_THEN_FETCH:
searchAsyncAction = new SearchQueryThenFetchAsyncAction(logger, searchTransportService, nodesLookup,
aliasFilter, searchPhaseController, executor, searchRequest, listener, shardIterators, startTime,
aliasFilter, concreteIndexBoosts, searchPhaseController, executor, searchRequest, listener, shardIterators, startTime,
clusterStateVersion, task);
break;
case DFS_QUERY_AND_FETCH:
searchAsyncAction = new SearchDfsQueryAndFetchAsyncAction(logger, searchTransportService, nodesLookup,
aliasFilter, searchPhaseController, executor, searchRequest, listener, shardIterators, startTime,
aliasFilter, concreteIndexBoosts, searchPhaseController, executor, searchRequest, listener, shardIterators, startTime,
clusterStateVersion, task);
break;
case QUERY_AND_FETCH:
searchAsyncAction = new SearchQueryAndFetchAsyncAction(logger, searchTransportService, nodesLookup,
aliasFilter, searchPhaseController, executor, searchRequest, listener, shardIterators, startTime,
aliasFilter, concreteIndexBoosts, searchPhaseController, executor, searchRequest, listener, shardIterators, startTime,
clusterStateVersion, task);
break;
default:
Expand All @@ -177,5 +204,4 @@ private void failIfOverShardCountLimit(ClusterService clusterService, int shardC
+ "] to a greater value if you really want to query that many shards at the same time.");
}
}

}
9 changes: 2 additions & 7 deletions core/src/main/java/org/elasticsearch/search/DefaultSearchContext.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -96,7 +96,7 @@ final class DefaultSearchContext extends SearchContext {
private final DfsSearchResult dfsResult;
private final QuerySearchResult queryResult;
private final FetchSearchResult fetchResult;
private float queryBoost = 1.0f;
private final float queryBoost;
private TimeValue timeout;
// terminate after count
private int terminateAfter = DEFAULT_TERMINATE_AFTER;
Expand Down Expand Up @@ -173,6 +173,7 @@ final class DefaultSearchContext extends SearchContext {
this.timeout = timeout;
queryShardContext = indexService.newQueryShardContext(request.shardId().id(), searcher.getIndexReader(), request::nowInMillis);
queryShardContext.setTypes(request.types());
queryBoost = request.indexBoost();
}

@Override
Expand Down Expand Up @@ -352,12 +353,6 @@ public float queryBoost() {
return queryBoost;
}

@Override
public SearchContext queryBoost(float queryBoost) {
this.queryBoost = queryBoost;
return this;
}

@Override
public long getOriginNanoTime() {
return originNanoTime;
Expand Down
8 changes: 0 additions & 8 deletions core/src/main/java/org/elasticsearch/search/SearchService.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,6 @@

package org.elasticsearch.search;

import com.carrotsearch.hppc.ObjectFloatHashMap;
import org.apache.lucene.search.FieldDoc;
import org.apache.lucene.search.TopDocs;
import org.apache.lucene.util.IOUtils;
Expand Down Expand Up @@ -671,13 +670,6 @@ private void parseSource(DefaultSearchContext context, SearchSourceBuilder sourc
QueryShardContext queryShardContext = context.getQueryShardContext();
context.from(source.from());
context.size(source.size());
ObjectFloatHashMap<String> indexBoostMap = source.indexBoost();
if (indexBoostMap != null) {
Float indexBoost = indexBoostMap.get(context.shardTarget().index());
if (indexBoost != null) {
context.queryBoost(indexBoost);
}
}
Map<String, InnerHitBuilder> innerHitBuilders = new HashMap<>();
if (source.query() != null) {
InnerHitBuilder.extractInnerHits(source.query(), innerHitBuilders);
Expand Down
Loading

0 comments on commit a0185c8

Please sign in to comment.