Merge remote branch 'rails/master' into identity_map

Conflicts:
	activerecord/examples/performance.rb
	activerecord/lib/active_record/association_preload.rb
	activerecord/lib/active_record/associations.rb
	activerecord/lib/active_record/associations/association_proxy.rb
	activerecord/lib/active_record/autosave_association.rb
	activerecord/lib/active_record/base.rb
	activerecord/lib/active_record/nested_attributes.rb
	activerecord/test/cases/relations_test.rb

14 files changed, 873 insertions, 1044 deletions

diff --git a/activerecord/lib/active_record/associations/association_collection.rb b/activerecord/lib/active_record/associations/association_collection.rb
index 11a7a725e5..ca350f51c9 100644
--- a/activerecord/lib/active_record/associations/association_collection.rb
+++ b/activerecord/lib/active_record/associations/association_collection.rb
@@ -1,4 +1,3 @@
-require 'set'
 require 'active_support/core_ext/array/wrap'
 
 module ActiveRecord
@@ -23,98 +22,55 @@ module ActiveRecord
 
       def select(select = nil)
         if block_given?
-          load_target
-          @target.select.each { |e| yield e }
+          load_target.select.each { |e| yield e }
         else
           scoped.select(select)
         end
       end
 
-      def scoped
-        with_scope(@scope) { @reflection.klass.scoped }
-      end
-
       def find(*args)
-        options = args.extract_options!
-
-        # If using a custom finder_sql, scan the entire collection.
         if @reflection.options[:finder_sql]
-          expects_array = args.first.kind_of?(Array)
-          ids           = args.flatten.compact.uniq.map { |arg| arg.to_i }
-
-          if ids.size == 1
-            id = ids.first
-            record = load_target.detect { |r| id == r.id }
-            expects_array ? [ record ] : record
-          else
-            load_target.select { |r| ids.include?(r.id) }
-          end
+          find_by_scan(*args)
         else
-          merge_options_from_reflection!(options)
-          construct_find_options!(options)
-
-          with_scope(:find => @scope[:find].slice(:conditions, :order)) do
-            relation = @reflection.klass.send(:construct_finder_arel, options, @reflection.klass.send(:current_scoped_methods))
-
-            case args.first
-            when :first, :last
-              relation.send(args.first)
-            when :all
-              records = relation.all
-              @reflection.options[:uniq] ? uniq(records) : records
-            else
-              relation.find(*args)
-            end
-          end
+          scoped.find(*args)
         end
       end
 
-      # Fetches the first one using SQL if possible.
       def first(*args)
-        if fetch_first_or_last_using_find?(args)
-          find(:first, *args)
-        else
-          load_target unless loaded?
-          args = args[1..-1] if args.first.kind_of?(Hash) && args.first.empty?
-          @target.first(*args)
-        end
+        first_or_last(:first, *args)
       end
 
-      # Fetches the last one using SQL if possible.
       def last(*args)
-        if fetch_first_or_last_using_find?(args)
-          find(:last, *args)
-        else
-          load_target unless loaded?
-          @target.last(*args)
-        end
+        first_or_last(:last, *args)
       end
 
       def to_ary
-        load_target
-        if @target.is_a?(Array)
-          @target.to_ary
-        else
-          Array.wrap(@target)
-        end
+        load_target.dup
       end
       alias_method :to_a, :to_ary
 
       def reset
-        reset_target!
-        reset_scopes_cache!
-        @loaded = false
+        @_scopes_cache = {}
+        @loaded        = false
+        @target        = []
       end
 
       def build(attributes = {}, &block)
-        if attributes.is_a?(Array)
-          attributes.collect { |attr| build(attr, &block) }
-        else
-          build_record(attributes) do |record|
-            block.call(record) if block_given?
-            set_belongs_to_association_for(record)
-          end
+        build_or_create(attributes, :build, &block)
+      end
+
+      def create(attributes = {}, &block)
+        unless @owner.persisted?
+          raise ActiveRecord::RecordNotSaved, "You cannot call create unless the parent is saved"
         end
+
+        build_or_create(attributes, :create, &block)
+      end
+
+      def create!(attrs = {}, &block)
+        record = create(attrs, &block)
+        Array.wrap(record).each(&:save!)
+        record
       end
 
       # Add +records+ to this association.  Returns +self+ so method calls may be chained.
@@ -124,9 +80,9 @@ module ActiveRecord
         load_target if @owner.new_record?
 
         transaction do
-          flatten_deeper(records).each do |record|
+          records.flatten.each do |record|
             raise_on_type_mismatch(record)
-            add_record_to_target_with_callbacks(record) do |r|
+            add_to_target(record) do |r|
               result &&= insert_record(record) unless @owner.new_record?
             end
           end
@@ -157,18 +113,35 @@ module ActiveRecord
       #
       # See delete for more info.
       def delete_all
-        load_target
-        delete(@target)
-        reset_target!
-        reset_scopes_cache!
+        delete(load_target).tap do
+          reset
+          loaded!
+        end
+      end
+
+      # Identical to delete_all, except that the return value is the association (for chaining)
+      # rather than the records which have been removed.
+      def clear
+        delete_all
+        self
+      end
+
+      # Destroy all the records from this association.
+      #
+      # See destroy for more info.
+      def destroy_all
+        destroy(load_target).tap do
+          reset
+          loaded!
+        end
       end
 
       # Calculate sum using SQL, not Enumerable
       def sum(*args)
         if block_given?
-          calculate(:sum, *args) { |*block_args| yield(*block_args) }
+          scoped.sum(*args) { |*block_args| yield(*block_args) }
         else
-          calculate(:sum, *args)
+          scoped.sum(*args)
         end
       end
 
@@ -185,14 +158,13 @@ module ActiveRecord
 
           @reflection.klass.count_by_sql(custom_counter_sql)
         else
-
           if @reflection.options[:uniq]
             # This is needed because 'SELECT count(DISTINCT *)..' is not valid SQL.
-            column_name = "#{@reflection.quoted_table_name}.#{@reflection.klass.primary_key}" unless column_name
+            column_name ||= @reflection.klass.primary_key
             options.merge!(:distinct => true)
           end
 
-          value = @reflection.klass.send(:with_scope, @scope) { @reflection.klass.count(column_name, options) }
+          value = scoped.count(column_name, options)
 
           limit  = @reflection.options[:limit]
           offset = @reflection.options[:offset]
@@ -213,10 +185,7 @@ 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)
-        remove_records(records) do |_records, old_records|
-          delete_records(old_records) if old_records.any?
-          _records.each { |record| @target.delete(record) }
-        end
+        delete_or_destroy(records, @reflection.options[:dependent])
       end
 
       # Destroy +records+ and remove them from this association calling
@@ -225,54 +194,8 @@ module ActiveRecord
       # Note that this method will _always_ remove records from the database
       # ignoring the +:dependent+ option.
       def destroy(*records)
-        records = find(records) if records.any? {|record| record.kind_of?(Fixnum) || record.kind_of?(String)}
-        remove_records(records) do |_records, old_records|
-          old_records.each { |record| record.destroy }
-        end
-
-        load_target
-      end
-
-      # Removes all records from this association.  Returns +self+ so method calls may be chained.
-      def clear
-        unless length.zero? # forces load_target if it hasn't happened already
-          if @reflection.options[:dependent] == :destroy
-            destroy_all
-          else
-            delete_all
-          end
-        end
-
-        self
-      end
-
-      # Destroy all the records from this association.
-      #
-      # See destroy for more info.
-      def destroy_all
-        load_target
-        destroy(@target).tap do
-          reset_target!
-          reset_scopes_cache!
-        end
-      end
-
-      def create(attrs = {})
-        if attrs.is_a?(Array)
-          attrs.collect { |attr| create(attr) }
-        else
-          create_record(attrs) do |record|
-            yield(record) if block_given?
-            record.save
-          end
-        end
-      end
-
-      def create!(attrs = {})
-        create_record(attrs) do |record|
-          yield(record) if block_given?
-          record.save!
-        end
+        records = find(records) if records.any? { |record| record.kind_of?(Fixnum) || record.kind_of?(String) }
+        delete_or_destroy(records, :destroy)
       end
 
       # Returns the size of the collection by executing a SELECT COUNT(*)
@@ -316,7 +239,7 @@ module ActiveRecord
 
       def any?
         if block_given?
-          method_missing(:any?) { |*block_args| yield(*block_args) }
+          load_target.any? { |*block_args| yield(*block_args) }
         else
           !empty?
         end
@@ -325,7 +248,7 @@ module ActiveRecord
       # Returns true if the collection has more than 1 record. Equivalent to collection.size > 1.
       def many?
         if block_given?
-          method_missing(:many?) { |*block_args| yield(*block_args) }
+          load_target.many? { |*block_args| yield(*block_args) }
         else
           size > 1
         end
@@ -342,108 +265,117 @@ module ActiveRecord
       # This will perform a diff and delete/add only records that have changed.
       def replace(other_array)
         other_array.each { |val| raise_on_type_mismatch(val) }
-
-        load_target
-        other   = other_array.size < 100 ? other_array : other_array.to_set
-        current = @target.size < 100 ? @target : @target.to_set
+        original_target = load_target.dup
 
         transaction do
-          delete(@target.select { |v| !other.include?(v) })
-          concat(other_array.select { |v| !current.include?(v) })
+          delete(@target - other_array)
+
+          unless concat(other_array - @target)
+            @target = original_target
+            raise RecordNotSaved, "Failed to replace #{@reflection.name} because one or more of the " \
+                                  "new records could not be saved."
+          end
         end
       end
 
       def include?(record)
-        return false unless record.is_a?(@reflection.klass)
-        return include_in_memory?(record) if record.new_record?
-        load_target if @reflection.options[:finder_sql] && !loaded?
-        loaded? ? @target.include?(record) : exists?(record)
+        if record.is_a?(@reflection.klass)
+          if record.new_record?
+            include_in_memory?(record)
+          else
+            load_target if @reflection.options[:finder_sql]
+            loaded? ? @target.include?(record) : scoped.exists?(record)
+          end
+        else
+          false
+        end
       end
 
-      def proxy_respond_to?(method, include_private = false)
+      def respond_to?(method, include_private = false)
         super || @reflection.klass.respond_to?(method, include_private)
       end
 
