Move schema cache from connection to pool

This PR proposes moving the schema cache from the connection to the pool
so the connection can ask the pool for the cache. In a future PR our
goal is to be able to read the yaml file from the pool so we can get
rid of the `active_record.check_schema_cache_dump` initializer. This
will fix the issues surrounding dumping the schema cache and mulitple
databases.

Why do we want to get rid of the initializer you ask?

Well I was looking at #34449 and trying to make it work for our usecase
and it revealed A LOT of problems. There are a few issues that I will
fix in remaining PRs with SchemaMigration, but there's a big glaring
issue with this initializer.

When you have an application with multiple databases we'll need to loop
through all the configurations and set the schema cache on those
connections. The problem is on initialization we only have one
connection - the one for Ar::Base. This is fine in a single db
application but not fine in multi-db. If we follow the pattern in #34449
and establish a connection to those other dbs we will end up setting the
cache on the _connection object_ rather than on all connections that
connect for that config.

So even though we looped through the configs and assigned the cache the
cache will not be set (or will be set wrong) once the app is booted
because the connection objects after boot are _different_ than the
connection objects we assigned the cache to.

After trying many different ways to set the schema cache `@tenderlove`
and I came to the conclusion that the initializer is problematic, as is
setting the schema cache twice.

This is part 1 to move the cache to the pool so the cache can read from
the schema cache yaml file instead of setting it when initializing the
app.

To do this we have created a `NullPool` that initializes an empty cache. I
put the `get_schema_cache` and `set_schema_cache` in an `AbstractPool`
so we can share code between `ConnectionPool` and `NullPool` instead of
duplicating code.

Now we only need to set the schema_cache on the pool rather than the
connection. In `discard!` we need to unset the connection from the
schema_cache - we still want the cache just not the connection.

activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb

@@ -20,6 +20,26 @@ class ExclusiveConnectionTimeoutError < ConnectionTimeoutError
   end
 
   module ConnectionAdapters
+    module AbstractPool # :nodoc:
+      def get_schema_cache(connection)
+        @schema_cache ||= SchemaCache.new(connection)
+        @schema_cache.connection = connection
+        @schema_cache
+      end
+
+      def set_schema_cache(cache)
+        @schema_cache = cache
+      end
+    end
+
+    class NullPool # :nodoc:
+      include ConnectionAdapters::AbstractPool
+
+      def initialize
+        @schema_cache = nil
+      end
+    end
+
     # Connection pool base class for managing Active Record database
     # connections.
     #
@@ -336,6 +356,7 @@ def run
 
       include MonitorMixin
       include QueryCache::ConnectionPoolConfiguration
+      include ConnectionAdapters::AbstractPool
 
       attr_accessor :automatic_reconnect, :checkout_timeout, :schema_cache
       attr_reader :spec, :connections, :size, :reaper
@@ -837,7 +858,6 @@ def remove_connection_from_thread_cache(conn, owner_thread = conn.owner)
 
         def new_connection
           Base.send(spec.adapter_method, spec.config).tap do |conn|
-            conn.schema_cache = schema_cache.dup if schema_cache
             conn.check_version
           end
         end

activerecord/lib/active_record/connection_adapters/abstract_adapter.rb

@@ -78,7 +78,7 @@ class AbstractAdapter
       SIMPLE_INT = /\A\d+\z/
 
       attr_accessor :pool
-      attr_reader :schema_cache, :visitor, :owner, :logger, :lock, :prepared_statements, :prevent_writes
+      attr_reader :visitor, :owner, :logger, :lock, :prepared_statements, :prevent_writes
       alias :in_use? :owner
 
       set_callback :checkin, :after, :enable_lazy_transactions!
@@ -114,9 +114,8 @@ def initialize(connection, logger = nil, config = {}) # :nodoc:
         @instrumenter        = ActiveSupport::Notifications.instrumenter
         @logger              = logger
         @config              = config
-        @pool                = nil
+        @pool                = ActiveRecord::ConnectionAdapters::NullPool.new
         @idle_since          = Concurrent.monotonic_time
-        @schema_cache        = SchemaCache.new self
         @quoted_column_names, @quoted_table_names = {}, {}
         @prevent_writes = false
         @visitor = arel_visitor
@@ -206,9 +205,13 @@ def lease
         @owner = Thread.current
       end
 
+      def schema_cache
+        @pool.get_schema_cache(self)
+      end
+
       def schema_cache=(cache)
         cache.connection = self
-        @schema_cache = cache
+        @pool.set_schema_cache(cache)
       end
 
       # this method must only be called while holding connection pool's mutex
@@ -487,6 +490,9 @@ def discard!
         #
         # Prevent @connection's finalizer from touching the socket, or
         # otherwise communicating with its server, when it is collected.
+        if schema_cache.connection == self
+          schema_cache.connection = nil
+        end
       end
 
       # Reset the state of this connection, directing the DBMS to clear

activerecord/lib/active_record/connection_adapters/mysql2_adapter.rb

@@ -109,6 +109,7 @@ def disconnect!
       end
 
       def discard! # :nodoc:
+        super
         @connection.automatic_close = false
         @connection = nil
       end

activerecord/lib/active_record/connection_adapters/postgresql_adapter.rb

@@ -302,6 +302,7 @@ def disconnect!
       end
 
       def discard! # :nodoc:
+        super
         @connection.socket_io.reopen(IO::NULL) rescue nil
         @connection = nil
       end

activerecord/lib/active_record/railtie.rb

@@ -134,7 +134,6 @@ class Railtie < Rails::Railtie # :nodoc:
 
               cache = YAML.load(File.read(filename))
               if cache.version == current_version
-                connection.schema_cache = cache
                 connection_pool.schema_cache = cache.dup
               else
                 warn "Ignoring db/schema_cache.yml because it has expired. The current schema version is #{current_version}, but the one in the cache is #{cache.version}."

activerecord/test/cases/connection_pool_test.rb

@@ -507,7 +507,6 @@ def test_pool_sets_connection_schema_cache
         pool.schema_cache = schema_cache
 
         pool.with_connection do |conn|
-          assert_not_same pool.schema_cache, conn.schema_cache
           assert_equal pool.schema_cache.size, conn.schema_cache.size
           assert_same pool.schema_cache.columns(:posts), conn.schema_cache.columns(:posts)
         end