Rewrote AssociationPreload.

16 files changed, 565 insertions, 439 deletions

diff --git a/activerecord/lib/active_record/association_preload.rb b/activerecord/lib/active_record/association_preload.rb
deleted file mode 100644
index a34a73cf5d..0000000000
--- a/activerecord/lib/active_record/association_preload.rb
+++ /dev/null
@@ -1,433 +0,0 @@
-require 'active_support/core_ext/array/wrap'
-require 'active_support/core_ext/enumerable'
-
-module ActiveRecord
-  # See ActiveRecord::AssociationPreload::ClassMethods for documentation.
-  module AssociationPreload #:nodoc:
-    extend ActiveSupport::Concern
-
-    # Implements the details of eager loading of Active Record associations.
-    # Application developers should not use this module directly.
-    #
-    # <tt>ActiveRecord::Base</tt> is extended with this module. The source code in
-    # <tt>ActiveRecord::Base</tt> references methods defined in this module.
-    #
-    # Note that 'eager loading' and 'preloading' are actually the same thing.
-    # However, there are two different eager loading strategies.
-    #
-    # The first one is by using table joins. This was only strategy available
-    # prior to Rails 2.1. Suppose that you have an Author model with columns
-    # 'name' and 'age', and a Book model with columns 'name' and 'sales'. Using
-    # this strategy, Active Record would try to retrieve all data for an author
-    # and all of its books via a single query:
-    #
-    #   SELECT * FROM authors
-    #   LEFT OUTER JOIN books ON authors.id = books.id
-    #   WHERE authors.name = 'Ken Akamatsu'
-    #
-    # However, this could result in many rows that contain redundant data. After
-    # having received the first row, we already have enough data to instantiate
-    # the Author object. In all subsequent rows, only the data for the joined
-    # 'books' table is useful; the joined 'authors' data is just redundant, and
-    # processing this redundant data takes memory and CPU time. The problem
-    # quickly becomes worse and worse as the level of eager loading increases
-    # (i.e. if Active Record is to eager load the associations' associations as
-    # well).
-    #
-    # The second strategy is to use multiple database queries, one for each
-    # level of association. Since Rails 2.1, this is the default strategy. In
-    # situations where a table join is necessary (e.g. when the +:conditions+
-    # option references an association's column), it will fallback to the table
-    # join strategy.
-    #
-    # See also ActiveRecord::Associations::ClassMethods, which explains eager
-    # loading in a more high-level (application developer-friendly) manner.
-    module ClassMethods
-      protected
-
-      # Eager loads the named associations for the given Active Record record(s).
-      #
-      # In this description, 'association name' shall refer to the name passed
-      # to an association creation method. For example, a model that specifies
-      # <tt>belongs_to :author</tt>, <tt>has_many :buyers</tt> has association
-      # names +:author+ and +:buyers+.
-      #
-      # == Parameters
-      # +records+ is an array of ActiveRecord::Base. This array needs not be flat,
-      # i.e. +records+ itself may also contain arrays of records. In any case,
-      # +preload_associations+ will preload the all associations records by
-      # flattening +records+.
-      #
-      # +associations+ specifies one or more associations that you want to
-      # preload. It may be:
-      # - a Symbol or a String which specifies a single association name. For
-      #   example, specifying +:books+ allows this method to preload all books
-      #   for an Author.
-      # - an Array which specifies multiple association names. This array
-      #   is processed recursively. For example, specifying <tt>[:avatar, :books]</tt>
-      #   allows this method to preload an author's avatar as well as all of his
-      #   books.
-      # - a Hash which specifies multiple association names, as well as
-      #   association names for the to-be-preloaded association objects. For
-      #   example, specifying <tt>{ :author => :avatar }</tt> will preload a
-      #   book's author, as well as that author's avatar.
-      #
-      # +:associations+ has the same format as the +:include+ option for
-      # <tt>ActiveRecord::Base.find</tt>. So +associations+ could look like this:
-      #
-      #   :books
-      #   [ :books, :author ]
-      #   { :author => :avatar }
-      #   [ :books, { :author => :avatar } ]
-      #
-      # +preload_options+ contains options that will be passed to ActiveRecord::Base#find
-      # (which is called under the hood for preloading records). But it is passed
-      # only one level deep in the +associations+ argument, i.e. it's not passed
-      # to the child associations when +associations+ is a Hash.
-      def preload_associations(records, associations, preload_options={})
-        records = Array.wrap(records).compact.uniq
-        return if records.empty?
-        case associations
-        when Array then associations.each {|association| preload_associations(records, association, preload_options)}
-        when Symbol, String then preload_one_association(records, associations.to_sym, preload_options)
-        when Hash then
-          associations.each do |parent, child|
-            raise "parent must be an association name" unless parent.is_a?(String) || parent.is_a?(Symbol)
-            preload_associations(records, parent, preload_options)
-            reflection = reflections[parent]
-            parents = records.sum { |record| Array.wrap(record.send(reflection.name)) }
-            unless parents.empty?
-              parents.first.class.preload_associations(parents, child)
-            end
-          end
-        end
-      end
-
-      private
-
-      # Preloads a specific named association for the given records. This is
-      # called by +preload_associations+ as its base case.
-      def preload_one_association(records, association, preload_options={})
-        class_to_reflection = {}
-        # Not all records have the same class, so group then preload
-        # group on the reflection itself so that if various subclass share the same association then
-        # we do not split them unnecessarily
-        records.group_by { |record| class_to_reflection[record.class] ||= record.class.reflections[association]}.each do |reflection, _records|
-          raise ConfigurationError, "Association named '#{ association }' was not found; perhaps you misspelled it?" unless reflection
-
-          # 'reflection.macro' can return 'belongs_to', 'has_many', etc. Thus,
-          # the following could call 'preload_belongs_to_association',
-          # 'preload_has_many_association', etc.
-          send("preload_#{reflection.macro}_association", _records, reflection, preload_options)
-        end
-      end
-
-      def add_preloaded_records_to_collection(parent_records, reflection_name, associated_record)
-        parent_records.each do |parent_record|
-          association = parent_record.association(reflection_name)
-          association.loaded!
-          association.target.concat(Array.wrap(associated_record))
-          association.set_inverse_instance(associated_record)
-        end
-      end
-
-      def add_preloaded_record_to_collection(parent_records, reflection_name, associated_record)
-        parent_records.each do |parent_record|
-          parent_record.association(reflection_name).target = associated_record
-        end
-      end
-
-      def set_association_collection_records(id_to_parent_map, reflection_name, associated_records, key)
-        associated_records.each do |associated_record|
-          parent_records = id_to_parent_map[associated_record[key].to_s]
-          add_preloaded_records_to_collection(parent_records, reflection_name, associated_record)
-        end
-      end
-
-      def set_association_single_records(id_to_record_map, reflection_name, associated_records, key)
-        seen_keys = {}
-        associated_records.each do |associated_record|
-          seen_key = associated_record[key].to_s
-
-          #this is a has_one or belongs_to: there should only be one record.
-          #Unfortunately we can't (in portable way) ask the database for
-          #'all records where foo_id in (x,y,z), but please
-          # only one row per distinct foo_id' so this where we enforce that
-          next if seen_keys.key? seen_key
-
-          seen_keys[seen_key] = true
-          mapped_records = id_to_record_map[seen_key]
-          mapped_records.each do |mapped_record|
-            association_proxy = mapped_record.association(reflection_name)
-            association_proxy.target = associated_record
-            association_proxy.send(:set_inverse_instance, associated_record)
-          end
-        end
-
-        id_to_record_map.each do |id, records|
-          next if seen_keys.include?(id)
-          add_preloaded_record_to_collection(records, reflection_name, nil)
-        end
-      end
-
-      # Given a collection of Active Record objects, constructs a Hash which maps
-      # the objects' IDs to the relevant objects. Returns a 2-tuple
-      # <tt>(id_to_record_map, ids)</tt> where +id_to_record_map+ is the Hash,
-      # and +ids+ is an Array of record IDs.
-      def construct_id_map(records, primary_key=nil)
-        records.group_by do |record|
-          primary_key ||= record.class.primary_key
-          record[primary_key].to_s
-        end
-      end
-
-      def preload_has_and_belongs_to_many_association(records, reflection, preload_options={})
-
-        left = reflection.klass.arel_table
-
-        id_to_record_map = construct_id_map(records)
-
-        records.each { |record| record.association(reflection.name).loaded! }
-        options = reflection.options
-
-        right = Arel::Table.new(options[:join_table]).alias('t0')
-
-        join_condition = left[reflection.klass.primary_key].eq(
-          right[reflection.association_foreign_key])
-
-        join = left.create_join(right, left.create_on(join_condition))
-        select = [
-          # FIXME: options[:select] is always nil in the tests.  Do we really
-          # need it?
-          options[:select] || left[Arel.star],
-          right[reflection.foreign_key].as(
-            Arel.sql('the_parent_record_id'))
-        ]
-
-        associated_records_proxy = reflection.klass.unscoped.
-            includes(options[:include]).
-            order(options[:order])
-
-        associated_records_proxy.joins_values = [join]
-        associated_records_proxy.select_values = select
-
-        custom_conditions = append_conditions(reflection, preload_options)
-
-        klass = associated_records_proxy.klass
-
-        associated_records(id_to_record_map.keys) { |some_ids|
-          method     = in_or_equal(some_ids)
-          conditions = right.create_and(
-            [right[reflection.foreign_key].send(*method)] +
-            custom_conditions)
-
-          relation = associated_records_proxy.where(conditions)
-          klass.connection.select_all(relation.arel.to_sql, 'SQL', relation.bind_values)
-        }.map! { |row|
-          parent_records = id_to_record_map[row['the_parent_record_id'].to_s]
-          associated_record = klass.instantiate row
-          add_preloaded_records_to_collection(
-            parent_records, reflection.name, associated_record)
-          associated_record
-        }
-      end
-
-      def preload_has_one_association(records, reflection, preload_options={})
-        return if records.first.association(reflection.name).loaded?
-        id_to_record_map = construct_id_map(records, reflection.options[:primary_key])
-        options = reflection.options
-
-        add_preloaded_record_to_collection(records, reflection.name, nil)
-
-        if options[:through]
-          through_records = preload_through_records(records, reflection, options[:through])
-
-          unless through_records.empty?
-            through_reflection = reflections[options[:through]]
-            through_primary_key = through_reflection.foreign_key
-            source = reflection.source_reflection.name
-            through_records.first.class.preload_associations(through_records, source)
-            if through_reflection.macro == :belongs_to
-              id_to_record_map    = construct_id_map(records, through_primary_key)
-              through_primary_key = through_reflection.klass.primary_key
-            end
-
-            through_records.each do |through_record|
-              add_preloaded_record_to_collection(id_to_record_map[through_record[through_primary_key].to_s],
-                                                 reflection.name, through_record.send(source))
-            end
-          end
-        else
-          set_association_single_records(id_to_record_map, reflection.name, find_associated_records(id_to_record_map.keys, reflection, preload_options), reflection.foreign_key)
-        end
-      end
-
-      def preload_has_many_association(records, reflection, preload_options={})
-        return if records.first.send(reflection.name).loaded?
-        options = reflection.options
-
-        foreign_key = reflection.through_reflection_foreign_key
-        id_to_record_map = construct_id_map(records, foreign_key || reflection.options[:primary_key])
-        records.each { |record| record.association(reflection.name).loaded! }
-
-        if options[:through]
-          through_records = preload_through_records(records, reflection, options[:through])
-          unless through_records.empty?
-            source = reflection.source_reflection.name
-            through_records.first.class.preload_associations(through_records, source, options)
-            through_records.each do |through_record|
-              through_record_id = through_record[reflection.through_reflection_primary_key].to_s
-              add_preloaded_records_to_collection(id_to_record_map[through_record_id], reflection.name, through_record.send(source))
-            end
-            records.each { |record| record.send(reflection.name).target.uniq! } if options[:uniq]
-          end
-
-        else
-          set_association_collection_records(id_to_record_map, reflection.name, find_associated_records(id_to_record_map.keys, reflection, preload_options),
-                                             reflection.foreign_key)
-        end
-      end
-
-      def preload_through_records(records, reflection, through_association)
-        if reflection.options[:source_type]
-          interface = reflection.source_reflection.foreign_type
-          preload_options = {:conditions => ["#{connection.quote_column_name interface} = ?", reflection.options[:source_type]]}
-
-          records.compact!
-          records.first.class.preload_associations(records, through_association, preload_options)
-
-          # Dont cache the association - we would only be caching a subset
-          records.map { |record|
-            proxy = record.association(through_association)
-
-            if proxy.respond_to?(:target)
-              Array.wrap(proxy.target).tap { proxy.reset }
-            else # this is a has_one :through reflection
-              [proxy].compact
-            end
-          }.flatten(1)
-        else
-          options = {}
-          options[:include] = reflection.options[:include] || reflection.options[:source] if reflection.options[:conditions]
-          options[:order] = reflection.options[:order]
-          options[:conditions] = reflection.options[:conditions]
-          records.first.class.preload_associations(records, through_association, options)
-
-          records.map { |record|
-            Array.wrap(record.send(through_association))
-          }.flatten(1)
-        end
-      end
-
-      def preload_belongs_to_association(records, reflection, preload_options={})
-        return if records.first.association(reflection.name).loaded?
-        options = reflection.options
-
-        klasses_and_ids = {}
-
-        if options[:polymorphic]
-          # Construct a mapping from klass to a list of ids to load and a mapping of those ids back
-          # to their parent_records
-          records.each do |record|
-            if klass = record.send(reflection.foreign_type)
-              klass_id = record.send(reflection.foreign_key)
-              if klass_id
-                id_map = klasses_and_ids[klass.constantize] ||= {}
-                (id_map[klass_id.to_s] ||= []) << record
-              end
-            end
-          end
-        else
-          id_map = records.group_by do |record|
-            key = record.send(reflection.foreign_key)
-            key && key.to_s
-          end
-          klasses_and_ids[reflection.klass] = id_map unless id_map.empty?
-        end
-
-        klasses_and_ids.each do |klass, _id_map|
-          primary_key = (reflection.options[:primary_key] || klass.primary_key).to_s
-          keys        = _id_map.keys.compact
-
-          unless keys.empty?
-            table       = klass.arel_table
-            method      = in_or_equal(keys)
-            conditions  = table[primary_key].send(*method)
-
-            custom_conditions = append_conditions(reflection, preload_options)
-            conditions = custom_conditions.inject(conditions) do |ast, cond|
-              ast.and cond
-            end
-
-            associated_records = klass.unscoped.where(conditions).apply_finder_options(options.slice(:include, :select, :joins, :order)).to_a
-          else
-            associated_records = []
-          end
-
-          set_association_single_records(_id_map, reflection.name, associated_records, primary_key)
-        end
-      end
-
-      def find_associated_records(ids, reflection, preload_options)
-        options    = reflection.options
-        table      = reflection.klass.arel_table
-
-        conditions = []
-
-        key = reflection.foreign_key
-
-        if interface = reflection.options[:as]
-          key = "#{interface}_id"
-          conditions << table["#{interface}_type"].eq(base_class.sti_name)
-        end
-
-        conditions.concat append_conditions(reflection, preload_options)
-
-        find_options = {
-          :select => preload_options[:select] || options[:select] || table[Arel.star],
-          :include => preload_options[:include] || options[:include],
-          :joins => options[:joins],
-          :group => preload_options[:group] || options[:group],
-          :order => preload_options[:order] || options[:order]
-        }
-
-        associated_records(ids) do |some_ids|
-          method = in_or_equal(some_ids)
-          where = table.create_and(conditions + [table[key].send(*method)])
-
-          reflection.klass.scoped.apply_finder_options(find_options.merge(:conditions => where)).to_a
-        end
-      end
-
-      def process_conditions(conditions, klass = self)
-        if conditions.respond_to?(:to_proc)
-          conditions = instance_eval(&conditions)
-        end
-
-        klass.send(:sanitize_sql, conditions)
-      end
-
-      def append_conditions(reflection, preload_options)
-        [
-          ('(' + process_conditions(reflection.options[:conditions], reflection.klass) + ')' if reflection.options[:conditions]),
-          ('(' + process_conditions(preload_options[:conditions]) + ')' if preload_options[:conditions]),
-        ].compact.map { |x| Arel.sql x }
-      end
-
-      def in_or_equal(ids)
-        ids.length == 1 ? ['eq', ids.first] : ['in', ids]
-      end
-
-      # Some databases impose a limit on the number of ids in a list (in Oracle its 1000)
-      # Make several smaller queries if necessary or make one query if the adapter supports it
-      def associated_records(ids)
-        in_clause_length = connection.in_clause_length || ids.size
-        records = []
-        ids.each_slice(in_clause_length) do |some_ids|
-          records.concat yield(some_ids)
-        end
-        records
-      end
-    end
-  end
-end
diff --git a/activerecord/lib/active_record/associations.rb b/activerecord/lib/active_record/associations.rb
index e0b4db498d..0fa9bcd908 100644
--- a/activerecord/lib/active_record/associations.rb
+++ b/activerecord/lib/active_record/associations.rb
@@ -143,6 +143,8 @@ module ActiveRecord
       autoload :HasAndBelongsToMany, 'active_record/associations/builder/has_and_belongs_to_many'
     end
 
