Merge remote branch 'rails/master' into identity_map

Conflicts:
	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/persistence.rb

13 files changed, 714 insertions, 98 deletions

diff --git a/activerecord/lib/active_record/associations/association_collection.rb b/activerecord/lib/active_record/associations/association_collection.rb
index 6090376bb8..11a7a725e5 100644
--- a/activerecord/lib/active_record/associations/association_collection.rb
+++ b/activerecord/lib/active_record/associations/association_collection.rb
@@ -102,7 +102,7 @@ module ActiveRecord
 
       def reset
         reset_target!
-        reset_named_scopes_cache!
+        reset_scopes_cache!
         @loaded = false
       end
 
@@ -121,13 +121,13 @@ module ActiveRecord
       # Since << flattens its argument list and inserts each record, +push+ and +concat+ behave identically.
       def <<(*records)
         result = true
-        load_target unless @owner.persisted?
+        load_target if @owner.new_record?
 
         transaction do
           flatten_deeper(records).each do |record|
             raise_on_type_mismatch(record)
             add_record_to_target_with_callbacks(record) do |r|
-              result &&= insert_record(record) if @owner.persisted?
+              result &&= insert_record(record) unless @owner.new_record?
             end
           end
         end
@@ -160,7 +160,7 @@ module ActiveRecord
         load_target
         delete(@target)
         reset_target!
-        reset_named_scopes_cache!
+        reset_scopes_cache!
       end
 
       # Calculate sum using SQL, not Enumerable
@@ -235,12 +235,12 @@ module ActiveRecord
 
       # Removes all records from this association.  Returns +self+ so method calls may be chained.
       def clear
-        return self if length.zero? # forces load_target if it hasn't happened already
-
-        if @reflection.options[:dependent] && @reflection.options[:dependent] == :destroy
-          destroy_all
-        else
-          delete_all
+        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
@@ -253,7 +253,7 @@ module ActiveRecord
         load_target
         destroy(@target).tap do
           reset_target!
-          reset_named_scopes_cache!
+          reset_scopes_cache!
         end
       end
 
@@ -286,12 +286,12 @@ module ActiveRecord
       # This method is abstract in the sense that it relies on
       # +count_records+, which is a method descendants have to provide.
       def size
-        if !@owner.persisted? || (loaded? && !@reflection.options[:uniq])
+        if @owner.new_record? || (loaded? && !@reflection.options[:uniq])
           @target.size
         elsif !loaded? && @reflection.options[:group]
           load_target.size
         elsif !loaded? && !@reflection.options[:uniq] && @target.is_a?(Array)
-          unsaved_records = @target.reject { |r| r.persisted? }
+          unsaved_records = @target.select { |r| r.new_record? }
           unsaved_records.size + count_records
         else
           count_records
@@ -355,10 +355,9 @@ module ActiveRecord
 
       def include?(record)
         return false unless record.is_a?(@reflection.klass)
-        return include_in_memory?(record) unless record.persisted?
+        return include_in_memory?(record) if record.new_record?
         load_target if @reflection.options[:finder_sql] && !loaded?
-        return @target.include?(record) if loaded?
-        exists?(record)
+        loaded? ? @target.include?(record) : exists?(record)
       end
 
       def proxy_respond_to?(method, include_private = false)
@@ -370,7 +369,7 @@ module ActiveRecord
         end
 
         def load_target
-          if @owner.persisted? || foreign_key_present
+          if !@owner.new_record? || foreign_key_present
             begin
               unless loaded?
                 if @target.is_a?(Array) && @target.any?
@@ -410,9 +409,9 @@ module ActiveRecord
           if @target.respond_to?(method) || (!@reflection.klass.respond_to?(method) && Class.respond_to?(method))
             super
           elsif @reflection.klass.scopes[method]
-            @_named_scopes_cache ||= {}
-            @_named_scopes_cache[method] ||= {}
-            @_named_scopes_cache[method][args] ||= with_scope(@scope) { @reflection.klass.send(method, *args) }
+            @_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?
@@ -443,8 +442,8 @@ module ActiveRecord
           @target = Array.new
         end
 
-        def reset_named_scopes_cache!
-          @_named_scopes_cache = {}
+        def reset_scopes_cache!
+          @_scopes_cache = {}
         end
 
         def find_target
@@ -479,7 +478,7 @@ module ActiveRecord
       private
         def create_record(attrs)
           attrs.update(@reflection.options[:conditions]) if @reflection.options[:conditions].is_a?(Hash)
-          ensure_owner_is_not_new
+          ensure_owner_is_persisted!
 
           scoped_where = scoped.where_values_hash
           create_scope = scoped_where ? @scope[:create].merge(scoped_where) : @scope[:create]