+      def method_missing(method, *args, &block)
+        match = DynamicFinderMatch.match(method)
+        if match && match.creator?
+          attributes = match.attribute_names
+          return send(:"find_by_#{attributes.join('_and_')}", *args) || create(Hash[attributes.zip(args)])
+        end
+
+        if @target.respond_to?(method) || (!@reflection.klass.respond_to?(method) && Class.respond_to?(method))
+          super
+        elsif @reflection.klass.scopes[method]
+          @_scopes_cache ||= {}
+          @_scopes_cache[method] ||= {}
+          @_scopes_cache[method][args] ||= scoped.readonly(nil).send(method, *args)
+        else
+          scoped.readonly(nil).send(method, *args, &block)
+        end
+      end
+
       protected
-        def construct_find_options!(options)
+
+        def association_scope
+          options = @reflection.options.slice(:order, :limit, :joins, :group, :having, :offset)
+          super.apply_finder_options(options)
         end
 
         def load_target
-          if !@owner.new_record? || foreign_key_present
+          if find_target?
+            targets = []
+
             begin
-              unless loaded?
-                if @target.is_a?(Array) && @target.any?
-                  @target = find_target.map do |f|
-                    i = @target.index(f)
-                    if i
-                      @target.delete_at(i).tap do |t|
-                        keys = ["id"] + t.changes.keys + (f.attribute_names - t.attribute_names)
-                        f.attributes.except(*keys).each do |k,v|
-                          t.send("#{k}=", v)
-                        end
-                      end
-                    else
-                      f
-                    end
-                  end + @target
-                else
-                  @target = find_target
-                end
-              end
+              targets = find_target
             rescue ActiveRecord::RecordNotFound
               reset
             end
+
+            @target = merge_target_lists(targets, @target)
           end
 
-          loaded if target
+          loaded!
           target
         end
 
-        def method_missing(method, *args)
-          match = DynamicFinderMatch.match(method)
-          if match && match.creator?
-            attributes = match.attribute_names
-            return send(:"find_by_#{attributes.join('_and_')}", *args) || create(Hash[attributes.zip(args)])
-          end
+        def add_to_target(record)
+          transaction do
+            callback(:before_add, record)
+            yield(record) if block_given?
 
-          if @target.respond_to?(method) || (!@reflection.klass.respond_to?(method) && Class.respond_to?(method))
-            super
-          elsif @reflection.klass.scopes[method]
-            @_scopes_cache ||= {}
-            @_scopes_cache[method] ||= {}
-            @_scopes_cache[method][args] ||= with_scope(@scope) { @reflection.klass.send(method, *args) }
-          else
-            with_scope(@scope) do
-              if block_given?
-                @reflection.klass.send(method, *args) { |*block_args| yield(*block_args) }
-              else
-                @reflection.klass.send(method, *args)
-              end
+            if @reflection.options[:uniq] && index = @target.index(record)
+              @target[index] = record
+            else
+              @target << record
             end
+
+            callback(:after_add, record)
+            set_inverse_instance(record)
           end
+
+          record
+        end
+
+      private
+
+        def select_value
+          super || uniq_select_value
+        end
+
+        def uniq_select_value
+          @reflection.options[:uniq] && "DISTINCT #{@reflection.quoted_table_name}.*"
         end
 
         def custom_counter_sql
           if @reflection.options[:counter_sql]
-            counter_sql = @reflection.options[:counter_sql]
+            interpolate(@reflection.options[:counter_sql])
           else
             # replace the SELECT clause with COUNT(*), preserving any hints within /* ... */
-            counter_sql = @reflection.options[:finder_sql].sub(/SELECT\b(\/\*.*?\*\/ )?(.*)\bFROM\b/im) { "SELECT #{$1}COUNT(*) FROM" }
+            interpolate(@reflection.options[:finder_sql]).sub(/SELECT\b(\/\*.*?\*\/ )?(.*)\bFROM\b/im) { "SELECT #{$1}COUNT(*) FROM" }
           end
-
-          interpolate_sql(counter_sql)
         end
 
         def custom_finder_sql
-          interpolate_sql(@reflection.options[:finder_sql])
-        end
-
-        def reset_target!
-          @target = Array.new
-        end
-
-        def reset_scopes_cache!
-          @_scopes_cache = {}
+          interpolate(@reflection.options[:finder_sql])
         end
 
         def find_target
@@ -455,65 +387,74 @@ module ActiveRecord
             end
 
           records = @reflection.options[:uniq] ? uniq(records) : records
-          records.each do |record|
-            set_inverse_instance(record, @owner)
-          end
+          records.each { |record| set_inverse_instance(record) }
           records
         end
 
-        def add_record_to_target_with_callbacks(record)
-          callback(:before_add, record)
-          yield(record) if block_given?
-          @target ||= [] unless loaded?
-          if index = @target.index(record)
-            @target[index] = record
-          else
-             @target << record
-          end
-          callback(:after_add, record)
-          set_inverse_instance(record, @owner)
-          record
+        def merge_target_lists(loaded, existing)
+          return loaded if existing.empty?
+          return existing if loaded.empty?
+
+          loaded.map do |f|
+            i = existing.index(f)
+            if i
+              existing.delete_at(i).tap do |t|
+                keys = ["id"] + t.changes.keys + (f.attribute_names - t.attribute_names)
+                # FIXME: this call to attributes causes many NoMethodErrors
+                attributes = f.attributes
+                (attributes.keys - keys).each do |k|
+                  t.send("#{k}=", attributes[k])
+                end
+              end
+            else
+              f
+            end
+          end + existing
         end
 
-      private
-        def create_record(attrs)
-          attrs.update(@reflection.options[:conditions]) if @reflection.options[:conditions].is_a?(Hash)
-          ensure_owner_is_persisted!
-
-          scoped_where = scoped.where_values_hash
-          create_scope = scoped_where ? @scope[:create].merge(scoped_where) : @scope[:create]
-          record = @reflection.klass.send(:with_scope, :create => create_scope) do
-            @reflection.build_association(attrs)
-          end
-          if block_given?
-            add_record_to_target_with_callbacks(record) { |*block_args| yield(*block_args) }
-          else
-            add_record_to_target_with_callbacks(record)
+        def build_or_create(attributes, method)
+          records = Array.wrap(attributes).map do |attrs|
+            record = build_record(attrs)
+
+            add_to_target(record) do
+              yield(record) if block_given?
+              insert_record(record) if method == :create
+            end
           end
+
+          attributes.is_a?(Array) ? records : records.first
         end
 
-        def build_record(attrs)
-          attrs.update(@reflection.options[:conditions]) if @reflection.options[:conditions].is_a?(Hash)
-          record = @reflection.build_association(attrs)
-          if block_given?
-            add_record_to_target_with_callbacks(record) { |*block_args| yield(*block_args) }
-          else
-            add_record_to_target_with_callbacks(record)
-          end
+        # Do the relevant stuff to insert the given record into the association collection.
+        def insert_record(record, validate = true)
+          raise NotImplementedError
         end
 
-        def remove_records(*records)
-          records = flatten_deeper(records)
+        def build_record(attributes)
+          @reflection.build_association(scoped.scope_for_create.merge(attributes))
+        end
+
+        def delete_or_destroy(records, method)
+          records = records.flatten
           records.each { |record| raise_on_type_mismatch(record) }
+          existing_records = records.reject { |r| r.new_record? }
 
           transaction do
             records.each { |record| callback(:before_remove, record) }
-            old_records = records.reject { |r| r.new_record? }
-            yield(records, old_records)
+
+            delete_records(existing_records, method) if existing_records.any?
+            records.each { |record| @target.delete(record) }
+
             records.each { |record| callback(:after_remove, record) }
           end
         end
 
+        # Delete the given records from the association, using one of the methods :destroy,
+        # :delete_all or :nullify (or nil, in which case a default is used).
+        def delete_records(records, method)
+          raise NotImplementedError
+        end
+
         def callback(method, record)
           callbacks_for(method).each do |callback|
             case callback
@@ -532,27 +473,61 @@ module ActiveRecord
           @owner.class.send(full_callback_name.to_sym) || []
         end
 
-        def ensure_owner_is_persisted!
-          unless @owner.persisted?
-            raise ActiveRecord::RecordNotSaved, "You cannot call create unless the parent is saved"
-          end
-        end
-
+        # Should we deal with assoc.first or assoc.last by issuing an independent query to
+        # the database, or by getting the target, and then taking the first/last item from that?
+        #
+        # If the args is just a non-empty options hash, go to the database.
+        #
+        # 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)
-          (args.first.kind_of?(Hash) && !args.first.empty?) || !(loaded? || @owner.new_record? || @reflection.options[:finder_sql] ||
-            @target.any? { |record| record.new_record? } || args.first.kind_of?(Integer))
+          if args.first.is_a?(Hash)
+            true
+          else
+            !(loaded? ||
+              @owner.new_record? ||
+              @reflection.options[:finder_sql] ||
+              @target.any? { |record| record.new_record? || record.changed? } ||
+              args.first.kind_of?(Integer))
+          end
         end
 
         def include_in_memory?(record)
           if @reflection.is_a?(ActiveRecord::Reflection::ThroughReflection)
-            @owner.send(proxy_reflection.through_reflection.name.to_sym).any? do |source|
+            @owner.send(proxy_reflection.through_reflection.name).any? { |source|
               target = source.send(proxy_reflection.source_reflection.name)
               target.respond_to?(:include?) ? target.include?(record) : target == record
-            end
+            } || @target.include?(record)
           else
             @target.include?(record)
           end
         end
+
+        # If using a custom finder_sql, #find scans the entire collection.
+        def find_by_scan(*args)
+          expects_array = args.first.kind_of?(Array)
+          ids           = args.flatten.compact.uniq.map { |arg| arg.to_i }
+
+          if ids.size == 1
+            id = ids.first
+            record = load_target.detect { |r| id == r.id }
+            expects_array ? [ record ] : record
+          else
+            load_target.select { |r| ids.include?(r.id) }
+          end
+        end
+
+        # Fetches the first/last using SQL if possible, otherwise from the target array.
+        def first_or_last(type, *args)
+          args.shift if args.first.is_a?(Hash) && args.first.empty?
+
+          collection = fetch_first_or_last_using_find?(args) ? scoped : load_target
+          collection.send(type, *args)
+        end
     end
   end
 end
diff --git a/activerecord/lib/active_record/associations/association_proxy.rb b/activerecord/lib/active_record/associations/association_proxy.rb
index 53ec5a0da6..d16cda5585 100644
--- a/activerecord/lib/active_record/associations/association_proxy.rb
+++ b/activerecord/lib/active_record/associations/association_proxy.rb
@@ -4,17 +4,18 @@ module ActiveRecord
   module Associations
     # = Active Record Associations
     #
