Clear all threads query cache when a connection is pinned

Ref: https://github.com/Shopify/maintenance_tasks/pull/983#issuecomment-1969407080
Ref: https://github.com/rails/rails/pull/51151

Now that query caches are owned by the pool, and assigned on connections
during checkout, when running multithreaded code inside transactional
tests (typically system tests), the two threads uses the same connection
but not the same cache.

So it's important that we do clear the caches for all threads when
a connection is pinned.

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

@@ -129,7 +129,14 @@ def query_cache_enabled
         end
 
         def clear_query_cache
-          query_cache.clear
+          if @pinned_connection
+            # With transactional fixtures, and especially systems test
+            # another thread may use the same connection, but with a different
+            # query cache. So we must clear them all.
+            @thread_query_caches.each_value(&:clear)
+          else
+            query_cache.clear
+          end
         end
 
         private

activerecord/test/cases/query_cache_test.rb

@@ -672,6 +672,40 @@ def test_clear_query_cache_is_called_on_all_connections
     end
   end
 
+  test "query cache is cleared for all thread when a connection is shared" do
+    ActiveRecord::Base.connection_pool.pin_connection!(ActiveSupport::IsolatedExecutionState.context)
+
+    begin
+      assert_cache :off
+      ActiveRecord::Base.connection.enable_query_cache!
+      assert_cache :clean
+
+      Post.first
+      assert_cache :dirty
+
+      thread_a = Thread.new do
+        middleware { |env|
+          assert_cache :dirty # The cache is shared with the main thread
+
+          Post.first
+          assert_cache :dirty
+
+          Post.delete_all
+
+          assert_cache :clean
+
+          [200, {}, nil]
+        }.call({})
+      end
+
+      thread_a.join
+
+      assert_cache :clean
+    ensure
+      ActiveRecord::Base.connection_pool.unpin_connection!
+    end
+  end
+
   private
     def with_temporary_connection_pool(&block)
       pool_config = ActiveRecord::Base.connection.pool.pool_config