@@ -509,7 +508,7 @@ module ActiveRecord
 
           transaction do
             records.each { |record| callback(:before_remove, record) }
-            old_records = records.select { |r| r.persisted? }
+            old_records = records.reject { |r| r.new_record? }
             yield(records, old_records)
             records.each { |record| callback(:after_remove, record) }
           end
@@ -530,18 +529,18 @@ module ActiveRecord
 
         def callbacks_for(callback_name)
           full_callback_name = "#{callback_name}_for_#{@reflection.name}"
-          @owner.class.read_inheritable_attribute(full_callback_name.to_sym) || []
+          @owner.class.send(full_callback_name.to_sym) || []
         end
 
-        def ensure_owner_is_not_new
+        def ensure_owner_is_persisted!
           unless @owner.persisted?
             raise ActiveRecord::RecordNotSaved, "You cannot call create unless the parent is saved"
           end
         end
 
         def fetch_first_or_last_using_find?(args)
-          (args.first.kind_of?(Hash) && !args.first.empty?) || !(loaded? || !@owner.persisted? || @reflection.options[:finder_sql] ||
-                                         !@target.all? { |record| record.persisted? } || args.first.kind_of?(Integer))
+          (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))
         end
 
         def include_in_memory?(record)
diff --git a/activerecord/lib/active_record/associations/association_proxy.rb b/activerecord/lib/active_record/associations/association_proxy.rb
index 9a9ffe9d62..53ec5a0da6 100644
--- a/activerecord/lib/active_record/associations/association_proxy.rb
+++ b/activerecord/lib/active_record/associations/association_proxy.rb
@@ -253,7 +253,7 @@ module ActiveRecord
         def load_target
           return nil unless defined?(@loaded)
 
-          if !loaded? and (@owner.persisted? || foreign_key_present)
+          if !loaded? && (!@owner.new_record? || foreign_key_present)
             if IdentityMap.enabled? && association_class
               @target = IdentityMap.get(association_class, @owner[@reflection.association_foreign_key])
             end
diff --git a/activerecord/lib/active_record/associations/belongs_to_association.rb b/activerecord/lib/active_record/associations/belongs_to_association.rb
index b624951cd9..b438620c8f 100644
--- a/activerecord/lib/active_record/associations/belongs_to_association.rb
+++ b/activerecord/lib/active_record/associations/belongs_to_association.rb
@@ -61,7 +61,7 @@ module ActiveRecord
           set_inverse_instance(the_target, @owner)
           the_target
         end
-        
+
         def construct_find_scope
           { :conditions => conditions }
         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
