Make schema agreement query pull only columns that are used

Currently schema agreement logic run `select * from system.peers`, while
only `schema_version` is used, it creates excessive load on cluster and
driver side.

Author: dkropachev

core/src/main/java/com/datastax/oss/driver/internal/core/metadata/SchemaAgreementChecker.java

@@ -21,6 +21,7 @@
 import com.datastax.oss.driver.api.core.config.DriverExecutionProfile;
 import com.datastax.oss.driver.api.core.metadata.Node;
 import com.datastax.oss.driver.api.core.metadata.NodeState;
+import com.datastax.oss.driver.api.core.metadata.TokenMap;
 import com.datastax.oss.driver.internal.core.adminrequest.AdminRequestHandler;
 import com.datastax.oss.driver.internal.core.adminrequest.AdminResult;
 import com.datastax.oss.driver.internal.core.adminrequest.AdminRow;
@@ -108,7 +109,11 @@ private void sendQueries() {
     } else {
       CompletionStage<AdminResult> localQuery =
           query("SELECT schema_version FROM system.local WHERE key='local'");
-      CompletionStage<AdminResult> peersQuery = query("SELECT * FROM system.peers");
+
+      // `tokens` column is excluded, it is served from
+      // context.getMetadataManager().getMetadata().getTokenMap()
+      CompletionStage<AdminResult> peersQuery =
+          query("SELECT host_id, schema_version, rpc_address, data_center, rack FROM system.peers");
 
       localQuery
           .thenCombine(peersQuery, this::extractSchemaVersions)
@@ -142,9 +147,20 @@ private Set<UUID> extractSchemaVersions(AdminResult controlNodeResult, AdminResu
           channel.getEndPoint());
     }
 
