Skip to content

Commit 83f12a3

Browse files
committed
CCS: skip empty search hits when minimizing round-trips (#40098)
When minimizing round-trips, each cluster returns its own independent search response. In case sort by field and/or field collapsing were requested, when one cluster has no results to return, the information about the field that sorting was based on (SortField array) as well as the field (and the values) that collapsing was performed on are missing in the search response. That causes problems as we can't build the proper `TopDocs` instance which would need to be either `TopFieldDocs` or `CollapseTopFieldDocs`. The merge routine expects that all the top docs are of the same exact type which can't be guaranteed. Given that the problematic results are empty, hence have no impact on the final results, we can simply skip them. Relates to #32125 Closes #40067
1 parent a11f1c8 commit 83f12a3

File tree

2 files changed

+85
-8
lines changed

2 files changed

+85
-8
lines changed

server/src/main/java/org/elasticsearch/action/search/SearchResponseMerger.java

+19-8
Original file line numberDiff line numberDiff line change
@@ -178,17 +178,23 @@ SearchResponse getMergedResponse(Clusters clusters) {
178178
assert trackTotalHits == null || trackTotalHits;
179179
trackTotalHits = true;
180180
}
181+
181182
TopDocs topDocs = searchHitsToTopDocs(searchHits, totalHits, shards);
182183
topDocsStats.add(new TopDocsAndMaxScore(topDocs, searchHits.getMaxScore()),
183184
searchResponse.isTimedOut(), searchResponse.isTerminatedEarly());
184-
topDocsList.add(topDocs);
185+
if (searchHits.getHits().length > 0) {
186+
//there is no point in adding empty search hits and merging them with the others. Also, empty search hits always come
187+
//without sort fields and collapse info, despite sort by field and/or field collapsing was requested, which causes
188+
//issues reconstructing the proper TopDocs instance and breaks mergeTopDocs which expects the same type for each result.
189+
topDocsList.add(topDocs);
190+
}
185191
}
186192

187-
//after going through all the hits and collecting all their distinct shards, we can assign shardIndex and set it to the ScoreDocs
193+
//after going through all the hits and collecting all their distinct shards, we assign shardIndex and set it to the ScoreDocs
188194
setTopDocsShardIndex(shards, topDocsList);
189-
setSuggestShardIndex(shards, groupedSuggestions);
190195
TopDocs topDocs = mergeTopDocs(topDocsList, size, from);
191196
SearchHits mergedSearchHits = topDocsToSearchHits(topDocs, topDocsStats);
197+
setSuggestShardIndex(shards, groupedSuggestions);
192198
Suggest suggest = groupedSuggestions.isEmpty() ? null : new Suggest(Suggest.reduce(groupedSuggestions));
193199
InternalAggregations reducedAggs = InternalAggregations.reduce(aggs, reduceContextFunction.apply(true));
194200
ShardSearchFailure[] shardFailures = failures.toArray(ShardSearchFailure.EMPTY_ARRAY);
@@ -330,12 +336,17 @@ private static void assignShardIndex(Map<ShardIdAndClusterAlias, Integer> shards
330336
}
331337