new file mode 100644
index 0000000000..a74d0a4ca8
--- /dev/null
+++ b/activerecord/lib/active_record/associations/class_methods/join_dependency.rb
@@ -0,0 +1,232 @@
+require 'active_record/associations/class_methods/join_dependency/join_part'
+require 'active_record/associations/class_methods/join_dependency/join_base'
+require 'active_record/associations/class_methods/join_dependency/join_association'
+
+module ActiveRecord
+  module Associations
+    module ClassMethods
+      class JoinDependency # :nodoc:
+        attr_reader :join_parts, :reflections, :table_aliases, :active_record
+
+        def initialize(base, associations, joins)
+          @active_record         = base
+          @table_joins           = joins
+          @join_parts            = [JoinBase.new(base)]
+          @associations          = {}
+          @reflections           = []
+          @table_aliases         = Hash.new do |h,name|
+            h[name] = count_aliases_from_table_joins(name.downcase)
+          end
+          @table_aliases[base.table_name] = 1
+          build(associations)
+        end
+
+        def graft(*associations)
+          associations.each do |association|
+            join_associations.detect {|a| association == a} ||
+              build(association.reflection.name, association.find_parent_in(self) || join_base, association.join_type)
+          end
+          self
+        end
+
+        def join_associations
+          join_parts.last(join_parts.length - 1)
+        end
+
+        def join_base
+          join_parts.first
+        end
+
+        def columns
+          join_parts.collect { |join_part|
+            table = join_part.aliased_table
+            join_part.column_names_with_alias.collect{ |column_name, aliased_name|
+              table[column_name].as Arel.sql(aliased_name)
+            }
+          }.flatten
+        end
+
+        def count_aliases_from_table_joins(name)
+          return 0 if Arel::Table === @table_joins
+
+          # quoted_name should be downcased as some database adapters (Oracle) return quoted name in uppercase
+          quoted_name = active_record.connection.quote_table_name(name).downcase
+
+          @table_joins.map { |join|
+            # Table names + table aliases
+            join.left.downcase.scan(
+              /join(?:\s+\w+)?\s+(\S+\s+)?#{quoted_name}\son/
+            ).size
+          }.sum
+        end
+
+        def instantiate(rows)
+          primary_key = join_base.aliased_primary_key
+          parents = {}
+
+          records = rows.map { |model|
+            primary_id = model[primary_key]
+            parent = parents[primary_id] ||= join_base.instantiate(model)
+            construct(parent, @associations, join_associations, model)
+            parent
+          }.uniq
+
+          remove_duplicate_results!(active_record, records, @associations)
+          records
+        end
+
+        def remove_duplicate_results!(base, records, associations)
+          case associations
+          when Symbol, String
+            reflection = base.reflections[associations]
+            remove_uniq_by_reflection(reflection, records)
+          when Array
+            associations.each do |association|
+              remove_duplicate_results!(base, records, association)
+            end
+          when Hash
+            associations.keys.each do |name|
+              reflection = base.reflections[name]
+              remove_uniq_by_reflection(reflection, records)
+
+              parent_records = []
+              records.each do |record|
+                if descendant = record.send(reflection.name)
+                  if reflection.collection?
+                    parent_records.concat descendant.target.uniq
+                  else
+                    parent_records << descendant
+                  end
+                end
+              end
+
+              remove_duplicate_results!(reflection.klass, parent_records, associations[name]) unless parent_records.empty?
+            end
+          end
+        end
+
+        protected
+
+        def cache_joined_association(association)
+          associations = []
+          parent = association.parent
+          while parent != join_base
+            associations.unshift(parent.reflection.name)
+            parent = parent.parent
+          end
+          ref = @associations
+          associations.each do |key|
+            ref = ref[key]
+          end
+          ref[association.reflection.name] ||= {}
+        end
+
+        def build(associations, parent = nil, join_type = Arel::InnerJoin)
+          parent ||= join_parts.last
+          case associations
+          when Symbol, String
+            reflection = parent.reflections[associations.to_s.intern] or
+            raise ConfigurationError, "Association named '#{ associations }' was not found; perhaps you misspelled it?"
+            unless join_association = find_join_association(reflection, parent)
+              @reflections << reflection
+              join_association = build_join_association(reflection, parent)
+              join_association.join_type = join_type
+              @join_parts << join_association
+              cache_joined_association(join_association)
+            end
+            join_association
+          when Array
+            associations.each do |association|
+              build(association, parent, join_type)
+            end
+          when Hash
+            associations.keys.sort_by { |a| a.to_s }.each do |name|
+              join_association = build(name, parent, join_type)
+              build(associations[name], join_association, join_type)
+            end
+          else
+            raise ConfigurationError, associations.inspect
+          end
+        end
+
+        def find_join_association(name_or_reflection, parent)
+          if String === name_or_reflection
+            name_or_reflection = name_or_reflection.to_sym
+          end
+
+          join_associations.detect { |j|
+            j.reflection == name_or_reflection && j.parent == parent
+          }
+        end
+
+        def remove_uniq_by_reflection(reflection, records)
+          if reflection && reflection.collection?
+            records.each { |record| record.send(reflection.name).target.uniq! }
+          end
+        end
+
+        def build_join_association(reflection, parent)
+          JoinAssociation.new(reflection, self, parent)
+        end
+
+        def construct(parent, associations, join_parts, row)
+          case associations
+          when Symbol, String
+            name = associations.to_s
+
+            join_part = join_parts.detect { |j|
+              j.reflection.name.to_s == name &&
+                j.parent_table_name    == parent.class.table_name }
+
+              raise(ConfigurationError, "No such association") unless join_part
+
+              join_parts.delete(join_part)
+              construct_association(parent, join_part, row)
+          when Array
+            associations.each do |association|
+              construct(parent, association, join_parts, row)
+            end
+          when Hash
+            associations.sort_by { |k,_| k.to_s }.each do |name, assoc|
+              association = construct(parent, name, join_parts, row)
+              construct(association, assoc, join_parts, row) if association
+            end
+          else
+            raise ConfigurationError, associations.inspect
+          end
+        end
+
+        def construct_association(record, join_part, row)
+          return if record.id.to_s != join_part.parent.record_id(row).to_s
+
+          macro = join_part.reflection.macro
+          if macro == :has_one
+            return if record.instance_variable_defined?("@#{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
+            return if row[join_part.aliased_primary_key].nil?
+            association = join_part.instantiate(row)
+            case macro
+            when :has_many, :has_and_belongs_to_many
+              collection = record.send(join_part.reflection.name)
+              collection.loaded
+              collection.target.push(association)
+              collection.__send__(:set_inverse_instance, association, record)
+            when :belongs_to
+              set_target_and_inverse(join_part, association, record)
+            else
+              raise ConfigurationError, "unknown macro: #{join_part.reflection.macro}"
+            end
+          end
+          association
+        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)
+        end
+      end
+    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
new file mode 100644
index 0000000000..694778008b
--- /dev/null
+++ b/activerecord/lib/active_record/associations/class_methods/join_dependency/join_association.rb
@@ -0,0 +1,274 @@
+module ActiveRecord
+  module Associations
+    module ClassMethods
+      class JoinDependency # :nodoc:
+        class JoinAssociation < JoinPart # :nodoc:
+          # The reflection of the association represented
+          attr_reader :reflection
+
+          # The JoinDependency object which this JoinAssociation exists within. This is mainly
+          # relevant for generating aliases which do not conflict with other joins which are
+          # part of the query.
+          attr_reader :join_dependency
+
+          # A JoinBase instance representing the active record we are joining onto.
+          # (So in Author.has_many :posts, the Author would be that base record.)
+          attr_reader :parent
+
+          # What type of join will be generated, either Arel::InnerJoin (default) or Arel::OuterJoin
+          attr_accessor :join_type
+
+          # These implement abstract methods from the superclass
+          attr_reader :aliased_prefix, :aliased_table_name
+
+          delegate :options, :through_reflection, :source_reflection, :to => :reflection
+          delegate :table, :table_name, :to => :parent, :prefix => :parent
+
+          def initialize(reflection, join_dependency, parent = nil)
+            reflection.check_validity!
+
+            if reflection.options[:polymorphic]
+              raise EagerLoadPolymorphicError.new(reflection)
+            end
+
+            super(reflection.klass)
+
+            @reflection      = reflection
+            @join_dependency = join_dependency
+            @parent          = parent
+            @join_type       = Arel::InnerJoin
+            @aliased_prefix  = "t#{ join_dependency.join_parts.size }"
+
+            # This must be done eagerly upon initialisation because the alias which is produced
+            # depends on the state of the join dependency, but we want it to work the same way
+            # every time.
+            allocate_aliases
+            @table = Arel::Table.new(
+              table_name, :as => aliased_table_name, :engine => arel_engine
+            )
+          end
+
+          def ==(other)
+            other.class == self.class &&
+              other.reflection == reflection &&
+              other.parent == parent
+          end
+
+          def find_parent_in(other_join_dependency)
+            other_join_dependency.join_parts.detect do |join_part|
+              parent == join_part
+            end
+          end
+
+          def join_to(relation)
+            send("join_#{reflection.macro}_to", relation)
+          end
+
+          def join_relation(joining_relation)
+            self.join_type = Arel::OuterJoin
+            joining_relation.joins(self)
+          end
+
+          attr_reader :table
+          # More semantic name given we are talking about associations
+          alias_method :target_table, :table
+
+          protected
+
+          def aliased_table_name_for(name, suffix = nil)
+            aliases = @join_dependency.table_aliases
+
+            if aliases[name] != 0 # We need an alias
+              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
+            end
+
+            aliases[name] += 1
+
+            name
+          end
+
+          def pluralize(table_name)
+            ActiveRecord::Base.pluralize_table_names ? table_name.to_s.pluralize : table_name
+          end
+
+          private
+
+          def allocate_aliases
+            @aliased_table_name = aliased_table_name_for(table_name)
+
+            if reflection.macro == :has_and_belongs_to_many
+              @aliased_join_table_name = aliased_table_name_for(reflection.options[:join_table], "_join")
+            elsif [:has_many, :has_one].include?(reflection.macro) && reflection.options[:through]
+              @aliased_join_table_name = aliased_table_name_for(reflection.through_reflection.klass.table_name, "_join")
+            end
+          end
+
+          def process_conditions(conditions, table_name)
+            Arel.sql(sanitize_sql(conditions, table_name))
+          end
+
+          def sanitize_sql(condition, table_name)
+            active_record.send(:sanitize_sql, condition, table_name)
+          end
+
+          def join_target_table(relation, condition)
+            conditions = [condition]
+
+            # If the target table is an STI model then we must be sure to only include records of
+            # its type and its sub-types.
+            unless active_record.descends_from_active_record?
+              sti_column    = target_table[active_record.inheritance_column]
+              subclasses    = active_record.descendants
+              sti_condition = sti_column.eq(active_record.sti_name)
+
+              conditions << subclasses.inject(sti_condition) { |attr,subclass|
+                attr.or(sti_column.eq(subclass.sti_name))
+              }
+            end
+
+            # If the reflection has conditions, add them
+            if options[:conditions]
+              conditions << process_conditions(options[:conditions], aliased_table_name)
+            end
+
+            ands = relation.create_and(conditions)
+
+            join = relation.create_join(
+              target_table,
+              relation.create_on(ands),
+              join_type)
+
+            relation.from join
+          end
+
+          def join_has_and_belongs_to_many_to(relation)
+            join_table = Arel::Table.new(
+              options[:join_table]
+            ).alias(@aliased_join_table_name)
+
+            fk       = options[:foreign_key]             || reflection.active_record.to_s.foreign_key
+            klass_fk = options[:association_foreign_key] || reflection.klass.to_s.foreign_key
+
+            relation = relation.join(join_table, join_type)
+            relation = relation.on(
+              join_table[fk].
+              eq(parent_table[reflection.active_record.primary_key])
+            )
+
+            join_target_table(
+              relation,
+              target_table[reflection.klass.primary_key].
+              eq(join_table[klass_fk])
+            )
+          end
+
+          def join_has_many_to(relation)
+            if reflection.options[:through]
+              join_has_many_through_to(relation)
+            elsif reflection.options[:as]
+              join_has_many_polymorphic_to(relation)
+            else
+              foreign_key = options[:foreign_key] || reflection.active_record.name.foreign_key
+              primary_key = options[:primary_key] || parent.primary_key
+
+              join_target_table(
+                relation,
+                target_table[foreign_key].
+                eq(parent_table[primary_key])
+              )
+            end
+          end
+          alias :join_has_one_to :join_has_many_to
+
+          def join_has_many_through_to(relation)
+            join_table = Arel::Table.new(
+              through_reflection.klass.table_name
+            ).alias @aliased_join_table_name
+
+            jt_conditions = []
+            jt_foreign_key = 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)
+            else
+              jt_foreign_key = through_reflection.primary_key_name
+            end
+
+            case source_reflection.macro
+            when :has_many
+              second_key = options[:foreign_key] || primary_key
+
+              if source_reflection.options[:as]
+                first_key = "#{source_reflection.options[:as]}_id"
+              else
+                first_key = through_reflection.klass.base_class.to_s.foreign_key
+              end
+
+              unless through_reflection.klass.descends_from_active_record?
+                jt_conditions <<
+                join_table[through_reflection.active_record.inheritance_column].
+                  eq(through_reflection.klass.sti_name)
+              end
+            when :belongs_to
+              first_key = primary_key
+
+              if reflection.options[:source_type]
+                second_key = source_reflection.association_foreign_key
+
+                jt_conditions <<
+                join_table[reflection.source_reflection.options[:foreign_type]].
+                  eq(reflection.options[:source_type])
+              else
+                second_key = source_reflection.primary_key_name
+              end
+            end
+
+            jt_conditions <<
+            parent_table[parent.primary_key].
+              eq(join_table[jt_foreign_key])
+
+            if through_reflection.options[:conditions]
+              jt_conditions << process_conditions(through_reflection.options[:conditions], aliased_table_name)
+            end
+
+            relation = relation.join(join_table, join_type).on(*jt_conditions)
+
+            join_target_table(
+              relation,
+              target_table[first_key].eq(join_table[second_key])
+            )
+          end
+
+          def join_has_many_polymorphic_to(relation)
+            join_target_table(
+              relation,
+              target_table["#{reflection.options[:as]}_id"].
+              eq(parent_table[parent.primary_key]).and(
+              target_table["#{reflection.options[:as]}_type"].
+              eq(parent.active_record.base_class.name))
+            )
+          end
+
+          def join_belongs_to_to(relation)
+            foreign_key = options[:foreign_key] || reflection.primary_key_name
+            primary_key = options[:primary_key] || reflection.klass.primary_key
+
+            join_target_table(
+              relation,
+              target_table[primary_key].eq(parent_table[foreign_key])
+            )
+          end
+        end
+      end
+    end
+  end
+end
diff --git a/activerecord/lib/active_record/associations/class_methods/join_dependency/join_base.rb b/activerecord/lib/active_record/associations/class_methods/join_dependency/join_base.rb
new file mode 100644
index 0000000000..67567f06df
--- /dev/null
+++ b/activerecord/lib/active_record/associations/class_methods/join_dependency/join_base.rb
@@ -0,0 +1,26 @@
+module ActiveRecord
+  module Associations
+    module ClassMethods
+      class JoinDependency # :nodoc:
+        class JoinBase < JoinPart # :nodoc:
+          def ==(other)
+            other.class == self.class &&
+              other.active_record == active_record
+          end
+
+          def aliased_prefix
+            "t0"
+          end
+
+          def table
+            Arel::Table.new(table_name, arel_engine)
+          end
+
+          def aliased_table_name
+            active_record.table_name
+          end
+        end
+      end
+    end
+  end
+end
diff --git a/activerecord/lib/active_record/associations/class_methods/join_dependency/join_part.rb b/activerecord/lib/active_record/associations/class_methods/join_dependency/join_part.rb
new file mode 100644
index 0000000000..cd16ae5a8b
--- /dev/null
+++ b/activerecord/lib/active_record/associations/class_methods/join_dependency/join_part.rb
@@ -0,0 +1,80 @@
+module ActiveRecord
+  module Associations
+    module ClassMethods
+      class JoinDependency # :nodoc:
+        # A JoinPart represents a part of a JoinDependency. It is an abstract class, inherited
+        # by JoinBase and JoinAssociation. A JoinBase represents the Active Record which
+        # everything else is being joined onto. A JoinAssociation represents an association which
+        # is joining to the base. A JoinAssociation may result in more than one actual join
+        # operations (for example a has_and_belongs_to_many JoinAssociation would result in
+        # two; one for the join table and one for the target table).
+        class JoinPart # :nodoc:
+          # The Active Record class which this join part is associated 'about'; for a JoinBase
+          # this is the actual base model, for a JoinAssociation this is the target model of the
+          # association.
+          attr_reader :active_record
+
+          delegate :table_name, :column_names, :primary_key, :reflections, :arel_engine, :to => :active_record
+
+          def initialize(active_record)
+            @active_record = active_record
+            @cached_record = {}
+            @column_names_with_alias = nil
+          end
+
+          def aliased_table
+            Arel::Nodes::TableAlias.new aliased_table_name, table
+          end
+
+          def ==(other)
+            raise NotImplementedError
+          end
+
+          # An Arel::Table for the active_record
+          def table
+            raise NotImplementedError
+          end
+
+          # The prefix to be used when aliasing columns in the active_record's table
+          def aliased_prefix
+            raise NotImplementedError
+          end
+
+          # The alias for the active_record's table
+          def aliased_table_name
+            raise NotImplementedError
+          end
+
+          # The alias for the primary key of the active_record's table
+          def aliased_primary_key
+            "#{aliased_prefix}_r0"
+          end
+
+          # An array of [column_name, alias] pairs for the table
+          def column_names_with_alias
+            unless @column_names_with_alias
+              @column_names_with_alias = []
+
+              ([primary_key] + (column_names - [primary_key])).each_with_index do |column_name, i|
+                @column_names_with_alias << [column_name, "#{aliased_prefix}_r#{i}"]
+              end
+            end
+            @column_names_with_alias
+          end
+
+          def extract_record(row)
+            Hash[column_names_with_alias.map{|cn, an| [cn, row[an]]}]
+          end
+
+          def record_id(row)
+            row[aliased_primary_key]
+          end
+
+          def instantiate(row)
+            @cached_record[record_id(row)] ||= active_record.send(:instantiate, extract_record(row))
+          end
+        end
+      end
+    end
+  end
+end
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 da742fa668..e2ce9aefcf 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
@@ -34,7 +34,7 @@ module ActiveRecord
         end
 
         def insert_record(record, force = true, validate = true)
