Merge pull request #20211 from kamipo/refactor_statement_pool

Eliminate the duplication code of `StatementPool`

activerecord/lib/active_record/connection_adapters/mysql_adapter.rb

@@ -71,34 +71,10 @@ class MysqlAdapter < AbstractMysqlAdapter
       ADAPTER_NAME = 'MySQL'.freeze
 
       class StatementPool < ConnectionAdapters::StatementPool
-        def initialize(connection, max = 1000)
-          super
-          @cache = Hash.new { |h,pid| h[pid] = {} }
-        end
-
-        def each(&block); cache.each(&block); end
-        def key?(key);    cache.key?(key); end
-        def [](key);      cache[key]; end
-        def length;       cache.length; end
-        def delete(key);  cache.delete(key); end
-
-        def []=(sql, key)
-          while @max <= cache.size
-            cache.shift.last[:stmt].close
-          end
-          cache[sql] = key
-        end
-
-        def clear
-          cache.each_value do |hash|
-            hash[:stmt].close
-          end
-          cache.clear
-        end
-
         private
-        def cache
-          @cache[Process.pid]
+
+        def dealloc(stmt)
+          stmt[:stmt].close
         end
       end
 
@@ -416,8 +392,11 @@ def exec_stmt(sql, name, binds)
             # place when an error occurs. To support older MySQL versions, we
             # need to close the statement and delete the statement from the
             # cache.
-            stmt.close
-            @statements.delete sql
+            if binds.empty?
+              stmt.close
+            else
+              @statements.delete sql
+            end
             raise e
           end
 

activerecord/lib/active_record/connection_adapters/postgresql_adapter.rb

@@ -211,44 +211,18 @@ class StatementPool < ConnectionAdapters::StatementPool
         def initialize(connection, max)
           super
           @counter = 0
-          @cache   = Hash.new { |h,pid| h[pid] = {} }
         end
 
-        def each(&block); cache.each(&block); end
-        def key?(key);    cache.key?(key); end
-        def [](key);      cache[key]; end
-        def length;       cache.length; end
-
         def next_key
           "a#{@counter + 1}"
         end
 
         def []=(sql, key)
-          while @max <= cache.size
-            dealloc(cache.shift.last)
-          end
-          @counter += 1
-          cache[sql] = key
-        end
-
-        def clear
-          cache.each_value do |stmt_key|
-            dealloc stmt_key
-          end
-          cache.clear
-        end
-
-        def delete(sql_key)
-          dealloc cache[sql_key]
-          cache.delete sql_key
+          super.tap { @counter += 1 }
         end
 
         private
 
-          def cache
-            @cache[Process.pid]
-          end
-
           def dealloc(key)
             @connection.query "DEALLOCATE #{key}" if connection_active?
           end

activerecord/lib/active_record/connection_adapters/sqlite3_adapter.rb

@@ -78,37 +78,10 @@ def <=>(version_string)
       end
 
       class StatementPool < ConnectionAdapters::StatementPool
-        def initialize(connection, max)
-          super
-          @cache = Hash.new { |h,pid| h[pid] = {} }
-        end
-
-        def each(&block); cache.each(&block); end
-        def key?(key);    cache.key?(key); end
-        def [](key);      cache[key]; end
-        def length;       cache.length; end
-
-        def []=(sql, key)
-          while @max <= cache.size
-            dealloc(cache.shift.last[:stmt])
-          end
-          cache[sql] = key
-        end
-
-        def clear
-          cache.each_value do |hash|
-            dealloc hash[:stmt]
-          end
-          cache.clear
-        end
-
         private
-        def cache
-          @cache[$$]
-        end
 
         def dealloc(stmt)
-          stmt.close unless stmt.closed?
+          stmt[:stmt].close unless stmt[:stmt].closed?
         end
       end
 

activerecord/lib/active_record/connection_adapters/statement_pool.rb

@@ -4,35 +4,53 @@ class StatementPool
       include Enumerable
 
       def initialize(connection, max = 1000)
+        @cache = Hash.new { |h,pid| h[pid] = {} }
         @connection = connection
         @max        = max
       end
 
-      def each
-        raise NotImplementedError
+      def each(&block)
+        cache.each(&block)
       end
 
       def key?(key)
-        raise NotImplementedError
+        cache.key?(key)
       end
 
       def [](key)
-        raise NotImplementedError
+        cache[key]
       end
 
       def length
-        raise NotImplementedError
+        cache.length
       end
 
-      def []=(sql, key)
-        raise NotImplementedError
+      def []=(sql, stmt)
+        while @max <= cache.size
+          dealloc(cache.shift.last)
+        end
+        cache[sql] = stmt
       end
 
       def clear
-        raise NotImplementedError
+        cache.each_value do |stmt|
+          dealloc stmt
+        end
+        cache.clear
       end
 
       def delete(key)
+        dealloc cache[key]
+        cache.delete(key)
+      end
+
+      private
+
+      def cache
+        @cache[Process.pid]
+      end
+
+      def dealloc(stmt)
         raise NotImplementedError
       end
     end