Merge branch 'master' into set_binds

* master: (2794 commits)
  doc, API example on how to use `Model#exists?` with multiple IDs. [ci skip]
  Restore DATABASE_URL even if it's nil in connection_handler test
  [ci skip] - error_messages_for has been deprecated since 2.3.8 - lets reduce any confusion for users
  Ensure Active Record connection consistency
  Revert "ask the fixture set for the sql statements"
  Check `respond_to` before delegation due to: https://github.com/ruby/ruby/commit/d781caaf313b8649948c107bba277e5ad7307314
  Adding Hash#compact and Hash#compact! methods
  MySQL version 4.1 was EOL on December 31, 2009 We should at least recommend modern versions of MySQL to users.
  clear cache on body close so that cache remains during rendering
  add a more restricted codepath for templates fixes #13390
  refactor generator tests to use block form of Tempfile
  Fix typo [ci skip]
  Move finish_template as the last public method in the generator
  Minor typos fix [ci skip]
  make `change_column_null` reversible. Closes #13576.
  create/drop test and development databases only if RAILS_ENV is nil
  Revert "Speedup String#to"
  typo fix in test name. [ci skip].
  `core_ext/string/access.rb` test what we are documenting.
  Fix typo in image_tag documentation
  ...

Conflicts:
	activerecord/lib/active_record/associations/join_dependency/join_association.rb
	activerecord/lib/active_record/relation/query_methods.rb

31 files changed, 979 insertions, 952 deletions

diff --git a/activerecord/lib/active_record/associations/association.rb b/activerecord/lib/active_record/associations/association.rb
index 608a6af16c..67ea489b22 100644
--- a/activerecord/lib/active_record/associations/association.rb
+++ b/activerecord/lib/active_record/associations/association.rb
@@ -13,11 +13,11 @@ module ActiveRecord
     #       BelongsToAssociation
     #         BelongsToPolymorphicAssociation
     #     CollectionAssociation
-    #       HasAndBelongsToManyAssociation
     #       HasManyAssociation
     #         HasManyThroughAssociation + ThroughAssociation
     class Association #:nodoc:
       attr_reader :owner, :target, :reflection
+      attr_accessor :inversed
 
       delegate :options, :to => :reflection
 
@@ -43,6 +43,7 @@ module ActiveRecord
         @loaded = false
         @target = nil
         @stale_state = nil
+        @inversed = false
       end
 
       # Reloads the \target and returns +self+ on success.
@@ -60,18 +61,19 @@ module ActiveRecord
 
       # Asserts the \target has been loaded setting the \loaded flag to +true+.
       def loaded!
-        @loaded      = true
+        @loaded = true
         @stale_state = stale_state
+        @inversed = false
       end
 
       # The target is stale if the target no longer points to the record(s) that the
       # relevant foreign_key(s) refers to. If stale, the association accessor method
       # on the owner will reload the target. It's up to subclasses to implement the
-      # state_state method if relevant.
+      # stale_state method if relevant.
       #
       # Note that if the target has not been loaded, it is not considered stale.
       def stale_target?
-        loaded? && @stale_state != stale_state
+        !inversed && loaded? && @stale_state != stale_state
       end
 
       # Sets the target of this association to <tt>\target</tt>, and the \loaded flag to +true+.
@@ -84,11 +86,6 @@ module ActiveRecord
         target_scope.merge(association_scope)
       end
 
-      def scoped
-        ActiveSupport::Deprecation.warn "#scoped is deprecated. use #scope instead."
-        scope
-      end
-
       # The scope for this association.
       #
       # Note that the association_scope is merged into the target_scope only when the
@@ -107,10 +104,12 @@ module ActiveRecord
 
       # Set the inverse association, if possible
       def set_inverse_instance(record)
-        if record && invertible_for?(record)
+        if invertible_for?(record)
           inverse = record.association(inverse_reflection_for(record).name)
           inverse.target = owner
+          inverse.inversed = true
         end
+        record
       end
 
       # Returns the class of the target. belongs_to polymorphic overrides this to look at the
@@ -122,11 +121,7 @@ module ActiveRecord
       # Can be overridden (i.e. in ThroughAssociation) to merge in other scopes (i.e. the
       # through association's scope)
       def target_scope
-        all = klass.all
-        scope = AssociationRelation.new(klass, klass.arel_table, self)
-        scope.merge! all
-        scope.default_scoped = all.default_scoped?
-        scope
+        AssociationRelation.create(klass, klass.arel_table, self).merge!(klass.all)
       end
 
       # Loads the \target if needed and returns it.
@@ -200,13 +195,14 @@ module ActiveRecord
           creation_attributes.each { |key, value| record[key] = value }
         end
 
-        # Should be true if there is a foreign key present on the owner which
+        # Returns true if there is a foreign key present on the owner which
         # references the target. This is used to determine whether we can load
         # the target if the owner is currently a new record (and therefore
-        # without a key).
+        # without a key). If the owner is a new record then foreign_key must
+        # be present in order to load target.
         #
         # Currently implemented by belongs_to (vanilla and polymorphic) and
-        # has_one/has_many :through associations which go through a belongs_to
+        # has_one/has_many :through associations which go through a belongs_to.
         def foreign_key_present?
           false
         end
@@ -231,7 +227,12 @@ module ActiveRecord
         # Returns true if inverse association on the given record needs to be set.
         # This method is redefined by subclasses.
         def invertible_for?(record)
-          inverse_reflection_for(record)
+          foreign_key_for?(record) && inverse_reflection_for(record)
+        end
+
+        # Returns true if record contains the foreign_key
+        def foreign_key_for?(record)
+          record.attributes.has_key? reflection.foreign_key
         end
 
         # This should be implemented to return the values of the relevant key(s) on the owner,
diff --git a/activerecord/lib/active_record/associations/association_scope.rb b/activerecord/lib/active_record/associations/association_scope.rb
index 21840ba74a..f455b6934e 100644
--- a/activerecord/lib/active_record/associations/association_scope.rb
+++ b/activerecord/lib/active_record/associations/association_scope.rb
@@ -44,18 +44,6 @@ module ActiveRecord
         chain.each_with_index do |reflection, i|
           table, foreign_table = tables.shift, tables.first
 
-          if reflection.source_macro == :has_and_belongs_to_many
-            join_table = tables.shift
-
-            scope = scope.joins(join(
-              join_table,
-              table[reflection.association_primary_key].
-                eq(join_table[reflection.association_foreign_key])
-            ))
-
-            table, foreign_table = join_table, tables.first
-          end
-
           if reflection.source_macro == :belongs_to
             if reflection.options[:polymorphic]
               key = reflection.association_primary_key(self.klass)
@@ -82,24 +70,30 @@ module ActiveRecord
             constraint = table[key].eq(foreign_table[foreign_key])
 
             if reflection.type
-              type = chain[i + 1].klass.base_class.name
-              constraint = constraint.and(table[reflection.type].eq(type))
+              value    = chain[i + 1].klass.base_class.name
+              bind_val = bind scope, table.table_name, reflection.type.to_s, value
+              scope    = scope.where(table[reflection.type].eq(bind_val))
             end
 
             scope = scope.joins(join(foreign_table, constraint))
           end
 
+          is_first_chain = i == 0
+          klass = is_first_chain ? self.klass : reflection.klass
+
           # Exclude the scope of the association itself, because that
           # was already merged in the #scope method.
           scope_chain[i].each do |scope_chain_item|
-            klass = i == 0 ? self.klass : reflection.klass
             item  = eval_scope(klass, scope_chain_item)
 
             if scope_chain_item == self.reflection.scope
-              scope.merge! item.except(:where, :includes)
+              scope.merge! item.except(:where, :includes, :bind)
+            end
+
+            if is_first_chain
+              scope.includes! item.includes_values
             end
 
-            scope.includes! item.includes_values
             scope.where_values += item.where_values
             scope.bind_values  += item.bind_values
             scope.order_values |= item.order_values
@@ -120,7 +114,7 @@ module ActiveRecord
           # the owner
           klass.table_name
         else
-          reflection.table_name
+          super
         end
       end
 
diff --git a/activerecord/lib/active_record/associations/belongs_to_association.rb b/activerecord/lib/active_record/associations/belongs_to_association.rb
index 8eec4f56af..8272a5584c 100644
--- a/activerecord/lib/active_record/associations/belongs_to_association.rb
+++ b/activerecord/lib/active_record/associations/belongs_to_association.rb
@@ -8,13 +8,16 @@ module ActiveRecord
       end
 
       def replace(record)
-        raise_on_type_mismatch!(record) if record
-
-        update_counters(record)
-        replace_keys(record)
-        set_inverse_instance(record)
-
-        @updated = true if record
+        if record
+          raise_on_type_mismatch!(record)
+          update_counters(record)
+          replace_keys(record)
+          set_inverse_instance(record)
+          @updated = true
+        else
+          decrement_counters
+          remove_keys
+        end
 
         self.target = record
       end
@@ -34,35 +37,41 @@ module ActiveRecord
           !loaded? && foreign_key_present? && klass
         end
 
-        def update_counters(record)
+        def with_cache_name
           counter_cache_name = reflection.counter_cache_column
+          return unless counter_cache_name && owner.persisted?
+          yield counter_cache_name
+        end
+
+        def update_counters(record)
+          with_cache_name do |name|
+            return unless different_target? record
+            record.class.increment_counter(name, record.id)
+            decrement_counter name
+          end
+        end
 
-          if counter_cache_name && owner.persisted? && different_target?(record)
-            if record
-              record.class.increment_counter(counter_cache_name, record.id)
-            end
+        def decrement_counters
+          with_cache_name { |name| decrement_counter name }
+        end
 
-            if foreign_key_present?
-              klass.decrement_counter(counter_cache_name, target_id)
-            end
+        def decrement_counter counter_cache_name
+          if foreign_key_present?
+            klass.decrement_counter(counter_cache_name, target_id)
           end
         end
 
         # Checks whether record is different to the current target, without loading it
         def different_target?(record)
-          if record.nil? 
-            owner[reflection.foreign_key] 
-          else
-            record.id != owner[reflection.foreign_key]
-          end
+          record.id != owner[reflection.foreign_key]
         end
 
         def replace_keys(record)
-          if record
-            owner[reflection.foreign_key] = record[reflection.association_primary_key(record.class)]
-          else
-            owner[reflection.foreign_key] = nil
-          end
+          owner[reflection.foreign_key] = record[reflection.association_primary_key(record.class)]
+        end
+
+        def remove_keys
+          owner[reflection.foreign_key] = nil
         end
 
         def foreign_key_present?
diff --git a/activerecord/lib/active_record/associations/belongs_to_polymorphic_association.rb b/activerecord/lib/active_record/associations/belongs_to_polymorphic_association.rb
index eae5eed3a1..b710cf6bdb 100644
--- a/activerecord/lib/active_record/associations/belongs_to_polymorphic_association.rb
+++ b/activerecord/lib/active_record/associations/belongs_to_polymorphic_association.rb
@@ -11,7 +11,12 @@ module ActiveRecord
 
         def replace_keys(record)
           super
-          owner[reflection.foreign_type] = record && record.class.base_class.name
+          owner[reflection.foreign_type] = record.class.base_class.name
+        end
+
+        def remove_keys
+          super
+          owner[reflection.foreign_type] = nil
         end
 
         def different_target?(record)
diff --git a/activerecord/lib/active_record/associations/builder/association.rb b/activerecord/lib/active_record/associations/builder/association.rb
index 6a3658f328..3911d1b520 100644
--- a/activerecord/lib/active_record/associations/builder/association.rb
+++ b/activerecord/lib/active_record/associations/builder/association.rb
@@ -1,62 +1,66 @@
+require 'active_support/core_ext/module/attribute_accessors'
+
 # This is the parent Association class which defines the variables
 # used by all associations.
 #
 # The hierarchy is defined as follows:
-#  Association 
+#  Association
 #    - SingularAssociation
 #      - BelongsToAssociation
 #      - HasOneAssociation
 #    - CollectionAssociation
 #      - HasManyAssociation
-#      - HasAndBelongsToManyAssociation
 
 module ActiveRecord::Associations::Builder
   class Association #:nodoc:
     class << self
+      attr_accessor :extensions
+      # TODO: This class accessor is needed to make activerecord-deprecated_finders work.
+      # We can move it to a constant in 5.0.
       attr_accessor :valid_options
     end
+    self.extensions = []
 
-    self.valid_options = [:class_name, :foreign_key, :validate]
+    self.valid_options = [:class_name, :class, :foreign_key, :validate]
 
-    attr_reader :model, :name, :scope, :options, :reflection
+    attr_reader :name, :scope, :options
 
-    def self.build(*args, &block)
-      new(*args, &block).build
+    def self.build(model, name, scope, options, &block)
+      builder = create_builder model, name, scope, options, &block
+      reflection = builder.build(model)
+      define_accessors model, reflection
+      define_callbacks model, reflection
+      builder.define_extensions model
+      reflection
     end
 
-    def initialize(model, name, scope, options)
+    def self.create_builder(model, name, scope, options, &block)
       raise ArgumentError, "association names must be a Symbol" unless name.kind_of?(Symbol)
 
-      @model   = model
-      @name    = name
+      new(model, name, scope, options, &block)
+    end
 