+    boolean allowZeroTokenNodes =
+        context
+            .getConfig()
+            .getDefaultProfile()
+            .getBoolean(DefaultDriverOption.METADATA_ALLOW_ZERO_TOKEN_PEERS);
+
+    TokenMap tokenMap =
+        context.getMetadataManager().getMetadata().getTokenMap().isPresent()
+            ? context.getMetadataManager().getMetadata().getTokenMap().get()
+            : null;
+
     Map<UUID, Node> nodes = context.getMetadataManager().getMetadata().getNodes();
     for (AdminRow peerRow : peersResult) {
-      if (isPeerValid(peerRow, nodes)) {
+      if (isPeerValid(peerRow, nodes, allowZeroTokenNodes, tokenMap)) {
         UUID schemaVersion = Objects.requireNonNull(peerRow.getUuid("schema_version"));
         schemaVersions.add(schemaVersion);
       }
@@ -189,13 +205,9 @@ protected CompletionStage<AdminResult> query(String queryString) {
         .start();
   }
 
-  protected boolean isPeerValid(AdminRow peerRow, Map<UUID, Node> nodes) {
-    if (PeerRowValidator.isValid(
-        peerRow,
-        context
-            .getConfig()
-            .getDefaultProfile()
-            .getBoolean(DefaultDriverOption.METADATA_ALLOW_ZERO_TOKEN_PEERS))) {
+  protected boolean isPeerValid(
+      AdminRow peerRow, Map<UUID, Node> nodes, boolean allowZeroTokenNodes, TokenMap tokenMap) {
+    if (PeerRowValidator.isValid(peerRow, true)) {
       UUID hostId = peerRow.getUuid("host_id");
       Node node = nodes.get(hostId);
       if (node == null) {
@@ -205,7 +217,10 @@ protected boolean isPeerValid(AdminRow peerRow, Map<UUID, Node> nodes) {
         LOG.debug("[{}] Peer {} is down, excluding from schema agreement check", logPrefix, hostId);
         return false;
       }
-      return true;
+      return allowZeroTokenNodes
+          || tokenMap == null
+          || tokenMap.getTokenRanges().isEmpty()
+          || !tokenMap.getTokenRanges(node).isEmpty();
     } else {
       LOG.warn(
           "[{}] Found invalid system.peers row for peer: {}, excluding from schema agreement check.",

core/src/test/java/com/datastax/oss/driver/internal/core/metadata/SchemaAgreementCheckerTest.java

@@ -34,6 +34,9 @@
 import com.datastax.oss.driver.internal.core.adminrequest.AdminRow;
 import com.datastax.oss.driver.internal.core.channel.DriverChannel;
 import com.datastax.oss.driver.internal.core.context.InternalDriverContext;
+import com.datastax.oss.driver.internal.core.metadata.token.DefaultTokenMap;
+import com.datastax.oss.driver.internal.core.metadata.token.Murmur3Token;
+import com.datastax.oss.driver.internal.core.metadata.token.Murmur3TokenRange;
 import com.datastax.oss.driver.internal.core.metrics.MetricsFactory;
 import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableMap;
 import com.datastax.oss.driver.shaded.guava.common.collect.Iterators;
@@ -46,6 +49,7 @@
 import java.util.Arrays;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Optional;
 import java.util.Queue;
 import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
@@ -56,6 +60,7 @@
 import org.junit.runner.RunWith;
 import org.mockito.Mock;
 import org.mockito.MockitoAnnotations;
+import org.mockito.internal.util.collections.Sets;
 
 @RunWith(DataProviderRunner.class)
 public class SchemaAgreementCheckerTest {
@@ -72,6 +77,7 @@ public class SchemaAgreementCheckerTest {
   @Mock private EventLoop eventLoop;
   @Mock private MetadataManager metadataManager;
   @Mock private MetricsFactory metricsFactory;
+  @Mock private DefaultTokenMap tokenMap;
   @Mock private Metadata metadata;
   @Mock private DefaultNode node1;
   @Mock private DefaultNode node2;
@@ -102,6 +108,12 @@ public void setup() {
             Objects.requireNonNull(node2.getHostId()),
             node2);
     when(metadata.getNodes()).thenReturn(nodes);
+    when(tokenMap.getTokenRanges())
+        .thenReturn(
+            Sets.newSet(
+                new Murmur3TokenRange(new Murmur3Token(1), new Murmur3Token(2)),
+                new Murmur3TokenRange(new Murmur3Token(3), new Murmur3Token(4))));
+    when(metadata.getTokenMap()).thenReturn(Optional.of(tokenMap));
     when(metadataManager.getMetadata()).thenReturn(metadata);
     when(context.getMetadataManager()).thenReturn(metadataManager);
 
@@ -139,7 +151,9 @@ public void should_succeed_if_only_one_node() {
         new StubbedQuery(
             "SELECT schema_version FROM system.local WHERE key='local'",
             mockResult(mockLocalRow(VERSION1))),
-        new StubbedQuery("SELECT * FROM system.peers", mockResult(/*empty*/ )));
+        new StubbedQuery(
+            "SELECT host_id, schema_version, rpc_address, data_center, rack FROM system.peers",
+            mockResult(/*empty*/ )));
 
     // When
     CompletionStage<Boolean> future = checker.run();
@@ -156,7 +170,9 @@ public void should_succeed_if_versions_match_on_first_try() {
         new StubbedQuery(
             "SELECT schema_version FROM system.local WHERE key='local'",
             mockResult(mockLocalRow(VERSION1))),
-        new StubbedQuery("SELECT * FROM system.peers", mockResult(mockValidPeerRow(VERSION1))));
+        new StubbedQuery(
+            "SELECT host_id, schema_version, rpc_address, data_center, rack FROM system.peers",
+            mockResult(mockValidPeerRow(VERSION1))));
 
     // When
     CompletionStage<Boolean> future = checker.run();
@@ -174,7 +190,9 @@ public void should_ignore_down_peers() {
         new StubbedQuery(
             "SELECT schema_version FROM system.local WHERE key='local'",
             mockResult(mockLocalRow(VERSION1))),
-        new StubbedQuery("SELECT * FROM system.peers", mockResult(mockValidPeerRow(VERSION2))));
+        new StubbedQuery(
+            "SELECT host_id, schema_version, rpc_address, data_center, rack FROM system.peers",
+            mockResult(mockValidPeerRow(VERSION2))));
 
     // When
     CompletionStage<Boolean> future = checker.run();
@@ -210,7 +228,9 @@ public void should_ignore_malformed_rows(AdminRow malformedPeer) {
         new StubbedQuery(
             "SELECT schema_version FROM system.local WHERE key='local'",
             mockResult(mockLocalRow(VERSION1))),
-        new StubbedQuery("SELECT * FROM system.peers", mockResult(malformedPeer)));
+        new StubbedQuery(
+            "SELECT host_id, schema_version, rpc_address, data_center, rack FROM system.peers",
+            mockResult(malformedPeer)));
 
     // When
     CompletionStage<Boolean> future = checker.run();
@@ -228,13 +248,17 @@ public void should_reschedule_if_versions_do_not_match_on_first_try() {
         new StubbedQuery(
             "SELECT schema_version FROM system.local WHERE key='local'",
             mockResult(mockLocalRow(VERSION1))),
-        new StubbedQuery("SELECT * FROM system.peers", mockResult(mockValidPeerRow(VERSION2))),
+        new StubbedQuery(
+            "SELECT host_id, schema_version, rpc_address, data_center, rack FROM system.peers",
+            mockResult(mockValidPeerRow(VERSION2))),
 
         // Second round
         new StubbedQuery(
             "SELECT schema_version FROM system.local WHERE key='local'",
             mockResult(mockLocalRow(VERSION1))),
-        new StubbedQuery("SELECT * FROM system.peers", mockResult(mockValidPeerRow(VERSION1))));
+        new StubbedQuery(
+            "SELECT host_id, schema_version, rpc_address, data_center, rack FROM system.peers",
+            mockResult(mockValidPeerRow(VERSION1))));
 
     // When
     CompletionStage<Boolean> future = checker.run();
@@ -253,7 +277,9 @@ public void should_fail_if_versions_do_not_match_after_timeout() {
         new StubbedQuery(
             "SELECT schema_version FROM system.local WHERE key='local'",
             mockResult(mockLocalRow(VERSION1))),
-        new StubbedQuery("SELECT * FROM system.peers", mockResult(mockValidPeerRow(VERSION1))));
+        new StubbedQuery(
+            "SELECT host_id, schema_version, rpc_address, data_center, rack FROM system.peers",
+            mockResult(mockValidPeerRow(VERSION1))));
 
     // When
     CompletionStage<Boolean> future = checker.run();