Merge pull request #200 from datastax/RUBY-257

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
@@ -1093,17 +1061,23 @@ def handle_response(response_future,
                                                     r.data_present,
                                                     retries)
             when Protocol::UnpreparedErrorResponse
-              cql = statement.cql
-
-              synchronize do
-                @preparing_statements[host].delete(cql)
-                @prepared_statements[host].delete(cql)
+              cql = nil
+              if statement.is_a?(Cassandra::Statements::Batch)
+                # Find the prepared statement with the prepared-statement-id reported by the node.
+                unprepared_child = statement.statements.select do |s|
+                  (s.is_a?(Cassandra::Statements::Prepared) || s.is_a?(Cassandra::Statements::Bound)) && s.id == r.id
+                end.first
+                cql = unprepared_child ? unprepared_child.cql : nil
+              else
+                # This is a normal statement, so we have everything we need.
+                cql = statement.cql
+                synchronize { @preparing_statements[host].delete(cql) }
               end
 
               prepare = prepare_statement(host, connection, cql, timeout)
               prepare.on_complete do |_|
                 if prepare.resolved?
-                  request.id = prepare.value
+                  request.id = prepare.value unless request.is_a?(Cassandra::Protocol::BatchRequest)
                   do_send_request_by_plan(host,
                                           connection,
                                           promise,
@@ -1198,7 +1172,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
 
@@ -1207,7 +1181,8 @@ def handle_response(response_future,
               pk_idx ||= @schema.get_pk_idx(metadata)
 
               promise.fulfill(
-                Statements::Prepared.new(r.custom_payload,
+                Statements::Prepared.new(r.id,
+                                         r.custom_payload,
                                          r.warnings,
                                          cql,
                                          metadata,
@@ -1531,7 +1506,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

lib/cassandra/statements/bound.rb

@@ -28,15 +28,19 @@ class Bound
       attr_reader :params
       # @private
       attr_reader :params_types, :result_metadata, :keyspace, :partition_key
+      # @private prepared-statement id
+      attr_reader :id
 
       # @private
-      def initialize(cql,
+      def initialize(id,
+                     cql,
                      params_types,
                      result_metadata,
                      params,
                      keyspace = nil,
                      partition_key = nil,
                      idempotent = false)
+        @id              = id
         @cql             = cql
         @params_types    = params_types
         @result_metadata = result_metadata

lib/cassandra/statements/prepared.rb

@@ -27,9 +27,12 @@ class Prepared
       attr_reader :cql
       # @private
       attr_reader :result_metadata
+      # @private prepared-statement id
+      attr_reader :id
 
       # @private
-      def initialize(payload,
+      def initialize(id,
+                     payload,
                      warnings,
                      cql,
                      params_metadata,
@@ -44,6 +47,7 @@ def initialize(payload,
                      retries,
                      client,
                      connection_options)
+        @id                 = id
         @payload            = payload
         @warnings           = warnings
         @cql                = cql
@@ -131,7 +135,8 @@ def bind(args = nil)
 
         partition_key = create_partition_key(params)
 
-        Bound.new(@cql,
+        Bound.new(@id,
+                  @cql,
                   param_types,
                   @result_metadata,
                   params,

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/cassandra/session_spec.rb

@@ -131,7 +131,7 @@ module Cassandra
         let(:cql)             { "INSERT INTO songs (id, title, album, artist, tags) VALUES (?, ?, ?, ?, ?)" }
         let(:result_metadata) { nil }
         let(:params_metadata) { Array.new(5) }
-        let(:statement)       { Statements::Prepared.new(nil, nil, cql, params_metadata, result_metadata, nil, nil, nil, nil, VOID_OPTIONS, nil, nil, nil, nil, nil) }
+        let(:statement)       { Statements::Prepared.new(nil, nil, nil, cql, params_metadata, result_metadata, nil, nil, nil, nil, VOID_OPTIONS, nil, nil, nil, nil, nil) }
 
         it 'binds and executes result' do
           promise         = double('promise')
@@ -151,7 +151,7 @@ module Cassandra
         let(:result_metadata) { nil }
         let(:params_metadata) { Array.new(5) }
         let(:params)          { [1,2,3,4,5] }
-        let(:statement)       { Statements::Bound.new(cql, params_metadata, result_metadata, params) }
+        let(:statement)       { Statements::Bound.new(nil, cql, params_metadata, result_metadata, params) }
 
         it 'executes statement' do
           promise = double('promise')

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,