-    # This is the root class of all association proxies:
+    # This is the root class of all association proxies ('+ Foo' signifies an included module Foo):
     #
     #   AssociationProxy
-    #     BelongsToAssociation
+    #     SingularAssociaton
     #       HasOneAssociation
-    #     BelongsToPolymorphicAssociation
+    #         HasOneThroughAssociation + ThroughAssociation
+    #       BelongsToAssociation
+    #         BelongsToPolymorphicAssociation
     #     AssociationCollection
     #       HasAndBelongsToManyAssociation
     #       HasManyAssociation
-    #         HasManyThroughAssociation
-    #            HasOneThroughAssociation
+    #         HasManyThroughAssociation + ThroughAssociation
     #
     # Association proxies in Active Record are middlemen between the object that
     # holds the association, known as the <tt>@owner</tt>, and the actual associated
@@ -50,10 +51,9 @@ module ActiveRecord
     # is computed directly through SQL and does not trigger by itself the
     # instantiation of the actual post records.
     class AssociationProxy #:nodoc:
-      alias_method :proxy_respond_to?, :respond_to?
       alias_method :proxy_extend, :extend
-      delegate :to_param, :to => :proxy_target
-      instance_methods.each { |m| undef_method m unless m.to_s =~ /^(?:nil\?|send|object_id|to_a)$|^__|^respond_to_missing|proxy_/ }
+
+      instance_methods.each { |m| undef_method m unless m.to_s =~ /^(?:nil\?|send|object_id|to_a)$|^__|^respond_to|proxy_/ }
 
       def initialize(owner, reflection)
         @owner, @reflection = owner, reflection
@@ -64,6 +64,10 @@ module ActiveRecord
         construct_scope
       end
 
+      def to_param
+        proxy_target.to_param
+      end
+
       # Returns the owner of the proxy.
       def proxy_owner
         @owner
@@ -75,21 +79,25 @@ module ActiveRecord
         @reflection
       end
 
-      # Returns the \target of the proxy, same as +target+.
-      def proxy_target
-        @target
-      end
-
       # Does the proxy or its \target respond to +symbol+?
       def respond_to?(*args)