-          unless record.persisted?
+          if record.new_record?
             if force
               record.save!
             else
@@ -49,7 +49,7 @@ module ActiveRecord
             timestamps = record_timestamp_columns(record)
             timezone   = record.send(:current_time_from_proper_timezone) if timestamps.any?
 
-            attributes = Hash[columns.map do |column|
+            attributes = columns.map do |column|
               name = column.name
               value = case name.to_s
                 when @reflection.primary_key_name.to_s
@@ -62,12 +62,13 @@ module ActiveRecord
                   @owner.send(:quote_value, record[name], column) if record.has_attribute?(name)
               end
               [relation[name], value] unless value.nil?
-            end]
+            end
 
-            relation.insert(attributes)
+            stmt = relation.compile_insert Hash[attributes]
+            @owner.connection.insert stmt.to_sql
           end
 
-          return true
+          true
         end
 
         def delete_records(records)
@@ -75,12 +76,13 @@ module ActiveRecord
             records.each { |record| @owner.connection.delete(interpolate_sql(sql, record)) }
           else
             relation = Arel::Table.new(@reflection.options[:join_table])
-            relation.where(relation[@reflection.primary_key_name].eq(@owner.id).
+            stmt = relation.where(relation[@reflection.primary_key_name].eq(@owner.id).
               and(relation[@reflection.association_foreign_key].in(records.map { |x| x.id }.compact))
-            ).delete
+            ).compile_delete
+            @owner.connection.delete stmt.to_sql
           end
         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
