Introduce `_delete_record` and use it for deleting a record

activerecord/lib/active_record/locking/optimistic.rb

@@ -112,24 +112,20 @@ def _update_record(attribute_names = self.attribute_names)
         end
 
         def destroy_row
-          affected_rows = super
+          return super unless locking_enabled?
 
-          if locking_enabled? && affected_rows != 1
-            raise ActiveRecord::StaleObjectError.new(self, "destroy")
-          end
+          locking_column = self.class.locking_column
 
-          affected_rows
-        end
+          affected_rows = self.class._delete_record(
+            self.class.primary_key => id_in_database,
+            locking_column => read_attribute_before_type_cast(locking_column)
+          )
 
-        def relation_for_destroy
-          relation = super
-
-          if locking_enabled?
-            locking_column = self.class.locking_column
-            relation = relation.where(locking_column => read_attribute_before_type_cast(locking_column))
+          if affected_rows != 1
+            raise ActiveRecord::StaleObjectError.new(self, "destroy")
           end
 
-          relation
+          affected_rows
         end
 
         module ClassMethods

activerecord/lib/active_record/persistence.rb

@@ -197,6 +197,16 @@ def _update_record(values, constraints) # :nodoc:
         connection.update(um, "#{self} Update")
       end
 
+      def _delete_record(constraints) # :nodoc:
+        constraints = _substitute_values(constraints).map { |attr, bind| attr.eq(bind) }
+
+        dm = Arel::DeleteManager.new
+        dm.from(arel_table)
+        dm.wheres = constraints
+
+        connection.delete(dm, "#{self} Destroy")
+      end
+
       private
         # Called by +instantiate+ to decide which class to use for a new
         # record instance.
@@ -311,7 +321,7 @@ def save!(*args, &block)
     # callbacks or any <tt>:dependent</tt> association
     # options, use <tt>#destroy</tt>.
     def delete
-      _relation_for_itself.delete_all if persisted?
+      _delete_row if persisted?
       @destroyed = true
       freeze
     end
@@ -690,15 +700,11 @@ def destroy_associations
     end
 
     def destroy_row
-      relation_for_destroy.delete_all
-    end
-
-    def relation_for_destroy
-      _relation_for_itself
+      _delete_row
     end
 
-    def _relation_for_itself
-      self.class.unscoped.where(self.class.primary_key => id_in_database)
+    def _delete_row
+      self.class._delete_record(self.class.primary_key => id_in_database)
     end
 
     def create_or_update(*args, &block)