Refactor Active Record to let Arel manage bind params

A common source of bugs and code bloat within Active Record has been the
need for us to maintain the list of bind values separately from the AST
they're associated with. This makes any sort of AST manipulation
incredibly difficult, as any time we want to potentially insert or
remove an AST node, we need to traverse the entire tree to find where
the associated bind parameters are.

With this change, the bind parameters now live on the AST directly.
Active Record does not need to know or care about them until the final
AST traversal for SQL construction. Rather than returning just the SQL,
the Arel collector will now return both the SQL and the bind parameters.
At this point the connection adapter will have all the values that it
had before.

A bit of this code is janky and something I'd like to refactor later. In
particular, I don't like how we're handling associations in the
predicate builder, the special casing of `StatementCache::Substitute` in
`QueryAttribute`, or generally how we're handling bind value replacement
in the statement cache when prepared statements are disabled.

This also mostly reverts #26378, as it moved all the code into a
location that I wanted to delete.

/cc @metaskills @yahonda, this change will affect the adapters

Fixes #29766.
Fixes #29804.
Fixes #26541.
Close #28539.
Close #24769.
Close #26468.
Close #26202.

There are probably other issues/PRs that can be closed because of this
commit, but that's all I could find on the first few pages.

8 files changed, 56 insertions, 62 deletions

diff --git a/activerecord/lib/active_record/connection_adapters/abstract/database_statements.rb b/activerecord/lib/active_record/connection_adapters/abstract/database_statements.rb
index f2715cdab0..f687d3a7e8 100644
--- a/activerecord/lib/active_record/connection_adapters/abstract/database_statements.rb
+++ b/activerecord/lib/active_record/connection_adapters/abstract/database_statements.rb
@@ -9,30 +9,36 @@ module ActiveRecord
       end
 
       # Converts an arel AST to SQL
-      def to_sql(arel, binds = [])
-        if arel.respond_to?(:ast)
-          collected = visitor.accept(arel.ast, collector)
-          collected.compile(binds, self).freeze
+      def to_sql(arel_or_sql_string, binds = [])
+        if arel_or_sql_string.respond_to?(:ast)
+          unless binds.empty?
+            raise "Passing bind parameters with an arel AST is forbidden. " \
+              "The values must be stored on the AST directly"
+          end
+          sql, binds = visitor.accept(arel_or_sql_string.ast, collector).value
+          [sql.freeze, binds || []]
         else
-          arel.dup.freeze
+          [arel_or_sql_string.dup.freeze, binds]
         end
       end
 
       # This is used in the StatementCache object. It returns an object that
       # can be used to query the database repeatedly.
       def cacheable_query(klass, arel) # :nodoc:
-        collected = visitor.accept(arel.ast, collector)
         if prepared_statements
-          klass.query(collected.value)
+          sql, binds = visitor.accept(arel.ast, collector).value
+          query = klass.query(sql)
         else
-          klass.partial_query(collected.value)
+          query = klass.partial_query(arel.ast)
+          binds = []
         end
+        [query, binds]
       end
 
       # Returns an ActiveRecord::Result instance.
       def select_all(arel, name = nil, binds = [], preparable: nil)
-        arel, binds = binds_from_relation arel, binds
-        sql = to_sql(arel, binds)
+        arel = arel_from_relation(arel)
+        sql, binds = to_sql(arel, binds)
         if !prepared_statements || (arel.is_a?(String) && preparable.nil?)
           preparable = false
         else
@@ -131,20 +137,23 @@ module ActiveRecord
       #
       # If the next id was calculated in advance (as in Oracle), it should be
       # passed in as +id_value+.
-      def insert(arel, name = nil, pk = nil, id_value = nil, sequence_name = nil, binds = [])
-        value = exec_insert(to_sql(arel, binds), name, binds, pk, sequence_name)
+      def insert(arel, name = nil, pk = nil, id_value = nil, sequence_name = nil)
+        sql, binds = to_sql(arel)
+        value = exec_insert(sql, name, binds, pk, sequence_name)
         id_value || last_inserted_id(value)
       end
       alias create insert
 
       # Executes the update statement and returns the number of rows affected.
-      def update(arel, name = nil, binds = [])
-        exec_update(to_sql(arel, binds), name, binds)
+      def update(arel, name = nil)
+        sql, binds = to_sql(arel)
+        exec_update(sql, name, binds)
       end
 
       # Executes the delete statement and returns the number of rows affected.