-        proxy_respond_to?(*args) || (load_target && @target.respond_to?(*args))
+        super || (load_target && @target.respond_to?(*args))
+      end
+
+      # Forwards any missing method call to the \target.
+      def method_missing(method, *args, &block)
+        if load_target
+          return super unless @target.respond_to?(method)
+          @target.send(method, *args, &block)
+        end
+      rescue NoMethodError => e
+        raise e, e.message.sub(/ for #<.*$/, " via proxy for #{@target}")
       end
 
       # Forwards <tt>===</tt> explicitly to the \target because the instance method
       # removal above doesn't catch it. Loads the \target if needed.
       def ===(other)
-        load_target
-        other === @target
+        other === load_target
       end
 
       # Returns the name of the table of the related class:
@@ -100,13 +108,6 @@ module ActiveRecord
         @reflection.klass.table_name
       end
 
-      # Returns the SQL string that corresponds to the <tt>:conditions</tt>
-      # option of the macro, if given, or +nil+ otherwise.
-      def conditions
-        @conditions ||= interpolate_sql(@reflection.sanitized_conditions) if @reflection.sanitized_conditions
-      end
-      alias :sql_conditions :conditions
-
       # Resets the \loaded flag to +false+ and sets the \target to +nil+.
       def reset
         @loaded = false
@@ -117,6 +118,7 @@ module ActiveRecord
       # Reloads the \target and returns +self+ on success.
       def reload
         reset
+        construct_scope
         load_target
         self unless @target.nil?
       end
@@ -127,117 +129,93 @@ module ActiveRecord
       end
 
       # Asserts the \target has been loaded setting the \loaded flag to +true+.
-      def loaded
-        @loaded = true
+      def loaded!
+        @loaded      = true
+        @stale_state = stale_state
       end
 
-      # Returns the target of this proxy, same as +proxy_target+.
-      def target
-        @target
+      # 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.
+      #
+      # Note that if the target has not been loaded, it is not considered stale.
+      def stale_target?
+        loaded? && @stale_state != stale_state
       end
 
+      # Returns the target of this proxy, same as +proxy_target+.
+      attr_reader :target
+
+      # Returns the \target of the proxy, same as +target+.
+      alias :proxy_target :target
+
       # Sets the target of this proxy to <tt>\target</tt>, and the \loaded flag to +true+.
       def target=(target)
         @target = target
-        loaded
+        loaded!
       end
 
       # Forwards the call to the target. Loads the \target if needed.
       def inspect
-        load_target
-        @target.inspect
+        load_target.inspect
       end
 
       def send(method, *args)
-        if proxy_respond_to?(method)
-          super
-        else
-          load_target
-          @target.send(method, *args)
-        end
+        return super if respond_to?(method)
+        load_target.send(method, *args)
       end
 
-      protected
-        # Does the association have a <tt>:dependent</tt> option?
-        def dependent?
-          @reflection.options[:dependent]
-        end
+      def scoped
+        target_scope.merge(@association_scope)
+      end
 
-        def interpolate_sql(sql, record = nil)
-          @owner.send(:interpolate_sql, sql, record)
-        end
+      protected
 
-        # Forwards the call to the reflection class.
-        def sanitize_sql(sql, table_name = @reflection.klass.table_name)
-          @reflection.klass.send(:sanitize_sql, sql, table_name)
+        # Construct the scope for this association.
+        #
+        # Note that the association_scope is merged into the targed_scope only when the
+        # scoped method is called. This is because at that point the call may be surrounded
+        # by scope.scoping { ... } or with_scope { ... } etc, which affects the scope which
+        # actually gets built.
+        def construct_scope
+          @association_scope = association_scope if target_klass
         end
 
-        # Assigns the ID of the owner to the corresponding foreign key in +record+.
-        # If the association is polymorphic the type of the owner is also set.
-        def set_belongs_to_association_for(record)
-          if @reflection.options[:as]
-            record["#{@reflection.options[:as]}_id"]   = @owner.id if @owner.persisted?
-            record["#{@reflection.options[:as]}_type"] = @owner.class.base_class.name.to_s
-          else
-            if @owner.persisted?
-              primary_key = @reflection.options[:primary_key] || :id
-              record[@reflection.primary_key_name] = @owner.send(primary_key)
-            end
+        def association_scope
+          scope = target_klass.unscoped
+          scope = scope.create_with(creation_attributes)
+          scope = scope.apply_finder_options(@reflection.options.slice(:readonly, :include))
+          scope = scope.where(interpolate(@reflection.options[:conditions]))
+          if select = select_value
+            scope = scope.select(select)
           end
+          scope = scope.extending(*Array.wrap(@reflection.options[:extend]))
+          scope.where(construct_owner_conditions)
         end
 
-        # Merges into +options+ the ones coming from the reflection.
-        def merge_options_from_reflection!(options)
-          options.reverse_merge!(
-            :group   => @reflection.options[:group],
-            :having  => @reflection.options[:having],
-            :limit   => @reflection.options[:limit],
-            :offset  => @reflection.options[:offset],
-            :joins   => @reflection.options[:joins],
-            :include => @reflection.options[:include],
-            :select  => @reflection.options[:select],
-            :readonly  => @reflection.options[:readonly]
-          )
+        def aliased_table
+          target_klass.arel_table
         end
 
-        # Forwards +with_scope+ to the reflection.
-        def with_scope(*args, &block)
-          @reflection.klass.send :with_scope, *args, &block
-        end
-
-        # Construct the scope used for find/create queries on the target
-        def construct_scope
-          @scope = {
-            :find   => construct_find_scope,
-            :create => construct_create_scope
-          }
+        # Set the inverse association, if possible
+        def set_inverse_instance(record)
+          if record && invertible_for?(record)
+            inverse = record.send(:association_proxy, inverse_reflection_for(record).name)
+            inverse.target = @owner
+          end
         end
 
-        # Implemented by subclasses
-        def construct_find_scope
-          raise NotImplementedError
+        # This class of the target. belongs_to polymorphic overrides this to look at the
+        # polymorphic_type field on the owner.
+        def target_klass
+          @reflection.klass
         end
 
-        # Implemented by (some) subclasses
-        def construct_create_scope
-          {}
-        end
-
-      private
-        # Forwards any missing method call to the \target.
-        def method_missing(method, *args)
-          if load_target
-            unless @target.respond_to?(method)
-              message = "undefined method `#{method.to_s}' for \"#{@target}\":#{@target.class.to_s}"
-              raise NoMethodError, message
-            end
-
-            if block_given?
-              @target.send(method, *args)  { |*block_args| yield(*block_args) }
-            else
-              @target.send(method, *args)
-            end
-          end
+        # Can be overridden (i.e. in ThroughAssociation) to merge in other scopes (i.e. the
+        # through association's scope)
+        def target_scope
+          target_klass.scoped
         end
 
         # Loads the \target if needed and returns it.
@@ -251,26 +229,84 @@ module ActiveRecord
         # ActiveRecord::RecordNotFound is rescued within the method, and it is
         # not reraised. The proxy is \reset and +nil+ is the return value.
         def load_target
-          return nil unless defined?(@loaded)
-
-          if !loaded? && (!@owner.new_record? || foreign_key_present)
-            if IdentityMap.enabled? && association_class
-              @target = IdentityMap.get(association_class, @owner[@reflection.association_foreign_key])
+          if find_target?
+            begin
+            if IdentityMap.enabled? && association_class && association_class.respond_to?(:base_class)
+              @target = IdentityMap.get(association_class, @owner[@reflection.foreign_key])
+            end
+            rescue NameError
+              nil
+            ensure
+              @target ||= find_target
             end
-            @target ||= find_target
           end
-
-          @loaded = true
-          @target
+          loaded!
+          target
         rescue ActiveRecord::RecordNotFound
           reset
         end
 
-        # Can be overwritten by associations that might have the foreign key
-        # available for an association without having the object itself (and
-        # still being a new record). Currently, only +belongs_to+ presents
-        # this scenario (both vanilla and polymorphic).
-        def foreign_key_present
+      private
+
+        def find_target?
+          !loaded? && (!@owner.new_record? || foreign_key_present?) && target_klass
+        end
+
+        def interpolate(sql, record = nil)
+          if sql.respond_to?(:to_proc)
+            @owner.send(:instance_exec, record, &sql)
+          else
+            sql
+          end
+        end
+
+        def select_value
+          @reflection.options[:select]
+        end
+
+        # Implemented by (some) subclasses
+        def creation_attributes
+          { }
+        end
+
+        # Returns a hash linking the owner to the association represented by the reflection
+        def construct_owner_attributes(reflection = @reflection)
+          attributes = {}
+          if reflection.macro == :belongs_to
+            attributes[reflection.association_primary_key] = @owner[reflection.foreign_key]
+          else
+            attributes[reflection.foreign_key] = @owner[reflection.active_record_primary_key]
+
+            if reflection.options[:as]
+              attributes["#{reflection.options[:as]}_type"] = @owner.class.base_class.name
+            end
+          end
+          attributes
+        end
+
+        # Builds an array of arel nodes from the owner attributes hash
+        def construct_owner_conditions(table = aliased_table, reflection = @reflection)
+          conditions = construct_owner_attributes(reflection).map do |attr, value|
+            table[attr].eq(value)
+          end
+          table.create_and(conditions)
+        end
+
+        # Sets the owner attributes on the given record
+        def set_owner_attributes(record)
+          if @owner.persisted?
+            construct_owner_attributes.each { |key, value| record[key] = value }
+          end
+        end
+
+        # Should be 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).
+        #
+        # Currently implemented by belongs_to (vanilla and polymorphic) and
+        # has_one/has_many :through associations which go through a belongs_to
+        def foreign_key_present?
           false
         end
 
@@ -284,34 +320,24 @@ module ActiveRecord
           end
         end
 
-        if RUBY_VERSION < '1.9.2'
-          # Array#flatten has problems with recursive arrays before Ruby 1.9.2.
-          # Going one level deeper solves the majority of the problems.
-          def flatten_deeper(array)
-            array.collect { |element| (element.respond_to?(:flatten) && !element.is_a?(Hash)) ? element.flatten : element }.flatten
-          end
-        else
-          def flatten_deeper(array)
-            array.flatten
-          end
+        # Can be redefined by subclasses, notably polymorphic belongs_to
+        # The record parameter is necessary to support polymorphic inverses as we must check for
+        # the association in the specific class of the record.
+        def inverse_reflection_for(record)
+          @reflection.inverse_of
         end
 
-        # Returns the ID of the owner, quoted if needed.
-        def owner_quoted_id
-          @owner.quoted_id
+        # Is this association invertible? Can be redefined by subclasses.
+        def invertible_for?(record)
+          inverse_reflection_for(record)
         end
 
-        def set_inverse_instance(record, instance)
-          return if record.nil? || !we_can_set_the_inverse_on_this?(record)
-          inverse_relationship = @reflection.inverse_of
-          unless inverse_relationship.nil?
-            record.send(:"set_#{inverse_relationship.name}_target", instance)
-          end
-        end
-
-        # Override in subclasses
-        def we_can_set_the_inverse_on_this?(record)
-          false
+        # This should be implemented to return the values of the relevant key(s) on the owner,
+        # so that when state_state is different from the value stored on the last find_target,
+        # the target is stale.
+        #
+        # This is only relevant to certain associations, which is why it returns nil by default.
+        def stale_state
         end
 
         def association_class
diff --git a/activerecord/lib/active_record/associations/belongs_to_association.rb b/activerecord/lib/active_record/associations/belongs_to_association.rb
index b438620c8f..178c7204f8 100644
--- a/activerecord/lib/active_record/associations/belongs_to_association.rb
+++ b/activerecord/lib/active_record/associations/belongs_to_association.rb
@@ -1,41 +1,17 @@
 module ActiveRecord
   # = Active Record Belongs To Associations
   module Associations
-    class BelongsToAssociation < AssociationProxy #:nodoc:
-      def create(attributes = {})
-        replace(@reflection.create_association(attributes))
-      end
-
-      def build(attributes = {})
-        replace(@reflection.build_association(attributes))
-      end
-
+    class BelongsToAssociation < SingularAssociation #:nodoc:
       def replace(record)
-        counter_cache_name = @reflection.counter_cache_column
-
-        if record.nil?
-          if counter_cache_name && @owner.persisted?
-            @reflection.klass.decrement_counter(counter_cache_name, previous_record_id) if @owner[@reflection.primary_key_name]
-          end
-
-          @target = @owner[@reflection.primary_key_name] = nil
-        else
-          raise_on_type_mismatch(record)
+        record = check_record(record)
 
-          if counter_cache_name && @owner.persisted? && record.id != @owner[@reflection.primary_key_name]
-            @reflection.klass.increment_counter(counter_cache_name, record.id)
-            @reflection.klass.decrement_counter(counter_cache_name, @owner[@reflection.primary_key_name]) if @owner[@reflection.primary_key_name]
-          end
-
-          @target = (AssociationProxy === record ? record.target : record)
-          @owner[@reflection.primary_key_name] = record_id(record) if record.persisted?
-          @updated = true
-        end
+        update_counters(record)
+        replace_keys(record)
+        set_inverse_instance(record)
 
-        set_inverse_instance(record, @owner)
+        @updated = true if record
 
-        loaded
-        record
+        self.target = record
       end
 
       def updated?
@@ -43,50 +19,52 @@ module ActiveRecord
       end
 
       private
-        def find_target
-          find_method = if @reflection.options[:primary_key]
-                          "find_by_#{@reflection.options[:primary_key]}"
-                        else
-                          "find"
-                        end
 
-          options = @reflection.options.dup.slice(:select, :include, :readonly)
+        def update_counters(record)
+          counter_cache_name = @reflection.counter_cache_column
+
+          if counter_cache_name && @owner.persisted? && different_target?(record)
+            if record
+              record.class.increment_counter(counter_cache_name, record.id)
+            end
 
-          the_target = with_scope(:find => @scope[:find]) do
-            @reflection.klass.send(find_method,
-              @owner[@reflection.primary_key_name],
-              options
-            ) if @owner[@reflection.primary_key_name]
+            if foreign_key_present?
+              target_klass.decrement_counter(counter_cache_name, target_id)
+            end
           end
-          set_inverse_instance(the_target, @owner)
-          the_target
         end
 
-        def construct_find_scope
-          { :conditions => conditions }
+        # Checks whether record is different to the current target, without loading it
+        def different_target?(record)
+          record.nil? && @owner[@reflection.foreign_key] ||
+          record.id   != @owner[@reflection.foreign_key]
         end
 
-        def foreign_key_present
-          !@owner[@reflection.primary_key_name].nil?
+        def replace_keys(record)
+          @owner[@reflection.foreign_key] = record && record[@reflection.association_primary_key]
+        end
+
+        def foreign_key_present?
+          @owner[@reflection.foreign_key]
         end
 
         # NOTE - for now, we're only supporting inverse setting from belongs_to back onto
         # has_one associations.
-        def we_can_set_the_inverse_on_this?(record)
-          @reflection.has_inverse? && @reflection.inverse_of.macro == :has_one
+        def invertible_for?(record)
+          inverse = inverse_reflection_for(record)
+          inverse && inverse.macro == :has_one
         end
 
-        def record_id(record)
-          record.send(@reflection.options[:primary_key] || :id)
+        def target_id
+          if @reflection.options[:primary_key]
+            @owner.send(@reflection.name).try(:id)
+          else
+            @owner[@reflection.foreign_key]
+          end
         end
 
-        def previous_record_id
-          @previous_record_id ||= if @reflection.options[:primary_key]
-                                    previous_record = @owner.send(@reflection.name)
-                                    previous_record.nil? ? nil : previous_record.id
-                                  else
-                                    @owner[@reflection.primary_key_name]
-                                  end
+        def stale_state
+          @owner[@reflection.foreign_key].to_s
         end
     end
   end
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 a0df860623..4f67b02d00 100644
--- a/activerecord/lib/active_record/associations/belongs_to_polymorphic_association.rb
+++ b/activerecord/lib/active_record/associations/belongs_to_polymorphic_association.rb
@@ -1,77 +1,33 @@
 module ActiveRecord
   # = Active Record Belongs To Polymorphic Association
   module Associations
-    class BelongsToPolymorphicAssociation < AssociationProxy #:nodoc:
-      def replace(record)
-        if record.nil?
-          @target = @owner[@reflection.primary_key_name] = @owner[@reflection.options[:foreign_type]] = nil
-        else
-          @target = (AssociationProxy === record ? record.target : record)
-
-          @owner[@reflection.primary_key_name] = record_id(record)
-          @owner[@reflection.options[:foreign_type]] = record.class.base_class.name.to_s
-
-          @updated = true
-        end
-
-        set_inverse_instance(record, @owner)
-        loaded
-        record
-      end
-
-      def updated?
-        @updated
-      end
-
+    class BelongsToPolymorphicAssociation < BelongsToAssociation #:nodoc:
       private
 
-        # NOTE - for now, we're only supporting inverse setting from belongs_to back onto
-        # has_one associations.
-        def we_can_set_the_inverse_on_this?(record)
-          if @reflection.has_inverse?
-            inverse_association = @reflection.polymorphic_inverse_of(record.class)
-            inverse_association && inverse_association.macro == :has_one
-          else
-            false
-          end
-        end
-
-        def set_inverse_instance(record, instance)
-          return if record.nil? || !we_can_set_the_inverse_on_this?(record)
-          inverse_relationship = @reflection.polymorphic_inverse_of(record.class)
-          if inverse_relationship
-            record.send(:"set_#{inverse_relationship.name}_target", instance)
-          end
+        def replace_keys(record)
+          super
+          @owner[@reflection.foreign_type] = record && record.class.base_class.name
         end
 
-        def construct_find_scope
-          { :conditions => conditions }
+        def different_target?(record)
+          super || record.class != target_klass
         end
 
-        def find_target
-          return nil if association_class.nil?
-
-          target = association_class.send(:with_scope, :find => @scope[:find]) do
-            association_class.find(
-              @owner[@reflection.primary_key_name],
-              :select  => @reflection.options[:select],
-              :include => @reflection.options[:include]
-            )
-          end
-          set_inverse_instance(target, @owner)
-          target
+        def inverse_reflection_for(record)
+          @reflection.polymorphic_inverse_of(record.class)
         end
 
-        def foreign_key_present
-          !@owner[@reflection.primary_key_name].nil?
+        def target_klass
+          type = @owner[@reflection.foreign_type]
+          type && type.constantize
         end
 
-        def record_id(record)
-          record.send(@reflection.options[:primary_key] || :id)
+        def raise_on_type_mismatch(record)
+          # A polymorphic association cannot have a type mismatch, by definition
         end
 
-        def association_class
-          @owner[@reflection.options[:foreign_type]] ? @owner[@reflection.options[:foreign_type]].constantize : nil
+        def stale_state
+          [super, @owner[@reflection.foreign_type].to_s]
         end
     end
   end
diff --git a/activerecord/lib/active_record/associations/class_methods/join_dependency.rb b/activerecord/lib/active_record/associations/class_methods/join_dependency.rb
index a74d0a4ca8..fdd4fe8946 100644
--- a/activerecord/lib/active_record/associations/class_methods/join_dependency.rb
+++ b/activerecord/lib/active_record/associations/class_methods/join_dependency.rb
@@ -176,7 +176,7 @@ module ActiveRecord
 
             join_part = join_parts.detect { |j|
               j.reflection.name.to_s == name &&
-                j.parent_table_name    == parent.class.table_name }
+                j.parent_table_name == parent.class.table_name }
 
               raise(ConfigurationError, "No such association") unless join_part
 