@@ -113,7 +115,7 @@ module ActiveRecord
       private
         def create_record(attributes, &block)
           # Can't use Base.create because the foreign key may be a protected attribute.
-          ensure_owner_is_not_new
+          ensure_owner_is_persisted!
           if attributes.is_a?(Array)
             attributes.collect { |attr| create(attr) }
           else
diff --git a/activerecord/lib/active_record/associations/has_many_association.rb b/activerecord/lib/active_record/associations/has_many_association.rb
index 830a82980d..4ff61fff45 100644
--- a/activerecord/lib/active_record/associations/has_many_association.rb
+++ b/activerecord/lib/active_record/associations/has_many_association.rb
@@ -42,11 +42,7 @@ module ActiveRecord
           # documented side-effect of the method that may avoid an extra SELECT.
           @target ||= [] and loaded if count == 0
 
-          if @reflection.options[:limit]
-            count = [ @reflection.options[:limit], count ].min
-          end
-
-          return count
+          [@reflection.options[:limit], count].compact.min
         end
 
         def has_cached_counter?
@@ -71,9 +67,10 @@ module ActiveRecord
               @reflection.klass.delete(records.map { |record| record.id })
             else
               relation = Arel::Table.new(@reflection.table_name)
-              relation.where(relation[@reflection.primary_key_name].eq(@owner.id).
+              stmt = relation.where(relation[@reflection.primary_key_name].eq(@owner.id).
                   and(relation[@reflection.klass.primary_key].in(records.map { |r| r.id }))
-              ).update(relation[@reflection.primary_key_name] => nil)
+              ).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?
           end