-      def delete(arel, name = nil, binds = [])
-        exec_delete(to_sql(arel, binds), name, binds)
+      def delete(arel, name = nil)
+        sql, binds = to_sql(arel)
+        exec_delete(sql, name, binds)
       end
 
       # Returns +true+ when the connection adapter supports prepared statement
@@ -430,11 +439,12 @@ module ActiveRecord
           row && row.first
         end
 
-        def binds_from_relation(relation, binds)
-          if relation.is_a?(Relation) && binds.empty?
-            relation, binds = relation.arel, relation.bound_attributes
+        def arel_from_relation(relation)
+          if relation.is_a?(Relation)
+            relation.arel
+          else
+            relation
           end
-          [relation, binds]
         end
 
         # Fixture value is quoted by Arel, however scalar values
diff --git a/activerecord/lib/active_record/connection_adapters/abstract/query_cache.rb b/activerecord/lib/active_record/connection_adapters/abstract/query_cache.rb
index 077e8beba9..ecf5201d12 100644
--- a/activerecord/lib/active_record/connection_adapters/abstract/query_cache.rb
+++ b/activerecord/lib/active_record/connection_adapters/abstract/query_cache.rb
@@ -92,8 +92,8 @@ module ActiveRecord
 
       def select_all(arel, name = nil, binds = [], preparable: nil)
         if @query_cache_enabled && !locked?(arel)
-          arel, binds = binds_from_relation arel, binds
-          sql = to_sql(arel, binds)
+          arel = arel_from_relation(arel)
+          sql, binds = to_sql(arel, binds)
           cache_sql(sql, name, binds) { super(sql, name, binds, preparable: preparable) }
         else
           super
diff --git a/activerecord/lib/active_record/connection_adapters/abstract/quoting.rb b/activerecord/lib/active_record/connection_adapters/abstract/quoting.rb
index b637518fc8..7b83bc319c 100644
--- a/activerecord/lib/active_record/connection_adapters/abstract/quoting.rb
+++ b/activerecord/lib/active_record/connection_adapters/abstract/quoting.rb
@@ -24,6 +24,10 @@ module ActiveRecord
           return value.quoted_id
         end
 
+        if value.respond_to?(:value_for_database)
+          value = value.value_for_database
+        end
+
         _quote(value)
       end
 
diff --git a/activerecord/lib/active_record/connection_adapters/abstract_adapter.rb b/activerecord/lib/active_record/connection_adapters/abstract_adapter.rb
index 04e32d03e1..e699a77230 100644
--- a/activerecord/lib/active_record/connection_adapters/abstract_adapter.rb
+++ b/activerecord/lib/active_record/connection_adapters/abstract_adapter.rb
@@ -7,7 +7,9 @@ require_relative "sql_type_metadata"
 require_relative "abstract/schema_dumper"
 require_relative "abstract/schema_creation"
 require "arel/collectors/bind"
+require "arel/collectors/composite"
 require "arel/collectors/sql_string"
+require "arel/collectors/substitute_binds"
 
 module ActiveRecord
   module ConnectionAdapters # :nodoc:
@@ -129,19 +131,6 @@ module ActiveRecord
         end
       end
 
-      class BindCollector < Arel::Collectors::Bind
-        def compile(bvs, conn)
-          casted_binds = bvs.map(&:value_for_database)
-          super(casted_binds.map { |value| conn.quote(value) })
-        end
-      end
-
-      class SQLString < Arel::Collectors::SQLString
-        def compile(bvs, conn)
-          super(bvs)
-        end
-      end
-
       def valid_type?(type) # :nodoc:
         !native_database_types[type].nil?
       end
@@ -432,14 +421,14 @@ module ActiveRecord
       end
 
       def case_sensitive_comparison(table, attribute, column, value) # :nodoc:
-        table[attribute].eq(value)
+        table[attribute].eq(Arel::Nodes::BindParam.new(value))
       end
 
       def case_insensitive_comparison(table, attribute, column, value) # :nodoc:
         if can_perform_case_insensitive_comparison_for?(column)
-          table[attribute].lower.eq(table.lower(value))
+          table[attribute].lower.eq(table.lower(Arel::Nodes::BindParam.new(value)))
         else