+    def initialize(model, name, scope, options)
+      # TODO: Move this to create_builder as soon we drop support to activerecord-deprecated_finders.
       if scope.is_a?(Hash)
-        @scope   = nil
-        @options = scope
-      else
-        @scope   = scope
-        @options = options
+        options = scope
+        scope   = nil
       end
 
-      if @scope && @scope.arity == 0
-        prev_scope = @scope
-        @scope = proc { instance_exec(&prev_scope) }
-      end
-    end
+      # TODO: Remove this model argument as soon we drop support to activerecord-deprecated_finders.
+      @name    = name
+      @scope   = scope
+      @options = options
 
-    def mixin
-      @model.generated_feature_methods
-    end
+      validate_options
 
-    include Module.new { def build; end }
+      if scope && scope.arity == 0
+        @scope = proc { instance_exec(&scope) }
+      end
+    end
 
-    def build
-      validate_options
-      define_accessors
-      configure_dependency if options[:dependent]
-      @reflection = model.create_reflection(macro, name, scope, options, model)
-      super # provides an extension point
-      @reflection
+    def build(model)
+      ActiveRecord::Reflection.create(macro, name, scope, options, model)
     end
 
     def macro
@@ -64,26 +68,37 @@ module ActiveRecord::Associations::Builder
     end
 
     def valid_options
-      Association.valid_options
+      Association.valid_options + Association.extensions.flat_map(&:valid_options)
     end
 
     def validate_options
       options.assert_valid_keys(valid_options)
     end
-    
+
+    def define_extensions(model)
+    end
+
+    def self.define_callbacks(model, reflection)
+      add_before_destroy_callbacks(model, reflection) if reflection.options[:dependent]
+      Association.extensions.each do |extension|
+        extension.build model, reflection
+      end
+    end
+
     # Defines the setter and getter methods for the association
     # class Post < ActiveRecord::Base
     #   has_many :comments
     # end
-    # 
+    #
     # Post.first.comments and Post.first.comments= methods are defined by this method...
-
-    def define_accessors
-      define_readers
-      define_writers
+    def self.define_accessors(model, reflection)
+      mixin = model.generated_association_methods
+      name = reflection.name
+      define_readers(mixin, name)
+      define_writers(mixin, name)
     end
 