@@ -112,8 +109,7 @@ module ActiveRecord
         end
 
         def we_can_set_the_inverse_on_this?(record)
-          inverse = @reflection.inverse_of
-          return !inverse.nil?
+          @reflection.inverse_of
         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 79c229d9c4..781aa7ef62 100644
--- a/activerecord/lib/active_record/associations/has_many_through_association.rb
+++ b/activerecord/lib/active_record/associations/has_many_through_association.rb
@@ -36,7 +36,7 @@ module ActiveRecord
 
       protected
         def create_record(attrs, force = true)
-          ensure_owner_is_not_new
+          ensure_owner_is_persisted!
 
           transaction do
             object = @reflection.klass.new(attrs)
@@ -54,12 +54,12 @@ module ActiveRecord
         end
 
         def construct_find_options!(options)
-          options[:joins]   = construct_joins(options[:joins])
+          options[:joins]   = [construct_joins] + Array.wrap(options[:joins])
           options[:include] = @reflection.source_reflection.options[:include] if options[:include].nil? && @reflection.source_reflection.options[:include]
         end
 
         def insert_record(record, force = true, validate = true)
-          unless record.persisted?
+          if record.new_record?
             if force
               record.save!
             else
@@ -81,7 +81,7 @@ module ActiveRecord
 
         def find_target
           return [] unless target_reflection_has_associated_record?
