Add "WHERE key='local'" to accelerate queries against system.local (#434)

Just like other drivers (Python, GoCQL), the Java 4.x driver can query with this simple WHERE clause added, to improve
query performance.

Fixes: #282

Signed-off-by: Yaniv Kaul <[email protected]>

Author: mykaul

core/src/main/java/com/datastax/oss/driver/internal/core/channel/ProtocolInitHandler.java

@@ -83,7 +83,7 @@
 class ProtocolInitHandler extends ConnectInitHandler {
   private static final Logger LOG = LoggerFactory.getLogger(ProtocolInitHandler.class);
   private static final Query CLUSTER_NAME_QUERY =
-      new Query("SELECT cluster_name FROM system.local");
+      new Query("SELECT cluster_name FROM system.local WHERE key='local'");
 
   private final InternalDriverContext context;
   private final long timeoutMillis;

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

@@ -170,7 +170,8 @@ public CompletionStage<Iterable<NodeInfo>> refreshNodeList() {
 
     savePort(channel);
 
-    CompletionStage<AdminResult> localQuery = query(channel, "SELECT * FROM system.local");
+    CompletionStage<AdminResult> localQuery =
+        query(channel, "SELECT * FROM system.local WHERE key='local'");
     CompletionStage<AdminResult> peersV2Query = query(channel, "SELECT * FROM system.peers_v2");
     CompletableFuture<AdminResult> peersQuery = new CompletableFuture<>();
 

core/src/test/java/com/datastax/oss/driver/internal/core/channel/ProtocolInitHandlerTest.java

@@ -434,7 +434,7 @@ public void should_check_cluster_name_if_provided() {
     requestFrame = readOutboundFrame();
     assertThat(requestFrame.message).isInstanceOf(Query.class);
     Query query = (Query) requestFrame.message;
-    assertThat(query.query).isEqualTo("SELECT cluster_name FROM system.local");
+    assertThat(query.query).isEqualTo("SELECT cluster_name FROM system.local WHERE key='local'");
     assertThat(connectFuture).isNotDone();
 
     writeInboundFrame(requestFrame, TestResponses.clusterNameResponse("expectedClusterName"));

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

@@ -337,7 +337,7 @@ public void should_refresh_node_list_from_local_and_peers() {
     AdminRow peer3 = mockPeersRow(3, UUID.randomUUID());
     AdminRow peer2 = mockPeersRow(2, node2.getHostId());
     topologyMonitor.stubQueries(
-        new StubbedQuery("SELECT * FROM system.local", mockResult(local)),
+        new StubbedQuery("SELECT * FROM system.local WHERE key='local'", mockResult(local)),
         new StubbedQuery("SELECT * FROM system.peers_v2", Collections.emptyMap(), null, true),
         new StubbedQuery("SELECT * FROM system.peers", mockResult(peer3, peer2)));
 
@@ -447,7 +447,7 @@ public void should_warn_when_control_host_found_in_system_peers() {
     AdminRow peer2 = mockPeersRow(2, node2.getHostId());
     AdminRow peer1 = mockPeersRow(1, node2.getHostId()); // invalid
     topologyMonitor.stubQueries(
-        new StubbedQuery("SELECT * FROM system.local", mockResult(local)),
+        new StubbedQuery("SELECT * FROM system.local WHERE key='local'", mockResult(local)),
         new StubbedQuery("SELECT * FROM system.peers_v2", Collections.emptyMap(), null, true),
         new StubbedQuery("SELECT * FROM system.peers", mockResult(peer3, peer2, peer1)));
 
@@ -478,7 +478,7 @@ public void should_warn_when_control_host_found_in_system_peers_v2() {
     AdminRow peer2 = mockPeersRow(2, node2.getHostId());
     AdminRow peer1 = mockPeersRow(1, node2.getHostId()); // invalid
     topologyMonitor.stubQueries(
-        new StubbedQuery("SELECT * FROM system.local", mockResult(local)),
+        new StubbedQuery("SELECT * FROM system.local WHERE key='local'", mockResult(local)),
         new StubbedQuery("SELECT * FROM system.peers_v2", mockResult(peer3, peer2, peer1)));
 
     // When

integration-tests/src/test/java/com/datastax/oss/driver/core/ProtocolVersionMixedClusterIT.java

@@ -76,8 +76,8 @@ public void should_downgrade_if_peer_does_not_support_negotiated_version() {
       assertThat(protocolQueries(contactPoint, 4))
           .containsExactly(
               // Initial connection with protocol v4
-              "SELECT cluster_name FROM system.local",
-              "SELECT * FROM system.local",
+              "SELECT cluster_name FROM system.local WHERE key='local'",
+              "SELECT * FROM system.local WHERE key='local'",
               "SELECT * FROM system.peers_v2",
               "SELECT * FROM system.peers");
     }
@@ -104,8 +104,8 @@ public void should_keep_current_if_supported_by_all_peers() {
       assertThat(protocolQueries(contactPoint, 4))
           .containsExactly(
               // Initial connection with protocol v4
-              "SELECT cluster_name FROM system.local",
-              "SELECT * FROM system.local",
+              "SELECT cluster_name FROM system.local WHERE key='local'",
+              "SELECT * FROM system.local WHERE key='local'",
               "SELECT * FROM system.peers_v2",
               "SELECT * FROM system.peers");
     }
@@ -155,8 +155,8 @@ public void should_not_downgrade_and_force_down_old_nodes_if_version_forced() {
       assertThat(protocolQueries(contactPoint, 4))
           .containsExactly(
               // Initial connection with protocol v4
-              "SELECT cluster_name FROM system.local",
-              "SELECT * FROM system.local",
+              "SELECT cluster_name FROM system.local WHERE key='local'",
+              "SELECT * FROM system.local WHERE key='local'",
               "SELECT * FROM system.peers_v2",
               "SELECT * FROM system.peers");
 

integration-tests/src/test/java/com/datastax/oss/driver/core/metadata/ZeroTokenNodesIT.java

@@ -92,7 +92,7 @@ public void should_not_discover_zero_token_DC_when_option_is_disabled() {
 
       Set<String> landedOn = new HashSet<>();
       for (int i = 0; i < 30; i++) {
-        ResultSet rs = session.execute("SELECT * FROM system.local");
+        ResultSet rs = session.execute("SELECT * FROM system.local WHERE key='local'");
         InetAddress broadcastRpcInetAddress =
             Objects.requireNonNull(rs.one()).getInetAddress("rpc_address");
         landedOn.add(Objects.requireNonNull(broadcastRpcInetAddress).toString());
@@ -130,7 +130,7 @@ public void should_discover_zero_token_DC_when_option_is_enabled() {
 
       Set<String> landedOn = new HashSet<>();
       for (int i = 0; i < 30; i++) {
-        ResultSet rs = session.execute("SELECT * FROM system.local");
+        ResultSet rs = session.execute("SELECT * FROM system.local WHERE key='local'");
         InetAddress broadcastRpcInetAddress =
             Objects.requireNonNull(rs.one()).getInetAddress("rpc_address");
         landedOn.add(Objects.requireNonNull(broadcastRpcInetAddress).toString());
@@ -201,7 +201,7 @@ public void should_connect_to_zero_token_DC() {
 
       Set<String> landedOn = new HashSet<>();
       for (int i = 0; i < 30; i++) {
-        ResultSet rs = session.execute("SELECT * FROM system.local");
+        ResultSet rs = session.execute("SELECT * FROM system.local WHERE key='local'");
         InetAddress broadcastRpcInetAddress =
             Objects.requireNonNull(rs.one()).getInetAddress("rpc_address");
         landedOn.add(Objects.requireNonNull(broadcastRpcInetAddress).toString());