Fix insert_fixtures_set to be restored original connection flags

#33363 has two regressions. First one is that `insert_fixtures_set` is
failed if flags is an array. Second one is that connection flags are not
restored if `set_server_option` is not supported.

activerecord/lib/active_record/connection_adapters/abstract_mysql_adapter.rb

@@ -558,35 +558,6 @@ module ActiveRecord
           @max_allowed_packet ||= (show_variable("max_allowed_packet") - bytes_margin)
         end
 
-        def with_multi_statements
-          if supports_set_server_option?
-            @connection.set_server_option(Mysql2::Client::OPTION_MULTI_STATEMENTS_ON)
-          elsif !supports_multi_statements?
-            previous_flags = @config[:flags]
-            @config[:flags] = Mysql2::Client::MULTI_STATEMENTS
-            reconnect!
-          end
-
-          yield
-        ensure
-          unless supports_multi_statements?
-            if supports_set_server_option?
-              @connection.set_server_option(Mysql2::Client::OPTION_MULTI_STATEMENTS_OFF)
-            else
-              @config[:flags] = previous_flags
-              reconnect!
-            end
-          end
-        end
-
-        def supports_multi_statements?
-          (@config[:flags] & Mysql2::Client::MULTI_STATEMENTS) != 0
-        end
-
-        def supports_set_server_option?
-          @connection.respond_to?(:set_server_option)
-        end
-
         def initialize_type_map(m = type_map)
           super
 

activerecord/lib/active_record/connection_adapters/mysql/database_statements.rb

@@ -62,6 +62,42 @@ module ActiveRecord
             @connection.abandon_results!
           end
 
+          def supports_set_server_option?
+            @connection.respond_to?(:set_server_option)
+          end
+
+          def multi_statements_enabled?(flags)
+            if flags.is_a?(Array)
+              flags.include?("MULTI_STATEMENTS")
+            else
+              (flags & Mysql2::Client::MULTI_STATEMENTS) != 0
+            end
+          end
+
+          def with_multi_statements
+            previous_flags = @config[:flags]
+
+            unless multi_statements_enabled?(previous_flags)
+              if supports_set_server_option?
+                @connection.set_server_option(Mysql2::Client::OPTION_MULTI_STATEMENTS_ON)
+              else
+                @config[:flags] = Mysql2::Client::MULTI_STATEMENTS
+                reconnect!
+              end
+            end
+
+            yield
+          ensure
+            unless multi_statements_enabled?(previous_flags)
+              if supports_set_server_option?
+                @connection.set_server_option(Mysql2::Client::OPTION_MULTI_STATEMENTS_OFF)
+              else
+                @config[:flags] = previous_flags
+                reconnect!
+              end
+            end
+          end
+
           def exec_stmt_and_free(sql, name, binds, cache_stmt: false)
             # make sure we carry over any changes to ActiveRecord::Base.default_timezone that have been
             # made since we established the connection

activerecord/test/cases/fixtures_test.rb

@@ -1,6 +1,7 @@
 # frozen_string_literal: true
 
 require "cases/helper"
+require "support/connection_helper"
 require "models/admin"
 require "models/admin/account"
 require "models/admin/randomly_named_c1"
@@ -32,6 +33,8 @@ require "models/treasure"
 require "tempfile"
 
 class FixturesTest < ActiveRecord::TestCase
+  include ConnectionHelper
+
   self.use_instantiated_fixtures = true
   self.use_transactional_tests = false
 
@@ -122,16 +125,88 @@ class FixturesTest < ActiveRecord::TestCase
         ]
       }
 
-      ActiveRecord::Base.transaction do
-        con = ActiveRecord::Base.connection
-        assert_equal 1, con.open_transactions
-        con.insert_fixtures_set(fixtures)
-        assert_equal 1, con.open_transactions
+      assert_difference "TrafficLight.count" do
+        ActiveRecord::Base.transaction do
+          conn = ActiveRecord::Base.connection
+          assert_equal 1, conn.open_transactions
+          conn.insert_fixtures_set(fixtures)
+          assert_equal 1, conn.open_transactions
+        end
       end
     end
   end
 
   if current_adapter?(:Mysql2Adapter)
+    def test_bulk_insert_with_multi_statements_enabled
+      run_without_connection do |orig_connection|
+        ActiveRecord::Base.establish_connection(
+          orig_connection.merge(flags: %w[MULTI_STATEMENTS])
+        )
+
+        fixtures = {
+          "traffic_lights" => [
+            { "location" => "US", "state" => ["NY"], "long_state" => ["a"] },
+          ]
+        }
+
+        ActiveRecord::Base.connection.stub(:supports_set_server_option?, false) do
+          assert_nothing_raised do
+            conn = ActiveRecord::Base.connection
+            conn.execute("SELECT 1; SELECT 2;")
+            conn.raw_connection.abandon_results!
+          end
+
+          assert_difference "TrafficLight.count" do
+            ActiveRecord::Base.transaction do
+              conn = ActiveRecord::Base.connection
+              assert_equal 1, conn.open_transactions
+              conn.insert_fixtures_set(fixtures)
+              assert_equal 1, conn.open_transactions
+            end
+          end
+
+          assert_nothing_raised do
+            conn = ActiveRecord::Base.connection
+            conn.execute("SELECT 1; SELECT 2;")
+            conn.raw_connection.abandon_results!
+          end
+        end
+      end
+    end
+
+    def test_bulk_insert_with_multi_statements_disabled
+      run_without_connection do |orig_connection|
+        ActiveRecord::Base.establish_connection(
+          orig_connection.merge(flags: [])
+        )
+
+        fixtures = {
+          "traffic_lights" => [
+            { "location" => "US", "state" => ["NY"], "long_state" => ["a"] },
+          ]
+        }
+
+        ActiveRecord::Base.connection.stub(:supports_set_server_option?, false) do
+          assert_raises(ActiveRecord::StatementInvalid) do
+            conn = ActiveRecord::Base.connection
+            conn.execute("SELECT 1; SELECT 2;")
+            conn.raw_connection.abandon_results!
+          end
+
+          assert_difference "TrafficLight.count" do
+            conn = ActiveRecord::Base.connection
+            conn.insert_fixtures_set(fixtures)
+          end
+
+          assert_raises(ActiveRecord::StatementInvalid) do
+            conn = ActiveRecord::Base.connection
+            conn.execute("SELECT 1; SELECT 2;")
+            conn.raw_connection.abandon_results!
+          end
+        end
+      end
+    end
+
     def test_insert_fixtures_set_raises_an_error_when_max_allowed_packet_is_smaller_than_fixtures_set_size
       conn = ActiveRecord::Base.connection
       mysql_margin = 2