-          with_scope(@scope) { @reflection.klass.find(:all) }
+          scoped.all
         end
 
         def has_cached_counter?
diff --git a/activerecord/lib/active_record/associations/has_one_association.rb b/activerecord/lib/active_record/associations/has_one_association.rb
index e6e037441f..c49fd6e66a 100644
--- a/activerecord/lib/active_record/associations/has_one_association.rb
+++ b/activerecord/lib/active_record/associations/has_one_association.rb
@@ -56,7 +56,7 @@ module ActiveRecord
         set_inverse_instance(obj, @owner)
         @loaded = true
 
-        unless !@owner.persisted? or obj.nil? or dont_save
+        unless !@owner.persisted? || obj.nil? || dont_save
           return (obj.save ? self : false)
         else
           return (obj.nil? ? nil : self)
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 6e98f7dffb..e8cf73976b 100644
--- a/activerecord/lib/active_record/associations/has_one_through_association.rb
+++ b/activerecord/lib/active_record/associations/has_one_through_association.rb
@@ -21,7 +21,7 @@ module ActiveRecord
         if current_object
            new_value ? current_object.update_attributes(construct_join_attributes(new_value)) : current_object.destroy
         elsif new_value
-          unless @owner.persisted?
+          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))
@@ -33,7 +33,7 @@ module ActiveRecord
 
     private
       def find_target
-        with_scope(@scope) { @reflection.klass.find(:first) }
+        scoped.first
       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
