Tie break search shard iterator comparisons on cluster alias (#38853)

`SearchShardIterator` inherits its `compareTo` implementation from `PlainShardIterator`. That is good in most of the cases, as such comparisons are based on the shard id which is unique, even when searching against indices with same names across multiple clusters (thanks to the index uuid being different). In case though the same cluster is registered multiple times with different aliases, the shard id is exactly the same, hence remote results will be returned before local ones with same shard id objects. That is because remote iterators are added before local ones, and we use a stable sorting method in `GroupShardIterators` constructor.

This PR enhances `compareTo` for `SearchShardIterator` to tie break on cluster alias and introduces consistent `equals` and `hashcode` methods. This allows to remove a TODO in `SearchResponseMerger` which otherwise has to handle this special case specifically. Also, while at it I added missing tests around equals/hashcode and compareTo and expanded existing ones.

Author: javanna

docs/reference/modules/cross-cluster-search.asciidoc

@@ -150,10 +150,10 @@ will be prefixed with their remote cluster name:
     "max_score": 1,
     "hits": [
       {
-        "_index": "cluster_one:twitter",
+        "_index": "twitter",
         "_type": "_doc",
         "_id": "0",
-        "_score": 1,
+        "_score": 2,
         "_source": {
           "user": "kimchy",
           "date": "2009-11-15T14:12:12",
@@ -162,10 +162,10 @@ will be prefixed with their remote cluster name:
         }
       },
       {
-        "_index": "twitter",
+        "_index": "cluster_one:twitter",
         "_type": "_doc",
         "_id": "0",
-        "_score": 2,
+        "_score": 1,
         "_source": {
           "user": "kimchy",
           "date": "2009-11-15T14:12:12",
@@ -243,10 +243,10 @@ GET /cluster_one:twitter,cluster_two:twitter,twitter/_search <1>
     "max_score": 1,
     "hits": [
       {
-        "_index": "cluster_one:twitter",
+        "_index": "twitter",
         "_type": "_doc",
         "_id": "0",
-        "_score": 1,
+        "_score": 2,
         "_source": {
           "user": "kimchy",
           "date": "2009-11-15T14:12:12",
@@ -255,10 +255,10 @@ GET /cluster_one:twitter,cluster_two:twitter,twitter/_search <1>
         }
       },
       {
-        "_index": "twitter",
+        "_index": "cluster_one:twitter",
         "_type": "_doc",
         "_id": "0",
-        "_score": 2,
+        "_score": 1,
         "_source": {
           "user": "kimchy",
           "date": "2009-11-15T14:12:12",

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

@@ -39,7 +39,6 @@
 import org.elasticsearch.search.profile.ProfileShardResult;
 import org.elasticsearch.search.profile.SearchProfileShardResults;
 import org.elasticsearch.search.suggest.Suggest;
-import org.elasticsearch.transport.RemoteClusterAware;
 
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -368,17 +367,7 @@ public int compareTo(ShardIdAndClusterAlias o) {
             if (shardIdCompareTo != 0) {
                 return shardIdCompareTo;
             }
-            int clusterAliasCompareTo = clusterAlias.compareTo(o.clusterAlias);
-            if (clusterAliasCompareTo != 0) {
-                //TODO we may want to fix this, CCS returns remote results before local ones (TransportSearchAction#mergeShardsIterators)
-                if (clusterAlias.equals(RemoteClusterAware.LOCAL_CLUSTER_GROUP_KEY)) {
-                    return 1;
-                }
-                if (o.clusterAlias.equals(RemoteClusterAware.LOCAL_CLUSTER_GROUP_KEY)) {
-                    return -1;
-                }
-            }
-            return clusterAliasCompareTo;
+            return clusterAlias.compareTo(o.clusterAlias);
         }
     }
 }

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

@@ -21,12 +21,14 @@
 
 import org.elasticsearch.action.OriginalIndices;
 import org.elasticsearch.cluster.routing.PlainShardIterator;
+import org.elasticsearch.cluster.routing.ShardIterator;
 import org.elasticsearch.cluster.routing.ShardRouting;
 import org.elasticsearch.common.Nullable;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.search.SearchShardTarget;
 
 import java.util.List;
+import java.util.Objects;
 
 /**
  * Extension of {@link PlainShardIterator} used in the search api, which also holds the {@link OriginalIndices}
@@ -93,4 +95,43 @@ void resetAndSkip() {
     boolean skip() {
         return skip;
     }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        if (super.equals(o) == false) {
+            return false;
+        }
+        SearchShardIterator that = (SearchShardIterator) o;
+        return Objects.equals(clusterAlias, that.clusterAlias);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(super.hashCode(), clusterAlias);
+    }
+
+    @Override
+    public int compareTo(ShardIterator o) {
+        int superCompareTo = super.compareTo(o);
+        if (superCompareTo != 0 || (o instanceof SearchShardIterator == false)) {
+            return superCompareTo;
+        }
+        SearchShardIterator searchShardIterator = (SearchShardIterator)o;
+        if (clusterAlias == null && searchShardIterator.getClusterAlias() == null) {
+            return 0;
+        }
+        if (clusterAlias == null) {
+            return -1;
+        }
+        if (searchShardIterator.getClusterAlias() == null) {
+            return 1;
+        }
+        return clusterAlias.compareTo(searchShardIterator.getClusterAlias());
+    }
 }

server/src/test/java/org/elasticsearch/action/OriginalIndicesTests.java

@@ -53,7 +53,7 @@ public void testOriginalIndicesSerialization() throws IOException {
         }
     }
 
-    private static OriginalIndices randomOriginalIndices() {
+    public static OriginalIndices randomOriginalIndices() {
         int numIndices = randomInt(10);
         String[] indices = new String[numIndices];
         for (int j = 0; j < indices.length; j++) {

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

@@ -632,12 +632,6 @@ public int compare(SearchHit a, SearchHit b) {
             }
             int clusterAliasCompareTo = aShard.getClusterAlias().compareTo(bShard.getClusterAlias());
             if (clusterAliasCompareTo != 0) {
-                if (aShard.getClusterAlias().equals(RemoteClusterAware.LOCAL_CLUSTER_GROUP_KEY)) {
-                    return 1;
-                }
-                if (bShard.getClusterAlias().equals(RemoteClusterAware.LOCAL_CLUSTER_GROUP_KEY)) {
-                    return -1;
-                }
                 return clusterAliasCompareTo;
             }
             return Integer.compare(a.docId(), b.docId());

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

@@ -20,12 +20,19 @@
 package org.elasticsearch.action.search;
 
 import org.elasticsearch.action.OriginalIndices;
+import org.elasticsearch.action.OriginalIndicesTests;
 import org.elasticsearch.action.support.IndicesOptions;
+import org.elasticsearch.cluster.routing.GroupShardsIteratorTests;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.search.SearchShardTarget;
 import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.test.EqualsHashCodeTestUtils;
+import org.hamcrest.Matchers;
 
+import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collections;
+import java.util.List;
 
 public class SearchShardIteratorTests extends ESTestCase {
 
@@ -64,4 +71,79 @@ public void testNewSearchShardTarget() {
         assertEquals(nodeId, searchShardTarget.getNodeId());
         assertSame(originalIndices, searchShardTarget.getOriginalIndices());
     }
+
+    public void testEqualsAndHashcode() {
+        EqualsHashCodeTestUtils.checkEqualsAndHashCode(randomSearchShardIterator(), s -> new SearchShardIterator(s.getClusterAlias(),
+            s.shardId(), s.getShardRoutings(), s.getOriginalIndices()), s -> {
+            if (randomBoolean()) {
+                String clusterAlias;
+                if (s.getClusterAlias() == null) {
+                    clusterAlias = randomAlphaOfLengthBetween(5, 10);
+                } else {
+                    clusterAlias = randomBoolean() ? null : s.getClusterAlias() + randomAlphaOfLength(3);
+                }
+                return new SearchShardIterator(clusterAlias, s.shardId(), s.getShardRoutings(), s.getOriginalIndices());
+            } else {
+                ShardId shardId = new ShardId(randomAlphaOfLengthBetween(5, 10), randomAlphaOfLength(10),
+                    randomIntBetween(0, Integer.MAX_VALUE));
+                return new SearchShardIterator(s.getClusterAlias(), shardId, s.getShardRoutings(), s.getOriginalIndices());
+            }
+        });
+    }
+
+    public void testCompareTo() {
+        String[] clusters = generateRandomStringArray(2, 10, false, false);
+        Arrays.sort(clusters);
+        String[] indices = generateRandomStringArray(3, 10, false, false);
+        Arrays.sort(indices);
+        String[] uuids = generateRandomStringArray(3, 10, false, false);
+        Arrays.sort(uuids);
+        List<SearchShardIterator> shardIterators = new ArrayList<>();
+        int numShards = randomIntBetween(1, 5);
+        for (int i = 0; i < numShards; i++) {
+            for (String index : indices) {
+                for (String uuid : uuids) {
+                    ShardId shardId = new ShardId(index, uuid, i);
+                    shardIterators.add(new SearchShardIterator(null, shardId, GroupShardsIteratorTests.randomShardRoutings(shardId),
+                        OriginalIndicesTests.randomOriginalIndices()));
+                    for (String cluster : clusters) {
+                        shardIterators.add(new SearchShardIterator(cluster, shardId, GroupShardsIteratorTests.randomShardRoutings(shardId),
+                            OriginalIndicesTests.randomOriginalIndices()));
+                    }
+
+                }
+            }
+        }
+        for (int i = 0; i < shardIterators.size(); i++) {
+            SearchShardIterator currentIterator = shardIterators.get(i);
+            for (int j = i + 1; j < shardIterators.size(); j++) {
+                SearchShardIterator greaterIterator = shardIterators.get(j);
+                assertThat(currentIterator, Matchers.lessThan(greaterIterator));
+                assertThat(greaterIterator, Matchers.greaterThan(currentIterator));
+                assertNotEquals(currentIterator, greaterIterator);
+            }
+            for (int j = i - 1; j >= 0; j--) {
+                SearchShardIterator smallerIterator = shardIterators.get(j);
+                assertThat(smallerIterator, Matchers.lessThan(currentIterator));
+                assertThat(currentIterator, Matchers.greaterThan(smallerIterator));
+                assertNotEquals(currentIterator, smallerIterator);
+            }
+        }
+    }
+
+    public void testCompareToEqualItems() {
+        SearchShardIterator shardIterator1 = randomSearchShardIterator();
+        SearchShardIterator shardIterator2 = new SearchShardIterator(shardIterator1.getClusterAlias(), shardIterator1.shardId(),
+            shardIterator1.getShardRoutings(), shardIterator1.getOriginalIndices());
+        assertEquals(shardIterator1, shardIterator2);
+        assertEquals(0, shardIterator1.compareTo(shardIterator2));
+        assertEquals(0, shardIterator2.compareTo(shardIterator1));
+    }
+
+    private static SearchShardIterator randomSearchShardIterator() {
+        String clusterAlias = randomBoolean() ? null : randomAlphaOfLengthBetween(5, 10);
+        ShardId shardId = new ShardId(randomAlphaOfLengthBetween(5, 10), randomAlphaOfLength(10), randomIntBetween(0, Integer.MAX_VALUE));
+        return new SearchShardIterator(clusterAlias, shardId, GroupShardsIteratorTests.randomShardRoutings(shardId),
+            OriginalIndicesTests.randomOriginalIndices());
+    }
 }