@@ -201,7 +201,7 @@ module ActiveRecord
 
           macro = join_part.reflection.macro
           if macro == :has_one
-            return if record.instance_variable_defined?("@#{join_part.reflection.name}")
+            return 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
@@ -210,9 +210,9 @@ module ActiveRecord
             case macro
             when :has_many, :has_and_belongs_to_many
               collection = record.send(join_part.reflection.name)
-              collection.loaded
+              collection.loaded!
               collection.target.push(association)
-              collection.__send__(:set_inverse_instance, association, record)
+              collection.send(:set_inverse_instance, association)
             when :belongs_to
               set_target_and_inverse(join_part, association, record)
             else
@@ -223,8 +223,9 @@ module ActiveRecord
         end
 
         def set_target_and_inverse(join_part, association, record)
-          association_proxy = record.send("set_#{join_part.reflection.name}_target", association)
-          association_proxy.__send__(:set_inverse_instance, association, record)
+          association_proxy = record.send(:association_proxy, join_part.reflection.name)
+          association_proxy.target = association
+          association_proxy.send(:set_inverse_instance, association)
         end
       end
     end
diff --git a/activerecord/lib/active_record/associations/class_methods/join_dependency/join_association.rb b/activerecord/lib/active_record/associations/class_methods/join_dependency/join_association.rb
index 694778008b..aaa475109e 100644
--- a/activerecord/lib/active_record/associations/class_methods/join_dependency/join_association.rb
+++ b/activerecord/lib/active_record/associations/class_methods/join_dependency/join_association.rb
@@ -82,12 +82,12 @@ module ActiveRecord
               connection = active_record.connection
 
               name = connection.table_alias_for "#{pluralize(reflection.name)}_#{parent_table_name}#{suffix}"
-              table_index = aliases[name] + 1
-              name = name[0, connection.table_alias_length-3] + "_#{table_index}" if table_index > 1
+              aliases[name] += 1
+              name = name[0, connection.table_alias_length-3] + "_#{aliases[name]}" if aliases[name] > 1
+            else
+              aliases[name] += 1
             end
 
-            aliases[name] += 1
-
             name
           end
 
@@ -108,6 +108,10 @@ module ActiveRecord
           end
 
           def process_conditions(conditions, table_name)
+            if conditions.respond_to?(:to_proc)
+              conditions = instance_eval(&conditions)
+            end
+
             Arel.sql(sanitize_sql(conditions, table_name))
           end
 
@@ -190,17 +194,20 @@ module ActiveRecord
             ).alias @aliased_join_table_name
 
             jt_conditions = []
-            jt_foreign_key = first_key = second_key = nil
+            first_key = second_key = nil
 
-            if through_reflection.options[:as] # has_many :through against a polymorphic join
-              as_key         = through_reflection.options[:as].to_s
-              jt_foreign_key = as_key + '_id'
-
-              jt_conditions <<
-              join_table[as_key + '_type'].
-                eq(parent.active_record.base_class.name)
+            if through_reflection.macro == :belongs_to
+              jt_primary_key = through_reflection.foreign_key
+              jt_foreign_key = through_reflection.association_primary_key
             else
-              jt_foreign_key = through_reflection.primary_key_name
+              jt_primary_key = through_reflection.active_record_primary_key
+              jt_foreign_key = through_reflection.foreign_key
+
+              if through_reflection.options[:as] # has_many :through against a polymorphic join
+                jt_conditions <<
+                join_table["#{through_reflection.options[:as]}_type"].
+                  eq(parent.active_record.base_class.name)
+              end
             end
 
             case source_reflection.macro
@@ -225,15 +232,15 @@ module ActiveRecord
                 second_key = source_reflection.association_foreign_key
 
                 jt_conditions <<
-                join_table[reflection.source_reflection.options[:foreign_type]].
+                join_table[reflection.source_reflection.foreign_type].
                   eq(reflection.options[:source_type])
               else
-                second_key = source_reflection.primary_key_name
+                second_key = source_reflection.foreign_key
               end
             end
 
             jt_conditions <<
-            parent_table[parent.primary_key].
+            parent_table[jt_primary_key].
               eq(join_table[jt_foreign_key])
 
             if through_reflection.options[:conditions]
@@ -259,7 +266,7 @@ module ActiveRecord
           end
 
           def join_belongs_to_to(relation)
