Merge pull request #22415 from kamipo/prepared_statements_for_mysql2_adapter

Add prepared statements support for `Mysql2Adapter`

4 files changed, 149 insertions, 151 deletions

diff --git a/activerecord/CHANGELOG.md b/activerecord/CHANGELOG.md
index 3724b1a387..e4e459c0f7 100644
--- a/activerecord/CHANGELOG.md
+++ b/activerecord/CHANGELOG.md
@@ -1,3 +1,7 @@
+*   Add prepared statements support for `Mysql2Adapter`.
+
+    *Ryuta Kamizono*
+
 *   Add schema dumping support for PostgreSQL geometric data types.
 
     *Ryuta Kamizono*
diff --git a/activerecord/lib/active_record/connection_adapters/abstract_mysql_adapter.rb b/activerecord/lib/active_record/connection_adapters/abstract_mysql_adapter.rb
index 735bc0e67a..b0a33ca634 100644
--- a/activerecord/lib/active_record/connection_adapters/abstract_mysql_adapter.rb
+++ b/activerecord/lib/active_record/connection_adapters/abstract_mysql_adapter.rb
@@ -2,6 +2,7 @@ require 'active_record/connection_adapters/abstract_adapter'
 require 'active_record/connection_adapters/mysql/schema_creation'
 require 'active_record/connection_adapters/mysql/schema_definitions'
 require 'active_record/connection_adapters/mysql/schema_dumper'
+require 'active_record/connection_adapters/statement_pool'
 
 require 'active_support/core_ext/string/strip'
 
@@ -141,6 +142,14 @@ module ActiveRecord
       INDEX_TYPES  = [:fulltext, :spatial]
       INDEX_USINGS = [:btree, :hash]
 
+      class StatementPool < ConnectionAdapters::StatementPool
+        private
+
+        def dealloc(stmt)
+          stmt[:stmt].close
+        end
+      end
+
       # FIXME: Make the first parameter more similar for the two adapters
       def initialize(connection, logger, connection_options, config)
         super(connection, logger)
@@ -148,6 +157,7 @@ module ActiveRecord
         @quoted_column_names, @quoted_table_names = {}, {}
 
         @visitor = Arel::Visitors::MySQL.new self
+        @statements = StatementPool.new(self.class.type_cast_config_to_integer(config.fetch(:statement_limit) { 1000 }))
 
         if self.class.type_cast_config_to_boolean(config.fetch(:prepared_statements) { true })
           @prepared_statements = true
@@ -184,6 +194,12 @@ module ActiveRecord
         true
       end
 
+      # Returns true, since this connection adapter supports prepared statement
+      # caching.
+      def supports_statement_cache?
+        true
+      end
+
       # Technically MySQL allows to create indexes with the sort order syntax
       # but at the moment (5.5) it doesn't yet implement them
       def supports_index_sort_order?
@@ -391,9 +407,20 @@ module ActiveRecord
         end
       end
 
+      def select_all(arel, name = nil, binds = [])
+        rows = if ExplainRegistry.collect? && prepared_statements
+          unprepared_statement { super }
+        else
+          super
+        end
+        @connection.next_result while @connection.more_results?
+        rows
+      end
+
+      # Clears the prepared statements cache.
       def clear_cache!
-        super
         reload_type_map
+        @statements.clear
       end
 
       # Executes the SQL statement in the context of this connection.
@@ -404,11 +431,26 @@ module ActiveRecord
       # MysqlAdapter has to free a result after using it, so we use this method to write
       # stuff in an abstract way without concerning ourselves about whether it needs to be
       # explicitly freed or not.
-      def execute_and_free(sql, name = nil) #:nodoc:
+      def execute_and_free(sql, name = nil) # :nodoc:
         yield execute(sql, name)
       end
 
-      def update_sql(sql, name = nil) #:nodoc:
+      def exec_delete(sql, name, binds) # :nodoc:
+        if without_prepared_statement?(binds)
+          execute_and_free(sql, name) { @connection.affected_rows }
+        else
+          exec_stmt_and_free(sql, name, binds) { |stmt| stmt.affected_rows }
+        end
+      end
+      alias :exec_update :exec_delete
+
+      def insert_sql(sql, name = nil, pk = nil, id_value = nil, sequence_name = nil) # :nodoc:
+        super
+        id_value || last_inserted_id
+      end
+      alias :create :insert_sql
+
+      def update_sql(sql, name = nil) # :nodoc:
         super
         @connection.affected_rows
       end