index bd8e304e99..5dc5b0c048 100644
--- a/activerecord/lib/active_record/associations/through_association_scope.rb
+++ b/activerecord/lib/active_record/associations/through_association_scope.rb
@@ -3,6 +3,13 @@ module ActiveRecord
   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
@@ -16,32 +23,38 @@ module ActiveRecord
           :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_name = @reflection.through_reflection.quoted_table_name
-        conditions = construct_quoted_owner_attributes(@reflection.through_reflection).map do |attr, value|
-          "#{table_name}.#{attr} = #{value}"
+        table = aliased_through_table
+        conditions = construct_owner_attributes(@reflection.through_reflection).map do |attr, value|
+          table[attr].eq(value)
         end
-        conditions << sql_conditions if sql_conditions
-        "(" + conditions.join(') AND (') + ")"
+        conditions << Arel.sql(sql_conditions) if sql_conditions
+        table.create_and(conditions)
       end
 
-      # Associate attributes pointing to owner, quoted.
-      def construct_quoted_owner_attributes(reflection)
+      # Associate attributes pointing to owner
+      def construct_owner_attributes(reflection)
         if as = reflection.options[:as]
-          { "#{as}_id" => owner_quoted_id,
-            "#{as}_type" => reflection.klass.quote_value(
-              @owner.class.base_class.name.to_s,
-              reflection.klass.columns_hash["#{as}_type"]) }
+          { "#{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.class.quote_value(@owner[reflection.primary_key_name]) }
+          { reflection.klass.primary_key => @owner[reflection.primary_key_name] }
         else
-          { reflection.primary_key_name => owner_quoted_id }
+          { reflection.primary_key_name => @owner[reflection.active_record_primary_key] }
         end
       end
 
@@ -51,47 +64,41 @@ module ActiveRecord
 
       def construct_select(custom_select = nil)
         distinct = "DISTINCT " if @reflection.options[:uniq]
-        selected = custom_select || @reflection.options[:select] || "#{distinct}#{@reflection.quoted_table_name}.*"
+        custom_select || @reflection.options[:select] || "#{distinct}#{@reflection.quoted_table_name}.*"
       end
 
-      def construct_joins(custom_joins = nil)
-        polymorphic_join = nil
+      def construct_joins
+        right = aliased_through_table
+        left  = @reflection.klass.arel_table
+
+        conditions = []
+
         if @reflection.source_reflection.macro == :belongs_to
-          reflection_primary_key = @reflection.klass.primary_key
+          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]
-            polymorphic_join = "AND %s.%s = %s" % [
-              @reflection.through_reflection.quoted_table_name, "#{@reflection.source_reflection.options[:foreign_type]}",
-              @owner.class.quote_value(@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.through_reflection.klass.primary_key
+          source_primary_key     = @reflection.source_reflection.options[:primary_key] ||
+                                   @reflection.through_reflection.klass.primary_key
           if @reflection.source_reflection.options[:as]
-            polymorphic_join = "AND %s.%s = %s" % [
-              @reflection.quoted_table_name, "#{@reflection.source_reflection.options[:as]}_type",
-              @owner.class.quote_value(@reflection.through_reflection.klass.name)
-            ]
+            column = "#{@reflection.source_reflection.options[:as]}_type"
+            conditions <<
+              left[column].eq(@reflection.through_reflection.klass.name)
           end
         end
 
-        "INNER JOIN %s ON %s.%s = %s.%s %s #{@reflection.options[:joins]} #{custom_joins}" % [
-          @reflection.through_reflection.quoted_table_name,
-          @reflection.quoted_table_name, reflection_primary_key,
-          @reflection.through_reflection.quoted_table_name, source_primary_key,
-          polymorphic_join
-        ]
-      end
+        conditions <<
+          left[reflection_primary_key].eq(right[source_primary_key])
 
-      # Construct attributes for associate pointing to owner.
-      def construct_owner_attributes(reflection)
-        if as = reflection.options[:as]
-          { "#{as}_id" => @owner.id,
-            "#{as}_type" => @owner.class.base_class.name.to_s }
-        else
-          { reflection.primary_key_name => @owner.id }
-        end
+        right.create_join(
+          right,
+          right.create_on(right.create_and(conditions)))
       end
 
       # Construct attributes for :through pointing to owner and associate.
@@ -102,7 +109,7 @@ module ActiveRecord
         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.to_s)
+          join_attributes.merge!(@reflection.source_reflection.options[:foreign_type] => associate.class.base_class.name)
         end
 
         if @reflection.through_reflection.options[:conditions].is_a?(Hash)