-            foreign_key = options[:foreign_key] || reflection.primary_key_name
+            foreign_key = options[:foreign_key] || reflection.foreign_key
             primary_key = options[:primary_key] || reflection.klass.primary_key
 
             join_target_table(
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
index e2ce9aefcf..b9c9919e7a 100644
--- 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
@@ -2,81 +2,48 @@ module ActiveRecord
   # = Active Record Has And Belongs To Many Association
   module Associations
     class HasAndBelongsToManyAssociation < AssociationCollection #:nodoc:
-      def create(attributes = {})
-        create_record(attributes) { |record| insert_record(record) }
-      end
-
-      def create!(attributes = {})
-        create_record(attributes) { |record| insert_record(record, true) }
-      end
-
-      def columns
-        @reflection.columns(@reflection.options[:join_table], "#{@reflection.options[:join_table]} Columns")
-      end
-
-      def reset_column_information
-        @reflection.reset_column_information
-      end
+      attr_reader :join_table
 
-      def has_primary_key?
-        @has_primary_key ||= @owner.connection.supports_primary_key? && @owner.connection.primary_key(@reflection.options[:join_table])
+      def initialize(owner, reflection)
+        @join_table = Arel::Table.new(reflection.options[:join_table])
+        super
       end
 
       protected
-        def construct_find_options!(options)
-          options[:joins]      = Arel::SqlLiteral.new(@scope[:find][:joins])
-          options[:readonly]   = finding_with_ambiguous_select?(options[:select] || @reflection.options[:select])
-          options[:select]   ||= (@reflection.options[:select] || Arel::SqlLiteral.new('*'))
-        end
 
-        def count_records
-          load_target.size
-        end
-
-        def insert_record(record, force = true, validate = true)
-          if record.new_record?
-            if force
-              record.save!
-            else
-              return false unless record.save(:validate => validate)
-            end
-          end
+        def insert_record(record, validate = true)
+          return if record.new_record? && !record.save(:validate => validate)
 
           if @reflection.options[:insert_sql]
-            @owner.connection.insert(interpolate_sql(@reflection.options[:insert_sql], record))
+            @owner.connection.insert(interpolate(@reflection.options[:insert_sql], record))
           else
-            relation   = Arel::Table.new(@reflection.options[:join_table])
-            timestamps = record_timestamp_columns(record)
-            timezone   = record.send(:current_time_from_proper_timezone) if timestamps.any?
-
-            attributes = columns.map do |column|
-              name = column.name
-              value = case name.to_s
-                when @reflection.primary_key_name.to_s
-                  @owner.id
-                when @reflection.association_foreign_key.to_s
-                  record.id
-                when *timestamps
-                  timezone
-                else
-                  @owner.send(:quote_value, record[name], column) if record.has_attribute?(name)
-              end
-              [relation[name], value] unless value.nil?
-            end
-
-            stmt = relation.compile_insert Hash[attributes]
+            stmt = join_table.compile_insert(
+              join_table[@reflection.foreign_key]             => @owner.id,
+              join_table[@reflection.association_foreign_key] => record.id
+            )
+
             @owner.connection.insert stmt.to_sql
           end
 
-          true
+          record
         end
 
-        def delete_records(records)
+        def association_scope
+          super.joins(construct_joins)
+        end
+
+      private
+
+        def count_records
+          load_target.size
+        end
+
+        def delete_records(records, method)
           if sql = @reflection.options[:delete_sql]
-            records.each { |record| @owner.connection.delete(interpolate_sql(sql, record)) }
+            records.each { |record| @owner.connection.delete(interpolate(sql, record)) }
           else
-            relation = Arel::Table.new(@reflection.options[:join_table])
-            stmt = relation.where(relation[@reflection.primary_key_name].eq(@owner.id).
+            relation = join_table
+            stmt = relation.where(relation[@reflection.foreign_key].eq(@owner.id).
               and(relation[@reflection.association_foreign_key].in(records.map { |x| x.id }.compact))
             ).compile_delete
             @owner.connection.delete stmt.to_sql
@@ -84,51 +51,25 @@ module ActiveRecord
         end
 
         def construct_joins
-          "INNER JOIN #{@owner.connection.quote_table_name @reflection.options[:join_table]} ON #{@reflection.quoted_table_name}.#{@reflection.klass.primary_key} = #{@owner.connection.quote_table_name @reflection.options[:join_table]}.#{@reflection.association_foreign_key}"
-        end
+          right = join_table
+          left  = @reflection.klass.arel_table
 
-        def construct_conditions
-          sql = "#{@owner.connection.quote_table_name @reflection.options[:join_table]}.#{@reflection.primary_key_name} = #{owner_quoted_id} "
-          sql << " AND (#{conditions})" if conditions
-          sql
-        end
+          condition = left[@reflection.klass.primary_key].eq(
+            right[@reflection.association_foreign_key])
 
-        def construct_find_scope
-          {
-            :conditions => construct_conditions,
-            :joins      => construct_joins,
-            :readonly   => false,
-            :order      => @reflection.options[:order],
-            :include    => @reflection.options[:include],
-            :limit      => @reflection.options[:limit]
-          }
+          right.create_join(right, right.create_on(condition))
         end
 
-        # Join tables with additional columns on top of the two foreign keys must be considered
-        # ambiguous unless a select clause has been explicitly defined. Otherwise you can get
-        # broken records back, if, for example, the join column also has an id column. This will
-        # then overwrite the id column of the records coming back.
-        def finding_with_ambiguous_select?(select_clause)
-          !select_clause && columns.size != 2
+        def construct_owner_conditions
+          super(join_table)
         end
 
-      private
-        def create_record(attributes, &block)
-          # Can't use Base.create because the foreign key may be a protected attribute.
-          ensure_owner_is_persisted!
-          if attributes.is_a?(Array)
-            attributes.collect { |attr| create(attr) }
-          else
-            build_record(attributes, &block)
-          end
+        def select_value
+          super || @reflection.klass.arel_table[Arel.star]
         end
 
-        def record_timestamp_columns(record)
-          if record.record_timestamps
-            record.send(:all_timestamp_attributes).map { |x| x.to_s }
-          else
-            []
-          end
+        def invertible_for?(record)
+          false
         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 4ff61fff45..543b073393 100644
--- a/activerecord/lib/active_record/associations/has_many_association.rb
+++ b/activerecord/lib/active_record/associations/has_many_association.rb
@@ -7,14 +7,14 @@ module ActiveRecord
     # is provided by its child HasManyThroughAssociation.
     class HasManyAssociation < AssociationCollection #:nodoc:
       protected
-        def owner_quoted_id
-          if @reflection.options[:primary_key]
-            @owner.class.quote_value(@owner.send(@reflection.options[:primary_key]))
-          else
-            @owner.quoted_id
-          end
+
+        def insert_record(record, validate = true)
+          set_owner_attributes(record)
+          record.save(:validate => validate)
         end
 
+      private
+
         # Returns the number of records in this collection.
         #
         # If the association has a counter cache it gets that value. Otherwise
@@ -34,83 +34,69 @@ module ActiveRecord
           elsif @reflection.options[:counter_sql] || @reflection.options[:finder_sql]
             @reflection.klass.count_by_sql(custom_counter_sql)
           else
-            @reflection.klass.count(@scope[:find].slice(:conditions, :joins, :include))
+            scoped.count
           end
 
           # If there's nothing in the database and @target has no new records
           # we are certain the current target is an empty array. This is a
           # documented side-effect of the method that may avoid an extra SELECT.
-          @target ||= [] and loaded if count == 0
+          @target ||= [] and loaded! if count == 0
 
           [@reflection.options[:limit], count].compact.min
         end
 
-        def has_cached_counter?
-          @owner.attribute_present?(cached_counter_attribute_name)
-        end
-
-        def cached_counter_attribute_name
-          "#{@reflection.name}_count"
+        def has_cached_counter?(reflection = @reflection)
+          @owner.attribute_present?(cached_counter_attribute_name(reflection))
         end
 
-        def insert_record(record, force = false, validate = true)
-          set_belongs_to_association_for(record)
-          force ? record.save! : record.save(:validate => validate)
+        def cached_counter_attribute_name(reflection = @reflection)
+          "#{reflection.name}_count"
         end
 
-        # Deletes the records according to the <tt>:dependent</tt> option.
-        def delete_records(records)
-          case @reflection.options[:dependent]
-            when :destroy
-              records.each { |r| r.destroy }
-            when :delete_all
-              @reflection.klass.delete(records.map { |record| record.id })
-            else
-              relation = Arel::Table.new(@reflection.table_name)
-              stmt = relation.where(relation[@reflection.primary_key_name].eq(@owner.id).
-                  and(relation[@reflection.klass.primary_key].in(records.map { |r| r.id }))
-              ).compile_update(relation[@reflection.primary_key_name] => nil)
-              @owner.connection.update stmt.to_sql
-
-              @owner.class.update_counters(@owner.id, cached_counter_attribute_name => -records.size) if has_cached_counter?
+        def update_counter(difference, reflection = @reflection)
+          if has_cached_counter?(reflection)
+            counter = cached_counter_attribute_name(reflection)
+            @owner.class.update_counters(@owner.id, counter => difference)
+            @owner[counter] += difference
+            @owner.changed_attributes.delete(counter) # eww
           end
         end
 
-        def target_obsolete?
-          false
+        # This shit is nasty. We need to avoid the following situation:
+        #
+        #   * An associated record is deleted via record.destroy
+        #   * Hence the callbacks run, and they find a belongs_to on the record with a
+        #     :counter_cache options which points back at our @owner. So they update the
+        #     counter cache.
+        #   * In which case, we must make sure to *not* update the counter cache, or else
+        #     it will be decremented twice.
+        #
+        # Hence this method.
+        def inverse_updates_counter_cache?(reflection = @reflection)
+          counter_name = cached_counter_attribute_name(reflection)
+          reflection.klass.reflect_on_all_associations(:belongs_to).any? { |inverse_reflection|
+            inverse_reflection.counter_cache_column == counter_name
+          }
         end
 
-        def construct_conditions
-          if @reflection.options[:as]
-            sql =
-              "#{@reflection.quoted_table_name}.#{@reflection.options[:as]}_id = #{owner_quoted_id} AND " +
-              "#{@reflection.quoted_table_name}.#{@reflection.options[:as]}_type = #{@owner.class.quote_value(@owner.class.base_class.name.to_s)}"
+        # Deletes the records according to the <tt>:dependent</tt> option.
+        def delete_records(records, method)
+          if method == :destroy
+            records.each { |r| r.destroy }
+            update_counter(-records.length) unless inverse_updates_counter_cache?
           else
-            sql = "#{@reflection.quoted_table_name}.#{@reflection.primary_key_name} = #{owner_quoted_id}"
-          end
-          sql << " AND (#{conditions})" if conditions
-          sql
-        end
-
-        def construct_find_scope
-          {
-            :conditions => construct_conditions,
-            :readonly   => false,
-            :order      => @reflection.options[:order],
-            :limit      => @reflection.options[:limit],
-            :include    => @reflection.options[:include]
-          }
-        end
+            keys  = records.map { |r| r[@reflection.association_primary_key] }
+            scope = scoped.where(@reflection.association_primary_key => keys)
 
-        def construct_create_scope
-          create_scoping = {}
-          set_belongs_to_association_for(create_scoping)
-          create_scoping
+            if method == :delete_all
+              update_counter(-scope.delete_all)
+            else
+              update_counter(-scope.update_all(@reflection.foreign_key => nil))
+            end
+          end
         end
 
-        def we_can_set_the_inverse_on_this?(record)
-          @reflection.inverse_of
-        end
+        alias creation_attributes construct_owner_attributes
     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 781aa7ef62..9f74d57c4d 100644
--- a/activerecord/lib/active_record/associations/has_many_through_association.rb
+++ b/activerecord/lib/active_record/associations/has_many_through_association.rb
@@ -1,82 +1,89 @@
-require "active_record/associations/through_association_scope"
 require 'active_support/core_ext/object/blank'
 
 module ActiveRecord
   # = Active Record Has Many Through Association
   module Associations
     class HasManyThroughAssociation < HasManyAssociation #:nodoc:
-      include ThroughAssociationScope
+      include ThroughAssociation
 
       alias_method :new, :build
 
-      def create!(attrs = nil)
-        create_record(attrs, true)
-      end
-
-      def create(attrs = nil)
-        create_record(attrs, false)
-      end
-
-      def destroy(*records)
-        transaction do
-          delete_records(flatten_deeper(records))
-          super
-        end
-      end
-
       # Returns the size of the collection by executing a SELECT COUNT(*) query if the collection hasn't been
       # loaded and calling collection.size if it has. If it's more likely than not that the collection does
       # have a size larger than zero, and you need to fetch that collection afterwards, it'll take one fewer
       # SELECT query if you use #length.
       def size
-        return @owner.send(:read_attribute, cached_counter_attribute_name) if has_cached_counter?
-        return @target.size if loaded?
-        return count
+        if has_cached_counter?
+          @owner.send(:read_attribute, cached_counter_attribute_name)
+        elsif loaded?
+          @target.size
+        else
+          count
+        end
+      end
+
+      def <<(*records)
+        unless @owner.new_record?
+          records.flatten.each do |record|
+            raise_on_type_mismatch(record)
+            record.save! if record.new_record?
+          end
+        end
+
+        super
       end
 
       protected
-        def create_record(attrs, force = true)
-          ensure_owner_is_persisted!
 
-          transaction do
-            object = @reflection.klass.new(attrs)
-            add_record_to_target_with_callbacks(object) {|r| insert_record(object, force) }
-            object
-          end
+        def insert_record(record, validate = true)
+          return if record.new_record? && !record.save(:validate => validate)
+
+          through_association = @owner.send(@reflection.through_reflection.name)
+          through_association.create!(construct_join_attributes(record))
+
+          update_counter(1)
+          record
         end
 
+      private
+
         def target_reflection_has_associated_record?
-          if @reflection.through_reflection.macro == :belongs_to && @owner[@reflection.through_reflection.primary_key_name].blank?
+          if @reflection.through_reflection.macro == :belongs_to && @owner[@reflection.through_reflection.foreign_key].blank?
             false
           else
             true
           end
         end
 
-        def construct_find_options!(options)
-          options[:joins]   = [construct_joins] + Array.wrap(options[:joins])
-          options[:include] = @reflection.source_reflection.options[:include] if options[:include].nil? && @reflection.source_reflection.options[:include]
+        def update_through_counter?(method)
+          case method
+          when :destroy
+            !inverse_updates_counter_cache?(@reflection.through_reflection)
+          when :nullify
+            false
+          else
+            true
+          end
         end
 
-        def insert_record(record, force = true, validate = true)
-          if record.new_record?
-            if force
-              record.save!
-            else
-              return false unless record.save(:validate => validate)
-            end
-          end
+        def delete_records(records, method)
+          through = @owner.send(:association_proxy, @reflection.through_reflection.name)
+          scope   = through.scoped.where(construct_join_attributes(*records))
 
-          through_association = @owner.send(@reflection.through_reflection.name)
-          through_association.create!(construct_join_attributes(record))
-        end
+          case method
+          when :destroy
+            count = scope.destroy_all.length
+          when :nullify
+            count = scope.update_all(@reflection.source_reflection.foreign_key => nil)
+          else
+            count = scope.delete_all
+          end
 
-        # TODO - add dependent option support
-        def delete_records(records)
-          klass = @reflection.through_reflection.klass
-          records.each do |associate|
-            klass.delete_all(construct_join_attributes(associate))
+          if @reflection.through_reflection.macro == :has_many && update_through_counter?(method)
+            update_counter(-count, @reflection.through_reflection)
           end
+
+          update_counter(-count)
         end
 
         def find_target
@@ -84,16 +91,8 @@ module ActiveRecord
           scoped.all
         end
 
-        def has_cached_counter?
-          @owner.attribute_present?(cached_counter_attribute_name)
-        end
-
-        def cached_counter_attribute_name
-          "#{@reflection.name}_count"
-        end
-
         # NOTE - not sure that we can actually cope with inverses here
-        def we_can_set_the_inverse_on_this?(record)
+        def invertible_for?(record)
           false
         end
     end
diff --git a/activerecord/lib/active_record/associations/has_one_association.rb b/activerecord/lib/active_record/associations/has_one_association.rb
index c49fd6e66a..a0828dcdea 100644
--- a/activerecord/lib/active_record/associations/has_one_association.rb
+++ b/activerecord/lib/active_record/associations/has_one_association.rb
@@ -1,135 +1,65 @@
 module ActiveRecord
   # = Active Record Belongs To Has One Association
   module Associations
-    class HasOneAssociation < AssociationProxy #:nodoc:
-      def create(attrs = {}, replace_existing = true)
-        new_record(replace_existing) do |reflection|
-          attrs = merge_with_conditions(attrs)
-          reflection.create_association(attrs)
-        end
-      end
-
-      def create!(attrs = {}, replace_existing = true)
-        new_record(replace_existing) do |reflection|
-          attrs = merge_with_conditions(attrs)
-          reflection.create_association!(attrs)
-        end
-      end
+    class HasOneAssociation < SingularAssociation #:nodoc:
+      def replace(record, save = true)
+        record = check_record(record)
+        load_target
 
-      def build(attrs = {}, replace_existing = true)
-        new_record(replace_existing) do |reflection|
-          attrs = merge_with_conditions(attrs)
-          reflection.build_association(attrs)
-        end
-      end
+        @reflection.klass.transaction do
+          if @target && @target != record
+            remove_target!(@reflection.options[:dependent])
+          end
 
-      def replace(obj, dont_save = false)
-        load_target
+          if record
+            set_inverse_instance(record)
+            set_owner_attributes(record)
 
-        unless @target.nil? || @target == obj
-          if dependent? && !dont_save
-            case @reflection.options[:dependent]
-            when :delete
-              @target.delete if @target.persisted?
-              @owner.clear_association_cache
-            when :destroy
-              @target.destroy if @target.persisted?
-              @owner.clear_association_cache
-            when :nullify
-              @target[@reflection.primary_key_name] = nil
-              @target.save if @owner.persisted? && @target.persisted?
+            if @owner.persisted? && save && !record.save
+              nullify_owner_attributes(record)
+              set_owner_attributes(@target)
+              raise RecordNotSaved, "Failed to save the new associated #{@reflection.name}."
             end
-          else
-            @target[@reflection.primary_key_name] = nil
-            @target.save if @owner.persisted? && @target.persisted?
           end
         end
 
-        if obj.nil?
-          @target = nil
-        else
-          raise_on_type_mismatch(obj)
-          set_belongs_to_association_for(obj)
-          @target = (AssociationProxy === obj ? obj.target : obj)
-        end
-
-        set_inverse_instance(obj, @owner)
-        @loaded = true
-
-        unless !@owner.persisted? || obj.nil? || dont_save
-          return (obj.save ? self : false)
-        else
-          return (obj.nil? ? nil : self)
-        end
+        self.target = record
       end
 
       protected
-        def owner_quoted_id
-          if @reflection.options[:primary_key]
-            @owner.class.quote_value(@owner.send(@reflection.options[:primary_key]))
-          else
-            @owner.quoted_id
-          end
+
+        def association_scope
+          super.order(@reflection.options[:order])
         end
 
       private
-        def find_target
-          options = @reflection.options.dup.slice(:select, :order, :include, :readonly)
-
-          the_target = with_scope(:find => @scope[:find]) do
-            @reflection.klass.find(:first, options)
-          end
-          set_inverse_instance(the_target, @owner)
-          the_target
-        end
 
-        def construct_find_scope
-          if @reflection.options[:as]
-            sql =
-              "#{@reflection.quoted_table_name}.#{@reflection.options[:as]}_id = #{owner_quoted_id} AND " +
-              "#{@reflection.quoted_table_name}.#{@reflection.options[:as]}_type = #{@owner.class.quote_value(@owner.class.base_class.name.to_s)}"
-          else
-            sql = "#{@reflection.quoted_table_name}.#{@reflection.primary_key_name} = #{owner_quoted_id}"
-          end
-          sql << " AND (#{conditions})" if conditions
-          { :conditions => sql }
-        end
+        alias creation_attributes construct_owner_attributes
 
-        def construct_create_scope
-          create_scoping = {}
-          set_belongs_to_association_for(create_scoping)
-          create_scoping
+        # The reason that the save param for replace is false, if for create (not just build),
+        # is because the setting of the foreign keys is actually handled by the scoping when
+        # the record is instantiated, and so they are set straight away and do not need to be
+        # updated within replace.
+        def set_new_record(record)
+          replace(record, false)
         end
 
-        def new_record(replace_existing)
-          # Make sure we load the target first, if we plan on replacing the existing
-          # instance. Otherwise, if the target has not previously been loaded
-          # elsewhere, the instance we create will get orphaned.
-          load_target if replace_existing
-          record = @reflection.klass.send(:with_scope, :create => @scope[:create]) do
-            yield @reflection
-          end
-
-          if replace_existing
-            replace(record, true)
+        def remove_target!(method)
+          if [:delete, :destroy].include?(method)
+            @target.send(method)
           else
-            record[@reflection.primary_key_name] = @owner.id if @owner.persisted?
-            self.target = record
-            set_inverse_instance(record, @owner)
-          end
+            nullify_owner_attributes(@target)
 
-          record
-        end
-
-        def we_can_set_the_inverse_on_this?(record)
-          inverse = @reflection.inverse_of
-          return !inverse.nil?
+            if @target.persisted? && @owner.persisted? && !@target.save
+              set_owner_attributes(@target)
+              raise RecordNotSaved, "Failed to remove the existing associated #{@reflection.name}. " +
+                                    "The record failed to save when after its foreign key was set to nil."
+            end
+          end
         end
 
-        def merge_with_conditions(attrs={})
-          attrs ||= {}
-          attrs.update(@reflection.options[:conditions]) if @reflection.options[:conditions].is_a?(Hash)
-          attrs
+        def nullify_owner_attributes(record)
+          record[@reflection.foreign_key] = nil
         end
     end
   end
diff --git a/activerecord/lib/active_record/associations/has_one_through_association.rb b/activerecord/lib/active_record/associations/has_one_through_association.rb
index e8cf73976b..69771afe50 100644
--- a/activerecord/lib/active_record/associations/has_one_through_association.rb
+++ b/activerecord/lib/active_record/associations/has_one_through_association.rb
@@ -1,40 +1,34 @@
-require "active_record/associations/through_association_scope"
-
 module ActiveRecord
   # = Active Record Has One Through Association
   module Associations
     class HasOneThroughAssociation < HasOneAssociation
-      include ThroughAssociationScope
+      include ThroughAssociation
 
-      def replace(new_value)
-        create_through_record(new_value)
-        @target = new_value
+      def replace(record)
+        create_through_record(record)
+        self.target = record
       end
 
       private
 
-      def create_through_record(new_value) #nodoc:
-        klass = @reflection.through_reflection.klass
+        def create_through_record(record)
+          through_proxy  = @owner.send(:association_proxy, @reflection.through_reflection.name)
+          through_record = through_proxy.send(:load_target)
 
-        current_object = @owner.send(@reflection.through_reflection.name)
+          if through_record && !record
+            through_record.destroy
+          elsif record
+            attributes = construct_join_attributes(record)
 
-        if current_object
-           new_value ? current_object.update_attributes(construct_join_attributes(new_value)) : current_object.destroy
-        elsif new_value
-          if @owner.new_record?
-            self.target = new_value
-            through_association = @owner.send(:association_instance_get, @reflection.through_reflection.name)
-            through_association.build(construct_join_attributes(new_value))
-          else
-            @owner.send(@reflection.through_reflection.name, klass.create(construct_join_attributes(new_value)))
+            if through_record
+              through_record.update_attributes(attributes)
+            elsif @owner.new_record?
+              through_proxy.build(attributes)
+            else
+              through_proxy.create(attributes)
+            end
           end
         end
-      end
-
-    private
-      def find_target
-        scoped.first
-      end
     end
   end
 end
diff --git a/activerecord/lib/active_record/associations/singular_association.rb b/activerecord/lib/active_record/associations/singular_association.rb
new file mode 100644
index 0000000000..7f92d9712a
--- /dev/null
+++ b/activerecord/lib/active_record/associations/singular_association.rb
@@ -0,0 +1,45 @@
+module ActiveRecord
+  module Associations
+    class SingularAssociation < AssociationProxy #:nodoc:
+      def create(attributes = {})
+        new_record(:create, attributes)
+      end
+
+      def create!(attributes = {})
+        build(attributes).tap { |record| record.save! }
+      end
+
+      def build(attributes = {})
+        new_record(:build, attributes)
+      end
+
+      private
+
+        def find_target
+          scoped.first.tap { |record| set_inverse_instance(record) }
+        end
+
+        # Implemented by subclasses
+        def replace(record)
+          raise NotImplementedError
+        end
+
+        def set_new_record(record)
+          replace(record)
+        end
+
+        def check_record(record)
+          record = record.target if AssociationProxy === record
+          raise_on_type_mismatch(record) if record
+          record
+        end
+
+        def new_record(method, attributes)
+          attributes = scoped.scope_for_create.merge(attributes || {})
+          record = @reflection.send("#{method}_association", attributes)
+          set_new_record(record)
+          record
+        end
+    end
+  end
+end
diff --git a/activerecord/lib/active_record/associations/through_association.rb b/activerecord/lib/active_record/associations/through_association.rb
new file mode 100644
index 0000000000..0550bae408
--- /dev/null
+++ b/activerecord/lib/active_record/associations/through_association.rb
@@ -0,0 +1,156 @@
+module ActiveRecord
+  # = Active Record Through Association
+  module Associations
+    module ThroughAssociation
+
+      protected
+
+        def target_scope
+          super.merge(@reflection.through_reflection.klass.scoped)
+        end
+
+        def association_scope
+          scope = super.joins(construct_joins)
+          scope = add_conditions(scope)
+          unless @reflection.options[:include]
+            scope = scope.includes(@reflection.source_reflection.options[:include])
+          end
+          scope
+        end
+
+      private
+
+        # This scope affects the creation of the associated records (not the join records). At the
+        # moment we only support creating on a :through association when the source reflection is a
+        # belongs_to. Thus it's not necessary to set a foreign key on the associated record(s), so
+        # this scope has can legitimately be empty.
+        def creation_attributes
+          { }
+        end
+
+        def aliased_through_table
+          name = @reflection.through_reflection.table_name
+
+          @reflection.table_name == name ?
+            @reflection.through_reflection.klass.arel_table.alias(name + "_join") :
+            @reflection.through_reflection.klass.arel_table
+        end
+
+        def construct_owner_conditions
+          super(aliased_through_table, @reflection.through_reflection)
+        end
+
+        def construct_joins
+          right = aliased_through_table
+          left  = @reflection.klass.arel_table
+
+          conditions = []
+
+          if @reflection.source_reflection.macro == :belongs_to
+            reflection_primary_key = @reflection.source_reflection.association_primary_key
+            source_primary_key     = @reflection.source_reflection.foreign_key
+
+            if @reflection.options[:source_type]
+              column = @reflection.source_reflection.foreign_type
+              conditions <<
+                right[column].eq(@reflection.options[:source_type])
+            end
+          else
+            reflection_primary_key = @reflection.source_reflection.foreign_key
+            source_primary_key     = @reflection.source_reflection.active_record_primary_key
+
+            if @reflection.source_reflection.options[:as]
+              column = "#{@reflection.source_reflection.options[:as]}_type"
+              conditions <<
+                left[column].eq(@reflection.through_reflection.klass.name)
+            end
+          end
+
+          conditions <<
+            left[reflection_primary_key].eq(right[source_primary_key])
+
+          right.create_join(
+            right,
+            right.create_on(right.create_and(conditions)))
+        end
+
+        # Construct attributes for :through pointing to owner and associate. This is used by the
+        # methods which create and delete records on the association.
+        #
+        # We only support indirectly modifying through associations which has a belongs_to source.
+        # This is the "has_many :tags, :through => :taggings" situation, where the join model
+        # typically has a belongs_to on both side. In other words, associations which could also
+        # be represented as has_and_belongs_to_many associations.
+        #
+        # We do not support creating/deleting records on the association where the source has
+        # some other type, because this opens up a whole can of worms, and in basically any
+        # situation it is more natural for the user to just create or modify their join records
+        # directly as required.
+        def construct_join_attributes(*records)
+          if @reflection.source_reflection.macro != :belongs_to
+            raise HasManyThroughCantAssociateThroughHasOneOrManyReflection.new(@owner, @reflection)
+          end
+
+          join_attributes = {
+            @reflection.source_reflection.foreign_key =>
+              records.map { |record|
+                record.send(@reflection.source_reflection.association_primary_key)
+              }
+          }
+
+          if @reflection.options[:source_type]
+            join_attributes[@reflection.source_reflection.foreign_type] =
+              records.map { |record| record.class.base_class.name }
+          end
+
+          if records.count == 1
+            Hash[join_attributes.map { |k, v| [k, v.first] }]
+          else
+            join_attributes
+          end
+        end
+
+        # The reason that we are operating directly on the scope here (rather than passing
+        # back some arel conditions to be added to the scope) is because scope.where([x, y])
+        # has a different meaning to scope.where(x).where(y) - the first version might
+        # perform some substitution if x is a string.
+        def add_conditions(scope)
+          unless @reflection.through_reflection.klass.descends_from_active_record?
+            scope = scope.where(@reflection.through_reflection.klass.send(:type_condition))
+          end
+
+          scope = scope.where(interpolate(@reflection.source_reflection.options[:conditions]))
+          scope.where(through_conditions)
+        end
+
+        # If there is a hash of conditions then we make sure the keys are scoped to the
+        # through table name if left ambiguous.
+        def through_conditions
+          conditions = interpolate(@reflection.through_reflection.options[:conditions])
+
+          if conditions.is_a?(Hash)
+            Hash[conditions.map { |key, value|
+              unless value.is_a?(Hash) || key.to_s.include?('.')
+                key = aliased_through_table.name + '.' + key.to_s
+              end
+
+              [key, value]
+            }]
+          else
+            conditions
+          end
+        end
+
+        def stale_state
+          if @reflection.through_reflection.macro == :belongs_to
+            @owner[@reflection.through_reflection.foreign_key].to_s
+          end
+        end
+
+        def foreign_key_present?
+          @reflection.through_reflection.macro == :belongs_to &&
+          !@owner[@reflection.through_reflection.foreign_key].nil?
+        end
+    end
+  end
+end
diff --git a/activerecord/lib/active_record/associations/through_association_scope.rb b/activerecord/lib/active_record/associations/through_association_scope.rb
deleted file mode 100644
index 5dc5b0c048..0000000000
--- a/activerecord/lib/active_record/associations/through_association_scope.rb
+++ /dev/null
@@ -1,165 +0,0 @@
-module ActiveRecord
-  # = Active Record Through Association Scope
-  module Associations
-    module ThroughAssociationScope
-
-      def scoped
-        with_scope(@scope) do
-          @reflection.klass.scoped &
-          @reflection.through_reflection.klass.scoped
-        end
-      end
-
-      protected
-
-      def construct_find_scope
-        {
-          :conditions => construct_conditions,
-          :joins      => construct_joins,
-          :include    => @reflection.options[:include] || @reflection.source_reflection.options[:include],
-          :select     => construct_select,
-          :order      => @reflection.options[:order],
-          :limit      => @reflection.options[:limit],
-          :readonly   => @reflection.options[:readonly]
-        }
-      end
-
-      def construct_create_scope
-        construct_owner_attributes(@reflection)
-      end
-
-      def aliased_through_table
-        name = @reflection.through_reflection.table_name
-
-        @reflection.table_name == name ?
-          @reflection.through_reflection.klass.arel_table.alias(name + "_join") :
-          @reflection.through_reflection.klass.arel_table
-      end
-
-      # Build SQL conditions from attributes, qualified by table name.
-      def construct_conditions
-        table = aliased_through_table
-        conditions = construct_owner_attributes(@reflection.through_reflection).map do |attr, value|
-          table[attr].eq(value)
-        end
-        conditions << Arel.sql(sql_conditions) if sql_conditions
-        table.create_and(conditions)
-      end
-
-      # Associate attributes pointing to owner
-      def construct_owner_attributes(reflection)
-        if as = reflection.options[:as]
-          { "#{as}_id"   => @owner[reflection.active_record_primary_key],
-            "#{as}_type" => @owner.class.base_class.name }
-        elsif reflection.macro == :belongs_to
-          { reflection.klass.primary_key => @owner[reflection.primary_key_name] }
-        else
-          { reflection.primary_key_name => @owner[reflection.active_record_primary_key] }
-        end
-      end
-
-      def construct_from
-        @reflection.table_name
-      end
-
-      def construct_select(custom_select = nil)
-        distinct = "DISTINCT " if @reflection.options[:uniq]
-        custom_select || @reflection.options[:select] || "#{distinct}#{@reflection.quoted_table_name}.*"
-      end
-
-      def construct_joins
-        right = aliased_through_table
-        left  = @reflection.klass.arel_table
-
-        conditions = []
-
-        if @reflection.source_reflection.macro == :belongs_to
-          reflection_primary_key = @reflection.source_reflection.options[:primary_key] ||
-                                   @reflection.klass.primary_key
-          source_primary_key     = @reflection.source_reflection.primary_key_name
-          if @reflection.options[:source_type]
-            column = @reflection.source_reflection.options[:foreign_type]
-            conditions <<
-              right[column].eq(@reflection.options[:source_type])
-          end
-        else
-          reflection_primary_key = @reflection.source_reflection.primary_key_name
-          source_primary_key     = @reflection.source_reflection.options[:primary_key] ||
-                                   @reflection.through_reflection.klass.primary_key
-          if @reflection.source_reflection.options[:as]
-            column = "#{@reflection.source_reflection.options[:as]}_type"
-            conditions <<
-              left[column].eq(@reflection.through_reflection.klass.name)
-          end
-        end
-
-        conditions <<
-          left[reflection_primary_key].eq(right[source_primary_key])
-
-        right.create_join(
-          right,
-          right.create_on(right.create_and(conditions)))
-      end
-
-      # Construct attributes for :through pointing to owner and associate.
-      def construct_join_attributes(associate)
-        # TODO: revisit this to allow it for deletion, supposing dependent option is supported
-        raise ActiveRecord::HasManyThroughCantAssociateThroughHasOneOrManyReflection.new(@owner, @reflection) if [:has_one, :has_many].include?(@reflection.source_reflection.macro)
-
-        join_attributes = construct_owner_attributes(@reflection.through_reflection).merge(@reflection.source_reflection.primary_key_name => associate.id)
-
-        if @reflection.options[:source_type]
-          join_attributes.merge!(@reflection.source_reflection.options[:foreign_type] => associate.class.base_class.name)
-        end
-
-        if @reflection.through_reflection.options[:conditions].is_a?(Hash)
-          join_attributes.merge!(@reflection.through_reflection.options[:conditions])
-        end
-
-        join_attributes
-      end
-
-      def conditions
-        @conditions = build_conditions unless defined?(@conditions)
-        @conditions
-      end
-
-      def build_conditions
-        association_conditions = @reflection.options[:conditions]
-        through_conditions = build_through_conditions
-        source_conditions = @reflection.source_reflection.options[:conditions]
-        uses_sti = !@reflection.through_reflection.klass.descends_from_active_record?
-
-        if association_conditions || through_conditions || source_conditions || uses_sti
-          all = []
-
-          [association_conditions, source_conditions].each do |conditions|
-            all << interpolate_sql(sanitize_sql(conditions)) if conditions
-          end
-
-          all << through_conditions  if through_conditions
-          all << build_sti_condition if uses_sti
-
-          all.map { |sql| "(#{sql})" } * ' AND '
-        end
-      end
-
-      def build_through_conditions
-        conditions = @reflection.through_reflection.options[:conditions]
-        if conditions.is_a?(Hash)
-          interpolate_sql(@reflection.through_reflection.klass.send(:sanitize_sql, conditions)).gsub(
-            @reflection.quoted_table_name,
-            @reflection.through_reflection.quoted_table_name)
-        elsif conditions
-          interpolate_sql(sanitize_sql(conditions))
-        end
-      end
-
-      def build_sti_condition
-        @reflection.through_reflection.klass.send(:type_condition).to_sql
-      end
-
-      alias_method :sql_conditions, :conditions
-    end
-  end
-end