diff --git a/activerecord/lib/active_record/connection_adapters/mysql2_adapter.rb b/activerecord/lib/active_record/connection_adapters/mysql2_adapter.rb
index 3944698910..273a9ee5fa 100644
--- a/activerecord/lib/active_record/connection_adapters/mysql2_adapter.rb
+++ b/activerecord/lib/active_record/connection_adapters/mysql2_adapter.rb
@@ -1,6 +1,6 @@
 require 'active_record/connection_adapters/abstract_mysql_adapter'
 
-gem 'mysql2', '>= 0.3.18', '< 0.5'
+gem 'mysql2', '>= 0.4.2', '< 0.5'
 require 'mysql2'
 
 module ActiveRecord
@@ -33,7 +33,6 @@ module ActiveRecord
 
       def initialize(connection, logger, connection_options, config)
         super
-        @prepared_statements = false
         configure_connection
       end
 
@@ -126,9 +125,13 @@ module ActiveRecord
       # Returns an array of arrays containing the field values.
       # Order is the same as that returned by +columns+.
       def select_rows(sql, name = nil, binds = [])
-        result = execute(sql, name)
+        rows = if without_prepared_statement?(binds)
+          execute_and_free(sql, name) { |result| result.to_a }
+        else
+          exec_stmt_and_free(sql, name, binds) { |stmt, result| result.to_a }
+        end
         @connection.next_result while @connection.more_results?
-        result.to_a
+        rows
       end
 
       # Executes the SQL statement in the context of this connection.
@@ -143,34 +146,58 @@ module ActiveRecord
       end
 
       def exec_query(sql, name = 'SQL', binds = [], prepare: false)
-        result = execute(sql, name)
-        @connection.next_result while @connection.more_results?
-        ActiveRecord::Result.new(result.fields, result.to_a)
+        if without_prepared_statement?(binds)
+          execute_and_free(sql, name) do |result|
+            ActiveRecord::Result.new(result.fields, result.to_a) if result
+          end
+        else
+          exec_stmt_and_free(sql, name, binds, cache_stmt: prepare) do |stmt, result|
+            ActiveRecord::Result.new(result.fields, result.to_a) if result
+          end
+        end
       end
 
-      alias exec_without_stmt exec_query
-
-      def insert_sql(sql, name = nil, pk = nil, id_value = nil, sequence_name = nil)
-        super
-        id_value || @connection.last_id
+      def last_inserted_id(result = nil)
+        @connection.last_id
       end
-      alias :create :insert_sql
 
-      def exec_insert(sql, name, binds, pk = nil, sequence_name = nil)
-        execute to_sql(sql, binds), name
-      end
+      private
 
-      def exec_delete(sql, name, binds)
-        execute to_sql(sql, binds), name
-        @connection.affected_rows
-      end
-      alias :exec_update :exec_delete
+      def exec_stmt_and_free(sql, name, binds, cache_stmt: false)
+        if @connection
+          # make sure we carry over any changes to ActiveRecord::Base.default_timezone that have been
+          # made since we established the connection
+          @connection.query_options[:database_timezone] = ActiveRecord::Base.default_timezone
+        end
 
-      def last_inserted_id(result)
-        @connection.last_id
-      end
+        type_casted_binds = binds.map { |attr| type_cast(attr.value_for_database) }
 
-      private
+        log(sql, name, binds) do
+          if !cache_stmt
+            stmt = @connection.prepare(sql)
+          else
+            cache = @statements[sql] ||= {
+              stmt: @connection.prepare(sql)
+            }
+            stmt = cache[:stmt]
+          end
+
+          begin
+            result = stmt.execute(*type_casted_binds)
+          rescue Mysql2::Error => e
+            if !cache_stmt
+              stmt.close
+            else
+              @statements.delete(sql)
+            end
+            raise e
+          end
+
+          ret = yield stmt, result
+          stmt.close if !cache_stmt
+          ret
+        end
+      end
 
       def connect
         @connection = Mysql2::Client.new(@config)