+    autoload :Preloader, 'active_record/associations/preloader'
+
     # Clears out the association cache.
     def clear_association_cache #:nodoc:
       @association_cache.clear if persisted?
diff --git a/activerecord/lib/active_record/associations/preloader.rb b/activerecord/lib/active_record/associations/preloader.rb
new file mode 100644
index 0000000000..fafed94ff2
--- /dev/null
+++ b/activerecord/lib/active_record/associations/preloader.rb
@@ -0,0 +1,177 @@
+module ActiveRecord
+  module Associations
+    # Implements the details of eager loading of Active Record associations.
+    #
+    # Note that 'eager loading' and 'preloading' are actually the same thing.
+    # However, there are two different eager loading strategies.
+    #
+    # The first one is by using table joins. This was only strategy available
+    # prior to Rails 2.1. Suppose that you have an Author model with columns
+    # 'name' and 'age', and a Book model with columns 'name' and 'sales'. Using
+    # this strategy, Active Record would try to retrieve all data for an author
+    # and all of its books via a single query:
+    #
+    #   SELECT * FROM authors
+    #   LEFT OUTER JOIN books ON authors.id = books.id
+    #   WHERE authors.name = 'Ken Akamatsu'
+    #
+    # However, this could result in many rows that contain redundant data. After
+    # having received the first row, we already have enough data to instantiate
+    # the Author object. In all subsequent rows, only the data for the joined
+    # 'books' table is useful; the joined 'authors' data is just redundant, and
+    # processing this redundant data takes memory and CPU time. The problem
+    # quickly becomes worse and worse as the level of eager loading increases
+    # (i.e. if Active Record is to eager load the associations' associations as
+    # well).
+    #
+    # The second strategy is to use multiple database queries, one for each
+    # level of association. Since Rails 2.1, this is the default strategy. In
+    # situations where a table join is necessary (e.g. when the +:conditions+
+    # option references an association's column), it will fallback to the table
+    # join strategy.
+    class Preloader #:nodoc:
+      autoload :Association,           'active_record/associations/preloader/association'
+      autoload :SingularAssociation,   'active_record/associations/preloader/singular_association'
+      autoload :CollectionAssociation, 'active_record/associations/preloader/collection_association'
+      autoload :ThroughAssociation,    'active_record/associations/preloader/through_association'
+
+      autoload :HasMany,             'active_record/associations/preloader/has_many'
+      autoload :HasManyThrough,      'active_record/associations/preloader/has_many_through'
+      autoload :HasOne,              'active_record/associations/preloader/has_one'
+      autoload :HasOneThrough,       'active_record/associations/preloader/has_one_through'
+      autoload :HasAndBelongsToMany, 'active_record/associations/preloader/has_and_belongs_to_many'
+      autoload :BelongsTo,           'active_record/associations/preloader/belongs_to'
+
+      attr_reader :records, :associations, :options, :model
+
+      # Eager loads the named associations for the given Active Record record(s).
+      #
+      # In this description, 'association name' shall refer to the name passed
+      # to an association creation method. For example, a model that specifies
+      # <tt>belongs_to :author</tt>, <tt>has_many :buyers</tt> has association
+      # names +:author+ and +:buyers+.
+      #
+      # == Parameters
+      # +records+ is an array of ActiveRecord::Base. This array needs not be flat,
+      # i.e. +records+ itself may also contain arrays of records. In any case,
+      # +preload_associations+ will preload the all associations records by
+      # flattening +records+.
+      #
+      # +associations+ specifies one or more associations that you want to
+      # preload. It may be:
+      # - a Symbol or a String which specifies a single association name. For
+      #   example, specifying +:books+ allows this method to preload all books
+      #   for an Author.
+      # - an Array which specifies multiple association names. This array
+      #   is processed recursively. For example, specifying <tt>[:avatar, :books]</tt>
+      #   allows this method to preload an author's avatar as well as all of his
+      #   books.
+      # - a Hash which specifies multiple association names, as well as
+      #   association names for the to-be-preloaded association objects. For
+      #   example, specifying <tt>{ :author => :avatar }</tt> will preload a
+      #   book's author, as well as that author's avatar.
+      #
+      # +:associations+ has the same format as the +:include+ option for
+      # <tt>ActiveRecord::Base.find</tt>. So +associations+ could look like this:
+      #
+      #   :books
+      #   [ :books, :author ]
+      #   { :author => :avatar }
+      #   [ :books, { :author => :avatar } ]
+      #
+      # +options+ contains options that will be passed to ActiveRecord::Base#find
+      # (which is called under the hood for preloading records). But it is passed
+      # only one level deep in the +associations+ argument, i.e. it's not passed
+      # to the child associations when +associations+ is a Hash.
+      def initialize(records, associations, options = {})
+        @records      = Array.wrap(records).compact.uniq
+        @associations = Array.wrap(associations)
+        @options      = options
+      end
+
+      def run
+        unless records.empty?
+          associations.each { |association| preload(association) }
+        end
+      end
+
+      private
+
+      def preload(association)
+        case association
+        when Hash
+          preload_hash(association)
+        when String, Symbol
+          preload_one(association.to_sym)
+        else
+          raise ArgumentError, "#{association.inspect} was not recognised for preload"
+        end
+      end
+
+      def preload_hash(association)
+        association.each do |parent, child|
+          Preloader.new(records, parent, options).run
+          Preloader.new(records.map { |record| record.send(parent) }.flatten, child).run
+        end
+      end
+
+      # Not all records have the same class, so group then preload group on the reflection
+      # itself so that if various subclass share the same association then we do not split
+      # them unnecessarily
+      #
+      # Additionally, polymorphic belongs_to associations can have multiple associated
+      # classes, depending on the polymorphic_type field. So we group by the classes as
+      # well.
+      def preload_one(association)
+        grouped_records(association).each do |reflection, klasses|
+          klasses.each do |klass, records|
+            preloader_for(reflection).new(klass, records, reflection, options).run
+          end
+        end
+      end
+
+      def grouped_records(association)
+        Hash[
+          records_by_reflection(association).map do |reflection, records|
+            [reflection, records.group_by { |record| association_klass(reflection, record) }]
+          end
+        ]
+      end
+
+      def records_by_reflection(association)
+        records.group_by do |record|
+          reflection = record.class.reflections[association]
+
+          unless reflection
+            raise ActiveRecord::ConfigurationError, "Association named '#{association}' was not found; " \
+                                                    "perhaps you misspelled it?"
+          end
+
+          reflection
+        end
+      end
+
+      def association_klass(reflection, record)
+        if reflection.macro == :belongs_to && reflection.options[:polymorphic]
+          klass = record.send(reflection.foreign_type)
+          klass && klass.constantize
+        else
+          reflection.klass
+        end
+      end
+
+      def preloader_for(reflection)
+        case reflection.macro
+        when :has_many
+          reflection.options[:through] ? HasManyThrough : HasMany
+        when :has_one
+          reflection.options[:through] ? HasOneThrough : HasOne
+        when :has_and_belongs_to_many
+          HasAndBelongsToMany
+        when :belongs_to
+          BelongsTo
+        end
+      end
+    end
+  end
+end
diff --git a/activerecord/lib/active_record/associations/preloader/association.rb b/activerecord/lib/active_record/associations/preloader/association.rb
new file mode 100644
index 0000000000..7256dd5288
--- /dev/null
+++ b/activerecord/lib/active_record/associations/preloader/association.rb
@@ -0,0 +1,126 @@
+module ActiveRecord
+  module Associations
+    class Preloader
+      class Association #:nodoc:
+        attr_reader :owners, :reflection, :preload_options, :model, :klass
+
+        def initialize(klass, owners, reflection, preload_options)
+          @klass           = klass
+          @owners          = owners
+          @reflection      = reflection
+          @preload_options = preload_options || {}
+          @model           = owners.first && owners.first.class
+          @scoped          = nil
+          @owners_by_key   = nil
+        end
+
+        def run
+          unless owners.first.association(reflection.name).loaded?
+            preload
+          end
+        end
+
+        def preload
+          raise NotImplementedError
+        end
+
+        def scoped
+          @scoped ||= build_scope
+        end
+
+        def records_for(ids)
+          scoped.where(association_key.in(ids))
+        end
+
+        def table
+          klass.arel_table
+        end
+
+        # The name of the key on the associated records
+        def association_key_name
+          raise NotImplementedError
+        end
+
+        # This is overridden by HABTM as the condition should be on the foreign_key column in
+        # the join table
+        def association_key
+          table[association_key_name]
+        end
+
+        # The name of the key on the model which declares the association
+        def owner_key_name
+          raise NotImplementedError
+        end
+
+        # We're converting to a string here because postgres will return the aliased association
+        # key in a habtm as a string (for whatever reason)
+        def owners_by_key
+          @owners_by_key ||= owners.group_by do |owner|
+            key = owner[owner_key_name]
+            key && key.to_s
+          end
+        end
+
+        def options
+          reflection.options
+        end
+
+        private
+
+        def associated_records_by_owner
+          owner_keys = owners.map { |owner| owner[owner_key_name] }.compact.uniq
+
+          if klass.nil? || owner_keys.empty?
+            records = []
+          else
+            # Some databases impose a limit on the number of ids in a list (in Oracle it's 1000)
+            # Make several smaller queries if necessary or make one query if the adapter supports it
+            sliced  = owner_keys.each_slice(model.connection.in_clause_length || owner_keys.size)
+            records = sliced.map { |slice| records_for(slice) }.flatten
+          end
+
+          # Each record may have multiple owners, and vice-versa
+          records_by_owner = Hash[owners.map { |owner| [owner, []] }]
+          records.each do |record|
+            owner_key = record[association_key_name].to_s
+
+            owners_by_key[owner_key].each do |owner|
+              records_by_owner[owner] << record
+            end
+          end
+          records_by_owner
+        end
+
+        def build_scope
+          scope = klass.scoped
+
+          scope = scope.where(process_conditions(options[:conditions]))
+          scope = scope.where(process_conditions(preload_options[:conditions]))
+
+          scope = scope.select(preload_options[:select] || options[:select] || table[Arel.star])
+          scope = scope.includes(preload_options[:include] || options[:include])
+
+          if options[:as]
+            scope = scope.where(
+              klass.table_name => {
+                reflection.type => model.base_class.sti_name
+              }
+            )
+          end
+
+          scope
+        end
+
+        def process_conditions(conditions)
+          if conditions.respond_to?(:to_proc)
+            conditions = klass.send(:instance_eval, &conditions)
+          end
+
+          if conditions
+            klass.send(:sanitize_sql, conditions)
+          end
+        end
+      end
+    end
+  end
+end
diff --git a/activerecord/lib/active_record/associations/preloader/belongs_to.rb b/activerecord/lib/active_record/associations/preloader/belongs_to.rb
new file mode 100644
index 0000000000..5091d4717a
--- /dev/null
+++ b/activerecord/lib/active_record/associations/preloader/belongs_to.rb
@@ -0,0 +1,17 @@
+module ActiveRecord
+  module Associations
+    class Preloader
+      class BelongsTo < SingularAssociation #:nodoc:
+
+        def association_key_name
+          reflection.options[:primary_key] || klass && klass.primary_key
+        end
+
+        def owner_key_name
+          reflection.foreign_key
+        end
+
+      end
+    end
+  end
+end
diff --git a/activerecord/lib/active_record/associations/preloader/collection_association.rb b/activerecord/lib/active_record/associations/preloader/collection_association.rb
new file mode 100644
index 0000000000..c248aeaaf6
--- /dev/null
+++ b/activerecord/lib/active_record/associations/preloader/collection_association.rb
@@ -0,0 +1,24 @@
+module ActiveRecord
+  module Associations
+    class Preloader
+      class CollectionAssociation < Association #:nodoc:
+
+        private
+
+        def build_scope
+          super.order(preload_options[:order] || options[:order])
+        end
+
+        def preload
+          associated_records_by_owner.each do |owner, records|
+            association = owner.association(reflection.name)
+            association.loaded!
+            association.target.concat(records)
+            records.each { |record| association.set_inverse_instance(record) }
+          end
+        end
+
+      end
+    end
+  end
+end
diff --git a/activerecord/lib/active_record/associations/preloader/has_and_belongs_to_many.rb b/activerecord/lib/active_record/associations/preloader/has_and_belongs_to_many.rb
new file mode 100644
index 0000000000..e794f05340
--- /dev/null
+++ b/activerecord/lib/active_record/associations/preloader/has_and_belongs_to_many.rb
@@ -0,0 +1,58 @@
+module ActiveRecord
+  module Associations
+    class Preloader
+      class HasAndBelongsToMany < CollectionAssociation #:nodoc:
+        attr_reader :join_table
+
+        def initialize(klass, records, reflection, preload_options)
+          super
+          @join_table = Arel::Table.new(options[:join_table]).alias('t0')
+        end
+
+        # Unlike the other associations, we want to get a raw array of rows so that we can
+        # access the aliased column on the join table
+        def records_for(ids)
+          scope = super
+          klass.connection.select_all(scope.arel.to_sql, 'SQL', scope.bind_values)
+        end
+
+        def owner_key_name
+          reflection.active_record_primary_key
+        end
+
+        def association_key_name
+          'ar_association_key_name'
+        end
+
+        def association_key
+          join_table[reflection.foreign_key]
+        end
+
+        private
+
+        # Once we have used the join table column (in super), we manually instantiate the
+        # actual records
+        def associated_records_by_owner
+          super.each do |owner_key, rows|
+            rows.map! { |row| klass.instantiate(row) }
+          end
+        end
+
+        def build_scope
+          super.joins(join).select(join_select)
+        end
+
+        def join_select
+          association_key.as(Arel.sql(association_key_name))
+        end
+
+        def join
+          condition = table[reflection.association_primary_key].eq(
+            join_table[reflection.association_foreign_key])
+
+          table.create_join(join_table, table.create_on(condition))
+        end
+      end
+    end
+  end
+end
diff --git a/activerecord/lib/active_record/associations/preloader/has_many.rb b/activerecord/lib/active_record/associations/preloader/has_many.rb
new file mode 100644
index 0000000000..3ea91a8c11
--- /dev/null
+++ b/activerecord/lib/active_record/associations/preloader/has_many.rb
@@ -0,0 +1,17 @@
+module ActiveRecord
+  module Associations
+    class Preloader
+      class HasMany < CollectionAssociation #:nodoc:
+
+        def association_key_name
+          reflection.foreign_key
+        end
+
+        def owner_key_name
+          reflection.active_record_primary_key
+        end
+
+      end
+    end
+  end
+end
diff --git a/activerecord/lib/active_record/associations/preloader/has_many_through.rb b/activerecord/lib/active_record/associations/preloader/has_many_through.rb
new file mode 100644
index 0000000000..c6e9ede356
--- /dev/null
+++ b/activerecord/lib/active_record/associations/preloader/has_many_through.rb
@@ -0,0 +1,15 @@
+module ActiveRecord
+  module Associations
+    class Preloader
+      class HasManyThrough < CollectionAssociation #:nodoc:
+        include ThroughAssociation
+
+        def associated_records_by_owner
+          super.each do |owner, records|
+            records.uniq! if options[:uniq]
+          end
+        end
+      end
+    end
+  end
+end
diff --git a/activerecord/lib/active_record/associations/preloader/has_one.rb b/activerecord/lib/active_record/associations/preloader/has_one.rb
new file mode 100644
index 0000000000..848448bb48
--- /dev/null
+++ b/activerecord/lib/active_record/associations/preloader/has_one.rb
@@ -0,0 +1,23 @@
+module ActiveRecord
+  module Associations
+    class Preloader
+      class HasOne < SingularAssociation #:nodoc:
+
+        def association_key_name
+          reflection.foreign_key
+        end
+
+        def owner_key_name
+          reflection.active_record_primary_key
+        end
+
+        private
+
+        def build_scope
+          super.order(preload_options[:order] || options[:order])
+        end
+
+      end
+    end
+  end
+end
diff --git a/activerecord/lib/active_record/associations/preloader/has_one_through.rb b/activerecord/lib/active_record/associations/preloader/has_one_through.rb
new file mode 100644
index 0000000000..f063f85574
--- /dev/null
+++ b/activerecord/lib/active_record/associations/preloader/has_one_through.rb
@@ -0,0 +1,9 @@
+module ActiveRecord
+  module Associations
+    class Preloader
+      class HasOneThrough < SingularAssociation #:nodoc:
+        include ThroughAssociation
+      end
+    end
+  end
+end
diff --git a/activerecord/lib/active_record/associations/preloader/singular_association.rb b/activerecord/lib/active_record/associations/preloader/singular_association.rb
new file mode 100644
index 0000000000..44e804d785
--- /dev/null
+++ b/activerecord/lib/active_record/associations/preloader/singular_association.rb
@@ -0,0 +1,21 @@
+module ActiveRecord
+  module Associations
+    class Preloader
+      class SingularAssociation < Association #:nodoc:
+
+        private
+
+        def preload
+          associated_records_by_owner.each do |owner, associated_records|
+            record = associated_records.first
+
+            association = owner.association(reflection.name)
+            association.target = record
+            association.set_inverse_instance(record)
+          end
+        end
+
+      end
+    end
+  end
+end
diff --git a/activerecord/lib/active_record/associations/preloader/through_association.rb b/activerecord/lib/active_record/associations/preloader/through_association.rb
new file mode 100644
index 0000000000..d630fc4c63
--- /dev/null
+++ b/activerecord/lib/active_record/associations/preloader/through_association.rb
@@ -0,0 +1,66 @@
+module ActiveRecord
+  module Associations
+    class Preloader
+      module ThroughAssociation #:nodoc:
+
+        def through_reflection
+          reflection.through_reflection
+        end
+
+        def source_reflection
+          reflection.source_reflection
+        end
+
+        def associated_records_by_owner
+          through_records = through_records_by_owner
+
+          ActiveRecord::Associations::Preloader.new(
+            through_records.values.flatten,
+            source_reflection.name, options
+          ).run
+
+          through_records.each do |owner, owner_through_records|
+            owner_through_records.map! { |r| r.send(source_reflection.name) }.flatten!
+          end
+        end
+
+        private
+
+        def through_records_by_owner
+          ActiveRecord::Associations::Preloader.new(
+            owners, through_reflection.name,
+            through_options
+          ).run
+
+          Hash[owners.map do |owner|
+            through_records = Array.wrap(owner.send(through_reflection.name))
+
+            # Dont cache the association - we would only be caching a subset
+            if reflection.options[:source_type] && through_reflection.collection?
+              owner.association(through_reflection.name).reset
+            end
+
+            [owner, through_records]
+          end]
+        end
+
+        def through_options
+          through_options = {}
+
+          if options[:source_type]
+            through_options[:conditions] = { reflection.foreign_type => options[:source_type] }
+          else
+            if options[:conditions]
+              through_options[:include]    = options[:include] || options[:source]
+              through_options[:conditions] = options[:conditions]
+            end
+
+            through_options[:order] = options[:order]
+          end
+
+          through_options
+        end
+      end
+    end
+  end
+end
diff --git a/activerecord/lib/active_record/base.rb b/activerecord/lib/active_record/base.rb
index 5c4998806b..b3204b2bda 100644
--- a/activerecord/lib/active_record/base.rb
+++ b/activerecord/lib/active_record/base.rb
@@ -1950,7 +1950,7 @@ MSG
     include AttributeMethods::Dirty
     include ActiveModel::MassAssignmentSecurity
     include Callbacks, ActiveModel::Observing, Timestamp