-          table[attribute].eq(value)
+          table[attribute].eq(Arel::Nodes::BindParam.new(value))
         end
       end
 
@@ -457,24 +446,6 @@ module ActiveRecord
         visitor.accept(node, collector).value
       end
 
-      def combine_bind_parameters(
-        from_clause: [],
-        join_clause: [],
-        where_clause: [],
-        having_clause: [],
-        limit: nil,
-        offset: nil
-      ) # :nodoc:
-        result = from_clause + join_clause + where_clause + having_clause
-        if limit
-          result << limit
-        end
-        if offset
-          result << offset
-        end
-        result
-      end
-
       def default_index_type?(index) # :nodoc:
         index.using.nil?
       end
@@ -609,9 +580,15 @@ module ActiveRecord
 
         def collector
           if prepared_statements
-            SQLString.new
+            Arel::Collectors::Composite.new(
+              Arel::Collectors::SQLString.new,
+              Arel::Collectors::Bind.new,
+            )
           else
-            BindCollector.new
+            Arel::Collectors::SubstituteBinds.new(
+              self,
+              Arel::Collectors::SQLString.new,
+            )
           end
         end
 
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 5915f0db2d..8a9c497918 100644
--- a/activerecord/lib/active_record/connection_adapters/abstract_mysql_adapter.rb
+++ b/activerecord/lib/active_record/connection_adapters/abstract_mysql_adapter.rb
@@ -177,7 +177,8 @@ module ActiveRecord
       #++
 
       def explain(arel, binds = [])
-        sql     = "EXPLAIN #{to_sql(arel, binds)}"
+        sql, binds = to_sql(arel, binds)
+        sql     = "EXPLAIN #{sql}"
         start   = Time.now
         result  = exec_query(sql, "EXPLAIN", binds)
         elapsed = Time.now - start
diff --git a/activerecord/lib/active_record/connection_adapters/mysql/database_statements.rb b/activerecord/lib/active_record/connection_adapters/mysql/database_statements.rb
index 00b56ee9ae..a058a72872 100644
--- a/activerecord/lib/active_record/connection_adapters/mysql/database_statements.rb
+++ b/activerecord/lib/active_record/connection_adapters/mysql/database_statements.rb
@@ -5,7 +5,7 @@ module ActiveRecord
     module MySQL
       module DatabaseStatements
         # Returns an ActiveRecord::Result instance.
-        def select_all(arel, name = nil, binds = [], preparable: nil) # :nodoc:
+        def select_all(*) # :nodoc:
           result = if ExplainRegistry.collect? && prepared_statements
             unprepared_statement { super }
           else
diff --git a/activerecord/lib/active_record/connection_adapters/postgresql/database_statements.rb b/activerecord/lib/active_record/connection_adapters/postgresql/database_statements.rb
index 8db2a645af..0dd4aac463 100644
--- a/activerecord/lib/active_record/connection_adapters/postgresql/database_statements.rb
+++ b/activerecord/lib/active_record/connection_adapters/postgresql/database_statements.rb
@@ -5,7 +5,8 @@ module ActiveRecord
     module PostgreSQL
       module DatabaseStatements
         def explain(arel, binds = [])
-          sql = "EXPLAIN #{to_sql(arel, binds)}"
+          sql, binds = to_sql(arel, binds)
+          sql = "EXPLAIN #{sql}"
           PostgreSQL::ExplainPrettyPrinter.new.pp(exec_query(sql, "EXPLAIN", binds))
         end
 
diff --git a/activerecord/lib/active_record/connection_adapters/sqlite3_adapter.rb b/activerecord/lib/active_record/connection_adapters/sqlite3_adapter.rb
index 10e80179ac..8c12cb09bd 100644
--- a/activerecord/lib/active_record/connection_adapters/sqlite3_adapter.rb
+++ b/activerecord/lib/active_record/connection_adapters/sqlite3_adapter.rb
@@ -203,7 +203,8 @@ module ActiveRecord
       #++
 
       def explain(arel, binds = [])
-        sql = "EXPLAIN QUERY PLAN #{to_sql(arel, binds)}"
+        sql, binds = to_sql(arel, binds)
+        sql = "EXPLAIN QUERY PLAN #{sql}"
         SQLite3::ExplainPrettyPrinter.new.pp(exec_query(sql, "EXPLAIN", []))
       end