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

* Fixed some issues in batch request handling.

Author: stamhankar999

lib/cassandra/cluster/client.rb

@@ -1061,16 +1061,23 @@ def handle_response(response_future,
                                                     r.data_present,
                                                     retries)
             when Protocol::UnpreparedErrorResponse
-              cql = statement.cql
-
-              synchronize do
-                @preparing_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,
@@ -1174,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,

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/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')