-    include Associations, AssociationPreload, NamedScope
+    include Associations, NamedScope
     include IdentityMap
     include ActiveModel::SecurePassword
 
diff --git a/activerecord/lib/active_record/reflection.rb b/activerecord/lib/active_record/reflection.rb
index 49659b3aa5..5de08953f9 100644
--- a/activerecord/lib/active_record/reflection.rb
+++ b/activerecord/lib/active_record/reflection.rb
@@ -210,6 +210,10 @@ module ActiveRecord
         @foreign_type ||= options[:foreign_type] || "#{name}_type"
       end
 
+      def type
+        @type ||= "#{options[:as]}_type"
+      end
+
       def primary_key_column
         @primary_key_column ||= klass.columns.find { |c| c.name == klass.primary_key }
       end
@@ -359,6 +363,8 @@ module ActiveRecord
     # Holds all the meta-data about a :through association as it was specified
     # in the Active Record class.
     class ThroughReflection < AssociationReflection #:nodoc:
+      delegate :association_primary_key, :foreign_type, :to => :source_reflection
+
       # Gets the source of the through reflection.  It checks both a singularized
       # and pluralized form for <tt>:belongs_to</tt> or <tt>:has_many</tt>.
       #
@@ -402,10 +408,6 @@ module ActiveRecord
         through_reflection.options
       end
 
-      def association_primary_key
-        source_reflection.association_primary_key
-      end
-
       def check_validity!
         if through_reflection.nil?
           raise HasManyThroughAssociationNotFoundError.new(active_record.name, self)
diff --git a/activerecord/lib/active_record/relation.rb b/activerecord/lib/active_record/relation.rb
index b828cffa77..f939bedc81 100644
--- a/activerecord/lib/active_record/relation.rb
+++ b/activerecord/lib/active_record/relation.rb
@@ -86,7 +86,9 @@ module ActiveRecord
 
       preload = @preload_values
       preload +=  @includes_values unless eager_loading?
-      preload.each {|associations| @klass.send(:preload_associations, @records, associations) }
+      preload.each do |associations|
+        ActiveRecord::Associations::Preloader.new(@records, associations).run
+      end
 
       # @readonly_value is true only if set explicitly. @implicit_readonly is true if there
       # are JOINS and no explicit SELECT.