diff --git a/activerecord/lib/active_record/connection_adapters/mysql_adapter.rb b/activerecord/lib/active_record/connection_adapters/mysql_adapter.rb
index f2d7b54105..7f1710b99c 100644
--- a/activerecord/lib/active_record/connection_adapters/mysql_adapter.rb
+++ b/activerecord/lib/active_record/connection_adapters/mysql_adapter.rb
@@ -1,5 +1,4 @@
 require 'active_record/connection_adapters/abstract_mysql_adapter'
-require 'active_record/connection_adapters/statement_pool'
 require 'active_support/core_ext/hash/keys'
 
 gem 'mysql', '~> 2.9'
@@ -70,27 +69,12 @@ module ActiveRecord
     class MysqlAdapter < AbstractMysqlAdapter
       ADAPTER_NAME = 'MySQL'.freeze
 
-      class StatementPool < ConnectionAdapters::StatementPool
-        private
-
-        def dealloc(stmt)
-          stmt[:stmt].close
-        end
-      end
-
       def initialize(connection, logger, connection_options, config)
         super
-        @statements = StatementPool.new(self.class.type_cast_config_to_integer(config.fetch(:statement_limit) { 1000 }))
         @client_encoding = nil
         connect
       end
 
-      # Returns true, since this connection adapter supports prepared statement
-      # caching.
-      def supports_statement_cache?
-        true
-      end
-
       # HELPER METHODS ===========================================
 
       def each_hash(result) # :nodoc:
@@ -166,27 +150,6 @@ module ActiveRecord
       # DATABASE STATEMENTS ======================================
       #++
 
-      def select_all(arel, name = nil, binds = [])
-        if ExplainRegistry.collect? && prepared_statements
-          unprepared_statement { super }
-        else
-          super
-        end
-      end
-
-      def select_rows(sql, name = nil, binds = [])
-        @connection.query_with_result = true
-        rows = exec_query(sql, name, binds).rows
-        @connection.more_results && @connection.next_result    # invoking stored procedures with CLIENT_MULTI_RESULTS requires this to tidy up else connection will be dropped
-        rows
-      end
-
-      # Clears the prepared statements cache.
-      def clear_cache!
-        super
-        @statements.clear
-      end
-
       # Taken from here:
       #   https://github.com/tmtm/ruby-mysql/blob/master/lib/mysql/charset.rb
       # Author: TOMITA Masahiro <tommy@tmtm.org>
@@ -232,27 +195,55 @@ module ActiveRecord
 
       # Get the client encoding for this database
       def client_encoding
-        return @client_encoding if @client_encoding
+        @client_encoding ||= ENCODINGS[select_value("SELECT @@character_set_client", 'SCHEMA')]
+      end
 
-        result = exec_query(
-          "select @@character_set_client",
-          'SCHEMA')
-        @client_encoding = ENCODINGS[result.rows.last.last]
+      def select_all(arel, name = nil, binds = [])
+        @connection.query_with_result = true
+        super
+      end
+
+      def select_rows(sql, name = nil, binds = [])
+        @connection.query_with_result = true
+        rows = if without_prepared_statement?(binds)
+          execute_and_free(sql, name) { |result| result.to_a }
+        else
+          exec_stmt_and_free(sql, name, binds) { |stmt| stmt.to_a }
+        end
+        @connection.next_result while @connection.more_results?
+        rows
       end
 
       def exec_query(sql, name = 'SQL', binds = [], prepare: false)
         if without_prepared_statement?(binds)
-          result_set, affected_rows = exec_without_stmt(sql, name)
+          execute_and_free(sql, name) do |result|
+            if result
+              types = {}
+              fields = []
+              result.fetch_fields.each { |field|
+                field_name = field.name
+                fields << field_name
+
+                if field.decimals > 0
+                  types[field_name] = Type::Decimal.new
+                else
+                  types[field_name] = Fields.find_type field
+                end
+              }
+              ActiveRecord::Result.new(fields, result.to_a, types)
+            end
+          end
         else