332338
private static SearchHits topDocsToSearchHits(TopDocs topDocs, TopDocsStats topDocsStats) {
333-
SearchHit[] searchHits = new SearchHit[topDocs.scoreDocs.length];
334-
for (int i = 0; i < topDocs.scoreDocs.length; i++) {
335-
FieldDocAndSearchHit scoreDoc = (FieldDocAndSearchHit)topDocs.scoreDocs[i];
336-
searchHits[i] = scoreDoc.searchHit;
339+
SearchHit[] searchHits;
340+
if (topDocs == null) {
341+
//merged TopDocs is null whenever all clusters have returned empty hits
342+
searchHits = new SearchHit[0];
343+
} else {
344+
searchHits = new SearchHit[topDocs.scoreDocs.length];
345+
for (int i = 0; i < topDocs.scoreDocs.length; i++) {
346+
FieldDocAndSearchHit scoreDoc = (FieldDocAndSearchHit)topDocs.scoreDocs[i];
347+
searchHits[i] = scoreDoc.searchHit;
348+
}
337349
}
338-
339350
SortField[] sortFields = null;
340351
String collapseField = null;
341352
Object[] collapseValues = null;

server/src/test/java/org/elasticsearch/action/search/SearchResponseMergerTests.java

+66
Original file line numberDiff line numberDiff line change
@@ -570,6 +570,72 @@ public void testMergeNoResponsesAdded() {
570570
assertEquals(0, response.getShardFailures().length);
571571
}
572572

573+
public void testMergeEmptySearchHitsWithNonEmpty() {
574+
long currentRelativeTime = randomLong();
575+
final SearchTimeProvider timeProvider = new SearchTimeProvider(randomLong(), 0, () -> currentRelativeTime);
576+
SearchResponseMerger merger = new SearchResponseMerger(0, 10, Integer.MAX_VALUE, timeProvider, flag -> null);
577+
SearchResponse.Clusters clusters = SearchResponseTests.randomClusters();
578+
int numFields = randomIntBetween(1, 3);
579+
SortField[] sortFields = new SortField[numFields];
580+
for (int i = 0; i < numFields; i++) {
581+
sortFields[i] = new SortField("field-" + i, SortField.Type.INT, randomBoolean());
582+
}
583+
PriorityQueue<SearchHit> priorityQueue = new PriorityQueue<>(new SearchHitComparator(sortFields));
584+
SearchHit[] hits = randomSearchHitArray(10, 1, "remote", new Index[]{new Index("index", "uuid")}, Float.NaN, 1,
585+
sortFields, priorityQueue);
586+
{
587+
SearchHits searchHits = new SearchHits(hits, new TotalHits(10, TotalHits.Relation.EQUAL_TO), Float.NaN, sortFields, null, null);
588+
InternalSearchResponse response = new InternalSearchResponse(searchHits, null, null, null, false, false, 1);
589+
SearchResponse searchResponse = new SearchResponse(response, null, 1, 1, 0, 1L,
590+
ShardSearchFailure.EMPTY_ARRAY, SearchResponse.Clusters.EMPTY);
591+
merger.add(searchResponse);
592+
}
593+
{
594+
SearchHits empty = new SearchHits(new SearchHit[0], new TotalHits(0, TotalHits.Relation.EQUAL_TO), Float.NaN, null, null, null);
595+
InternalSearchResponse response = new InternalSearchResponse(empty, null, null, null, false, false, 1);
596+
SearchResponse searchResponse = new SearchResponse(response, null, 1, 1, 0, 1L,
597+
ShardSearchFailure.EMPTY_ARRAY, SearchResponse.Clusters.EMPTY);
598+
merger.add(searchResponse);
599+
}
600+
assertEquals(2, merger.numResponses());
601+
SearchResponse mergedResponse = merger.getMergedResponse(clusters);
602+
assertEquals(10, mergedResponse.getHits().getTotalHits().value);
603+
assertEquals(10, mergedResponse.getHits().getHits().length);
604+
assertEquals(2, mergedResponse.getTotalShards());
605+
assertEquals(2, mergedResponse.getSuccessfulShards());
606+
assertEquals(0, mergedResponse.getSkippedShards());
607+
assertArrayEquals(sortFields, mergedResponse.getHits().getSortFields());
608+
assertArrayEquals(hits, mergedResponse.getHits().getHits());
609+
assertEquals(clusters, mergedResponse.getClusters());
610+
}
611+
612+
public void testMergeOnlyEmptyHits() {
613+
long currentRelativeTime = randomLong();
614+
final SearchTimeProvider timeProvider = new SearchTimeProvider(randomLong(), 0, () -> currentRelativeTime);
615+
SearchResponse.Clusters clusters = SearchResponseTests.randomClusters();
616+
Tuple<Integer, TotalHits.Relation> randomTrackTotalHits = randomTrackTotalHits();
617+
int trackTotalHitsUpTo = randomTrackTotalHits.v1();
618+
TotalHits.Relation totalHitsRelation = randomTrackTotalHits.v2();
619+
SearchResponseMerger merger = new SearchResponseMerger(0, 10, trackTotalHitsUpTo, timeProvider, flag -> null);
620+
int numResponses = randomIntBetween(1, 5);
621+
TotalHits expectedTotalHits = null;
622+
for (int i = 0; i < numResponses; i++) {
623+
TotalHits totalHits = null;
624+
if (trackTotalHitsUpTo != SearchContext.TRACK_TOTAL_HITS_DISABLED) {
625+
totalHits = new TotalHits(randomLongBetween(0, 1000), totalHitsRelation);
626+
long previousValue = expectedTotalHits == null ? 0 : expectedTotalHits.value;
627+
expectedTotalHits = new TotalHits(Math.min(previousValue + totalHits.value, trackTotalHitsUpTo), totalHitsRelation);
628+
}
629+
SearchHits empty = new SearchHits(new SearchHit[0], totalHits, Float.NaN, null, null, null);
630+
InternalSearchResponse response = new InternalSearchResponse(empty, null, null, null, false, false, 1);
631+
SearchResponse searchResponse = new SearchResponse(response, null, 1, 1, 0, 1L,
632+
ShardSearchFailure.EMPTY_ARRAY, SearchResponse.Clusters.EMPTY);
633+
merger.add(searchResponse);
634+
}
635+
SearchResponse mergedResponse = merger.getMergedResponse(clusters);
636+
assertEquals(expectedTotalHits, mergedResponse.getHits().getTotalHits());
637+
}
638+
573639
private static Tuple<Integer, TotalHits.Relation> randomTrackTotalHits() {
574640
switch(randomIntBetween(0, 2)) {
575641
case 0:

0 commit comments

Comments
 (0)