RUBY-257 - prepared-statement cache should not be host-scoped.

Author: stamhankar999

CHANGELOG.md

@@ -5,6 +5,10 @@ Features:
 * Update Table metadata to include trigger-collection and view-collection metadata.
 * Added execution profiles to encapsulate a group of request execution options.
 * Added support for v5 beta protocol.
+* Make prepared statement cache not be scoped by host and optimistically execute prepared statements on hosts where
+  we are not sure the statement is already prepared. The motivation is that in the steady state, all nodes have
+  prepared statements already, so there is no need to prepare statements before executing them. If the guess is wrong,
+  the client will prepare and execute at that point.
 
 Bug Fixes:
 * [RUBY-255](https://datastax-oss.atlassian.net/browse/RUBY-255) ControlConnection.peer_ip ignores peers that are missing critical information in system.peers.

lib/cassandra/cluster/client.rb

@@ -47,7 +47,7 @@ def initialize(logger,
         @futures                     = futures_factory
         @connections                 = ::Hash.new
         @prepared_statements         = ::Hash.new
-        @preparing_statements        = ::Hash.new
+        @preparing_statements        = ::Hash.new {|hash, host| hash[host] = {}}
         @pending_connections         = ::Hash.new
         @keyspace                    = nil
         @state                       = :idle
@@ -83,7 +83,6 @@ def connect
 
             connecting_hosts[host] = pool_size
             @pending_connections[host] = 0
-            @prepared_statements[host] = {}
             @preparing_statements[host] = {}
             @connections[host] = ConnectionPool.new
           end
@@ -182,7 +181,6 @@ def host_up(host)
           end
 
           @pending_connections[host] ||= 0
-          @prepared_statements[host] = {}
           @preparing_statements[host] = {}
           @connections[host] = ConnectionPool.new
         end
@@ -197,7 +195,6 @@ def host_down(host)
           return Ione::Future.resolved unless @connections.key?(host)
 
           @pending_connections.delete(host) unless @pending_connections[host] > 0
-          @prepared_statements.delete(host)
           @preparing_statements.delete(host)
           pool = @connections.delete(host)
         end
@@ -632,15 +629,13 @@ def prepare_and_send_request_by_plan(host,
                                            errors,
                                            hosts)
         cql = statement.cql
-        id = nil
-        host_is_up = true
-        synchronize do
-          if @prepared_statements[host].nil?
-            host_is_up = false
-          else
-            id = @prepared_statements[host][cql]
-          end
-        end
+
+        # Get the prepared statement id for this statement from our cache if possible. We are optimistic
+        # that the statement has previously been prepared on all hosts, so the id will be valid. However, if
+        # we're in the midst of preparing the statement on the given host, we know that executing with the id
+        # will fail. So, act like we don't have the prepared-statement id in that case.
+
+        id = synchronize { @preparing_statements[host][cql] ? nil : @prepared_statements[cql] }
 
         if id
           request.id = id
@@ -655,19 +650,6 @@ def prepare_and_send_request_by_plan(host,
                                   timeout,
                                   errors,
                                   hosts)
-        elsif !host_is_up
-          # We've hit a race condition where the plan says we can query this host, but the host has gone
-          # down in the mean time. Just execute the plan again on the next host.
-          @logger.debug("#{host} is down; executing plan on next host")
-          execute_by_plan(promise,
-                          keyspace,
-                          statement,
-                          options,
-                          request,
-                          plan,
-                          timeout,
-                          errors,
-                          hosts)
         else
           prepare = prepare_statement(host, connection, cql, timeout)
           prepare.on_complete do |_|
@@ -825,29 +807,15 @@ def batch_and_send_request_by_plan(host,
           cql = statement.cql
 
           if statement.is_a?(Statements::Bound)
-            host_is_up = true
-            id = nil
-            synchronize do
-              if @prepared_statements[host].nil?
-                host_is_up = false
-              else
-                id = @prepared_statements[host][cql]
-              end
-            end
+            # Get the prepared statement id for this statement from our cache if possible. We are optimistic
+            # that the statement has previously been prepared on all hosts, so the id will be valid. However, if
+            # we're in the midst of preparing the statement on the given host, we know that executing with the id
+            # will fail. So, act like we don't have the prepared-statement id in that case.
+
+            id = synchronize { @preparing_statements[host][cql] ? nil : @prepared_statements[cql] }
 
             if id
               request.add_prepared(id, statement.params, statement.params_types)
-            elsif !host_is_up
-              @logger.debug("#{host} is down; executing on next host in plan")
-              return batch_by_plan(promise,
-                                   keyspace,
-                                   batch_statement,
-                                   options,
-                                   request,
-                                   plan,
-                                   timeout,
-                                   errors,
-                                   hosts)
             else
               unprepared[cql] << statement
             end
@@ -1097,7 +1065,6 @@ def handle_response(response_future,
 
               synchronize do
                 @preparing_statements[host].delete(cql)
-                @prepared_statements[host].delete(cql)
               end
 
               prepare = prepare_statement(host, connection, cql, timeout)
@@ -1198,7 +1165,7 @@ def handle_response(response_future,
             when Protocol::PreparedResultResponse
               cql = request.cql
               synchronize do
-                @prepared_statements[host][cql] = r.id
+                @prepared_statements[cql] = r.id
                 @preparing_statements[host].delete(cql)
               end
 
@@ -1531,7 +1498,7 @@ def prepare_statement(host, connection, cql, timeout)
           when Protocol::PreparedResultResponse
             id = r.id
             synchronize do
-              @prepared_statements[host][cql] = id
+              @prepared_statements[cql] = id
               @preparing_statements[host].delete(cql)
             end
             id

spec/cassandra/cluster/client_spec.rb

@@ -695,7 +695,7 @@ class Cluster
           expect(sent).to be_truthy
         end
 
-        it 're-prepares a statement on new connection' do
+        it 'does not re-prepare a statement on new connection if the new host already has the statement prepared' do
           count = 0
           io_reactor.on_connection do |connection|
             connection.handle_request do |request|
@@ -719,7 +719,12 @@ class Cluster
           cluster_registry.hosts.delete(cluster_registry.hosts.first)
 
           client.execute(statement.bind, execution_options).get
-          expect(count).to eq(2)
+
+          # Expected sequence of events:
+          # 1. prepare on host1
+          # 2. execute on host2. Since execute succeeds, the implication is that host2 already had the statement
+          #    prepared.
+          expect(count).to eq(1)
         end
 
         it 're-prepares a statement on unprepared error' do
@@ -752,7 +757,14 @@ class Cluster
           cluster_registry.hosts.delete(statement.execution_info.hosts.first)
 
           client.execute(statement.bind, execution_options).get
-          expect(count).to eq(3)
+
+
+          # Expected sequence of events:
+          # 1. prepare on host1
+          # 2. execute on host2, yielding unprepared error.
+          # 3. prepare on host2
+          # 4. execute on host2.
+          expect(count).to eq(2)
           expect(error).to be(false)
         end
 
@@ -937,7 +949,7 @@ class Cluster
           expect(sent).to be_truthy
         end
 
-        it 'automatically re-prepares statements' do
+        it 'does not automatically re-prepare statements' do
           sent = false
           count = 0
           batch = Statements::Batch::Logged.new(driver.execution_options)
@@ -983,7 +995,11 @@ class Cluster
 
           client.batch(batch, execution_options.override(consistency: :one)).get
           expect(sent).to be_truthy
-          expect(count).to eq(2)
+
+          # Expected sequence of events:
+          # 1. prepare on host1
+          # 2. run batch on host2 with prepared-statement id's. This succeeds, so no more work to do.
+          expect(count).to eq(1)
         end
 
         it 'follows the plan on failure' do

spec/regressions/RUBY-189_spec.rb

@@ -56,19 +56,17 @@ class Cluster
       let(:request) { double('request') }
       let(:batch_statement) { double('batch_statement') }
       let(:bound_statement) { double('bound_statement') }
+      let(:response) { double('response future')}
+      let(:prepared_id) { 1234 }
 
       it 'RUBY-189 - handles node down after prepare' do
         expect(promise).to_not receive(:break)
         expect(statement).to receive(:cql).and_return('select * from foo')
-        expect(client).to receive(:execute_by_plan).with(promise,
-                                                         'keyspace',
-                                                         statement,
-                                                         options,
-                                                         'request',
-                                                         plan,
-                                                         12,
-                                                         errors,
-                                                         hosts)
+
+        # We expect the down node to receive a request attempt.
+        expect(connection).to receive(:send_request).and_return(response)
+        expect(response).to receive(:map).and_return(response)
+        expect(response).to receive(:on_complete)
         client.send(:prepare_and_send_request_by_plan,
                     'down_host',
                     connection,
@@ -85,19 +83,14 @@ class Cluster
 
       it 'RUBY-189 - handles node down after prepare in batch' do
         expect(request).to receive(:clear)
+        expect(connection).to receive(:send_request).and_return(response)
+        expect(response).to receive(:map).and_return(response)
+        expect(response).to receive(:on_complete)
+
         expect(bound_statement).to receive(:is_a?).and_return(true)
         expect(bound_statement).to receive(:cql).and_return('select * from foo')
         expect(batch_statement).to receive(:statements).and_return([bound_statement])
         expect(promise).to_not receive(:break)
-        expect(client).to receive(:batch_by_plan).with(promise,
-                                                       'keyspace',
-                                                       batch_statement,
-                                                       options,
-                                                       request,
-                                                       plan,
-                                                       12,
-                                                       errors,
-                                                       hosts)
         client.send(:batch_and_send_request_by_plan,
                     'down_host',
                     connection,