-          result_set, affected_rows = exec_stmt(sql, name, binds, cache_stmt: prepare)
+          exec_stmt_and_free(sql, name, binds, cache_stmt: prepare) do |stmt, result|
+            if result
+              fields = result.fetch_fields.map(&:name)
+              ActiveRecord::Result.new(fields, stmt.to_a)
+            end
+          end
         end
-
-        yield affected_rows if block_given?
-
-        result_set
       end
 
-      def last_inserted_id(result)
+      def last_inserted_id(result = nil)
         @connection.insert_id
       end
 
@@ -322,69 +313,16 @@ module ActiveRecord
         register_class_with_precision m, %r(time)i,     Fields::Time
       end
 
-      def exec_without_stmt(sql, name = 'SQL') # :nodoc:
-        # Some queries, like SHOW CREATE TABLE don't work through the prepared
-        # statement API. For those queries, we need to use this method. :'(
-        log(sql, name) do
-          result = @connection.query(sql)
-          affected_rows = @connection.affected_rows
-
-          if result
-            types = {}
-            fields = []
-            result.fetch_fields.each { |field|
-              field_name = field.name
-              fields << field_name
-
-              if field.decimals > 0
-                types[field_name] = Type::Decimal.new
-              else
-                types[field_name] = Fields.find_type field
-              end
-            }
-
-            result_set = ActiveRecord::Result.new(fields, result.to_a, types)
-            result.free
-          else
-            result_set = ActiveRecord::Result.new([], [])
-          end
-
-          [result_set, affected_rows]
-        end
-      end
-
       def execute_and_free(sql, name = nil) # :nodoc:
         result = execute(sql, name)
         ret = yield result
-        result.free
+        result.free if result
         ret
       end
 
-      def insert_sql(sql, name = nil, pk = nil, id_value = nil, sequence_name = nil) #:nodoc:
-        super sql, name
-        id_value || @connection.insert_id
-      end
-      alias :create :insert_sql
-
-      def exec_delete(sql, name, binds) # :nodoc:
-        affected_rows = 0
-
-        exec_query(sql, name, binds) do |n|
-          affected_rows = n
-        end
-
-        affected_rows
-      end
-      alias :exec_update :exec_delete
-
-      def begin_db_transaction #:nodoc:
-        exec_query "BEGIN"
-      end
-
       private
 
-      def exec_stmt(sql, name, binds, cache_stmt: false)
-        cache = {}
+      def exec_stmt_and_free(sql, name, binds, cache_stmt: false)
         type_casted_binds = binds.map { |attr| type_cast(attr.value_for_database) }
 
         log(sql, name, binds) do
@@ -392,7 +330,7 @@ module ActiveRecord
             stmt = @connection.prepare(sql)
           else
             cache = @statements[sql] ||= {
-              :stmt => @connection.prepare(sql)
+              stmt: @connection.prepare(sql)
             }
             stmt = cache[:stmt]
           end
@@ -407,24 +345,18 @@ module ActiveRecord
             if !cache_stmt
               stmt.close
             else
-              @statements.delete sql
+              @statements.delete(sql)
             end
             raise e
           end
 
-          cols = nil
-          if metadata = stmt.result_metadata
-            cols = cache[:cols] ||= metadata.fetch_fields.map(&:name)
-            metadata.free
-          end
-
-          result_set = ActiveRecord::Result.new(cols, stmt.to_a) if cols
-          affected_rows = stmt.affected_rows
+          result = stmt.result_metadata
+          ret = yield stmt, result
+          result.free if result
 
           stmt.free_result
           stmt.close if !cache_stmt
-
-          [result_set, affected_rows]
+          ret
         end
       end
 
@@ -456,13 +388,6 @@ module ActiveRecord
         super
       end
 
-      def select(sql, name = nil, binds = [])
-        @connection.query_with_result = true
-        rows = super
-        @connection.more_results && @connection.next_result    # invoking stored procedures with CLIENT_MULTI_RESULTS requires this to tidy up else connection will be dropped
-        rows
-      end
-
       # Returns the full version of the connected MySQL server.
       def full_version
         @full_version ||= @connection.server_info