-    def define_readers
+    def self.define_readers(mixin, name)
       mixin.class_eval <<-CODE, __FILE__, __LINE__ + 1
         def #{name}(*args)
           association(:#{name}).reader(*args)
@@ -91,7 +106,7 @@ module ActiveRecord::Associations::Builder
       CODE
     end
 
-    def define_writers
+    def self.define_writers(mixin, name)
       mixin.class_eval <<-CODE, __FILE__, __LINE__ + 1
         def #{name}=(value)
           association(:#{name}).writer(value)
@@ -99,29 +114,19 @@ module ActiveRecord::Associations::Builder
       CODE
     end
 
-    def configure_dependency
-      unless valid_dependent_options.include? options[:dependent]
-        raise ArgumentError, "The :dependent option must be one of #{valid_dependent_options}, but is :#{options[:dependent]}"
-      end
-
-      if options[:dependent] == :restrict
-        ActiveSupport::Deprecation.warn(
-          "The :restrict option is deprecated. Please use :restrict_with_exception instead, which " \
-          "provides the same functionality."
-        )
-      end
+    def self.valid_dependent_options
+      raise NotImplementedError
+    end
 
-      mixin.class_eval <<-CODE, __FILE__, __LINE__ + 1
-        def #{macro}_dependent_for_#{name}
-          association(:#{name}).handle_dependency
-        end
-      CODE
+    private
 
-      model.before_destroy "#{macro}_dependent_for_#{name}"
-    end
+    def self.add_before_destroy_callbacks(model, reflection)
+      unless valid_dependent_options.include? reflection.options[:dependent]
+        raise ArgumentError, "The :dependent option must be one of #{valid_dependent_options}, but is :#{reflection.options[:dependent]}"
+      end
 
-    def valid_dependent_options
-      raise NotImplementedError
+      name = reflection.name
+      model.before_destroy lambda { |o| o.association(name).handle_dependency }
     end
   end
 end
diff --git a/activerecord/lib/active_record/associations/builder/belongs_to.rb b/activerecord/lib/active_record/associations/builder/belongs_to.rb
index 63e9526436..5ccaa55a32 100644
--- a/activerecord/lib/active_record/associations/builder/belongs_to.rb
+++ b/activerecord/lib/active_record/associations/builder/belongs_to.rb
@@ -5,96 +5,134 @@ module ActiveRecord::Associations::Builder
     end
 
     def valid_options
-      super + [:foreign_type, :polymorphic, :touch]
+      super + [:foreign_type, :polymorphic, :touch, :counter_cache]
     end
 
-    def constructable?
-      !options[:polymorphic]
+    def self.valid_dependent_options
+      [:destroy, :delete]
     end
 
-    def build
-      reflection = super
-      add_counter_cache_callbacks(reflection) if options[:counter_cache]
-      add_touch_callbacks(reflection)         if options[:touch]
-      reflection
+    def self.define_callbacks(model, reflection)
+      super
+      add_counter_cache_callbacks(model, reflection) if reflection.options[:counter_cache]
+      add_touch_callbacks(model, reflection)         if reflection.options[:touch]
     end
 
-    def add_counter_cache_callbacks(reflection)
-      cache_column = reflection.counter_cache_column
-      foreign_key = reflection.foreign_key
+    def self.define_accessors(mixin, reflection)
+      super
+      add_counter_cache_methods mixin
+    end
+
+    private
+
+    def self.add_counter_cache_methods(mixin)
+      return if mixin.method_defined? :belongs_to_counter_cache_after_create
 
-      mixin.class_eval <<-CODE, __FILE__, __LINE__ + 1
-        def belongs_to_counter_cache_after_create_for_#{name}
-          if record = #{name}
-            record.class.increment_counter(:#{cache_column}, record.id)
+      mixin.class_eval do
+        def belongs_to_counter_cache_after_create(reflection)
+          if record = send(reflection.name)
+            cache_column = reflection.counter_cache_column
+            record.class.increment_counter(cache_column, record.id)
             @_after_create_counter_called = true
           end
         end
 
-        def belongs_to_counter_cache_before_destroy_for_#{name}
-          unless destroyed_by_association && destroyed_by_association.foreign_key.to_sym == #{foreign_key.to_sym.inspect}
-            record = #{name}
+        def belongs_to_counter_cache_before_destroy(reflection)
+          foreign_key = reflection.foreign_key.to_sym
+          unless destroyed_by_association && destroyed_by_association.foreign_key.to_sym == foreign_key
+            record = send reflection.name
             if record && !self.destroyed?
-              record.class.decrement_counter(:#{cache_column}, record.id)
+              cache_column = reflection.counter_cache_column
+              record.class.decrement_counter(cache_column, record.id)
             end
           end
         end
 
-        def belongs_to_counter_cache_after_update_for_#{name}
+        def belongs_to_counter_cache_after_update(reflection)
+          foreign_key  = reflection.foreign_key
+          cache_column = reflection.counter_cache_column
+
           if (@_after_create_counter_called ||= false)
             @_after_create_counter_called = false
-          elsif self.#{foreign_key}_changed? && !new_record? && defined?(#{name.to_s.camelize})
-            model = #{name.to_s.camelize}
-            foreign_key_was = self.#{foreign_key}_was
-            foreign_key = self.#{foreign_key}
+          elsif attribute_changed?(foreign_key) && !new_record? && reflection.constructable?
+            model           = reflection.klass
+            foreign_key_was = attribute_was foreign_key
+            foreign_key     = attribute foreign_key
 
             if foreign_key && model.respond_to?(:increment_counter)
-              model.increment_counter(:#{cache_column}, foreign_key)
+              model.increment_counter(cache_column, foreign_key)
             end
             if foreign_key_was && model.respond_to?(:decrement_counter)
-              model.decrement_counter(:#{cache_column}, foreign_key_was)
+              model.decrement_counter(cache_column, foreign_key_was)
             end
           end
         end
-      CODE
+      end
+    end
+
+    def self.add_counter_cache_callbacks(model, reflection)
+      cache_column = reflection.counter_cache_column
+
+      model.after_create lambda { |record|
+        record.belongs_to_counter_cache_after_create(reflection)
+      }
 
-      model.after_create   "belongs_to_counter_cache_after_create_for_#{name}"
-      model.before_destroy "belongs_to_counter_cache_before_destroy_for_#{name}"
-      model.after_update   "belongs_to_counter_cache_after_update_for_#{name}"
+      model.before_destroy lambda { |record|
+        record.belongs_to_counter_cache_before_destroy(reflection)
+      }
+
+      model.after_update lambda { |record|
+        record.belongs_to_counter_cache_after_update(reflection)
+      }
 
       klass = reflection.class_name.safe_constantize
       klass.attr_readonly cache_column if klass && klass.respond_to?(:attr_readonly)
     end
 
-    def add_touch_callbacks(reflection)
-      mixin.class_eval <<-CODE, __FILE__, __LINE__ + 1
-        def belongs_to_touch_after_save_or_destroy_for_#{name}
-          foreign_key_field = #{reflection.foreign_key.inspect}
-          old_foreign_id    = attribute_was(foreign_key_field)
-
-          if old_foreign_id
-            klass      = association(#{name.inspect}).klass
-            old_record = klass.find_by(klass.primary_key => old_foreign_id)
+    def self.touch_record(o, foreign_key, name, touch) # :nodoc:
+      old_foreign_id = o.changed_attributes[foreign_key]
 
-            if old_record
-              old_record.touch #{options[:touch].inspect if options[:touch] != true}
-            end
-          end
+      if old_foreign_id
+        association = o.association(name)
+        reflection = association.reflection
+        if reflection.polymorphic?
+          klass = o.public_send("#{reflection.foreign_type}_was").constantize
+        else
+          klass = association.klass
+        end
+        old_record = klass.find_by(klass.primary_key => old_foreign_id)
 
-          record = #{name}
-          unless record.nil? || record.new_record?
-            record.touch #{options[:touch].inspect if options[:touch] != true}
+        if old_record
+          if touch != true
+            old_record.touch touch
+          else
+            old_record.touch
           end
         end
-      CODE
-
-      model.after_save    "belongs_to_touch_after_save_or_destroy_for_#{name}"
-      model.after_touch   "belongs_to_touch_after_save_or_destroy_for_#{name}"
-      model.after_destroy "belongs_to_touch_after_save_or_destroy_for_#{name}"
+      end
+
+      record = o.send name
+      if record && record.persisted?
+        if touch != true
+          record.touch touch
+        else
+          record.touch
+        end
+      end
     end
 
-    def valid_dependent_options
-      [:destroy, :delete]
+    def self.add_touch_callbacks(model, reflection)
+      foreign_key = reflection.foreign_key
+      n           = reflection.name
+      touch       = reflection.options[:touch]
+
+      callback = lambda { |record|
+        BelongsTo.touch_record(record, foreign_key, n, touch)
+      }
+
+      model.after_save    callback
+      model.after_touch   callback
+      model.after_destroy callback
     end
   end
 end
diff --git a/activerecord/lib/active_record/associations/builder/collection_association.rb b/activerecord/lib/active_record/associations/builder/collection_association.rb
index 9c6690b721..bc15a49996 100644
--- a/activerecord/lib/active_record/associations/builder/collection_association.rb
+++ b/activerecord/lib/active_record/associations/builder/collection_association.rb
@@ -1,4 +1,4 @@
-# This class is inherited by the has_many and has_many_and_belongs_to_many association classes 
+# This class is inherited by the has_many and has_many_and_belongs_to_many association classes
 
 require 'active_record/associations'
 
@@ -8,69 +8,57 @@ module ActiveRecord::Associations::Builder
     CALLBACKS = [:before_add, :after_add, :before_remove, :after_remove]
 
     def valid_options
-      super + [:table_name, :finder_sql, :counter_sql, :before_add,
+      super + [:table_name, :before_add,
                :after_add, :before_remove, :after_remove, :extend]
     end
 
-    attr_reader :block_extension, :extension_module
+    attr_reader :block_extension
 
-    def initialize(*args, &extension)
-      super(*args)
-      @block_extension = extension
-    end
-
-    def build
-      show_deprecation_warnings
-      wrap_block_extension
-      reflection = super
-      CALLBACKS.each { |callback_name| define_callback(callback_name) }
-      reflection
+    def initialize(model, name, scope, options)
+      super
+      @mod = nil
+      if block_given?
+        @mod = Module.new(&Proc.new)
+        @scope = wrap_scope @scope, @mod
+      end
     end
 
-    def writable?
-      true
+    def self.define_callbacks(model, reflection)
+      super
+      name    = reflection.name
+      options = reflection.options
+      CALLBACKS.each { |callback_name|
+        define_callback(model, callback_name, name, options)
+      }
     end
 
-    def show_deprecation_warnings
-      [:finder_sql, :counter_sql].each do |name|
-        if options.include? name
-          ActiveSupport::Deprecation.warn("The :#{name} association option is deprecated. Please find an alternative (such as using scopes).")
-        end
+    def define_extensions(model)
+      if @mod
+        extension_module_name = "#{model.name.demodulize}#{name.to_s.camelize}AssociationExtension"
+        model.parent.const_set(extension_module_name, @mod)
       end
     end
 
-    def wrap_block_extension
-      if block_extension
-        @extension_module = mod = Module.new(&block_extension)
-        silence_warnings do
-          model.parent.const_set(extension_module_name, mod)
-        end
-
-        prev_scope = @scope
+    def self.define_callback(model, callback_name, name, options)
+      full_callback_name = "#{callback_name}_for_#{name}"
 
-        if prev_scope
-          @scope = proc { |owner| instance_exec(owner, &prev_scope).extending(mod) }
+      # TODO : why do i need method_defined? I think its because of the inheritance chain
+      model.class_attribute full_callback_name unless model.method_defined?(full_callback_name)
+      callbacks = Array(options[callback_name.to_sym]).map do |callback|
+        case callback
+        when Symbol
+          ->(method, owner, record) { owner.send(callback, record) }
+        when Proc
+          ->(method, owner, record) { callback.call(owner, record) }
         else
-          @scope = proc { extending(mod) }
+          ->(method, owner, record) { callback.send(method, owner, record) }
         end
       end
-    end
-
-    def extension_module_name
-      @extension_module_name ||= "#{model.name.demodulize}#{name.to_s.camelize}AssociationExtension"
-    end
-
-    def define_callback(callback_name)
-      full_callback_name = "#{callback_name}_for_#{name}"
-
-      # TODO : why do i need method_defined? I think its because of the inheritance chain
-      model.class_attribute full_callback_name.to_sym unless model.method_defined?(full_callback_name)
-      model.send("#{full_callback_name}=", Array(options[callback_name.to_sym]))
+      model.send "#{full_callback_name}=", callbacks
     end
 
     # Defines the setter and getter methods for the collection_singular_ids.
-
-    def define_readers
+    def self.define_readers(mixin, name)
       super
 
       mixin.class_eval <<-CODE, __FILE__, __LINE__ + 1
@@ -80,7 +68,7 @@ module ActiveRecord::Associations::Builder
       CODE
     end
 
-    def define_writers
+    def self.define_writers(mixin, name)
       super
 
       mixin.class_eval <<-CODE, __FILE__, __LINE__ + 1
@@ -89,5 +77,15 @@ module ActiveRecord::Associations::Builder
         end
       CODE
     end
+
+    private
+
+    def wrap_scope(scope, mod)
+      if scope
+        proc { |owner| instance_exec(owner, &scope).extending(mod) }
+      else
+        proc { extending(mod) }
+      end
+    end
   end
 end
diff --git a/activerecord/lib/active_record/associations/builder/has_and_belongs_to_many.rb b/activerecord/lib/active_record/associations/builder/has_and_belongs_to_many.rb
index bdac02b5bf..e472277374 100644
--- a/activerecord/lib/active_record/associations/builder/has_and_belongs_to_many.rb
+++ b/activerecord/lib/active_record/associations/builder/has_and_belongs_to_many.rb
@@ -1,39 +1,121 @@
 module ActiveRecord::Associations::Builder
-  class HasAndBelongsToMany < CollectionAssociation #:nodoc:
-    def macro
-      :has_and_belongs_to_many
-    end
+  class HasAndBelongsToMany # :nodoc:
+    class JoinTableResolver
+      KnownTable = Struct.new :join_table
+
+      class KnownClass
+        def initialize(lhs_class, rhs_class_name)
+          @lhs_class      = lhs_class
+          @rhs_class_name = rhs_class_name
+          @join_table     = nil
+        end
+
+        def join_table
+          @join_table ||= [@lhs_class.table_name, klass.table_name].sort.join("\0").gsub(/^(.*_)(.+)\0\1(.+)/, '\1\2_\3').gsub("\0", "_")
+        end
+
+        private
+        def klass; @rhs_class_name.constantize; end
+      end
 
-    def valid_options
-      super + [:join_table, :association_foreign_key, :delete_sql, :insert_sql]
+      def self.build(lhs_class, name, options)
+        if options[:join_table]
+          KnownTable.new options[:join_table].to_s
+        else
+          class_name = options.fetch(:class_name) {
+            name.to_s.camelize.singularize
+          }
+          KnownClass.new lhs_class, class_name
+        end
+      end
     end
 
-    def build
-      reflection = super
-      define_destroy_hook
-      reflection
+    attr_reader :lhs_model, :association_name, :options
+
+    def initialize(association_name, lhs_model, options)
+      @association_name = association_name
+      @lhs_model = lhs_model
+      @options = options
     end
 
-    def show_deprecation_warnings
-      super
+    def through_model
+      habtm = JoinTableResolver.build lhs_model, association_name, options
+
+      join_model = Class.new(ActiveRecord::Base) {
+        class << self;
+          attr_accessor :class_resolver
+          attr_accessor :name
+          attr_accessor :table_name_resolver
+          attr_accessor :left_reflection
+          attr_accessor :right_reflection
+        end
+
+        def self.table_name
+          table_name_resolver.join_table
+        end
+
+        def self.compute_type(class_name)
+          class_resolver.compute_type class_name
+        end
 
-      [:delete_sql, :insert_sql].each do |name|
-        if options.include? name
-          ActiveSupport::Deprecation.warn("The :#{name} association option is deprecated. Please find an alternative (such as using has_many :through).")
+        def self.add_left_association(name, options)
+          belongs_to name, options
+          self.left_reflection = reflect_on_association(name)
         end
+
+        def self.add_right_association(name, options)
+          rhs_name = name.to_s.singularize.to_sym
+          belongs_to rhs_name, options
+          self.right_reflection = reflect_on_association(rhs_name)
+        end
+
+      }
+
+      join_model.name                = "HABTM_#{association_name.to_s.camelize}"
+      join_model.table_name_resolver = habtm
+      join_model.class_resolver      = lhs_model
+
+      join_model.add_left_association :left_side, class: lhs_model
+      join_model.add_right_association association_name, belongs_to_options(options)
+      join_model
+    end
+
+    def middle_reflection(join_model)
+      middle_name = [lhs_model.name.downcase.pluralize,
+                     association_name].join('_').gsub(/::/, '_').to_sym
+      middle_options = middle_options join_model
+      hm_builder = HasMany.create_builder(lhs_model,
+                                          middle_name,
+                                          nil,
+                                          middle_options)
+      hm_builder.build lhs_model
+    end
+
+    private
+
+    def middle_options(join_model)
+      middle_options = {}
+      middle_options[:class] = join_model
+      middle_options[:source] = join_model.left_reflection.name
+      if options.key? :foreign_key
+        middle_options[:foreign_key] = options[:foreign_key]
       end
+      middle_options
     end
 
-    def define_destroy_hook
-      name = self.name
-      model.send(:include, Module.new {
-        class_eval <<-RUBY, __FILE__, __LINE__ + 1
-          def destroy_associations
-            association(:#{name}).delete_all
-            super
-          end
-        RUBY
-      })
+    def belongs_to_options(options)
+      rhs_options = {}
+
+      if options.key? :class_name
+        rhs_options[:foreign_key] = options[:class_name].foreign_key
+        rhs_options[:class_name] = options[:class_name]
+      end
+
+      if options.key? :association_foreign_key
+        rhs_options[:foreign_key] = options[:association_foreign_key]
+      end
+
+      rhs_options
     end
   end
 end
diff --git a/activerecord/lib/active_record/associations/builder/has_many.rb b/activerecord/lib/active_record/associations/builder/has_many.rb
index 429def5455..7909b93622 100644
--- a/activerecord/lib/active_record/associations/builder/has_many.rb
+++ b/activerecord/lib/active_record/associations/builder/has_many.rb
@@ -5,11 +5,11 @@ module ActiveRecord::Associations::Builder
     end
 
     def valid_options
-      super + [:primary_key, :dependent, :as, :through, :source, :source_type, :inverse_of, :automatic_inverse_of, :counter_cache]
+      super + [:primary_key, :dependent, :as, :through, :source, :source_type, :inverse_of, :counter_cache]
     end
 
-    def valid_dependent_options
-      [:destroy, :delete_all, :nullify, :restrict, :restrict_with_error, :restrict_with_exception]
+    def self.valid_dependent_options
+      [:destroy, :delete_all, :nullify, :restrict_with_error, :restrict_with_exception]
     end
   end
 end
diff --git a/activerecord/lib/active_record/associations/builder/has_one.rb b/activerecord/lib/active_record/associations/builder/has_one.rb
index 0da564f402..f359efd496 100644
--- a/activerecord/lib/active_record/associations/builder/has_one.rb
+++ b/activerecord/lib/active_record/associations/builder/has_one.rb
@@ -10,16 +10,14 @@ module ActiveRecord::Associations::Builder
       valid
     end
 
-    def constructable?
-      !options[:through]
+    def self.valid_dependent_options
+      [:destroy, :delete, :nullify, :restrict_with_error, :restrict_with_exception]
     end
 
-    def configure_dependency
-      super unless options[:through]
-    end
+    private
 
-    def valid_dependent_options
-      [:destroy, :delete, :nullify, :restrict, :restrict_with_error, :restrict_with_exception]
+    def self.add_before_destroy_callbacks(model, reflection)
+      super unless reflection.options[:through]
     end
   end
 end
diff --git a/activerecord/lib/active_record/associations/builder/singular_association.rb b/activerecord/lib/active_record/associations/builder/singular_association.rb
index 96ccbeb8a3..e655c389a6 100644
--- a/activerecord/lib/active_record/associations/builder/singular_association.rb
+++ b/activerecord/lib/active_record/associations/builder/singular_association.rb
@@ -1,23 +1,18 @@
-# This class is inherited by the has_one and belongs_to association classes 
+# This class is inherited by the has_one and belongs_to association classes
 
 module ActiveRecord::Associations::Builder
   class SingularAssociation < Association #:nodoc:
     def valid_options
-      super + [:remote, :dependent, :counter_cache, :primary_key, :inverse_of, :automatic_inverse_of]
+      super + [:remote, :dependent, :primary_key, :inverse_of]
     end
 
-    def constructable?
-      true
-    end
-
-    def define_accessors
+    def self.define_accessors(model, reflection)
       super
-      define_constructors if constructable?
+      define_constructors(model.generated_association_methods, reflection.name) if reflection.constructable?
     end
 
     # Defines the (build|create)_association methods for belongs_to or has_one association
-    
-    def define_constructors
+    def self.define_constructors(mixin, name)
       mixin.class_eval <<-CODE, __FILE__, __LINE__ + 1
         def build_#{name}(*args, &block)
           association(:#{name}).build(*args, &block)
diff --git a/activerecord/lib/active_record/associations/collection_association.rb b/activerecord/lib/active_record/associations/collection_association.rb
index efd7ecb97c..52531a3520 100644
--- a/activerecord/lib/active_record/associations/collection_association.rb
+++ b/activerecord/lib/active_record/associations/collection_association.rb
@@ -4,10 +4,9 @@ module ActiveRecord
     #
     # CollectionAssociation is an abstract class that provides common stuff to
     # ease the implementation of association proxies that represent
-    # collections. See the class hierarchy in AssociationProxy.
+    # collections. See the class hierarchy in Association.
     #
     #   CollectionAssociation:
-    #     HasAndBelongsToManyAssociation => has_and_belongs_to_many
     #     HasManyAssociation => has_many
     #       HasManyThroughAssociation + ThroughAssociation => has_many :through
     #
@@ -34,7 +33,7 @@ module ActiveRecord
           reload
         end
 
-        @proxy ||= CollectionProxy.new(klass, self)
+        @proxy ||= CollectionProxy.create(klass, self)
       end
 
       # Implements the writer method, e.g. foo.items= for Foo.has_many :items
@@ -44,7 +43,7 @@ module ActiveRecord
 
       # Implements the ids reader method, e.g. foo.item_ids for Foo.has_many :items
       def ids_reader
-        if loaded? || options[:finder_sql]
+        if loaded?
           load_target.map do |record|
             record.send(reflection.association_primary_key)
           end
@@ -67,11 +66,11 @@ module ActiveRecord
         @target = []
       end
 
-      def select(select = nil)
+      def select(*fields)
         if block_given?
           load_target.select.each { |e| yield e }
         else
-          scope.select(select)
+          scope.select(*fields)
         end
       end
 
@@ -79,17 +78,14 @@ module ActiveRecord
         if block_given?
           load_target.find(*args) { |*block_args| yield(*block_args) }
         else
-          if options[:finder_sql]
-            find_by_scan(*args)
-          elsif options[:inverse_of]
-            args = args.flatten
-            raise RecordNotFound, "Couldn't find #{scope.klass.name} without an ID" if args.blank?
-
+          if options[:inverse_of] && loaded?
+            args_flatten = args.flatten
+            raise RecordNotFound, "Couldn't find #{scope.klass.name} without an ID" if args_flatten.blank?
             result = find_by_scan(*args)
 
             result_size = Array(result).size
-            if !result || result_size != args.size
-              scope.raise_record_not_found_exception!(args, result_size, args.size)
+            if !result || result_size != args_flatten.size
+              scope.raise_record_not_found_exception!(args_flatten, result_size, args_flatten.size)
             else
               result
             end
@@ -153,11 +149,33 @@ module ActiveRecord
         end
       end
 
-      # Remove all records from this association.
+      # Removes all records from the association without calling callbacks
+      # on the associated records. It honors the `:dependent` option. However
+      # if the `:dependent` value is `:destroy` then in that case the `:delete_all`
+      # deletion strategy for the association is applied.
+      #
+      # You can force a particular deletion strategy by passing a parameter.
+      #
+      # Example:
+      #
+      # @author.books.delete_all(:nullify)
+      # @author.books.delete_all(:delete_all)
       #
       # See delete for more info.
-      def delete_all
-        delete(:all).tap do
+      def delete_all(dependent = nil)
+        if dependent.present? && ![:nullify, :delete_all].include?(dependent)
+          raise ArgumentError, "Valid values are :nullify or :delete_all"
+        end
+
+        dependent = if dependent.present?
+                      dependent
+                    elsif options[:dependent] == :destroy
+                      :delete_all
+                    else
+                      options[:dependent]
+                    end
+
+        delete(:all, dependent: dependent).tap do
           reset
           loaded!
         end
@@ -173,36 +191,29 @@ module ActiveRecord
         end
       end
 
-      # Count all records using SQL. If the +:counter_sql+ or +:finder_sql+ option is set for the
-      # association, it will be used for the query. Otherwise, construct options and pass them with
+      # Count all records using SQL.  Construct options and pass them with
       # scope to the target class's +count+.
       def count(column_name = nil, count_options = {})
+        # TODO: Remove count_options argument as soon we remove support to
+        # activerecord-deprecated_finders.
         column_name, count_options = nil, column_name if column_name.is_a?(Hash)
 
-        if options[:counter_sql] || options[:finder_sql]
-          unless count_options.blank?
-            raise ArgumentError, "If finder_sql/counter_sql is used then options cannot be passed"
-          end
-
-          reflection.klass.count_by_sql(custom_counter_sql)
-        else
-          relation = scope
-          if association_scope.distinct_value
-            # This is needed because 'SELECT count(DISTINCT *)..' is not valid SQL.
-            column_name ||= reflection.klass.primary_key
-            relation = relation.distinct
-          end
+        relation = scope
+        if association_scope.distinct_value
+          # This is needed because 'SELECT count(DISTINCT *)..' is not valid SQL.
+          column_name ||= reflection.klass.primary_key
+          relation = relation.distinct
+        end
 
-          value = relation.count(column_name)
+        value = relation.count(column_name)
 
-          limit  = options[:limit]
-          offset = options[:offset]
+        limit  = options[:limit]
+        offset = options[:offset]
 
-          if limit || offset
-            [ [value - offset.to_i, 0].max, limit.to_i ].min
-          else
-            value
-          end
+        if limit || offset
+          [ [value - offset.to_i, 0].max, limit.to_i ].min
+        else
+          value
         end
       end
 
@@ -214,18 +225,10 @@ module ActiveRecord
       # are actually removed from the database, that depends precisely on
       # +delete_records+. They are in any case removed from the collection.
       def delete(*records)
-        dependent = options[:dependent]
+        _options = records.extract_options!
+        dependent = _options[:dependent] || options[:dependent]
 
         if records.first == :all
-
-          if dependent && dependent == :destroy
-            message = 'In Rails 4.1 delete_all on associations would not fire callbacks. ' \
-                      'It means if the :dependent option is :destroy then the associated ' \
-                      'records would be deleted without loading and invoking callbacks.'
-
-            ActiveRecord::Base.logger ? ActiveRecord::Base.logger.warn(message) : $stderr.puts(message)
-          end
-
           if loaded? || dependent == :destroy
             delete_or_destroy(load_target, dependent)
           else
@@ -285,14 +288,14 @@ module ActiveRecord
 
       # Returns true if the collection is empty.
       #
-      # If the collection has been loaded or the <tt>:counter_sql</tt> option
-      # is provided, it is equivalent to <tt>collection.size.zero?</tt>. If the
+      # If the collection has been loaded
+      # it is equivalent to <tt>collection.size.zero?</tt>. If the
       # collection has not been loaded, it is equivalent to
       # <tt>collection.exists?</tt>. If the collection has not already been
       # loaded and you are going to fetch the records anyway it is better to
       # check <tt>collection.length.zero?</tt>.
       def empty?
-        if loaded? || options[:counter_sql]
+        if loaded?
           size.zero?
         else
           @target.blank? && !scope.exists?
@@ -345,7 +348,6 @@ module ActiveRecord
           if record.new_record?
             include_in_memory?(record)
           else
-            load_target if options[:finder_sql]
             loaded? ? target.include?(record) : scope.exists?(record)
           end
         else
@@ -362,8 +364,8 @@ module ActiveRecord
         target
       end
 
-      def add_to_target(record)
-        callback(:before_add, record)
+      def add_to_target(record, skip_callbacks = false)
+        callback(:before_add, record) unless skip_callbacks
         yield(record) if block_given?
 
         if association_scope.distinct_value && index = @target.index(record)
@@ -372,7 +374,7 @@ module ActiveRecord
           @target << record
         end
 
-        callback(:after_add, record)
+        callback(:after_add, record) unless skip_callbacks
         set_inverse_instance(record)
 
         record
@@ -390,31 +392,8 @@ module ActiveRecord
 
       private
 
-        def custom_counter_sql
-          if options[:counter_sql]
-            interpolate(options[:counter_sql])
-          else
-            # replace the SELECT clause with COUNT(SELECTS), preserving any hints within /* ... */
-            interpolate(options[:finder_sql]).sub(/SELECT\b(\/\*.*?\*\/ )?(.*)\bFROM\b/im) do
-              count_with = $2.to_s
-              count_with = '*' if count_with.blank? || count_with =~ /,/ || count_with =~ /\.\*/
-              "SELECT #{$1}COUNT(#{count_with}) FROM"
-            end
-          end
-        end
-
-        def custom_finder_sql
-          interpolate(options[:finder_sql])
-        end
-
         def find_target
-          records =
-            if options[:finder_sql]
-              reflection.klass.find_by_sql(custom_finder_sql)
-            else
-              scope.to_a
-            end
-
+          records = scope.to_a
           records.each { |record| set_inverse_instance(record) }
           records
         end
@@ -529,20 +508,13 @@ module ActiveRecord
 
         def callback(method, record)
           callbacks_for(method).each do |callback|
-            case callback
-            when Symbol
-              owner.send(callback, record)
-            when Proc
-              callback.call(owner, record)
-            else
-              callback.send(method, owner, record)
-            end
+            callback.call(method, owner, record)
           end
         end
 
         def callbacks_for(callback_name)
           full_callback_name = "#{callback_name}_for_#{reflection.name}"
-          owner.class.send(full_callback_name.to_sym) || []
+          owner.class.send(full_callback_name)
         end
 
         # Should we deal with assoc.first or assoc.last by issuing an independent query to
@@ -553,24 +525,21 @@ module ActiveRecord
         # Otherwise, go to the database only if none of the following are true:
         #   * target already loaded
         #   * owner is new record
-        #   * custom :finder_sql exists
         #   * target contains new or changed record(s)
-        #   * the first arg is an integer (which indicates the number of records to be returned)
         def fetch_first_or_last_using_find?(args)
           if args.first.is_a?(Hash)
             true
           else
             !(loaded? ||
               owner.new_record? ||
-              options[:finder_sql] ||
-              target.any? { |record| record.new_record? || record.changed? } ||
-              args.first.kind_of?(Integer))
+              target.any? { |record| record.new_record? || record.changed? })
           end
         end
 
         def include_in_memory?(record)
           if reflection.is_a?(ActiveRecord::Reflection::ThroughReflection)
-            owner.send(reflection.through_reflection.name).any? { |source|
+            assoc = owner.association(reflection.through_reflection.name)
+            assoc.reader.any? { |source|
               target = source.send(reflection.source_reflection.name)
               target.respond_to?(:include?) ? target.include?(record) : target == record
             } || target.include?(record)
@@ -579,18 +548,18 @@ module ActiveRecord
           end
         end
 
-        # If using a custom finder_sql or if the :inverse_of option has been
+        # If the :inverse_of option has been
         # specified, then #find scans the entire collection.
         def find_by_scan(*args)
           expects_array = args.first.kind_of?(Array)
-          ids           = args.flatten.compact.map{ |arg| arg.to_i }.uniq
+          ids           = args.flatten.compact.map{ |arg| arg.to_s }.uniq
 
           if ids.size == 1
             id = ids.first
-            record = load_target.detect { |r| id == r.id }
+            record = load_target.detect { |r| id == r.id.to_s }
             expects_array ? [ record ] : record
           else
-            load_target.select { |r| ids.include?(r.id) }
+            load_target.select { |r| ids.include?(r.id.to_s) }
           end
         end
 
diff --git a/activerecord/lib/active_record/associations/collection_proxy.rb b/activerecord/lib/active_record/associations/collection_proxy.rb
index e82c195335..e3fc908444 100644
--- a/activerecord/lib/active_record/associations/collection_proxy.rb
+++ b/activerecord/lib/active_record/associations/collection_proxy.rb
@@ -33,7 +33,6 @@ module ActiveRecord
       def initialize(klass, association) #:nodoc:
         @association = association
         super klass, klass.arel_table
-        self.default_scoped = true
         merge! association.scope(nullify: false)
       end
 
@@ -76,7 +75,7 @@ module ActiveRecord
       #   #      #<Pet id: nil, name: "Choo-Choo">
       #   #    ]
       #
-      #   person.pets.select([:id, :name])
+      #   person.pets.select(:id, :name )
       #   # => [
       #   #      #<Pet id: 1, name: "Fancy-Fancy">,
       #   #      #<Pet id: 2, name: "Spook">,
@@ -107,13 +106,13 @@ module ActiveRecord
       #   #      #<Pet id: 2, name: "Spook">,
       #   #      #<Pet id: 3, name: "Choo-Choo">
       #   #    ]
-      def select(select = nil, &block)
-        @association.select(select, &block)
+      def select(*fields, &block)
+        @association.select(*fields, &block)
       end
 
       # Finds an object in the collection responding to the +id+. Uses the same
       # rules as <tt>ActiveRecord::Base.find</tt>. Returns <tt>ActiveRecord::RecordNotFound</tt>
-      # error if the object can not be found.
+      # error if the object cannot be found.
       #
       #   class Person < ActiveRecord::Base
       #     has_many :pets
@@ -282,7 +281,7 @@ module ActiveRecord
       # so method calls may be chained.
       #
       #   class Person < ActiveRecord::Base
-      #     pets :has_many
+      #     has_many :pets
       #   end
       #
       #   person.pets.size # => 0
@@ -418,8 +417,8 @@ module ActiveRecord
       #
       #   Pet.find(1, 2, 3)
       #   # => ActiveRecord::RecordNotFound
-      def delete_all
-        @association.delete_all
+      def delete_all(dependent = nil)
+        @association.delete_all(dependent)
       end
 
       # Deletes the records of the collection directly from the database
@@ -671,6 +670,8 @@ module ActiveRecord
       #   #       #<Pet id: 3, name: "Choo-Choo", person_id: 1>
       #   #    ]
       def count(column_name = nil, options = {})
+        # TODO: Remove options argument as soon we remove support to
+        # activerecord-deprecated_finders.
         @association.count(column_name, options)
       end
 
@@ -727,7 +728,7 @@ module ActiveRecord
       end
 
       # Returns +true+ if the collection is empty. If the collection has been
-      # loaded or the <tt>:counter_sql</tt> option is provided, it is equivalent
+      # loaded it is equivalent
       # to <tt>collection.size.zero?</tt>. If the collection has not been loaded,
       # it is equivalent to <tt>collection.exists?</tt>. If the collection has
       # not already been loaded and you are going to fetch the records anyway it
@@ -788,12 +789,12 @@ module ActiveRecord
       #     has_many :pets
       #   end
       #
-      #   person.pets.count #=> 1
-      #   person.pets.many? #=> false
+      #   person.pets.count # => 1
+      #   person.pets.many? # => false
       #
       #   person.pets << Pet.new(name: 'Snoopy')
-      #   person.pets.count #=> 2
-      #   person.pets.many? #=> true
+      #   person.pets.count # => 2
+      #   person.pets.many? # => true
       #
       # You can also pass a block to define criteria. The
       # behavior is the same, it returns true if the collection
@@ -830,7 +831,7 @@ module ActiveRecord
       #   person.pets.include?(Pet.find(20)) # => true
       #   person.pets.include?(Pet.find(21)) # => false
       def include?(record)
-        @association.include?(record)
+        !!@association.include?(record)
       end
 
       def proxy_association
@@ -849,8 +850,6 @@ module ActiveRecord
       def scope
         @association.scope
       end
-
-      # :nodoc:
       alias spawn scope
 
       # Equivalent to <tt>Array#==</tt>. Returns +true+ if the two arrays
diff --git a/activerecord/lib/active_record/associations/has_and_belongs_to_many_association.rb b/activerecord/lib/active_record/associations/has_and_belongs_to_many_association.rb
deleted file mode 100644
index bb3e3db379..0000000000
--- a/activerecord/lib/active_record/associations/has_and_belongs_to_many_association.rb
+++ /dev/null
@@ -1,65 +0,0 @@
-module ActiveRecord
-  # = Active Record Has And Belongs To Many Association
-  module Associations
-    class HasAndBelongsToManyAssociation < CollectionAssociation #:nodoc:
-      attr_reader :join_table
-
-      def initialize(owner, reflection)
-        @join_table = Arel::Table.new(reflection.join_table)
-        super
-      end
-
-      def insert_record(record, validate = true, raise = false)
-        if record.new_record?
-          if raise
-            record.save!(:validate => validate)
-          else
-            return unless record.save(:validate => validate)
-          end
-        end
-
-        if options[:insert_sql]
-          owner.connection.insert(interpolate(options[:insert_sql], record))
-        else
-          stmt = join_table.compile_insert(
-            join_table[reflection.foreign_key]             => owner.id,
-            join_table[reflection.association_foreign_key] => record.id
-          )
-
-          owner.class.connection.insert stmt
-        end
-
-        record
-      end
-
-      private
-
-        def count_records
-          load_target.size
-        end
-
-        def delete_records(records, method)
-          if sql = options[:delete_sql]
-            records = load_target if records == :all
-            records.each { |record| owner.class.connection.delete(interpolate(sql, record)) }
-          else
-            relation  = join_table
-            condition = relation[reflection.foreign_key].eq(owner.id)
-
-            unless records == :all
-              condition = condition.and(
-                relation[reflection.association_foreign_key]
-                  .in(records.map { |x| x.id }.compact)
-              )
-            end
-
-            owner.class.connection.delete(relation.where(condition).compile_delete)
-          end
-        end
-
-        def invertible_for?(record)
-          false
-        end
-    end
-  end
-end
diff --git a/activerecord/lib/active_record/associations/has_many_association.rb b/activerecord/lib/active_record/associations/has_many_association.rb
index 29fae809da..72e0891702 100644
--- a/activerecord/lib/active_record/associations/has_many_association.rb
+++ b/activerecord/lib/active_record/associations/has_many_association.rb
@@ -9,7 +9,7 @@ module ActiveRecord
 
       def handle_dependency
         case options[:dependent]
-        when :restrict, :restrict_with_exception
+        when :restrict_with_exception
           raise ActiveRecord::DeleteRestrictionError.new(reflection.name) unless empty?
 
         when :restrict_with_error
@@ -32,6 +32,7 @@ module ActiveRecord
 
       def insert_record(record, validate = true, raise = false)
         set_owner_attributes(record)
+        set_inverse_instance(record)
 
         if raise
           record.save!(:validate => validate)
@@ -58,8 +59,6 @@ module ActiveRecord
         def count_records
           count = if has_cached_counter?
             owner.send(:read_attribute, cached_counter_attribute_name)
-          elsif options[:counter_sql] || options[:finder_sql]
-            reflection.klass.count_by_sql(custom_counter_sql)
           else
             scope.count
           end
@@ -109,14 +108,13 @@ module ActiveRecord
         # Deletes the records according to the <tt>:dependent</tt> option.
         def delete_records(records, method)
           if method == :destroy
-            records.each { |r| r.destroy }
+            records.each(&:destroy!)
             update_counter(-records.length) unless inverse_updates_counter_cache?
           else
             if records == :all
               scope = self.scope
             else
-              keys  = records.map { |r| r[reflection.association_primary_key] }
-              scope = self.scope.where(reflection.association_primary_key => keys)
+              scope = self.scope.where(reflection.klass.primary_key => records)
             end
 
             if method == :delete_all
@@ -128,7 +126,11 @@ module ActiveRecord
         end
 
         def foreign_key_present?
-          owner.attribute_present?(reflection.association_primary_key)
+          if reflection.klass.primary_key
+            owner.attribute_present?(reflection.association_primary_key)
+          else
+            false
+          end
         end
     end
   end
diff --git a/activerecord/lib/active_record/associations/has_many_through_association.rb b/activerecord/lib/active_record/associations/has_many_through_association.rb
index a74dd1cdab..31b8d27892 100644
--- a/activerecord/lib/active_record/associations/has_many_through_association.rb
+++ b/activerecord/lib/active_record/associations/has_many_through_association.rb
@@ -140,7 +140,21 @@ module ActiveRecord
 
           case method
           when :destroy
-            count = scope.destroy_all.length
+            if scope.klass.primary_key
+              count = scope.destroy_all.length
+            else
+              scope.to_a.each do |record|
+                record.run_callbacks :destroy
+              end
+
+              arel = scope.arel
+
+              stmt = Arel::DeleteManager.new arel.engine
+              stmt.from scope.klass.arel_table
+              stmt.wheres = arel.constraints
+
+              count = scope.klass.connection.delete(stmt, 'SQL', scope.bind_values)
+            end
           when :nullify
             count = scope.update_all(source_reflection.foreign_key => nil)
           else
@@ -149,7 +163,7 @@ module ActiveRecord
 
           delete_through_records(records)
 
-          if source_reflection.options[:counter_cache]
+          if source_reflection.options[:counter_cache] && method != :destroy
             counter = source_reflection.counter_cache_column
             klass.decrement_counter counter, records.map(&:id)
           end
diff --git a/activerecord/lib/active_record/associations/has_one_association.rb b/activerecord/lib/active_record/associations/has_one_association.rb
index 920038a543..944caacab6 100644
--- a/activerecord/lib/active_record/associations/has_one_association.rb
+++ b/activerecord/lib/active_record/associations/has_one_association.rb
@@ -6,7 +6,7 @@ module ActiveRecord
 
       def handle_dependency
         case options[:dependent]
-        when :restrict, :restrict_with_exception
+        when :restrict_with_exception
           raise ActiveRecord::DeleteRestrictionError.new(reflection.name) if load_target
 
         when :restrict_with_error
@@ -26,15 +26,19 @@ module ActiveRecord
         load_target
 
         return self.target if !(target || record)
-        if (target != record) || record.changed?
+
+        assigning_another_record = target != record
+        if assigning_another_record || record.changed?
+          save &&= owner.persisted?
+
           transaction_if(save) do
-            remove_target!(options[:dependent]) if target && !target.destroyed?
+            remove_target!(options[:dependent]) if target && !target.destroyed? && assigning_another_record
 
             if record
               set_owner_attributes(record)
               set_inverse_instance(record)
 
-              if owner.persisted? && save && !record.save
+              if save && !record.save
                 nullify_owner_attributes(record)
                 set_owner_attributes(target) if target
                 raise RecordNotSaved, "Failed to save the new associated #{reflection.name}."
diff --git a/activerecord/lib/active_record/associations/join_dependency.rb b/activerecord/lib/active_record/associations/join_dependency.rb
index 5b2f2d1902..295dccf34e 100644
--- a/activerecord/lib/active_record/associations/join_dependency.rb
+++ b/activerecord/lib/active_record/associations/join_dependency.rb
@@ -1,15 +1,79 @@
 module ActiveRecord
   module Associations
     class JoinDependency # :nodoc:
-      autoload :JoinPart,        'active_record/associations/join_dependency/join_part'
       autoload :JoinBase,        'active_record/associations/join_dependency/join_base'
       autoload :JoinAssociation, 'active_record/associations/join_dependency/join_association'
 
-      attr_reader :join_parts, :reflections, :alias_tracker, :base_klass
+      class Aliases # :nodoc:
+        def initialize(tables)
+          @tables = tables
+          @alias_cache = tables.each_with_object({}) { |table,h|
+            h[table.node] = table.columns.each_with_object({}) { |column,i|
+              i[column.name] = column.alias
+            }
+          }
+          @name_and_alias_cache = tables.each_with_object({}) { |table,h|
+            h[table.node] = table.columns.map { |column|
+              [column.name, column.alias]
+            }
+          }
+        end
+
+        def columns
+          @tables.flat_map { |t| t.column_aliases }
+        end
+
+        # An array of [column_name, alias] pairs for the table
+        def column_aliases(node)
+          @name_and_alias_cache[node]
+        end
+
+        def column_alias(node, column)
+          @alias_cache[node][column]
+        end
+
+        class Table < Struct.new(:node, :columns)
+          def table
+            Arel::Nodes::TableAlias.new node.table, node.aliased_table_name
+          end
+
+          def column_aliases
+            t = table
+            columns.map { |column| t[column.name].as Arel.sql column.alias }
+          end
+        end
+        Column = Struct.new(:name, :alias)
+      end
+
+      attr_reader :alias_tracker, :base_klass, :join_root
+
+      def self.make_tree(associations)
+        hash = {}
+        walk_tree associations, hash
+        hash
+      end
+
+      def self.walk_tree(associations, hash)
+        case associations
+        when Symbol, String
+          hash[associations.to_sym] ||= {}
+        when Array
+          associations.each do |assoc|
+            walk_tree assoc, hash
+          end
+        when Hash
+          associations.each do |k,v|
+            cache = hash[k] ||= {}
+            walk_tree v, cache
+          end
+        else
+          raise ConfigurationError, associations.inspect
+        end
+      end
 
       # base is the base class on which operation is taking place.
       # associations is the list of associations which are joined using hash, symbol or array.
-      # joins is the list of all string join commnads and arel nodes.
+      # joins is the list of all string join commands and arel nodes.
       #
       #  Example :
       #
@@ -19,223 +83,191 @@ module ActiveRecord
       #  end
       #
       #  If I execute `@physician.patients.to_a` then
-      #    base #=> Physician
-      #    associations #=> []
-      #    joins #=>  [#<Arel::Nodes::InnerJoin: ...]
+      #    base # => Physician
+      #    associations # => []
+      #    joins # =>  [#<Arel::Nodes::InnerJoin: ...]
       #
       #  However if I execute `Physician.joins(:appointments).to_a` then
-      #    base #=> Physician
-      #    associations #=> [:appointments]
-      #    joins #=>  []
+      #    base # => Physician
+      #    associations # => [:appointments]
+      #    joins # =>  []
       #
       def initialize(base, associations, joins)
-        @base_klass    = base
-        @table_joins   = joins
-        @join_parts    = [JoinBase.new(base)]
-        @associations  = {}
-        @reflections   = []
         @alias_tracker = AliasTracker.new(base.connection, joins)
         @alias_tracker.aliased_name_for(base.table_name) # Updates the count for base.table_name to 1
-        build(associations)
+        tree = self.class.make_tree associations
+        @join_root = JoinBase.new base, build(tree, base)
+        @join_root.children.each { |child| construct_tables! @join_root, child }
       end
 
-      def graft(*associations)
-        associations.each do |association|
-          join_associations.detect {|a| association == a} ||
-            build(association.reflection.name, association.find_parent_in(self) || join_base, association.join_type)
-        end
-        self
+      def reflections
+        join_root.drop(1).map!(&:reflection)
       end
 
-      def join_associations
-        join_parts.last(join_parts.length - 1)
-      end
+      def join_constraints(outer_joins)
+        joins = join_root.children.flat_map { |child|
+          make_inner_joins join_root, child
+        }
 
-      def join_base
-        join_parts.first
+        joins.concat outer_joins.flat_map { |oj|
+          if join_root.match? oj.join_root
+            walk join_root, oj.join_root
+          else
+            oj.join_root.children.flat_map { |child|
+              make_outer_joins oj.join_root, child
+            }
+          end
+        }
       end
 
-      def join_relation(relation)
-        join_associations.each do |association|
-          relation = association.join_relation(relation)
-        end
-        relation
+      def aliases
+        Aliases.new join_root.each_with_index.map { |join_part,i|
+          columns = join_part.column_names.each_with_index.map { |column_name,j|
+            Aliases::Column.new column_name, "t#{i}_r#{j}"
+          }
+          Aliases::Table.new(join_part, columns)
+        }
       end
 
-      def columns
-        join_parts.collect { |join_part|
-          table = join_part.aliased_table
-          join_part.column_names_with_alias.collect{ |column_name, aliased_name|
-            table[column_name].as Arel.sql(aliased_name)
+      def instantiate(result_set, aliases)
+        primary_key = aliases.column_alias(join_root, join_root.primary_key)
+        type_caster = result_set.column_type primary_key
+
+        seen = Hash.new { |h,parent_klass|
+          h[parent_klass] = Hash.new { |i,parent_id|
+            i[parent_id] = Hash.new { |j,child_klass| j[child_klass] = {} }
           }
-        }.flatten
-      end
+        }
 
-      def instantiate(rows)
-        primary_key = join_base.aliased_primary_key
-        parents = {}
+        model_cache = Hash.new { |h,klass| h[klass] = {} }
+        parents = model_cache[join_root]
+        column_aliases = aliases.column_aliases join_root
 
-        records = rows.map { |model|
-          primary_id = model[primary_key]
-          parent = parents[primary_id] ||= join_base.instantiate(model)
-          construct(parent, @associations, join_associations, model)
-          parent
-        }.uniq
+        result_set.each { |row_hash|
+          primary_id = type_caster.type_cast row_hash[primary_key]
+          parent = parents[primary_id] ||= join_root.instantiate(row_hash, column_aliases)
+          construct(parent, join_root, row_hash, result_set, seen, model_cache, aliases)
+        }
 
-        remove_duplicate_results!(base_klass, records, @associations)
-        records
+        parents.values
       end
 
-      protected
-
-      def remove_duplicate_results!(base, records, associations)
-        case associations
-        when Symbol, String
-          reflection = base.reflections[associations]
-          remove_uniq_by_reflection(reflection, records)
-        when Array
-          associations.each do |association|
-            remove_duplicate_results!(base, records, association)
-          end
-        when Hash
-          associations.each_key do |name|
-            reflection = base.reflections[name]
-            remove_uniq_by_reflection(reflection, records)
-
-            parent_records = []
-            records.each do |record|
-              if descendant = record.send(reflection.name)
-                if reflection.collection?
-                  parent_records.concat descendant.target.uniq
-                else
-                  parent_records << descendant
-                end
-              end
-            end
+      private
 
-            remove_duplicate_results!(reflection.klass, parent_records, associations[name]) unless parent_records.empty?
-          end
-        end
+      def make_constraints(parent, child, tables, join_type)
+        chain         = child.reflection.chain
+        foreign_table = parent.table
+        foreign_klass = parent.base_klass
+        child.join_constraints(foreign_table, foreign_klass, child, join_type, tables, child.reflection.scope_chain, chain)
       end
 
-      def cache_joined_association(association)
-        associations = []
-        parent = association.parent
-        while parent != join_base
-          associations.unshift(parent.reflection.name)
-          parent = parent.parent
-        end
-        ref = @associations
-        associations.each do |key|
-          ref = ref[key]
-        end
-        ref[association.reflection.name] ||= {}
-      end
+      def make_outer_joins(parent, child)
+        tables    = table_aliases_for(parent, child)
+        join_type = Arel::OuterJoin
+        joins     = make_constraints parent, child, tables, join_type
 
-      def build(associations, parent = nil, join_type = Arel::InnerJoin)
-        parent ||= join_parts.last
-        case associations
-        when Symbol, String
-          reflection = parent.reflections[associations.intern] or
-          raise ConfigurationError, "Association named '#{ associations }' was not found on #{ parent.base_klass.name }; perhaps you misspelled it?"
-          unless join_association = find_join_association(reflection, parent)
-            @reflections << reflection
-            join_association = build_join_association(reflection, parent)
-            join_association.join_type = join_type
-            @join_parts << join_association
-            cache_joined_association(join_association)
-          end
-          join_association
-        when Array
-          associations.each do |association|
-            build(association, parent, join_type)
-          end
-        when Hash
-          associations.keys.sort_by { |a| a.to_s }.each do |name|
-            join_association = build(name, parent, join_type)
-            build(associations[name], join_association, join_type)
-          end
-        else
-          raise ConfigurationError, associations.inspect
-        end
+        joins.concat child.children.flat_map { |c| make_outer_joins(child, c) }
       end
 
-      def find_join_association(name_or_reflection, parent)
-        if String === name_or_reflection
-          name_or_reflection = name_or_reflection.to_sym
-        end
+      def make_inner_joins(parent, child)
+        tables    = child.tables
+        join_type = Arel::InnerJoin
+        joins     = make_constraints parent, child, tables, join_type
 
-        join_associations.detect { |j|
-          j.reflection == name_or_reflection && j.parent == parent
+        joins.concat child.children.flat_map { |c| make_inner_joins(child, c) }
+      end
+
+      def table_aliases_for(parent, node)
+        node.reflection.chain.map { |reflection|
+          alias_tracker.aliased_table_for(
+            reflection.table_name,
+            table_alias_for(reflection, parent, reflection != node.reflection)
+          )
         }
       end
 
-      def remove_uniq_by_reflection(reflection, records)
-        if reflection && reflection.collection?
-          records.each { |record| record.send(reflection.name).target.uniq! }
-        end
+      def construct_tables!(parent, node)
+        node.tables = table_aliases_for(parent, node)
+        node.children.each { |child| construct_tables! node, child }
       end
 
-      def build_join_association(reflection, parent)
-        JoinAssociation.new(reflection, self, parent)
+      def table_alias_for(reflection, parent, join)
+        name = "#{reflection.plural_name}_#{parent.table_name}"
+        name << "_join" if join
+        name
       end
 
-      def construct(parent, associations, join_parts, row)
-        case associations
-        when Symbol, String
-          name = associations.to_s
+      def walk(left, right)
+        intersection, missing = right.children.map { |node1|
+          [left.children.find { |node2| node1.match? node2 }, node1]
+        }.partition(&:first)
 
-          join_part = join_parts.detect { |j|
-            j.reflection.name.to_s == name &&
-              j.parent_table_name == parent.class.table_name }
+        ojs = missing.flat_map { |_,n| make_outer_joins left, n }
+        intersection.flat_map { |l,r| walk l, r }.concat ojs
+      end
 
-            raise(ConfigurationError, "No such association") unless join_part
+      def find_reflection(klass, name)
+        klass.reflect_on_association(name) or
+          raise ConfigurationError, "Association named '#{ name }' was not found on #{ klass.name }; perhaps you misspelled it?"
+      end
 
-            join_parts.delete(join_part)
-            construct_association(parent, join_part, row)
-        when Array
-          associations.each do |association|
-            construct(parent, association, join_parts, row)
-          end
-        when Hash
-          associations.sort_by { |k,_| k.to_s }.each do |association_name, assoc|
-            association = construct(parent, association_name, join_parts, row)
-            construct(association, assoc, join_parts, row) if association
+      def build(associations, base_klass)
+        associations.map do |name, right|
+          reflection = find_reflection base_klass, name
+          reflection.check_validity!
+
+          if reflection.options[:polymorphic]
+            raise EagerLoadPolymorphicError.new(reflection)
           end
-        else
-          raise ConfigurationError, associations.inspect
+
+          JoinAssociation.new reflection, build(right, reflection.klass)
         end
       end
 
-      def construct_association(record, join_part, row)
-        return if record.id.to_s != join_part.parent.record_id(row).to_s
+      def construct(ar_parent, parent, row, rs, seen, model_cache, aliases)
+        primary_id  = ar_parent.id
 
-        macro = join_part.reflection.macro
-        if macro == :has_one
-          return record.association(join_part.reflection.name).target if record.association_cache.key?(join_part.reflection.name)
-          association = join_part.instantiate(row) unless row[join_part.aliased_primary_key].nil?
-          set_target_and_inverse(join_part, association, record)
-        else
-          association = join_part.instantiate(row) unless row[join_part.aliased_primary_key].nil?
-          case macro
-          when :has_many, :has_and_belongs_to_many
-            other = record.association(join_part.reflection.name)
+        parent.children.each do |node|
+          if node.reflection.collection?
+            other = ar_parent.association(node.reflection.name)
             other.loaded!
-            other.target.push(association) if association
-            other.set_inverse_instance(association)
-          when :belongs_to
-            set_target_and_inverse(join_part, association, record)
           else
-            raise ConfigurationError, "unknown macro: #{join_part.reflection.macro}"
+            if ar_parent.association_cache.key?(node.reflection.name)
+              model = ar_parent.association(node.reflection.name).target
+              construct(model, node, row, rs, seen, model_cache, aliases)
+              next
+            end
+          end
+
+          key = aliases.column_alias(node, node.primary_key)
+          id = row[key]
+          next if id.nil?
+
+          model = seen[parent.base_klass][primary_id][node.base_klass][id]
+
+          if model
+            construct(model, node, row, rs, seen, model_cache, aliases)
+          else
+            model = construct_model(ar_parent, node, row, model_cache, id, aliases)
+            seen[parent.base_klass][primary_id][node.base_klass][id] = model
+            construct(model, node, row, rs, seen, model_cache, aliases)
           end
         end
-        association
       end
 
-      def set_target_and_inverse(join_part, association, record)
-        other = record.association(join_part.reflection.name)
-        other.target = association
-        other.set_inverse_instance(association)
+      def construct_model(record, node, row, model_cache, id, aliases)
+        model = model_cache[node][id] ||= node.instantiate(row,
+                                                           aliases.column_aliases(node))
+        other = record.association(node.reflection.name)
+
+        if node.reflection.collection?
+          other.target.push(model)
+        else
+          other.target = model
+        end
+
+        other.set_inverse_instance(model)
+        model
       end
     end
   end
diff --git a/activerecord/lib/active_record/associations/join_dependency/join_association.rb b/activerecord/lib/active_record/associations/join_dependency/join_association.rb
index 33b363339c..84e18684d8 100644
--- a/activerecord/lib/active_record/associations/join_dependency/join_association.rb
+++ b/activerecord/lib/active_record/associations/join_dependency/join_association.rb
@@ -1,132 +1,80 @@
+require 'active_record/associations/join_dependency/join_part'
+
 module ActiveRecord
   module Associations
     class JoinDependency # :nodoc:
       class JoinAssociation < JoinPart # :nodoc:
-        include JoinHelper
-
         # The reflection of the association represented
         attr_reader :reflection
 
-        # The JoinDependency object which this JoinAssociation exists within. This is mainly
-        # relevant for generating aliases which do not conflict with other joins which are
-        # part of the query.
-        attr_reader :join_dependency
-
-        # A JoinBase instance representing the active record we are joining onto.
-        # (So in Author.has_many :posts, the Author would be that base record.)
-        attr_reader :parent
-
-        # What type of join will be generated, either Arel::InnerJoin (default) or Arel::OuterJoin
-        attr_accessor :join_type
-
-        # These implement abstract methods from the superclass
-        attr_reader :aliased_prefix
-
-        attr_reader :tables
-
-        delegate :options, :through_reflection, :source_reflection, :chain, :to => :reflection
-        delegate :table, :table_name, :to => :parent, :prefix => :parent
-        delegate :alias_tracker, :to => :join_dependency
-
-        alias :alias_suffix :parent_table_name
+        attr_accessor :tables
 
-        def initialize(reflection, join_dependency, parent = nil)
-          reflection.check_validity!
-
-          if reflection.options[:polymorphic]
-            raise EagerLoadPolymorphicError.new(reflection)
-          end
-
-          super(reflection.klass)
+        def initialize(reflection, children)
+          super(reflection.klass, children)
 
           @reflection      = reflection
-          @join_dependency = join_dependency
-          @parent          = parent
-          @join_type       = Arel::InnerJoin
-          @aliased_prefix  = "t#{ join_dependency.join_parts.size }"
-          @tables          = construct_tables.reverse
+          @tables          = nil
         end
 
-        def ==(other)
-          other.class == self.class &&
-            other.reflection == reflection &&
-            other.parent == parent
+        def match?(other)
+          return true if self == other
+          super && reflection == other.reflection
         end
 
-        def find_parent_in(other_join_dependency)
-          other_join_dependency.join_parts.detect do |join_part|
-            case parent
-            when JoinBase
-              parent.base_klass == join_part.base_klass
-            else
-              parent == join_part
-            end
-          end
-        end
+        def join_constraints(foreign_table, foreign_klass, node, join_type, tables, scope_chain, chain)
+          joins         = []
+          tables        = tables.reverse
 
-        def join_to(manager, relation)
-          tables        = @tables.dup
-          foreign_table = parent_table
-          foreign_klass = parent.base_klass
+          scope_chain_iter = scope_chain.reverse_each
 
           # The chain starts with the target table, but we want to end with it here (makes
           # more sense in this context), so we reverse
-          chain.reverse.each_with_index do |reflection, i|
+          chain.reverse_each do |reflection|
             table = tables.shift
+            klass = reflection.klass
 
             case reflection.source_macro
             when :belongs_to
               key         = reflection.association_primary_key
               foreign_key = reflection.foreign_key
-            when :has_and_belongs_to_many
-              # Join the join table first...
-              manager.from(join(
-                table,
-                table[reflection.foreign_key].
-                  eq(foreign_table[reflection.active_record_primary_key])
-              ))
-
-              foreign_table, table = table, tables.shift
-
-              key         = reflection.association_primary_key
-              foreign_key = reflection.association_foreign_key
             else
               key         = reflection.foreign_key
               foreign_key = reflection.active_record_primary_key
             end
 
-            constraint = build_constraint(reflection, table, key, foreign_table, foreign_key)
+            constraint = build_constraint(klass, table, key, foreign_table, foreign_key)
 
-            scope_chain_items = scope_chain[i]
+            scope_chain_items = scope_chain_iter.next.map do |item|
+              if item.is_a?(Relation)
+                item
+              else
+                ActiveRecord::Relation.create(klass, table).instance_exec(node, &item)
+              end
+            end
 
             if reflection.type
-              scope_chain_items += [
-                ActiveRecord::Relation.new(reflection.klass, table)
+              scope_chain_items <<
+                ActiveRecord::Relation.create(klass, table)
                   .where(reflection.type => foreign_klass.base_class.name)
-              ]
             end
 
-            constraint = scope_chain_items.inject(constraint) do |chain, item|
-              unless item.is_a?(Relation)
-                item = ActiveRecord::Relation.new(reflection.klass, table).instance_exec(self, &item)
-              end
+            scope_chain_items.concat [klass.send(:build_default_scope)].compact
 
-              relation.bind_values += item.bind_values
+            rel = scope_chain_items.inject(scope_chain_items.shift) do |left, right|
+              left.merge right
+            end
 
-              if item.arel.constraints.empty?
-                chain
-              else
-                chain.and(item.arel.constraints)
-              end
+            if rel && !rel.arel.constraints.empty?
+              constraint = constraint.and rel.arel.constraints
             end
 
-            manager.from(join(table, constraint))
+            joins << table.create_join(table, table.create_on(constraint), join_type)
 
             # The current table in this iteration becomes the foreign table in the next
-            foreign_table, foreign_klass = table, reflection.klass
+            foreign_table, foreign_klass = table, klass
           end
 
-          manager
+          joins
         end
 
         #  Builds equality condition.
@@ -138,42 +86,32 @@ module ActiveRecord
         #  end
         #
         #  If I execute `Physician.joins(:appointments).to_a` then
-        #    reflection    #=> #<ActiveRecord::Reflection::AssociationReflection @macro=:has_many ...>
-        #    table         #=> #<Arel::Table @name="appointments" ...>
-        #    key           #=>  physician_id
-        #    foreign_table #=> #<Arel::Table @name="physicians" ...>
-        #    foreign_key   #=> id
+        #    reflection    # => #<ActiveRecord::Reflection::AssociationReflection @macro=:has_many ...>
+        #    table         # => #<Arel::Table @name="appointments" ...>
+        #    key           # =>  physician_id
+        #    foreign_table # => #<Arel::Table @name="physicians" ...>
+        #    foreign_key   # => id
         #
-        def build_constraint(reflection, table, key, foreign_table, foreign_key)
+        def build_constraint(klass, table, key, foreign_table, foreign_key)
           constraint = table[key].eq(foreign_table[foreign_key])
 
-          if reflection.klass.finder_needs_type_condition?
+          if klass.finder_needs_type_condition?
             constraint = table.create_and([
               constraint,
-              reflection.klass.send(:type_condition, table)
+              klass.send(:type_condition, table)
             ])
           end
 
           constraint
         end
 
-        def join_relation(joining_relation)
-          self.join_type = Arel::OuterJoin
-          joining_relation.joins(self)
-        end
-
         def table
-          tables.last
+          tables.first
         end
 
         def aliased_table_name
           table.table_alias || table.name
         end
-
-        def scope_chain
-          @scope_chain ||= reflection.scope_chain.reverse
-        end
-
       end
     end
   end
diff --git a/activerecord/lib/active_record/associations/join_dependency/join_base.rb b/activerecord/lib/active_record/associations/join_dependency/join_base.rb
index a7dacdbfd6..3a26c25737 100644
--- a/activerecord/lib/active_record/associations/join_dependency/join_base.rb
+++ b/activerecord/lib/active_record/associations/join_dependency/join_base.rb
@@ -1,18 +1,16 @@
+require 'active_record/associations/join_dependency/join_part'
+
 module ActiveRecord
   module Associations
     class JoinDependency # :nodoc:
       class JoinBase < JoinPart # :nodoc:
-        def ==(other)
-          other.class == self.class &&
-            other.base_klass == base_klass
-        end
-
-        def aliased_prefix
-          "t0"
+        def match?(other)
+          return true if self == other
+          super && base_klass == other.base_klass
         end
 
         def table
-          Arel::Table.new(table_name, arel_engine)
+          base_klass.arel_table
         end
 
         def aliased_table_name
diff --git a/activerecord/lib/active_record/associations/join_dependency/join_part.rb b/activerecord/lib/active_record/associations/join_dependency/join_part.rb
index b534569063..91e1c6a9d7 100644
--- a/activerecord/lib/active_record/associations/join_dependency/join_part.rb
+++ b/activerecord/lib/active_record/associations/join_dependency/join_part.rb
@@ -8,34 +8,36 @@ module ActiveRecord
       # operations (for example a has_and_belongs_to_many JoinAssociation would result in
       # two; one for the join table and one for the target table).
       class JoinPart # :nodoc:
+        include Enumerable
+
         # The Active Record class which this join part is associated 'about'; for a JoinBase
         # this is the actual base model, for a JoinAssociation this is the target model of the
         # association.
-        attr_reader :base_klass
+        attr_reader :base_klass, :children
 
-        delegate :table_name, :column_names, :primary_key, :reflections, :arel_engine, :to => :base_klass
+        delegate :table_name, :column_names, :primary_key, :to => :base_klass
 
-        def initialize(base_klass)
+        def initialize(base_klass, children)
           @base_klass = base_klass
-          @cached_record = {}
           @column_names_with_alias = nil
+          @children = children
         end
 
-        def aliased_table
-          Arel::Nodes::TableAlias.new table, aliased_table_name
+        def name
+          reflection.name
         end
 
-        def ==(other)
-          raise NotImplementedError
+        def match?(other)
+          self.class == other.class
         end
 
-        # An Arel::Table for the active_record
-        def table
-          raise NotImplementedError
+        def each(&block)
+          yield self
+          children.each { |child| child.each(&block) }
         end
 
-        # The prefix to be used when aliasing columns in the active_record's table
-        def aliased_prefix
+        # An Arel::Table for the active_record
+        def table
           raise NotImplementedError
         end
 
@@ -44,33 +46,25 @@ module ActiveRecord
           raise NotImplementedError
         end
 
-        # The alias for the primary key of the active_record's table
-        def aliased_primary_key
-          "#{aliased_prefix}_r0"
-        end
+        def extract_record(row, column_names_with_alias)
+          # This code is performance critical as it is called per row.
+          # see: https://github.com/rails/rails/pull/12185
+          hash = {}
 
-        # An array of [column_name, alias] pairs for the table
-        def column_names_with_alias
-          unless @column_names_with_alias
-            @column_names_with_alias = []
+          index = 0
+          length = column_names_with_alias.length
 
-            ([primary_key] + (column_names - [primary_key])).compact.each_with_index do |column_name, i|
-              @column_names_with_alias << [column_name, "#{aliased_prefix}_r#{i}"]
-            end
+          while index < length
+            column_name, alias_name = column_names_with_alias[index]
+            hash[column_name] = row[alias_name]
+            index += 1
           end
-          @column_names_with_alias
-        end
-
-        def extract_record(row)
-          Hash[column_names_with_alias.map{|cn, an| [cn, row[an]]}]
-        end
 
-        def record_id(row)
-          row[aliased_primary_key]
+          hash
         end
 
-        def instantiate(row)
-          @cached_record[record_id(row)] ||= base_klass.instantiate(extract_record(row))
+        def instantiate(row, aliases)
+          base_klass.instantiate(extract_record(row, aliases))
         end
       end
     end
diff --git a/activerecord/lib/active_record/associations/join_helper.rb b/activerecord/lib/active_record/associations/join_helper.rb
index 5a41b40c8f..f345d16841 100644
--- a/activerecord/lib/active_record/associations/join_helper.rb
+++ b/activerecord/lib/active_record/associations/join_helper.rb
@@ -10,21 +10,12 @@ module ActiveRecord
       private
 
       def construct_tables
-        tables = []
-        chain.each do |reflection|
-          tables << alias_tracker.aliased_table_for(
+        chain.map do |reflection|
+          alias_tracker.aliased_table_for(
             table_name_for(reflection),
             table_alias_for(reflection, reflection != self.reflection)
           )
-
-          if reflection.source_macro == :has_and_belongs_to_many
-            tables << alias_tracker.aliased_table_for(
-              (reflection.source_reflection || reflection).join_table,
-              table_alias_for(reflection, true)
-            )
-          end
         end
-        tables
       end
 
       def table_name_for(reflection)
diff --git a/activerecord/lib/active_record/associations/preloader.rb b/activerecord/lib/active_record/associations/preloader.rb
index 82bf426b22..e49fc5d5c4 100644
--- a/activerecord/lib/active_record/associations/preloader.rb
+++ b/activerecord/lib/active_record/associations/preloader.rb
@@ -42,12 +42,9 @@ module ActiveRecord
         autoload :HasManyThrough,      'active_record/associations/preloader/has_many_through'
         autoload :HasOne,              'active_record/associations/preloader/has_one'
         autoload :HasOneThrough,       'active_record/associations/preloader/has_one_through'
-        autoload :HasAndBelongsToMany, 'active_record/associations/preloader/has_and_belongs_to_many'
         autoload :BelongsTo,           'active_record/associations/preloader/belongs_to'
       end
 
-      attr_reader :records, :associations, :preload_scope, :model
-
       # Eager loads the named associations for the given Active Record record(s).
       #
       # In this description, 'association name' shall refer to the name passed
@@ -82,38 +79,47 @@ module ActiveRecord
       #   [ :books, :author ]
       #   { author: :avatar }
       #   [ :books, { author: :avatar } ]
-      def initialize(records, associations, preload_scope = nil)
-        @records       = Array.wrap(records).compact.uniq
-        @associations  = Array.wrap(associations)
-        @preload_scope = preload_scope || Relation.new(nil, nil)
-      end
 
-      def run
-        unless records.empty?
-          associations.each { |association| preload(association) }
+      NULL_RELATION = Struct.new(:values, :bind_values).new({}, [])
+
+      def preload(records, associations, preload_scope = nil)
+        records       = Array.wrap(records).compact.uniq
+        associations  = Array.wrap(associations)
+        preload_scope = preload_scope || NULL_RELATION
+
+        if records.empty?
+          []
+        else
+          associations.flat_map { |association|
+            preloaders_on association, records, preload_scope
+          }
         end
       end
 
       private
 
-      def preload(association)
+      def preloaders_on(association, records, scope)
         case association
         when Hash
-          preload_hash(association)
+          preloaders_for_hash(association, records, scope)
         when Symbol
-          preload_one(association)
+          preloaders_for_one(association, records, scope)
         when String
-          preload_one(association.to_sym)
+          preloaders_for_one(association.to_sym, records, scope)
         else
           raise ArgumentError, "#{association.inspect} was not recognised for preload"
         end
       end
 
-      def preload_hash(association)
-        association.each do |parent, child|
-          Preloader.new(records, parent, preload_scope).run
-          Preloader.new(records.map { |record| record.send(parent) }.flatten, child).run
-        end
+      def preloaders_for_hash(association, records, scope)
+        parent, child = association.to_a.first # hash should only be of length 1
+
+        loaders = preloaders_for_one parent, records, scope
+
+        recs = loaders.flat_map(&:preloaded_records).uniq
+        loaders.concat Array.wrap(child).flat_map { |assoc|
+          preloaders_on assoc, recs, scope
+        }
       end
 
       # Not all records have the same class, so group then preload group on the reflection
@@ -123,52 +129,81 @@ module ActiveRecord
       # Additionally, polymorphic belongs_to associations can have multiple associated
       # classes, depending on the polymorphic_type field. So we group by the classes as
       # well.
-      def preload_one(association)
-        grouped_records(association).each do |reflection, klasses|
-          klasses.each do |klass, records|
-            preloader_for(reflection).new(klass, records, reflection, preload_scope).run
+      def preloaders_for_one(association, records, scope)
+        grouped_records(association, records).flat_map do |reflection, klasses|
+          klasses.map do |rhs_klass, rs|
+            loader = preloader_for(reflection, rs, rhs_klass).new(rhs_klass, rs, reflection, scope)
+            loader.run self
+            loader
           end
         end
       end
 
-      def grouped_records(association)
-        Hash[
-          records_by_reflection(association).map do |reflection, records|
-            [reflection, records.group_by { |record| association_klass(reflection, record) }]
-          end
-        ]
+      def grouped_records(association, records)
+        reflection_records = records_by_reflection(association, records)
+
+        reflection_records.each_with_object({}) do |(reflection, r_records),h|
+          h[reflection] = r_records.group_by { |record|
+            association_klass(reflection, record)
+          }
+        end
       end
 
-      def records_by_reflection(association)
+      def records_by_reflection(association, records)
         records.group_by do |record|
-          reflection = record.class.reflections[association]
+          reflection = record.class.reflect_on_association(association)
 
-          unless reflection
-            raise ActiveRecord::ConfigurationError, "Association named '#{association}' was not found; " \
-                                                    "perhaps you misspelled it?"
-          end
-
-          reflection
+          reflection || raise_config_error(record, association)
         end
       end
 
+      def raise_config_error(record, association)
+        raise ActiveRecord::ConfigurationError,
+              "Association named '#{association}' was not found on #{record.class.name}; " \
+              "perhaps you misspelled it?"
+      end
+
       def association_klass(reflection, record)
         if reflection.macro == :belongs_to && reflection.options[:polymorphic]
-          klass = record.send(reflection.foreign_type)
+          klass = record.read_attribute(reflection.foreign_type.to_s)
           klass && klass.constantize
         else
           reflection.klass
         end
       end
 
-      def preloader_for(reflection)
+      class AlreadyLoaded
+        attr_reader :owners, :reflection
+
+        def initialize(klass, owners, reflection, preload_scope)
+          @owners = owners
+          @reflection = reflection
+        end
+
+        def run(preloader); end
+
+        def preloaded_records
+          owners.flat_map { |owner| owner.association(reflection.name).target }
+        end
+      end
+
+      class NullPreloader
+        def self.new(klass, owners, reflection, preload_scope); self; end
+        def self.run(preloader); end
+      end
+
+      def preloader_for(reflection, owners, rhs_klass)
+        return NullPreloader unless rhs_klass
+
+        if owners.first.association(reflection.name).loaded?
+          return AlreadyLoaded
+        end
+
         case reflection.macro
         when :has_many
           reflection.options[:through] ? HasManyThrough : HasMany
         when :has_one
           reflection.options[:through] ? HasOneThrough : HasOne
-        when :has_and_belongs_to_many
-          HasAndBelongsToMany
         when :belongs_to
           BelongsTo
         end
diff --git a/activerecord/lib/active_record/associations/preloader/association.rb b/activerecord/lib/active_record/associations/preloader/association.rb
index f55b4824e5..83c69586e6 100644
--- a/activerecord/lib/active_record/associations/preloader/association.rb
+++ b/activerecord/lib/active_record/associations/preloader/association.rb
@@ -3,6 +3,7 @@ module ActiveRecord
     class Preloader
       class Association #:nodoc:
         attr_reader :owners, :reflection, :preload_scope, :model, :klass
+        attr_reader :preloaded_records
 
         def initialize(klass, owners, reflection, preload_scope)
           @klass         = klass
@@ -12,15 +13,14 @@ module ActiveRecord
           @model         = owners.first && owners.first.class
           @scope         = nil
           @owners_by_key = nil
+          @preloaded_records = []
         end
 
-        def run
-          unless owners.first.association(reflection.name).loaded?
-            preload
-          end
+        def run(preloader)
+          preload(preloader)
         end
 
-        def preload
+        def preload(preloader)
           raise NotImplementedError
         end
 
@@ -29,6 +29,10 @@ module ActiveRecord
         end
 
         def records_for(ids)
+          query_scope(ids)
+        end
+
+        def query_scope(ids)
           scope.where(association_key.in(ids))
         end
 
@@ -52,12 +56,9 @@ module ActiveRecord
           raise NotImplementedError
         end
 
-        # We're converting to a string here because postgres will return the aliased association
-        # key in a habtm as a string (for whatever reason)
         def owners_by_key
           @owners_by_key ||= owners.group_by do |owner|
-            key = owner[owner_key_name]
-            key && key.to_s
+            owner[owner_key_name]
           end
         end
 
@@ -67,38 +68,47 @@ module ActiveRecord
 
         private
 
-        def associated_records_by_owner
+        def associated_records_by_owner(preloader)
           owners_map = owners_by_key
           owner_keys = owners_map.keys.compact
 
-          if klass.nil? || owner_keys.empty?
-            records = []
-          else
+          # Each record may have multiple owners, and vice-versa
+          records_by_owner = owners.each_with_object({}) do |owner,h|
+            h[owner] = []
+          end
+
+          if owner_keys.any?
             # Some databases impose a limit on the number of ids in a list (in Oracle it's 1000)
             # Make several smaller queries if necessary or make one query if the adapter supports it
             sliced  = owner_keys.each_slice(klass.connection.in_clause_length || owner_keys.size)
-            records = sliced.map { |slice| records_for(slice).to_a }.flatten
-          end
-
-          # Each record may have multiple owners, and vice-versa
-          records_by_owner = Hash[owners.map { |owner| [owner, []] }]
-          records.each do |record|
-            owner_key = record[association_key_name].to_s
 
-            owners_map[owner_key].each do |owner|
-              records_by_owner[owner] << record
+            records = load_slices sliced
+            records.each do |record, owner_key|
+              owners_map[owner_key].each do |owner|
+                records_by_owner[owner] << record
+              end
             end
           end
+
           records_by_owner
         end
 
+        def load_slices(slices)
+          @preloaded_records = slices.flat_map { |slice|
+            records_for(slice)
+          }
+
+          @preloaded_records.map { |record|
+            [record, record[association_key_name]]
+          }
+        end
+
         def reflection_scope
           @reflection_scope ||= reflection.scope ? klass.unscoped.instance_exec(nil, &reflection.scope) : klass.unscoped
         end
 
         def build_scope
           scope = klass.unscoped
-          scope.default_scoped = true
 
           values         = reflection_scope.values
           reflection_binds = reflection_scope.bind_values
@@ -112,11 +122,19 @@ module ActiveRecord
           scope.select!   preload_values[:select] || values[:select] || table[Arel.star]
           scope.includes! preload_values[:includes] || values[:includes]
 
+          if preload_values.key? :order
+            scope.order! preload_values[:order]
+          else
+            if values.key? :order
+              scope.order! values[:order]
+            end
+          end
+
           if options[:as]
             scope.where!(klass.table_name => { reflection.type => model.base_class.sti_name })
           end
 
-          scope
+          klass.default_scoped.merge(scope)
         end
       end
     end
diff --git a/activerecord/lib/active_record/associations/preloader/collection_association.rb b/activerecord/lib/active_record/associations/preloader/collection_association.rb
index e6cd35e7a1..5adffcd831 100644
--- a/activerecord/lib/active_record/associations/preloader/collection_association.rb
+++ b/activerecord/lib/active_record/associations/preloader/collection_association.rb
@@ -9,8 +9,8 @@ module ActiveRecord
           super.order(preload_scope.values[:order] || reflection_scope.values[:order])
         end
 
-        def preload
-          associated_records_by_owner.each do |owner, records|
+        def preload(preloader)
+          associated_records_by_owner(preloader).each do |owner, records|
             association = owner.association(reflection.name)
             association.loaded!
             association.target.concat(records)
diff --git a/activerecord/lib/active_record/associations/preloader/has_and_belongs_to_many.rb b/activerecord/lib/active_record/associations/preloader/has_and_belongs_to_many.rb
deleted file mode 100644
index 9a3fada380..0000000000
--- a/activerecord/lib/active_record/associations/preloader/has_and_belongs_to_many.rb
+++ /dev/null
@@ -1,60 +0,0 @@
-module ActiveRecord
-  module Associations
-    class Preloader
-      class HasAndBelongsToMany < CollectionAssociation #:nodoc:
-        attr_reader :join_table
-
-        def initialize(klass, records, reflection, preload_options)
-          super
-          @join_table = Arel::Table.new(reflection.join_table).alias('t0')
-        end
-
-        # Unlike the other associations, we want to get a raw array of rows so that we can
-        # access the aliased column on the join table
-        def records_for(ids)
-          scope = super
-          klass.connection.select_all(scope.arel, 'SQL', scope.bind_values)
-        end
-
-        def owner_key_name
-          reflection.active_record_primary_key
-        end
-
-        def association_key_name
-          'ar_association_key_name'
-        end
-
-        def association_key
-          join_table[reflection.foreign_key]
-        end
-
-        private
-
-        # Once we have used the join table column (in super), we manually instantiate the
-        # actual records, ensuring that we don't create more than one instances of the same
-        # record
-        def associated_records_by_owner
-          records = {}
-          super.each_value do |rows|
-            rows.map! { |row| records[row[klass.primary_key]] ||= klass.instantiate(row) }
-          end
-        end
-
-        def build_scope
-          super.joins(join).select(join_select)
-        end
-
-        def join_select
-          association_key.as(Arel.sql(association_key_name))
-        end
-
-        def join
-          condition = table[reflection.association_primary_key].eq(
-            join_table[reflection.association_foreign_key])
-
-          table.create_join(join_table, table.create_on(condition))
-        end
-      end
-    end
-  end
-end
diff --git a/activerecord/lib/active_record/associations/preloader/has_many_through.rb b/activerecord/lib/active_record/associations/preloader/has_many_through.rb
index 157b627ad5..7b37b5942d 100644
--- a/activerecord/lib/active_record/associations/preloader/has_many_through.rb
+++ b/activerecord/lib/active_record/associations/preloader/has_many_through.rb
@@ -4,7 +4,7 @@ module ActiveRecord
       class HasManyThrough < CollectionAssociation #:nodoc:
         include ThroughAssociation
 
-        def associated_records_by_owner
+        def associated_records_by_owner(preloader)
           records_by_owner = super
 
           if reflection_scope.distinct_value
diff --git a/activerecord/lib/active_record/associations/preloader/singular_association.rb b/activerecord/lib/active_record/associations/preloader/singular_association.rb
index 44e804d785..f60647a81e 100644
--- a/activerecord/lib/active_record/associations/preloader/singular_association.rb
+++ b/activerecord/lib/active_record/associations/preloader/singular_association.rb
@@ -5,13 +5,13 @@ module ActiveRecord
 
         private
 
-        def preload
-          associated_records_by_owner.each do |owner, associated_records|
+        def preload(preloader)
+          associated_records_by_owner(preloader).each do |owner, associated_records|
             record = associated_records.first
 
             association = owner.association(reflection.name)
             association.target = record
-            association.set_inverse_instance(record)
+            association.set_inverse_instance(record) if record
           end
         end
 
diff --git a/activerecord/lib/active_record/associations/preloader/through_association.rb b/activerecord/lib/active_record/associations/preloader/through_association.rb
index c4b50ab306..2a8530af62 100644
--- a/activerecord/lib/active_record/associations/preloader/through_association.rb
+++ b/activerecord/lib/active_record/associations/preloader/through_association.rb
@@ -2,7 +2,6 @@ module ActiveRecord
   module Associations
     class Preloader
       module ThroughAssociation #:nodoc:
-
         def through_reflection
           reflection.through_reflection
         end
@@ -11,51 +10,84 @@ module ActiveRecord
           reflection.source_reflection
         end
 
-        def associated_records_by_owner
-          through_records = through_records_by_owner
+        def associated_records_by_owner(preloader)
+          preloader.preload(owners,
+                            through_reflection.name,
+                            through_scope)
 
-          Preloader.new(through_records.values.flatten, source_reflection.name, reflection_scope).run
+          through_records = owners.map do |owner|
+            association = owner.association through_reflection.name
 
-          through_records.each do |owner, records|
-            records.map! { |r| r.send(source_reflection.name) }.flatten!
-            records.compact!
+            [owner, Array(association.reader)]
           end
-        end
 
-        private
+          reset_association owners, through_reflection.name
+
+          middle_records = through_records.map { |(_,rec)| rec }.flatten
+
+          preloaders = preloader.preload(middle_records,
+                                         source_reflection.name,
+                                         reflection_scope)
 
-        def through_records_by_owner
-          Preloader.new(owners, through_reflection.name, through_scope).run
+          @preloaded_records = preloaders.flat_map(&:preloaded_records)
+
+          middle_to_pl = preloaders.each_with_object({}) do |pl,h|
+            pl.owners.each { |middle|
+              h[middle] = pl
+            }
+          end
+
+          record_offset = {}
+          @preloaded_records.each_with_index do |record,i|
+            record_offset[record] = i
+          end
 
-          Hash[owners.map do |owner|
-            through_records = Array.wrap(owner.send(through_reflection.name))
+          through_records.each_with_object({}) { |(lhs,center),records_by_owner|
+            pl_to_middle = center.group_by { |record| middle_to_pl[record] }
 
-            # Dont cache the association - we would only be caching a subset
-            if (through_scope != through_reflection.klass.unscoped) ||
-               (reflection.options[:source_type] && through_reflection.collection?)
-              owner.association(through_reflection.name).reset
+            records_by_owner[lhs] = pl_to_middle.flat_map do |pl, middles|
+              rhs_records = middles.flat_map { |r|
+                association = r.association source_reflection.name
+
+                association.reader
+              }.compact
+
+              rhs_records.sort_by { |rhs| record_offset[rhs] }
             end
+          }
+        end
+
+        private
+
+        def reset_association(owners, association_name)
+          should_reset = (through_scope != through_reflection.klass.unscoped) ||
+             (reflection.options[:source_type] && through_reflection.collection?)
 
-            [owner, through_records]
-          end]
+          # Dont cache the association - we would only be caching a subset
+          if should_reset
+            owners.each { |owner|
+              owner.association(association_name).reset
+            }
+          end
         end
 
+
         def through_scope
-          through_scope = through_reflection.klass.unscoped
+          scope = through_reflection.klass.unscoped
 
           if options[:source_type]
-            through_scope.where! reflection.foreign_type => options[:source_type]
+            scope.where! reflection.foreign_type => options[:source_type]
           else
             unless reflection_scope.where_values.empty?
-              through_scope.includes_values = Array(reflection_scope.values[:includes] || options[:source])
-              through_scope.where_values    = reflection_scope.values[:where]
+              scope.includes_values = Array(reflection_scope.values[:includes] || options[:source])
+              scope.where_values    = reflection_scope.values[:where]
             end
 
-            through_scope.references! reflection_scope.values[:references]
-            through_scope.order! reflection_scope.values[:order] if through_scope.eager_loading?
+            scope.references! reflection_scope.values[:references]
+            scope.order! reflection_scope.values[:order] if scope.eager_loading?
           end
 
-          through_scope
+          scope
         end
       end
     end
diff --git a/activerecord/lib/active_record/associations/singular_association.rb b/activerecord/lib/active_record/associations/singular_association.rb
index 10238555f0..e4500af5b2 100644
--- a/activerecord/lib/active_record/associations/singular_association.rb
+++ b/activerecord/lib/active_record/associations/singular_association.rb
@@ -39,10 +39,11 @@ module ActiveRecord
         end
 
         def find_target
-          scope.first.tap { |record| set_inverse_instance(record) }
+          if record = scope.first
+            set_inverse_instance record
+          end
         end
 
-        # Implemented by subclasses
         def replace(record)
           raise NotImplementedError, "Subclasses must implement a replace(record) method"
         end
diff --git a/activerecord/lib/active_record/associations/through_association.rb b/activerecord/lib/active_record/associations/through_association.rb
index 35f29b37a2..ba7d2a3782 100644
--- a/activerecord/lib/active_record/associations/through_association.rb
+++ b/activerecord/lib/active_record/associations/through_association.rb
@@ -13,9 +13,9 @@ module ActiveRecord
         #   2. To get the type conditions for any STI models in the chain
         def target_scope
           scope = super
-          chain[1..-1].each do |reflection|
+          chain.drop(1).each do |reflection|
             scope.merge!(
-              reflection.klass.all.with_default_scope.
+              reflection.klass.all.
                 except(:select, :create_with, :includes, :preload, :joins, :eager_load)
             )
           end