Merge remote branch 'rails/master' into identity_map

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

60 files changed, 3041 insertions, 2156 deletions

diff --git a/activerecord/lib/active_record/aggregations.rb b/activerecord/lib/active_record/aggregations.rb
index 8cd7389005..90d3b58c78 100644
--- a/activerecord/lib/active_record/aggregations.rb
+++ b/activerecord/lib/active_record/aggregations.rb
@@ -4,9 +4,7 @@ module ActiveRecord
     extend ActiveSupport::Concern
 
     def clear_aggregation_cache #:nodoc:
-      self.class.reflect_on_all_aggregations.to_a.each do |assoc|
-        instance_variable_set "@#{assoc.name}", nil
-      end if self.persisted?
+      @aggregation_cache.clear if persisted?
     end
 
     # Active Record implements aggregation through a macro-like class method called +composed_of+
@@ -222,53 +220,32 @@ module ActiveRecord
 
       private
         def reader_method(name, class_name, mapping, allow_nil, constructor)
-          module_eval do
-            define_method(name) do |*args|
-              force_reload = args.first || false
-
-              unless instance_variable_defined?("@#{name}")
-                instance_variable_set("@#{name}", nil)
-              end
-
-              if (instance_variable_get("@#{name}").nil? || force_reload) && (!allow_nil || mapping.any? {|pair| !read_attribute(pair.first).nil? })
-                attrs = mapping.collect {|pair| read_attribute(pair.first)}
-                object = case constructor
-                  when Symbol
-                    class_name.constantize.send(constructor, *attrs)
-                  when Proc, Method
-                    constructor.call(*attrs)
-                  else
-                    raise ArgumentError, 'Constructor must be a symbol denoting the constructor method to call or a Proc to be invoked.'
-                  end
-                instance_variable_set("@#{name}", object)
-              end
-              instance_variable_get("@#{name}")
+          define_method(name) do
+            if @aggregation_cache[name].nil? && (!allow_nil || mapping.any? {|pair| !read_attribute(pair.first).nil? })
+              attrs = mapping.collect {|pair| read_attribute(pair.first)}
+              object = constructor.respond_to?(:call) ?
+                constructor.call(*attrs) :
+                class_name.constantize.send(constructor, *attrs)
+              @aggregation_cache[name] = object
             end
+            @aggregation_cache[name]
           end
-
         end
 
         def writer_method(name, class_name, mapping, allow_nil, converter)
-          module_eval do
-            define_method("#{name}=") do |part|
-              if part.nil? && allow_nil
-                mapping.each { |pair| self[pair.first] = nil }
-                instance_variable_set("@#{name}", nil)
-              else
-                unless part.is_a?(class_name.constantize) || converter.nil?
-                  part = case converter
-                    when Symbol
-                     class_name.constantize.send(converter, part)
-                    when Proc, Method
-                      converter.call(part)
-                    else
-                      raise ArgumentError, 'Converter must be a symbol denoting the converter method to call or a Proc to be invoked.'
-                    end
-                end
-
-                mapping.each { |pair| self[pair.first] = part.send(pair.last) }
-                instance_variable_set("@#{name}", part.freeze)
+          define_method("#{name}=") do |part|
+            if part.nil? && allow_nil
+              mapping.each { |pair| self[pair.first] = nil }
+              @aggregation_cache[name] = nil
+            else
+              unless part.is_a?(class_name.constantize) || converter.nil?
+                part = converter.respond_to?(:call) ?
+                  converter.call(part) :
+                  class_name.constantize.send(converter, part)
               end
+
+              mapping.each { |pair| self[pair.first] = part.send(pair.last) }
+              @aggregation_cache[name] = part.freeze
             end
           end
         end
diff --git a/activerecord/lib/active_record/association_preload.rb b/activerecord/lib/active_record/association_preload.rb
index 373532704f..52d2c49836 100644
--- a/activerecord/lib/active_record/association_preload.rb
+++ b/activerecord/lib/active_record/association_preload.rb
@@ -125,45 +125,48 @@ module ActiveRecord
       def add_preloaded_records_to_collection(parent_records, reflection_name, associated_record)
         parent_records.each do |parent_record|
           association_proxy = parent_record.send(reflection_name)
-          association_proxy.loaded
-          association_proxy.target = association_proxy.target | [*Array.wrap(associated_record)]
-
-          association_proxy.__send__(:set_inverse_instance, associated_record, parent_record)
+          association_proxy.loaded!
+          association_proxy.target.concat(Array.wrap(associated_record))
+          association_proxy.send(: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.send("set_#{reflection_name}_target", associated_record)
+          parent_record.send(:association_proxy, reflection_name).target = associated_record
         end
       end
 
-      def set_association_collection_records(id_to_record_map, reflection_name, associated_records, key)
+      def set_association_collection_records(id_to_parent_map, reflection_name, associated_records, key)
         associated_records.each do |associated_record|
-          mapped_records = id_to_record_map[associated_record[key].to_s]
-          add_preloaded_records_to_collection(mapped_records, reflection_name, 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[associated_record[key].to_s]
-          seen_keys[associated_record[key].to_s] = true
-          mapped_records = id_to_record_map[associated_record[key].to_s]
+          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.send("set_#{reflection_name}_target", associated_record)
-            association_proxy.__send__(:set_inverse_instance, associated_record, mapped_record)
+            association_proxy = mapped_record.send(:association_proxy, 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.to_s)
-          records.each {|record| record.send("set_#{reflection_name}_target", nil) }
+          next if seen_keys.include?(id)
+          add_preloaded_record_to_collection(records, reflection_name, nil)
         end
       end
 
@@ -172,57 +175,80 @@ module ActiveRecord
       # <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)
-        id_to_record_map = {}
-        ids = []
-        records.each do |record|
+        records.group_by do |record|
           primary_key ||= record.class.primary_key
-          ids << record[primary_key]
-          mapped_records = (id_to_record_map[ids.last.to_s] ||= [])
-          mapped_records << record
+          record[primary_key].to_s
         end
-        ids.uniq!
-        return id_to_record_map, ids
       end
 
       def preload_has_and_belongs_to_many_association(records, reflection, preload_options={})
-        table_name = reflection.klass.quoted_table_name
-        id_to_record_map, ids = construct_id_map(records)
-        records.each {|record| record.send(reflection.name).loaded}
+
+        left = reflection.klass.arel_table
+
+        id_to_record_map = construct_id_map(records)
+
+        records.each { |record| record.send(reflection.name).loaded! }
         options = reflection.options
 
-        conditions = "t0.#{reflection.primary_key_name} #{in_or_equals_for_ids(ids)}"
-        conditions << append_conditions(reflection, preload_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]).
-            joins("INNER JOIN #{connection.quote_table_name options[:join_table]} t0 ON #{reflection.klass.quoted_table_name}.#{reflection.klass.primary_key} = t0.#{reflection.association_foreign_key}").
-            select("#{options[:select] || table_name+'.*'}, t0.#{reflection.primary_key_name} as the_parent_record_id").
             order(options[:order])
 
-        all_associated_records = ActiveRecord::IdentityMap.without do
-          associated_records(ids) do |some_ids|
-            associated_records_proxy.where([conditions, ids]).to_a
-          end
-        end
+        associated_records_proxy.joins_values = [join]
+        associated_records_proxy.select_values = select
+
+        custom_conditions = append_conditions(reflection, preload_options)
+
+        klass = associated_records_proxy.klass
 
-        set_association_collection_records(id_to_record_map, reflection.name, all_associated_records, 'the_parent_record_id')
+        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.send("loaded_#{reflection.name}?")
-        id_to_record_map, ids = construct_id_map(records, reflection.options[:primary_key])
+        return if records.first.send(:association_proxy, reflection.name).loaded?
+        id_to_record_map = construct_id_map(records, reflection.options[:primary_key])
         options = reflection.options
-        records.each {|record| record.send("set_#{reflection.name}_target", nil)}
+
+        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.primary_key_name
+            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).first
+              id_to_record_map    = construct_id_map(records, through_primary_key)
               through_primary_key = through_reflection.klass.primary_key
             end
 
@@ -232,7 +258,7 @@ module ActiveRecord
             end
           end
         else
-          set_association_single_records(id_to_record_map, reflection.name, find_associated_records(ids, reflection, preload_options), reflection.primary_key_name)
+          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
 
@@ -240,9 +266,9 @@ module ActiveRecord
         return if records.first.send(reflection.name).loaded?
         options = reflection.options
 
-        primary_key_name = reflection.through_reflection_primary_key_name
-        id_to_record_map, ids = construct_id_map(records, primary_key_name || reflection.options[:primary_key])
-        records.each {|record| record.send(reflection.name).loaded}
+        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.send(reflection.name).loaded! }
 
         if options[:through]
           through_records = preload_through_records(records, reflection, options[:through])
@@ -257,14 +283,14 @@ module ActiveRecord
           end
 
         else
-          set_association_collection_records(id_to_record_map, reflection.name, find_associated_records(ids, reflection, preload_options),
-                                             reflection.primary_key_name)
+          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.options[:foreign_type]
+          interface = reflection.source_reflection.foreign_type
           preload_options = {:conditions => ["#{connection.quote_column_name interface} = ?", reflection.options[:source_type]]}
 
           records.compact!
@@ -294,76 +320,71 @@ module ActiveRecord
       end
 
       def preload_belongs_to_association(records, reflection, preload_options={})
-        return if records.first.send("loaded_#{reflection.name}?")
+        return if records.first.send(:association_proxy, reflection.name).loaded?
         options = reflection.options
-        primary_key_name = reflection.primary_key_name
 
         klasses_and_ids = {}
 
         if options[:polymorphic]
-          polymorph_type = options[:foreign_type]
-
           # 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(polymorph_type)
-              klass_id = record.send(primary_key_name)
+            if klass = record.send(reflection.foreign_type)
+              klass_id = record.send(reflection.foreign_key)
               if klass_id
-                id_map = klasses_and_ids[klass] ||= {}
+                id_map = klasses_and_ids[klass.constantize] ||= {}
                 (id_map[klass_id.to_s] ||= []) << record
               end
             end
           end
         else
-          id_map = {}
-          records.each do |record|
-            key = record.send(primary_key_name)
-            (id_map[key.to_s] ||= []) << record if key
+          id_map = records.group_by do |record|
+            key = record.send(reflection.foreign_key)
+            key && key.to_s
           end
-          klasses_and_ids[reflection.klass.name] = id_map unless id_map.empty?
+          klasses_and_ids[reflection.klass] = id_map unless id_map.empty?
         end
 
-        klasses_and_ids.each do |klass_name, _id_map|
-          klass = klass_name.constantize
-
-          table_name = klass.quoted_table_name
+        klasses_and_ids.each do |klass, _id_map|
           primary_key = (reflection.options[:primary_key] || klass.primary_key).to_s
-          column_type = klass.columns.detect{|c| c.name == primary_key}.type
-
-          ids = _id_map.keys.map do |id|
-            if column_type == :integer
-              id.to_i
-            elsif column_type == :float
-              id.to_f
-            else
-              id
-            end
-          end
+          keys        = _id_map.keys.compact
 
-          conditions = "#{table_name}.#{connection.quote_column_name(primary_key)} #{in_or_equals_for_ids(ids)}"
-          conditions << append_conditions(reflection, preload_options)
+          unless keys.empty?
+            table       = klass.arel_table
+            method      = in_or_equal(keys)
+            conditions  = table[primary_key].send(*method)
 
-          associated_records = klass.unscoped.where([conditions, ids]).apply_finder_options(options.slice(:include, :select, :joins, :order)).to_a
+            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_name = reflection.klass.quoted_table_name
+        options    = reflection.options
+        table      = reflection.klass.arel_table
+
+        conditions = []
+
+        key = reflection.foreign_key
 
         if interface = reflection.options[:as]
-          conditions = "#{reflection.klass.quoted_table_name}.#{connection.quote_column_name "#{interface}_id"} #{in_or_equals_for_ids(ids)} and #{reflection.klass.quoted_table_name}.#{connection.quote_column_name "#{interface}_type"} = '#{self.base_class.sti_name}'"
-        else
-          foreign_key = reflection.primary_key_name
-          conditions = "#{reflection.klass.quoted_table_name}.#{foreign_key} #{in_or_equals_for_ids(ids)}"
+          key = "#{interface}_id"
+          conditions << table["#{interface}_type"].eq(base_class.sti_name)
         end
 
-        conditions << append_conditions(reflection, preload_options)
+        conditions.concat append_conditions(reflection, preload_options)
 
         find_options = {
-          :select => preload_options[:select] || options[:select] || Arel::SqlLiteral.new("#{table_name}.*"),
+          :select => preload_options[:select] || options[:select] || table[Arel.star],
           :include => preload_options[:include] || options[:include],
           :joins => options[:joins],
           :group => preload_options[:group] || options[:group],
@@ -371,24 +392,30 @@ module ActiveRecord
         }
 
         associated_records(ids) do |some_ids|
-          reflection.klass.scoped.apply_finder_options(find_options.merge(:conditions => [conditions, some_ids])).to_a
+          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
 
-      def interpolate_sql_for_preload(sql)
-        instance_eval("%@#{sql.gsub('@', '\@')}@", __FILE__, __LINE__)
+        klass.send(:sanitize_sql, conditions)
       end
 
       def append_conditions(reflection, preload_options)
-        sql = ""
-        sql << " AND (#{interpolate_sql_for_preload(reflection.sanitized_conditions)})" if reflection.sanitized_conditions
-        sql << " AND (#{sanitize_sql preload_options[:conditions]})" if preload_options[:conditions]
-        sql
+        [
+          ('(' + 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_equals_for_ids(ids)
-        ids.size > 1 ? "IN (?)" : "= ?"
+      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)
@@ -397,7 +424,7 @@ module ActiveRecord
         in_clause_length = connection.in_clause_length || ids.size
         records = []
         ids.each_slice(in_clause_length) do |some_ids|
-          records += yield(some_ids)
+          records.concat yield(some_ids)
         end
         records
       end
diff --git a/activerecord/lib/active_record/associations.rb b/activerecord/lib/active_record/associations.rb
index 371dad9104..a2db592c7d 100644
--- a/activerecord/lib/active_record/associations.rb
+++ b/activerecord/lib/active_record/associations.rb
@@ -20,18 +20,30 @@ module ActiveRecord
     end
   end
 
-  class HasManyThroughAssociationPolymorphicError < ActiveRecordError #:nodoc:
+  class HasManyThroughAssociationPolymorphicSourceError < ActiveRecordError #:nodoc:
     def initialize(owner_class_name, reflection, source_reflection)
       super("Cannot have a has_many :through association '#{owner_class_name}##{reflection.name}' on the polymorphic object '#{source_reflection.class_name}##{source_reflection.name}'.")
     end
   end
 
+  class HasManyThroughAssociationPolymorphicThroughError < ActiveRecordError #:nodoc:
+    def initialize(owner_class_name, reflection)
+      super("Cannot have a has_many :through association '#{owner_class_name}##{reflection.name}' which goes through the polymorphic association '#{owner_class_name}##{reflection.through_reflection.name}'.")
+    end
+  end
+
   class HasManyThroughAssociationPointlessSourceTypeError < ActiveRecordError #:nodoc:
     def initialize(owner_class_name, reflection, source_reflection)
       super("Cannot have a has_many :through association '#{owner_class_name}##{reflection.name}' with a :source_type option if the '#{reflection.through_reflection.class_name}##{source_reflection.name}' is not polymorphic.  Try removing :source_type on your association.")
     end
   end
 
+  class HasOneThroughCantAssociateThroughCollection < ActiveRecordError #:nodoc:
+    def initialize(owner_class_name, reflection, through_reflection)
+      super("Cannot have a has_one :through association '#{owner_class_name}##{reflection.name}' where the :through association '#{owner_class_name}##{through_reflection.name}' is a collection. Specify a has_one or belongs_to association in the :through option instead.")
+    end
+  end
+
   class HasManyThroughSourceAssociationNotFoundError < ActiveRecordError #:nodoc:
     def initialize(reflection)
       through_reflection      = reflection.through_reflection
@@ -107,7 +119,9 @@ module ActiveRecord
     # These classes will be loaded when associations are created.
     # So there is no need to eager load them.
     autoload :AssociationCollection, 'active_record/associations/association_collection'
+    autoload :SingularAssociation, 'active_record/associations/singular_association'
     autoload :AssociationProxy, 'active_record/associations/association_proxy'
+    autoload :ThroughAssociation, 'active_record/associations/through_association'
     autoload :BelongsToAssociation, 'active_record/associations/belongs_to_association'
     autoload :BelongsToPolymorphicAssociation, 'active_record/associations/belongs_to_polymorphic_association'
     autoload :HasAndBelongsToManyAssociation, 'active_record/associations/has_and_belongs_to_many_association'
@@ -118,30 +132,40 @@ module ActiveRecord
 
     # Clears out the association cache.
     def clear_association_cache #:nodoc:
-      self.class.reflect_on_all_associations.to_a.each do |assoc|
-        if IdentityMap.enabled? && instance_variable_defined?("@#{assoc.name}")
-          Array(instance_variable_get("@#{assoc.name}")).each do |t|
-            next unless t.respond_to?(:target)
-            IdentityMap.remove t.target unless t.target.nil?
-          end
-        end
-        instance_variable_set "@#{assoc.name}", nil
-      end if self.persisted?
+      @association_cache.clear if persisted?
     end
 
+    # :nodoc:
+    attr_reader :association_cache
+
+    protected
+
+      # Returns the proxy for the given association name, instantiating it if it doesn't
+      # already exist
+      def association_proxy(name)
+        association = association_instance_get(name)
+
+        if association.nil?
+          reflection  = self.class.reflect_on_association(name)
+          association = reflection.proxy_class.new(self, reflection)
+          association_instance_set(name, association)
+        end
+
+        association
+      end
+
     private
       # Returns the specified association instance if it responds to :loaded?, nil otherwise.
       def association_instance_get(name)
-        ivar = "@#{name}"
-        if instance_variable_defined?(ivar)
-          association = instance_variable_get(ivar)
+        if @association_cache.key? name
+          association = @association_cache[name]
           association if association.respond_to?(:loaded?)
         end
       end
 
       # Set the specified association instance.
       def association_instance_set(name, association)
-        instance_variable_set("@#{name}", association)
+        @association_cache[name] = association
       end
 
     # Associations are a set of macro-like class methods for tying objects together through
@@ -185,7 +209,7 @@ module ActiveRecord
     #   other=(other)                     |     X      |      X       |    X
     #   build_other(attributes={})        |     X      |              |    X
     #   create_other(attributes={})       |     X      |              |    X
-    #   other.create!(attributes={})      |            |              |    X
+    #   create_other!(attributes={})      |     X      |              |    X
     #
     # ===Collection associations (one-to-many / many-to-many)
     #                                     |       |          | has_many
@@ -208,10 +232,9 @@ module ActiveRecord
     #   others.empty?                     |   X   |    X     |    X
     #   others.clear                      |   X   |    X     |    X
     #   others.delete(other,other,...)    |   X   |    X     |    X
-    #   others.delete_all                 |   X   |    X     |
+    #   others.delete_all                 |   X   |    X     |    X
     #   others.destroy_all                |   X   |    X     |    X
     #   others.find(*args)                |   X   |    X     |    X
-    #   others.find_first                 |   X   |          |
     #   others.exists?                    |   X   |    X     |    X
     #   others.uniq                       |   X   |    X     |    X
     #   others.reset                      |   X   |    X     |    X
@@ -325,26 +348,31 @@ module ActiveRecord
     # === One-to-one associations
     #
     # * Assigning an object to a +has_one+ association automatically saves that object and
-    # the object being replaced (if there is one), in order to update their primary
-    # keys - except if the parent object is unsaved (<tt>new_record? == true</tt>).
-    # * If either of these saves fail (due to one of the objects being invalid) the assignment
-    # statement returns +false+ and the assignment is cancelled.
+    #   the object being replaced (if there is one), in order to update their foreign
+    #   keys - except if the parent object is unsaved (<tt>new_record? == true</tt>).
+    # * If either of these saves fail (due to one of the objects being invalid), an
+    #   <tt>ActiveRecord::RecordNotSaved</tt> exception is raised and the assignment is
+    #   cancelled.
     # * If you wish to assign an object to a +has_one+ association without saving it,
-    # use the <tt>association.build</tt> method (documented below).
+    #   use the <tt>build_association</tt> method (documented below). The object being
+    #   replaced will still be saved to update its foreign key.
     # * Assigning an object to a +belongs_to+ association does not save the object, since
-    # the foreign key field belongs on the parent. It does not save the parent either.
+    #   the foreign key field belongs on the parent. It does not save the parent either.
     #
     # === Collections
     #
     # * Adding an object to a collection (+has_many+ or +has_and_belongs_to_many+) automatically
-    # saves that object, except if the parent object (the owner of the collection) is not yet
-    # stored in the database.
+    #   saves that object, except if the parent object (the owner of the collection) is not yet
+    #   stored in the database.
     # * If saving any of the objects being added to a collection (via <tt>push</tt> or similar)
-    # fails, then <tt>push</tt> returns +false+.
+    #   fails, then <tt>push</tt> returns +false+.
+    # * If saving fails while replacing the collection (via <tt>association=</tt>), an
+    #   <tt>ActiveRecord::RecordNotSaved</tt> exception is raised and the assignment is
+    #   cancelled.
     # * You can add an object to a collection without automatically saving it by using the
-    # <tt>collection.build</tt> method (documented below).
+    #   <tt>collection.build</tt> method (documented below).
     # * All unsaved (<tt>new_record? == true</tt>) members of the collection are automatically
-    # saved when the parent is saved.
+    #   saved when the parent is saved.
     #
     # === Association callbacks
     #
@@ -804,6 +832,73 @@ module ActiveRecord
     # * does not work with <tt>:polymorphic</tt> associations.
     # * for +belongs_to+ associations +has_many+ inverse associations are ignored.
     #
+    # == Deleting from associations
+    #
+    # === Dependent associations
+    #
+    # +has_many+, +has_one+ and +belongs_to+ associations support the <tt>:dependent</tt> option.
+    # This allows you to specify that associated records should be deleted when the owner is
+    # deleted.
+    #
+    # For example:
+    #
+    #     class Author
+    #       has_many :posts, :dependent => :destroy
+    #     end
+    #     Author.find(1).destroy # => Will destroy all of the author's posts, too
+    #
+    # The <tt>:dependent</tt> option can have different values which specify how the deletion
+    # is done. For more information, see the documentation for this option on the different
+    # specific association types.
+    #
+    # === Delete or destroy?
+    #
+    # +has_many+ and +has_and_belongs_to_many+ associations have the methods <tt>destroy</tt>,
+    # <tt>delete</tt>, <tt>destroy_all</tt> and <tt>delete_all</tt>.
+    #
+    # For +has_and_belongs_to_many+, <tt>delete</tt> and <tt>destroy</tt> are the same: they
+    # cause the records in the join table to be removed.
+    #
+    # For +has_many+, <tt>destroy</tt> will always call the <tt>destroy</tt> method of the
+    # record(s) being removed so that callbacks are run. However <tt>delete</tt> will either
+    # do the deletion according to the strategy specified by the <tt>:dependent</tt> option, or
+    # if no <tt>:dependent</tt> option is given, then it will follow the default strategy.
+    # The default strategy is <tt>:nullify</tt> (set the foreign keys to <tt>nil</tt>), except for
+    # +has_many+ <tt>:through</tt>, where the default strategy is <tt>delete_all</tt> (delete
+    # the join records, without running their callbacks).
+    #
+    # There is also a <tt>clear</tt> method which is the same as <tt>delete_all</tt>, except that
+    # it returns the association rather than the records which have been deleted.
+    #
+    # === What gets deleted?
+    #
+    # There is a potential pitfall here: +has_and_belongs_to_many+ and +has_many+ <tt>:through</tt>
+    # associations have records in join tables, as well as the associated records. So when we
+    # call one of these deletion methods, what exactly should be deleted?
+    #
+    # The answer is that it is assumed that deletion on an association is about removing the
+    # <i>link</i> between the owner and the associated object(s), rather than necessarily the
+    # associated objects themselves. So with +has_and_belongs_to_many+ and +has_many+
+    # <tt>:through</tt>, the join records will be deleted, but the associated records won't.
+    #
+    # This makes sense if you think about it: if you were to call <tt>post.tags.delete(Tag.find_by_name('food'))</tt>
+    # you would want the 'food' tag to be unlinked from the post, rather than for the tag itself
+    # to be removed from the database.
+    #
+    # However, there are examples where this strategy doesn't make sense. For example, suppose
+    # a person has many projects, and each project has many tasks. If we deleted one of a person's
+    # tasks, we would probably not want the project to be deleted. In this scenario, the delete method
+    # won't actually work: it can only be used if the association on the join model is a
+    # +belongs_to+. In other situations you are expected to perform operations directly on
+    # either the associated records or the <tt>:through</tt> association.
+    #
+    # With a regular +has_many+ there is no distinction between the "associated records"
+    # and the "link", so there is only one choice for what gets deleted.
+    #
+    # With +has_and_belongs_to_many+ and +has_many+ <tt>:through</tt>, if you want to delete the
+    # associated records themselves, you can always do something along the lines of
+    # <tt>person.tasks.each(&:destroy)</tt>.
+    #
     # == Type safety with <tt>ActiveRecord::AssociationTypeMismatch</tt>
     #
     # If you attempt to assign an object to an association that doesn't match the inferred
@@ -828,6 +923,10 @@ module ActiveRecord
       #   Removes one or more objects from the collection by setting their foreign keys to +NULL+.
       #   Objects will be in addition destroyed if they're associated with <tt>:dependent => :destroy</tt>,
       #   and deleted if they're associated with <tt>:dependent => :delete_all</tt>.
+      #
+      #   If the <tt>:through</tt> option is used, then the join records are deleted (rather than
+      #   nullified) by default, but you can specify <tt>:dependent => :destroy</tt> or
+      #   <tt>:dependent => :nullify</tt> to override this.
       # [collection=objects]
       #   Replaces the collections content by deleting and adding objects as appropriate. If the <tt>:through</tt>
       #   option is true callbacks in the join models are triggered except destroy callbacks, since deletion is
@@ -883,7 +982,7 @@ module ActiveRecord
       # * <tt>Firm#clients.create</tt> (similar to <tt>c = Client.new("firm_id" => id); c.save; c</tt>)
       # The declaration can also include an options hash to specialize the behavior of the association.
       #
-      # === Supported options
+      # === Options
       # [:class_name]
       #   Specify the class name of the association. Use it only if that name can't be inferred
       #   from the association name. So <tt>has_many :products</tt> will by default be linked
@@ -911,7 +1010,9 @@ module ActiveRecord
       #   objects' foreign keys are set to +NULL+ *without* calling their +save+ callbacks. If set to
       #   <tt>:restrict</tt> this object cannot be deleted if it has any associated object.
       #
-      #   *Warning:* This option is ignored when used with <tt>:through</tt> option.
+      #   If using with the <tt>:through</tt> option, the association on the join model must be
+      #   a +belongs_to+, and the records which get deleted are the join records, rather than
+      #   the associated records.
       #
       # [:finder_sql]
       #   Specify a complete SQL statement to fetch the association. This is a good way to go for complex
@@ -991,12 +1092,7 @@ module ActiveRecord
         reflection = create_has_many_reflection(association_id, options, &extension)
         configure_dependency_for_has_many(reflection)
         add_association_callbacks(reflection.name, reflection.options)
-
-        if options[:through]
-          collection_accessor_methods(reflection, HasManyThroughAssociation)
-        else
-          collection_accessor_methods(reflection, HasManyAssociation)
-        end
+        collection_accessor_methods(reflection)
       end
 
       # Specifies a one-to-one association with another class. This method should only be used
@@ -1014,12 +1110,14 @@ module ActiveRecord
       # [build_association(attributes = {})]
       #   Returns a new object of the associated type that has been instantiated
       #   with +attributes+ and linked to this object through a foreign key, but has not
-      #   yet been saved. <b>Note:</b> This ONLY works if an association already exists.
-      #   It will NOT work if the association is +nil+.
+      #   yet been saved.
       # [create_association(attributes = {})]
       #   Returns a new object of the associated type that has been instantiated
       #   with +attributes+, linked to this object through a foreign key, and that
       #   has already been saved (if it passed the validation).
+      # [create_association!(attributes = {})]
+      #   Does the same as <tt>create_association</tt>, but raises <tt>ActiveRecord::RecordInvalid</tt>
+      #   if the record is invalid.
       #
       # (+association+ is replaced with the symbol passed as the first argument, so
       # <tt>has_one :manager</tt> would add among others <tt>manager.nil?</tt>.)
@@ -1031,6 +1129,7 @@ module ActiveRecord
       # * <tt>Account#beneficiary=(beneficiary)</tt> (similar to <tt>beneficiary.account_id = account.id; beneficiary.save</tt>)
       # * <tt>Account#build_beneficiary</tt> (similar to <tt>Beneficiary.new("account_id" => id)</tt>)
       # * <tt>Account#create_beneficiary</tt> (similar to <tt>b = Beneficiary.new("account_id" => id); b.save; b</tt>)
+      # * <tt>Account#create_beneficiary!</tt> (similar to <tt>b = Beneficiary.new("account_id" => id); b.save!; b</tt>)
       #
       # === Options
       #
@@ -1108,14 +1207,12 @@ module ActiveRecord
       def has_one(association_id, options = {})
         if options[:through]
           reflection = create_has_one_through_reflection(association_id, options)
-          association_accessor_methods(reflection, ActiveRecord::Associations::HasOneThroughAssociation)
         else
           reflection = create_has_one_reflection(association_id, options)
-          association_accessor_methods(reflection, HasOneAssociation)
-          association_constructor_method(:build,  reflection, HasOneAssociation)
-          association_constructor_method(:create, reflection, HasOneAssociation)
+          association_constructor_methods(reflection)
           configure_dependency_for_has_one(reflection)
         end
+        association_accessor_methods(reflection)
       end
 
       # Specifies a one-to-one association with another class. This method should only be used
@@ -1137,6 +1234,9 @@ module ActiveRecord
       #   Returns a new object of the associated type that has been instantiated
       #   with +attributes+, linked to this object through a foreign key, and that
       #   has already been saved (if it passed the validation).
+      # [create_association!(attributes = {})]
+      #   Does the same as <tt>create_association</tt>, but raises <tt>ActiveRecord::RecordInvalid</tt>
+      #   if the record is invalid.
       #
       # (+association+ is replaced with the symbol passed as the first argument, so
       # <tt>belongs_to :author</tt> would add among others <tt>author.nil?</tt>.)
@@ -1148,6 +1248,7 @@ module ActiveRecord
       # * <tt>Post#author=(author)</tt> (similar to <tt>post.author_id = author.id</tt>)
       # * <tt>Post#build_author</tt> (similar to <tt>post.author = Author.new</tt>)
       # * <tt>Post#create_author</tt> (similar to <tt>post.author = Author.new; post.author.save; post.author</tt>)
+      # * <tt>Post#create_author!</tt> (similar to <tt>post.author = Author.new; post.author.save!; post.author</tt>)
       # The declaration can also include an options hash to specialize the behavior of the association.
       #
       # === Options
@@ -1169,6 +1270,11 @@ module ActiveRecord
       #   association will use "person_id" as the default <tt>:foreign_key</tt>. Similarly,
       #   <tt>belongs_to :favorite_person, :class_name => "Person"</tt> will use a foreign key
       #   of "favorite_person_id".
+      # [:foreign_type]
+      #   Specify the column used to store the associated object's type, if this is a polymorphic
+      #   association. By default this is guessed to be the name of the association with a "_type"
+      #   suffix. So a class that defines a <tt>belongs_to :taggable, :polymorphic => true</tt>
+      #   association will use "taggable_type" as the default <tt>:foreign_type</tt>.
       # [:primary_key]
       #   Specify the method that returns the primary key of associated object used for the association.
       #   By default this is id.
@@ -1227,16 +1333,12 @@ module ActiveRecord
       def belongs_to(association_id, options = {})
         reflection = create_belongs_to_reflection(association_id, options)
 
-        if reflection.options[:polymorphic]
-          association_accessor_methods(reflection, BelongsToPolymorphicAssociation)
-          association_foreign_type_setter_method(reflection)
-        else
-          association_accessor_methods(reflection, BelongsToAssociation)
-          association_constructor_method(:build,  reflection, BelongsToAssociation)
-          association_constructor_method(:create, reflection, BelongsToAssociation)
+        association_accessor_methods(reflection)
+
+        unless reflection.options[:polymorphic]
+          association_constructor_methods(reflection)
         end
 
-        association_foreign_key_setter_method(reflection)
         add_counter_cache_callbacks(reflection)          if options[:counter_cache]
         add_touch_callbacks(reflection, options[:touch]) if options[:touch]
 
@@ -1271,12 +1373,6 @@ module ActiveRecord
       #     end
       #   end
       #
-      # Deprecated: Any additional fields added to the join table will be placed as attributes when
-      # pulling records out through +has_and_belongs_to_many+ associations. Records returned from join
-      # tables with additional attributes will be marked as readonly (because we can't save changes
-      # to the additional attributes). It's strongly recommended that you upgrade any
-      # associations with attributes to a real join model (see introduction).
-      #
       # Adds the following methods for retrieval and query:
       #
       # [collection(force_reload = false)]
@@ -1419,15 +1515,15 @@ module ActiveRecord
       #   'DELETE FROM developers_projects WHERE active=1 AND developer_id = #{id} AND project_id = #{record.id}'
       def has_and_belongs_to_many(association_id, options = {}, &extension)
         reflection = create_has_and_belongs_to_many_reflection(association_id, options, &extension)
-        collection_accessor_methods(reflection, HasAndBelongsToManyAssociation)
+        collection_accessor_methods(reflection)
 
         # Don't use a before_destroy callback since users' before_destroy
         # callbacks will be executed after the association is wiped out.
         include Module.new {
           class_eval <<-RUBY, __FILE__, __LINE__ + 1
             def destroy                     # def destroy
-              #{reflection.name}.clear      #   posts.clear
               super                         #   super
+              #{reflection.name}.clear      #   posts.clear
             end                             # end
           RUBY
         }
@@ -1451,60 +1547,36 @@ module ActiveRecord
           table_name_prefix + join_table + table_name_suffix
         end
 
-        def association_accessor_methods(reflection, association_proxy_class)
+        def association_accessor_methods(reflection)
           redefine_method(reflection.name) do |*params|
             force_reload = params.first unless params.empty?
-            association = association_instance_get(reflection.name)
-
-            if association.nil? || force_reload
-              association = association_proxy_class.new(self, reflection)
-              retval = force_reload ? reflection.klass.uncached { association.reload } : association.reload
-              if retval.nil? and association_proxy_class == BelongsToAssociation
-                association_instance_set(reflection.name, nil)
-                return nil
-              end
-              association_instance_set(reflection.name, association)
-            end
-
-            association.target.nil? ? nil : association
-          end
-
-          redefine_method("loaded_#{reflection.name}?") do
-            association = association_instance_get(reflection.name)
-            association && association.loaded?
-          end
-
-          redefine_method("#{reflection.name}=") do |new_value|
-            association = association_instance_get(reflection.name)
+            association  = association_proxy(reflection.name)
 
-            if association.nil? || association.target != new_value
-              association = association_proxy_class.new(self, reflection)
+            if force_reload
+              reflection.klass.uncached { association.reload }
+            elsif !association.loaded? || association.stale_target?
+              association.reload
             end
 
-            association.replace(new_value)
-            association_instance_set(reflection.name, new_value.nil? ? nil : association)
+            association.target.nil? ? nil : association
           end
 
-          redefine_method("set_#{reflection.name}_target") do |target|
-            return if target.nil? and association_proxy_class == BelongsToAssociation
-            association = association_proxy_class.new(self, reflection)
-            association.target = target
-            association_instance_set(reflection.name, association)
+          redefine_method("#{reflection.name}=") do |record|
+            association_proxy(reflection.name).replace(record)
           end
         end
 
-        def collection_reader_method(reflection, association_proxy_class)
+        def collection_reader_method(reflection)
           redefine_method(reflection.name) do |*params|
             force_reload = params.first unless params.empty?
-            association = association_instance_get(reflection.name)
+            association  = association_proxy(reflection.name)
 
-            unless association
-              association = association_proxy_class.new(self, reflection)
-              association_instance_set(reflection.name, association)
+            if force_reload
+              reflection.klass.uncached { association.reload }
+            elsif association.stale_target?
+              association.reload
             end
 
-            reflection.klass.uncached { association.reload } if force_reload
-
             association
           end
 
@@ -1512,27 +1584,17 @@ module ActiveRecord
             if send(reflection.name).loaded? || reflection.options[:finder_sql]
               send(reflection.name).map { |r| r.id }
             else
-              if reflection.through_reflection && reflection.source_reflection.belongs_to?
-                through = reflection.through_reflection
-                primary_key = reflection.source_reflection.primary_key_name
-                send(through.name).select("DISTINCT #{through.quoted_table_name}.#{primary_key}").map! { |r| r.send(primary_key) }
-              else
-                send(reflection.name).select("#{reflection.quoted_table_name}.#{reflection.klass.primary_key}").except(:includes).map! { |r| r.id }
-              end
+              send(reflection.name).select("#{reflection.quoted_table_name}.#{reflection.klass.primary_key}").except(:includes).map! { |r| r.id }
             end
           end
-
         end
 
-        def collection_accessor_methods(reflection, association_proxy_class, writer = true)
-          collection_reader_method(reflection, association_proxy_class)
+        def collection_accessor_methods(reflection, writer = true)
+          collection_reader_method(reflection)
 
           if writer
             redefine_method("#{reflection.name}=") do |new_value|
-              # Loads proxy class instance (defined in collection_reader_method) if not already loaded
-              association = send(reflection.name)
-              association.replace(new_value)
-              association
+              association_proxy(reflection.name).replace(new_value)
             end
 
             redefine_method("#{reflection.name.to_s.singularize}_ids=") do |new_value|
@@ -1544,62 +1606,19 @@ module ActiveRecord
           end
         end
 
-        def association_constructor_method(constructor, reflection, association_proxy_class)
-          redefine_method("#{constructor}_#{reflection.name}") do |*params|
-            attributees      = params.first unless params.empty?
-            replace_existing = params[1].nil? ? true : params[1]
-            association      = association_instance_get(reflection.name)
-
-            unless association
-              association = association_proxy_class.new(self, reflection)
-              association_instance_set(reflection.name, association)
-            end
-
-            if association_proxy_class == HasOneAssociation
-              association.send(constructor, attributees, replace_existing)
-            else
-              association.send(constructor, attributees)
-            end
-          end
-        end
-
-        def association_foreign_key_setter_method(reflection)
-          setters = reflect_on_all_associations(:belongs_to).map do |belongs_to_reflection|
-            if belongs_to_reflection.primary_key_name == reflection.primary_key_name
-              "association_instance_set(:#{belongs_to_reflection.name}, nil);"
+        def association_constructor_methods(reflection)
+          constructors = {
+            "build_#{reflection.name}"   => "build",
+            "create_#{reflection.name}"  => "create",
+            "create_#{reflection.name}!" => "create!"
+          }
+
+          constructors.each do |name, proxy_name|
+            redefine_method(name) do |*params|
+              attributes = params.first unless params.empty?
+              association_proxy(reflection.name).send(proxy_name, attributes)
             end
-          end.compact.join
-
-          if method_defined?(:"#{reflection.primary_key_name}=")
-            undef_method :"#{reflection.primary_key_name}="
           end
-
-          class_eval <<-FILE, __FILE__, __LINE__ + 1
-            def #{reflection.primary_key_name}=(new_id)
-              write_attribute :#{reflection.primary_key_name}, new_id
-              if #{reflection.primary_key_name}_changed?
-                #{ setters }
-              end
-            end
-          FILE
-        end
-
-        def association_foreign_type_setter_method(reflection)
-          setters = reflect_on_all_associations(:belongs_to).map do |belongs_to_reflection|
-            if belongs_to_reflection.options[:foreign_type] == reflection.options[:foreign_type]
-              "association_instance_set(:#{belongs_to_reflection.name}, nil);"
-            end
-          end.compact.join
-
-          field = reflection.options[:foreign_type]
-          class_eval <<-FILE, __FILE__, __LINE__ + 1
-            def #{field}=(new_id)
-              write_attribute :#{field}, new_id
-              if #{field}_changed?
-                #{ setters }
-              end
-            end
-          FILE
         end
 
         def add_counter_cache_callbacks(reflection)
@@ -1648,56 +1667,39 @@ module ActiveRecord
         #  - set the foreign key to NULL if the option is set to :nullify
         #  - do not delete the parent record if there is any child record if the
         #    option is set to :restrict
-        #
-        # The +extra_conditions+ parameter, which is not used within the main
-        # Active Record codebase, is meant to allow plugins to define extra
-        # finder conditions.
-        def configure_dependency_for_has_many(reflection, extra_conditions = nil)
-          if reflection.options.include?(:dependent)
+        def configure_dependency_for_has_many(reflection)
+          if reflection.options[:dependent]
+            method_name = "has_many_dependent_for_#{reflection.name}"
+
             case reflection.options[:dependent]
-              when :destroy
-                method_name = "has_many_dependent_destroy_for_#{reflection.name}".to_sym
-                define_method(method_name) do
+            when :destroy, :delete_all, :nullify
+              define_method(method_name) do
+                if reflection.options[:dependent] == :destroy
                   send(reflection.name).each do |o|
                     # No point in executing the counter update since we're going to destroy the parent anyway
                     counter_method = ('belongs_to_counter_cache_before_destroy_for_' + self.class.name.downcase).to_sym
-                    if(o.respond_to? counter_method) then
+                    if o.respond_to?(counter_method)
                       class << o
                         self
                       end.send(:define_method, counter_method, Proc.new {})
                     end
-                    o.destroy
                   end
                 end
-                before_destroy method_name
-              when :delete_all
-                before_destroy do |record|
-                  self.class.send(:delete_all_has_many_dependencies,
-                  record,
-                  reflection.name,
-                  reflection.klass,
-                  reflection.dependent_conditions(record, self.class, extra_conditions))
-                end
-              when :nullify
-                before_destroy do |record|
-                  self.class.send(:nullify_has_many_dependencies,
-                  record,
-                  reflection.name,
-                  reflection.klass,
-                  reflection.primary_key_name,
-                  reflection.dependent_conditions(record, self.class, extra_conditions))
-                end
-              when :restrict
-                method_name = "has_many_dependent_restrict_for_#{reflection.name}".to_sym
-                define_method(method_name) do
-                  unless send(reflection.name).empty?
-                    raise DeleteRestrictionError.new(reflection)
-                  end
+
+                # AssociationProxy#delete_all looks at the :dependent option and acts accordingly
+                send(reflection.name).delete_all
+              end
+            when :restrict
+              define_method(method_name) do
+                unless send(reflection.name).empty?
+                  raise DeleteRestrictionError.new(reflection)
                 end
-                before_destroy method_name
-              else
-                raise ArgumentError, "The :dependent option expects either :destroy, :delete_all, :nullify or :restrict (#{reflection.options[:dependent].inspect})"
+              end
+            else
+              raise ArgumentError, "The :dependent option expects either :destroy, :delete_all, :nullify or :restrict (#{reflection.options[:dependent].inspect})"
             end
+
+            before_destroy method_name
           end
         end
 
@@ -1722,7 +1724,7 @@ module ActiveRecord
                 class_eval <<-eoruby, __FILE__, __LINE__ + 1
                   def #{method_name}
                     association = #{reflection.name}
-                    association.update_attribute(#{reflection.primary_key_name.inspect}, nil) if association
+                    association.update_attribute(#{reflection.foreign_key.inspect}, nil) if association
                   end
                 eoruby
               when :restrict
@@ -1760,14 +1762,6 @@ module ActiveRecord
           end
         end
 
-        def delete_all_has_many_dependencies(record, reflection_name, association_class, dependent_conditions)
-          association_class.delete_all(dependent_conditions)
-        end
-
-        def nullify_has_many_dependencies(record, reflection_name, association_class, primary_key_name, dependent_conditions)
-          association_class.update_all("#{primary_key_name} = NULL", dependent_conditions)
-        end
-
         mattr_accessor :valid_keys_for_has_many_association
         @@valid_keys_for_has_many_association = [
           :class_name, :table_name, :foreign_key, :primary_key,
@@ -1816,13 +1810,7 @@ module ActiveRecord
 
         def create_belongs_to_reflection(association_id, options)
           options.assert_valid_keys(valid_keys_for_belongs_to_association)
-          reflection = create_reflection(:belongs_to, association_id, options, self)
-
-          if options[:polymorphic]
-            reflection.options[:foreign_type] ||= reflection.class_name.underscore + "_type"
-          end
-
-          reflection
+          create_reflection(:belongs_to, association_id, options, self)
         end
 
         mattr_accessor :valid_keys_for_has_and_belongs_to_many_association
@@ -1842,7 +1830,7 @@ module ActiveRecord
 
           reflection = create_reflection(:has_and_belongs_to_many, association_id, options, self)
 
-          if reflection.association_foreign_key == reflection.primary_key_name
+          if reflection.association_foreign_key == reflection.foreign_key
             raise HasAndBelongsToManyAssociationForeignKeyNeeded.new(reflection)
           end
 
diff --git a/activerecord/lib/active_record/associations/association_collection.rb b/activerecord/lib/active_record/associations/association_collection.rb
index 11a7a725e5..ca350f51c9 100644
--- a/activerecord/lib/active_record/associations/association_collection.rb
+++ b/activerecord/lib/active_record/associations/association_collection.rb
@@ -1,4 +1,3 @@
-require 'set'
 require 'active_support/core_ext/array/wrap'
 
 module ActiveRecord
@@ -23,98 +22,55 @@ module ActiveRecord
 
       def select(select = nil)
         if block_given?
-          load_target
-          @target.select.each { |e| yield e }
+          load_target.select.each { |e| yield e }
         else
           scoped.select(select)
         end
       end
 
-      def scoped
-        with_scope(@scope) { @reflection.klass.scoped }
-      end
-
       def find(*args)
-        options = args.extract_options!
-
-        # If using a custom finder_sql, scan the entire collection.
         if @reflection.options[:finder_sql]
-          expects_array = args.first.kind_of?(Array)
-          ids           = args.flatten.compact.uniq.map { |arg| arg.to_i }
-
-          if ids.size == 1
-            id = ids.first
-            record = load_target.detect { |r| id == r.id }
-            expects_array ? [ record ] : record
-          else
-            load_target.select { |r| ids.include?(r.id) }
-          end
+          find_by_scan(*args)
         else
-          merge_options_from_reflection!(options)
-          construct_find_options!(options)
-
-          with_scope(:find => @scope[:find].slice(:conditions, :order)) do
-            relation = @reflection.klass.send(:construct_finder_arel, options, @reflection.klass.send(:current_scoped_methods))
-
-            case args.first
-            when :first, :last
-              relation.send(args.first)
-            when :all
-              records = relation.all
-              @reflection.options[:uniq] ? uniq(records) : records
-            else
-              relation.find(*args)
-            end
-          end
+          scoped.find(*args)
         end
       end
 
-      # Fetches the first one using SQL if possible.
       def first(*args)
-        if fetch_first_or_last_using_find?(args)
-          find(:first, *args)
-        else
-          load_target unless loaded?
-          args = args[1..-1] if args.first.kind_of?(Hash) && args.first.empty?
-          @target.first(*args)
-        end
+        first_or_last(:first, *args)
       end
 
-      # Fetches the last one using SQL if possible.
       def last(*args)
-        if fetch_first_or_last_using_find?(args)
-          find(:last, *args)
-        else
-          load_target unless loaded?
-          @target.last(*args)
-        end
+        first_or_last(:last, *args)
       end
 
       def to_ary
-        load_target
-        if @target.is_a?(Array)
-          @target.to_ary
-        else
-          Array.wrap(@target)
-        end
+        load_target.dup
       end
       alias_method :to_a, :to_ary
 
       def reset
-        reset_target!
-        reset_scopes_cache!
-        @loaded = false
+        @_scopes_cache = {}
+        @loaded        = false
+        @target        = []
       end
 
       def build(attributes = {}, &block)
-        if attributes.is_a?(Array)
-          attributes.collect { |attr| build(attr, &block) }
-        else
-          build_record(attributes) do |record|
-            block.call(record) if block_given?
-            set_belongs_to_association_for(record)
-          end
+        build_or_create(attributes, :build, &block)
+      end
+
+      def create(attributes = {}, &block)
+        unless @owner.persisted?
+          raise ActiveRecord::RecordNotSaved, "You cannot call create unless the parent is saved"
         end
+
+        build_or_create(attributes, :create, &block)
+      end
+
+      def create!(attrs = {}, &block)
+        record = create(attrs, &block)
+        Array.wrap(record).each(&:save!)
+        record
       end
 
       # Add +records+ to this association.  Returns +self+ so method calls may be chained.
@@ -124,9 +80,9 @@ module ActiveRecord
         load_target if @owner.new_record?
 
         transaction do
-          flatten_deeper(records).each do |record|
+          records.flatten.each do |record|
             raise_on_type_mismatch(record)
-            add_record_to_target_with_callbacks(record) do |r|
+            add_to_target(record) do |r|
               result &&= insert_record(record) unless @owner.new_record?
             end
           end
@@ -157,18 +113,35 @@ module ActiveRecord
       #
       # See delete for more info.
       def delete_all
-        load_target
-        delete(@target)
-        reset_target!
-        reset_scopes_cache!
+        delete(load_target).tap do
+          reset
+          loaded!
+        end
+      end
+
+      # Identical to delete_all, except that the return value is the association (for chaining)
+      # rather than the records which have been removed.
+      def clear
+        delete_all
+        self
+      end
+
+      # Destroy all the records from this association.
+      #
+      # See destroy for more info.
+      def destroy_all
+        destroy(load_target).tap do
+          reset
+          loaded!
+        end
       end
 
       # Calculate sum using SQL, not Enumerable
       def sum(*args)
         if block_given?
-          calculate(:sum, *args) { |*block_args| yield(*block_args) }
+          scoped.sum(*args) { |*block_args| yield(*block_args) }
         else
-          calculate(:sum, *args)
+          scoped.sum(*args)
         end
       end
 
@@ -185,14 +158,13 @@ module ActiveRecord
 
           @reflection.klass.count_by_sql(custom_counter_sql)
         else
-
           if @reflection.options[:uniq]
             # This is needed because 'SELECT count(DISTINCT *)..' is not valid SQL.
-            column_name = "#{@reflection.quoted_table_name}.#{@reflection.klass.primary_key}" unless column_name
+            column_name ||= @reflection.klass.primary_key
             options.merge!(:distinct => true)
           end
 
-          value = @reflection.klass.send(:with_scope, @scope) { @reflection.klass.count(column_name, options) }
+          value = scoped.count(column_name, options)
 
           limit  = @reflection.options[:limit]
           offset = @reflection.options[:offset]
@@ -213,10 +185,7 @@ module ActiveRecord
       # are actually removed from the database, that depends precisely on
       # +delete_records+. They are in any case removed from the collection.
       def delete(*records)
-        remove_records(records) do |_records, old_records|
-          delete_records(old_records) if old_records.any?
-          _records.each { |record| @target.delete(record) }
-        end
+        delete_or_destroy(records, @reflection.options[:dependent])
       end
 
       # Destroy +records+ and remove them from this association calling
@@ -225,54 +194,8 @@ module ActiveRecord
       # Note that this method will _always_ remove records from the database
       # ignoring the +:dependent+ option.
       def destroy(*records)
-        records = find(records) if records.any? {|record| record.kind_of?(Fixnum) || record.kind_of?(String)}
-        remove_records(records) do |_records, old_records|
-          old_records.each { |record| record.destroy }
-        end
-
-        load_target
-      end
-
-      # Removes all records from this association.  Returns +self+ so method calls may be chained.
-      def clear
-        unless length.zero? # forces load_target if it hasn't happened already
-          if @reflection.options[:dependent] == :destroy
-            destroy_all
-          else
-            delete_all
-          end
-        end
-
-        self
-      end
-
-      # Destroy all the records from this association.
-      #
-      # See destroy for more info.
-      def destroy_all
-        load_target
-        destroy(@target).tap do
-          reset_target!
-          reset_scopes_cache!
-        end
-      end
-
-      def create(attrs = {})
-        if attrs.is_a?(Array)
-          attrs.collect { |attr| create(attr) }
-        else
-          create_record(attrs) do |record|
-            yield(record) if block_given?
-            record.save
-          end
-        end
-      end
-
-      def create!(attrs = {})
-        create_record(attrs) do |record|
-          yield(record) if block_given?
-          record.save!
-        end
+        records = find(records) if records.any? { |record| record.kind_of?(Fixnum) || record.kind_of?(String) }
+        delete_or_destroy(records, :destroy)
       end
 
       # Returns the size of the collection by executing a SELECT COUNT(*)
@@ -316,7 +239,7 @@ module ActiveRecord
 
       def any?
         if block_given?
-          method_missing(:any?) { |*block_args| yield(*block_args) }
+          load_target.any? { |*block_args| yield(*block_args) }
         else
           !empty?
         end
@@ -325,7 +248,7 @@ module ActiveRecord
       # Returns true if the collection has more than 1 record. Equivalent to collection.size > 1.
       def many?
         if block_given?
-          method_missing(:many?) { |*block_args| yield(*block_args) }
+          load_target.many? { |*block_args| yield(*block_args) }
         else
           size > 1
         end
@@ -342,108 +265,117 @@ module ActiveRecord
       # This will perform a diff and delete/add only records that have changed.
       def replace(other_array)
         other_array.each { |val| raise_on_type_mismatch(val) }
-
-        load_target
-        other   = other_array.size < 100 ? other_array : other_array.to_set
-        current = @target.size < 100 ? @target : @target.to_set
+        original_target = load_target.dup
 
         transaction do
-          delete(@target.select { |v| !other.include?(v) })
-          concat(other_array.select { |v| !current.include?(v) })
+          delete(@target - other_array)
+
+          unless concat(other_array - @target)
+            @target = original_target
+            raise RecordNotSaved, "Failed to replace #{@reflection.name} because one or more of the " \
+                                  "new records could not be saved."
+          end
         end
       end
 
       def include?(record)
-        return false unless record.is_a?(@reflection.klass)
-        return include_in_memory?(record) if record.new_record?
-        load_target if @reflection.options[:finder_sql] && !loaded?
-        loaded? ? @target.include?(record) : exists?(record)
+        if record.is_a?(@reflection.klass)
+          if record.new_record?
+            include_in_memory?(record)
+          else
+            load_target if @reflection.options[:finder_sql]
+            loaded? ? @target.include?(record) : scoped.exists?(record)
+          end
+        else
+          false
+        end
       end
 
-      def proxy_respond_to?(method, include_private = false)
+      def respond_to?(method, include_private = false)
         super || @reflection.klass.respond_to?(method, include_private)
       end
 
+      def method_missing(method, *args, &block)
+        match = DynamicFinderMatch.match(method)
+        if match && match.creator?
+          attributes = match.attribute_names
+          return send(:"find_by_#{attributes.join('_and_')}", *args) || create(Hash[attributes.zip(args)])
+        end
+
+        if @target.respond_to?(method) || (!@reflection.klass.respond_to?(method) && Class.respond_to?(method))
+          super
+        elsif @reflection.klass.scopes[method]
+          @_scopes_cache ||= {}
+          @_scopes_cache[method] ||= {}
+          @_scopes_cache[method][args] ||= scoped.readonly(nil).send(method, *args)
+        else
+          scoped.readonly(nil).send(method, *args, &block)
+        end
+      end
+
       protected
-        def construct_find_options!(options)
+
+        def association_scope
+          options = @reflection.options.slice(:order, :limit, :joins, :group, :having, :offset)
+          super.apply_finder_options(options)
         end
 
         def load_target
-          if !@owner.new_record? || foreign_key_present
+          if find_target?
+            targets = []
+
             begin
-              unless loaded?
-                if @target.is_a?(Array) && @target.any?
-                  @target = find_target.map do |f|
-                    i = @target.index(f)
-                    if i
-                      @target.delete_at(i).tap do |t|
-                        keys = ["id"] + t.changes.keys + (f.attribute_names - t.attribute_names)
-                        f.attributes.except(*keys).each do |k,v|
-                          t.send("#{k}=", v)
-                        end
-                      end
-                    else
-                      f
-                    end
-                  end + @target
-                else
-                  @target = find_target
-                end
-              end
+              targets = find_target
             rescue ActiveRecord::RecordNotFound
               reset
             end
+
+            @target = merge_target_lists(targets, @target)
           end
 
-          loaded if target
+          loaded!
           target
         end
 
-        def method_missing(method, *args)
-          match = DynamicFinderMatch.match(method)
-          if match && match.creator?
-            attributes = match.attribute_names
-            return send(:"find_by_#{attributes.join('_and_')}", *args) || create(Hash[attributes.zip(args)])
-          end
+        def add_to_target(record)
+          transaction do
+            callback(:before_add, record)
+            yield(record) if block_given?
 
-          if @target.respond_to?(method) || (!@reflection.klass.respond_to?(method) && Class.respond_to?(method))
-            super
-          elsif @reflection.klass.scopes[method]
-            @_scopes_cache ||= {}
-            @_scopes_cache[method] ||= {}
-            @_scopes_cache[method][args] ||= with_scope(@scope) { @reflection.klass.send(method, *args) }
-          else
-            with_scope(@scope) do
-              if block_given?
-                @reflection.klass.send(method, *args) { |*block_args| yield(*block_args) }
-              else
-                @reflection.klass.send(method, *args)
-              end
+            if @reflection.options[:uniq] && index = @target.index(record)
+              @target[index] = record
+            else
+              @target << record
             end
+
+            callback(:after_add, record)
+            set_inverse_instance(record)
           end
+
+          record
+        end
+
+      private
+
+        def select_value
+          super || uniq_select_value
+        end
+
+        def uniq_select_value
+          @reflection.options[:uniq] && "DISTINCT #{@reflection.quoted_table_name}.*"
         end
 
         def custom_counter_sql
           if @reflection.options[:counter_sql]
-            counter_sql = @reflection.options[:counter_sql]
+            interpolate(@reflection.options[:counter_sql])
           else
             # replace the SELECT clause with COUNT(*), preserving any hints within /* ... */
-            counter_sql = @reflection.options[:finder_sql].sub(/SELECT\b(\/\*.*?\*\/ )?(.*)\bFROM\b/im) { "SELECT #{$1}COUNT(*) FROM" }
+            interpolate(@reflection.options[:finder_sql]).sub(/SELECT\b(\/\*.*?\*\/ )?(.*)\bFROM\b/im) { "SELECT #{$1}COUNT(*) FROM" }
           end
-
-          interpolate_sql(counter_sql)
         end
 
         def custom_finder_sql
-          interpolate_sql(@reflection.options[:finder_sql])
-        end
-
-        def reset_target!
-          @target = Array.new
-        end
-
-        def reset_scopes_cache!
-          @_scopes_cache = {}
+          interpolate(@reflection.options[:finder_sql])
         end
 
         def find_target
@@ -455,65 +387,74 @@ module ActiveRecord
             end
 
           records = @reflection.options[:uniq] ? uniq(records) : records
-          records.each do |record|
-            set_inverse_instance(record, @owner)
-          end
+          records.each { |record| set_inverse_instance(record) }
           records
         end
 
-        def add_record_to_target_with_callbacks(record)
-          callback(:before_add, record)
-          yield(record) if block_given?
-          @target ||= [] unless loaded?
-          if index = @target.index(record)
-            @target[index] = record
-          else
-             @target << record
-          end
-          callback(:after_add, record)
-          set_inverse_instance(record, @owner)
-          record
+        def merge_target_lists(loaded, existing)
+          return loaded if existing.empty?
+          return existing if loaded.empty?
+
+          loaded.map do |f|
+            i = existing.index(f)
+            if i
+              existing.delete_at(i).tap do |t|
+                keys = ["id"] + t.changes.keys + (f.attribute_names - t.attribute_names)
+                # FIXME: this call to attributes causes many NoMethodErrors
+                attributes = f.attributes
+                (attributes.keys - keys).each do |k|
+                  t.send("#{k}=", attributes[k])
+                end
+              end
+            else
+              f
+            end
+          end + existing
         end
 
-      private
-        def create_record(attrs)
-          attrs.update(@reflection.options[:conditions]) if @reflection.options[:conditions].is_a?(Hash)
-          ensure_owner_is_persisted!
-
-          scoped_where = scoped.where_values_hash
-          create_scope = scoped_where ? @scope[:create].merge(scoped_where) : @scope[:create]
-          record = @reflection.klass.send(:with_scope, :create => create_scope) do
-            @reflection.build_association(attrs)
-          end
-          if block_given?
-            add_record_to_target_with_callbacks(record) { |*block_args| yield(*block_args) }
-          else
-            add_record_to_target_with_callbacks(record)
+        def build_or_create(attributes, method)
+          records = Array.wrap(attributes).map do |attrs|
+            record = build_record(attrs)
+
+            add_to_target(record) do
+              yield(record) if block_given?
+              insert_record(record) if method == :create
+            end
           end
+
+          attributes.is_a?(Array) ? records : records.first
         end
 
-        def build_record(attrs)
-          attrs.update(@reflection.options[:conditions]) if @reflection.options[:conditions].is_a?(Hash)
-          record = @reflection.build_association(attrs)
-          if block_given?
-            add_record_to_target_with_callbacks(record) { |*block_args| yield(*block_args) }
-          else
-            add_record_to_target_with_callbacks(record)
-          end
+        # Do the relevant stuff to insert the given record into the association collection.
+        def insert_record(record, validate = true)
+          raise NotImplementedError
         end
 
-        def remove_records(*records)
-          records = flatten_deeper(records)
+        def build_record(attributes)
+          @reflection.build_association(scoped.scope_for_create.merge(attributes))
+        end
+
+        def delete_or_destroy(records, method)
+          records = records.flatten
           records.each { |record| raise_on_type_mismatch(record) }
+          existing_records = records.reject { |r| r.new_record? }
 
           transaction do
             records.each { |record| callback(:before_remove, record) }
-            old_records = records.reject { |r| r.new_record? }
-            yield(records, old_records)
+
+            delete_records(existing_records, method) if existing_records.any?
+            records.each { |record| @target.delete(record) }
+
             records.each { |record| callback(:after_remove, record) }
           end
         end
 
+        # Delete the given records from the association, using one of the methods :destroy,
+        # :delete_all or :nullify (or nil, in which case a default is used).
+        def delete_records(records, method)
+          raise NotImplementedError
+        end
+
         def callback(method, record)
           callbacks_for(method).each do |callback|
             case callback
@@ -532,27 +473,61 @@ module ActiveRecord
           @owner.class.send(full_callback_name.to_sym) || []
         end
 
-        def ensure_owner_is_persisted!
-          unless @owner.persisted?
-            raise ActiveRecord::RecordNotSaved, "You cannot call create unless the parent is saved"
-          end
-        end
-
+        # Should we deal with assoc.first or assoc.last by issuing an independent query to
+        # the database, or by getting the target, and then taking the first/last item from that?
+        #
+        # If the args is just a non-empty options hash, go to the database.
+        #
+        # Otherwise, go to the database only if none of the following are true:
+        #   * target already loaded
+        #   * owner is new record
+        #   * custom :finder_sql exists
+        #   * target contains new or changed record(s)
+        #   * the first arg is an integer (which indicates the number of records to be returned)
         def fetch_first_or_last_using_find?(args)
-          (args.first.kind_of?(Hash) && !args.first.empty?) || !(loaded? || @owner.new_record? || @reflection.options[:finder_sql] ||
-            @target.any? { |record| record.new_record? } || args.first.kind_of?(Integer))
+          if args.first.is_a?(Hash)
+            true
+          else
+            !(loaded? ||
+              @owner.new_record? ||
+              @reflection.options[:finder_sql] ||
+              @target.any? { |record| record.new_record? || record.changed? } ||
+              args.first.kind_of?(Integer))
+          end
         end
 
         def include_in_memory?(record)
           if @reflection.is_a?(ActiveRecord::Reflection::ThroughReflection)
-            @owner.send(proxy_reflection.through_reflection.name.to_sym).any? do |source|
+            @owner.send(proxy_reflection.through_reflection.name).any? { |source|
               target = source.send(proxy_reflection.source_reflection.name)
               target.respond_to?(:include?) ? target.include?(record) : target == record
-            end
+            } || @target.include?(record)
           else
             @target.include?(record)
           end
         end
+
+        # If using a custom finder_sql, #find scans the entire collection.
+        def find_by_scan(*args)
+          expects_array = args.first.kind_of?(Array)
+          ids           = args.flatten.compact.uniq.map { |arg| arg.to_i }
+
+          if ids.size == 1
+            id = ids.first
+            record = load_target.detect { |r| id == r.id }
+            expects_array ? [ record ] : record
+          else
+            load_target.select { |r| ids.include?(r.id) }
+          end
+        end
+
+        # Fetches the first/last using SQL if possible, otherwise from the target array.
+        def first_or_last(type, *args)
+          args.shift if args.first.is_a?(Hash) && args.first.empty?
+
+          collection = fetch_first_or_last_using_find?(args) ? scoped : load_target
+          collection.send(type, *args)
+        end
     end
   end
 end
diff --git a/activerecord/lib/active_record/associations/association_proxy.rb b/activerecord/lib/active_record/associations/association_proxy.rb
index 53ec5a0da6..d16cda5585 100644
--- a/activerecord/lib/active_record/associations/association_proxy.rb
+++ b/activerecord/lib/active_record/associations/association_proxy.rb
@@ -4,17 +4,18 @@ module ActiveRecord
   module Associations
     # = Active Record Associations
     #
-    # This is the root class of all association proxies:
+    # This is the root class of all association proxies ('+ Foo' signifies an included module Foo):
     #
     #   AssociationProxy
-    #     BelongsToAssociation
+    #     SingularAssociaton
     #       HasOneAssociation
-    #     BelongsToPolymorphicAssociation
+    #         HasOneThroughAssociation + ThroughAssociation
+    #       BelongsToAssociation
+    #         BelongsToPolymorphicAssociation
     #     AssociationCollection
     #       HasAndBelongsToManyAssociation
     #       HasManyAssociation
-    #         HasManyThroughAssociation
-    #            HasOneThroughAssociation
+    #         HasManyThroughAssociation + ThroughAssociation
     #
     # Association proxies in Active Record are middlemen between the object that
     # holds the association, known as the <tt>@owner</tt>, and the actual associated
@@ -50,10 +51,9 @@ module ActiveRecord
     # is computed directly through SQL and does not trigger by itself the
     # instantiation of the actual post records.
     class AssociationProxy #:nodoc:
-      alias_method :proxy_respond_to?, :respond_to?
       alias_method :proxy_extend, :extend
-      delegate :to_param, :to => :proxy_target
-      instance_methods.each { |m| undef_method m unless m.to_s =~ /^(?:nil\?|send|object_id|to_a)$|^__|^respond_to_missing|proxy_/ }
+
+      instance_methods.each { |m| undef_method m unless m.to_s =~ /^(?:nil\?|send|object_id|to_a)$|^__|^respond_to|proxy_/ }
 
       def initialize(owner, reflection)
         @owner, @reflection = owner, reflection
@@ -64,6 +64,10 @@ module ActiveRecord
         construct_scope
       end
 
+      def to_param
+        proxy_target.to_param
+      end
+
       # Returns the owner of the proxy.
       def proxy_owner
         @owner
@@ -75,21 +79,25 @@ module ActiveRecord
         @reflection
       end
 
-      # Returns the \target of the proxy, same as +target+.
-      def proxy_target
-        @target
-      end
-
       # Does the proxy or its \target respond to +symbol+?
       def respond_to?(*args)
-        proxy_respond_to?(*args) || (load_target && @target.respond_to?(*args))
+        super || (load_target && @target.respond_to?(*args))
+      end
+
+      # Forwards any missing method call to the \target.
+      def method_missing(method, *args, &block)
+        if load_target
+          return super unless @target.respond_to?(method)
+          @target.send(method, *args, &block)
+        end
+      rescue NoMethodError => e
+        raise e, e.message.sub(/ for #<.*$/, " via proxy for #{@target}")
       end
 
       # Forwards <tt>===</tt> explicitly to the \target because the instance method
       # removal above doesn't catch it. Loads the \target if needed.
       def ===(other)
-        load_target
-        other === @target
+        other === load_target
       end
 
       # Returns the name of the table of the related class:
@@ -100,13 +108,6 @@ module ActiveRecord
         @reflection.klass.table_name
       end
 
-      # Returns the SQL string that corresponds to the <tt>:conditions</tt>
-      # option of the macro, if given, or +nil+ otherwise.
-      def conditions
-        @conditions ||= interpolate_sql(@reflection.sanitized_conditions) if @reflection.sanitized_conditions
-      end
-      alias :sql_conditions :conditions
-
       # Resets the \loaded flag to +false+ and sets the \target to +nil+.
       def reset
         @loaded = false
@@ -117,6 +118,7 @@ module ActiveRecord
       # Reloads the \target and returns +self+ on success.
       def reload
         reset
+        construct_scope
         load_target
         self unless @target.nil?
       end
@@ -127,117 +129,93 @@ module ActiveRecord
       end
 
       # Asserts the \target has been loaded setting the \loaded flag to +true+.
-      def loaded
-        @loaded = true
+      def loaded!
+        @loaded      = true
+        @stale_state = stale_state
       end
 
-      # Returns the target of this proxy, same as +proxy_target+.
-      def target
-        @target
+      # The target is stale if the target no longer points to the record(s) that the
+      # relevant foreign_key(s) refers to. If stale, the association accessor method
+      # on the owner will reload the target. It's up to subclasses to implement the
+      # state_state method if relevant.
+      #
+      # Note that if the target has not been loaded, it is not considered stale.
+      def stale_target?
+        loaded? && @stale_state != stale_state
       end
 
+      # Returns the target of this proxy, same as +proxy_target+.
+      attr_reader :target
+
+      # Returns the \target of the proxy, same as +target+.
+      alias :proxy_target :target
+
       # Sets the target of this proxy to <tt>\target</tt>, and the \loaded flag to +true+.
       def target=(target)
         @target = target
-        loaded
+        loaded!
       end
 
       # Forwards the call to the target. Loads the \target if needed.
       def inspect
-        load_target
-        @target.inspect
+        load_target.inspect
       end
 
       def send(method, *args)
-        if proxy_respond_to?(method)
-          super
-        else
-          load_target
-          @target.send(method, *args)
-        end
+        return super if respond_to?(method)
+        load_target.send(method, *args)
       end
 
-      protected
-        # Does the association have a <tt>:dependent</tt> option?
-        def dependent?
-          @reflection.options[:dependent]
-        end
+      def scoped
+        target_scope.merge(@association_scope)
+      end
 
-        def interpolate_sql(sql, record = nil)
-          @owner.send(:interpolate_sql, sql, record)
-        end
+      protected
 
-        # Forwards the call to the reflection class.
-        def sanitize_sql(sql, table_name = @reflection.klass.table_name)
-          @reflection.klass.send(:sanitize_sql, sql, table_name)
+        # Construct the scope for this association.
+        #
+        # Note that the association_scope is merged into the targed_scope only when the
+        # scoped method is called. This is because at that point the call may be surrounded
+        # by scope.scoping { ... } or with_scope { ... } etc, which affects the scope which
+        # actually gets built.
+        def construct_scope
+          @association_scope = association_scope if target_klass
         end
 
-        # Assigns the ID of the owner to the corresponding foreign key in +record+.
-        # If the association is polymorphic the type of the owner is also set.
-        def set_belongs_to_association_for(record)
-          if @reflection.options[:as]
-            record["#{@reflection.options[:as]}_id"]   = @owner.id if @owner.persisted?
-            record["#{@reflection.options[:as]}_type"] = @owner.class.base_class.name.to_s
-          else
-            if @owner.persisted?
-              primary_key = @reflection.options[:primary_key] || :id
-              record[@reflection.primary_key_name] = @owner.send(primary_key)
-            end
+        def association_scope
+          scope = target_klass.unscoped
+          scope = scope.create_with(creation_attributes)
+          scope = scope.apply_finder_options(@reflection.options.slice(:readonly, :include))
+          scope = scope.where(interpolate(@reflection.options[:conditions]))
+          if select = select_value
+            scope = scope.select(select)
           end
+          scope = scope.extending(*Array.wrap(@reflection.options[:extend]))
+          scope.where(construct_owner_conditions)
         end
 
-        # Merges into +options+ the ones coming from the reflection.
-        def merge_options_from_reflection!(options)
-          options.reverse_merge!(
-            :group   => @reflection.options[:group],
-            :having  => @reflection.options[:having],
-            :limit   => @reflection.options[:limit],
-            :offset  => @reflection.options[:offset],
-            :joins   => @reflection.options[:joins],
-            :include => @reflection.options[:include],
-            :select  => @reflection.options[:select],
-            :readonly  => @reflection.options[:readonly]
-          )
+        def aliased_table
+          target_klass.arel_table
         end
 
-        # Forwards +with_scope+ to the reflection.
-        def with_scope(*args, &block)
-          @reflection.klass.send :with_scope, *args, &block
-        end
-
-        # Construct the scope used for find/create queries on the target
-        def construct_scope
-          @scope = {
-            :find   => construct_find_scope,
-            :create => construct_create_scope
-          }
+        # Set the inverse association, if possible
+        def set_inverse_instance(record)
+          if record && invertible_for?(record)
+            inverse = record.send(:association_proxy, inverse_reflection_for(record).name)
+            inverse.target = @owner
+          end
         end
 
-        # Implemented by subclasses
-        def construct_find_scope
-          raise NotImplementedError
+        # This class of the target. belongs_to polymorphic overrides this to look at the
+        # polymorphic_type field on the owner.
+        def target_klass
+          @reflection.klass
         end
 
-        # Implemented by (some) subclasses
-        def construct_create_scope
-          {}
-        end
-
-      private
-        # Forwards any missing method call to the \target.
-        def method_missing(method, *args)
-          if load_target
-            unless @target.respond_to?(method)
-              message = "undefined method `#{method.to_s}' for \"#{@target}\":#{@target.class.to_s}"
-              raise NoMethodError, message
-            end
-
-            if block_given?
-              @target.send(method, *args)  { |*block_args| yield(*block_args) }
-            else
-              @target.send(method, *args)
-            end
-          end
+        # Can be overridden (i.e. in ThroughAssociation) to merge in other scopes (i.e. the
+        # through association's scope)
+        def target_scope
+          target_klass.scoped
         end
 
         # Loads the \target if needed and returns it.
@@ -251,26 +229,84 @@ module ActiveRecord
         # ActiveRecord::RecordNotFound is rescued within the method, and it is
         # not reraised. The proxy is \reset and +nil+ is the return value.
         def load_target
-          return nil unless defined?(@loaded)
-
-          if !loaded? && (!@owner.new_record? || foreign_key_present)
-            if IdentityMap.enabled? && association_class
-              @target = IdentityMap.get(association_class, @owner[@reflection.association_foreign_key])
+          if find_target?
+            begin
+            if IdentityMap.enabled? && association_class && association_class.respond_to?(:base_class)
+              @target = IdentityMap.get(association_class, @owner[@reflection.foreign_key])
+            end
+            rescue NameError
+              nil
+            ensure
+              @target ||= find_target
             end
-            @target ||= find_target
           end
-
-          @loaded = true
-          @target
+          loaded!
+          target
         rescue ActiveRecord::RecordNotFound
           reset
         end
 
-        # Can be overwritten by associations that might have the foreign key
-        # available for an association without having the object itself (and
-        # still being a new record). Currently, only +belongs_to+ presents
-        # this scenario (both vanilla and polymorphic).
-        def foreign_key_present
+      private
+
+        def find_target?
+          !loaded? && (!@owner.new_record? || foreign_key_present?) && target_klass
+        end
+
+        def interpolate(sql, record = nil)
+          if sql.respond_to?(:to_proc)
+            @owner.send(:instance_exec, record, &sql)
+          else
+            sql
+          end
+        end
+
+        def select_value
+          @reflection.options[:select]
+        end
+
+        # Implemented by (some) subclasses
+        def creation_attributes
+          { }
+        end
+
+        # Returns a hash linking the owner to the association represented by the reflection
+        def construct_owner_attributes(reflection = @reflection)
+          attributes = {}
+          if reflection.macro == :belongs_to
+            attributes[reflection.association_primary_key] = @owner[reflection.foreign_key]
+          else
+            attributes[reflection.foreign_key] = @owner[reflection.active_record_primary_key]
+
+            if reflection.options[:as]
+              attributes["#{reflection.options[:as]}_type"] = @owner.class.base_class.name
+            end
+          end
+          attributes
+        end
+
+        # Builds an array of arel nodes from the owner attributes hash
+        def construct_owner_conditions(table = aliased_table, reflection = @reflection)
+          conditions = construct_owner_attributes(reflection).map do |attr, value|
+            table[attr].eq(value)
+          end
+          table.create_and(conditions)
+        end
+
+        # Sets the owner attributes on the given record
+        def set_owner_attributes(record)
+          if @owner.persisted?
+            construct_owner_attributes.each { |key, value| record[key] = value }
+          end
+        end
+
+        # Should be true if there is a foreign key present on the @owner which
+        # references the target. This is used to determine whether we can load
+        # the target if the @owner is currently a new record (and therefore
+        # without a key).
+        #
+        # Currently implemented by belongs_to (vanilla and polymorphic) and
+        # has_one/has_many :through associations which go through a belongs_to
+        def foreign_key_present?
           false
         end
 
@@ -284,34 +320,24 @@ module ActiveRecord
           end
         end
 
-        if RUBY_VERSION < '1.9.2'
-          # Array#flatten has problems with recursive arrays before Ruby 1.9.2.
-          # Going one level deeper solves the majority of the problems.
-          def flatten_deeper(array)
-            array.collect { |element| (element.respond_to?(:flatten) && !element.is_a?(Hash)) ? element.flatten : element }.flatten
-          end
-        else
-          def flatten_deeper(array)
-            array.flatten
-          end
+        # Can be redefined by subclasses, notably polymorphic belongs_to
+        # The record parameter is necessary to support polymorphic inverses as we must check for
+        # the association in the specific class of the record.
+        def inverse_reflection_for(record)
+          @reflection.inverse_of
         end
 
-        # Returns the ID of the owner, quoted if needed.
-        def owner_quoted_id
-          @owner.quoted_id
+        # Is this association invertible? Can be redefined by subclasses.
+        def invertible_for?(record)
+          inverse_reflection_for(record)
         end
 
-        def set_inverse_instance(record, instance)
-          return if record.nil? || !we_can_set_the_inverse_on_this?(record)
-          inverse_relationship = @reflection.inverse_of
-          unless inverse_relationship.nil?
-            record.send(:"set_#{inverse_relationship.name}_target", instance)
-          end
-        end
-
-        # Override in subclasses
-        def we_can_set_the_inverse_on_this?(record)
-          false
+        # This should be implemented to return the values of the relevant key(s) on the owner,
+        # so that when state_state is different from the value stored on the last find_target,
+        # the target is stale.
+        #
+        # This is only relevant to certain associations, which is why it returns nil by default.
+        def stale_state
         end
 
         def association_class
diff --git a/activerecord/lib/active_record/associations/belongs_to_association.rb b/activerecord/lib/active_record/associations/belongs_to_association.rb
index b438620c8f..178c7204f8 100644
--- a/activerecord/lib/active_record/associations/belongs_to_association.rb
+++ b/activerecord/lib/active_record/associations/belongs_to_association.rb
@@ -1,41 +1,17 @@
 module ActiveRecord
   # = Active Record Belongs To Associations
   module Associations
-    class BelongsToAssociation < AssociationProxy #:nodoc:
-      def create(attributes = {})
-        replace(@reflection.create_association(attributes))
-      end
-
-      def build(attributes = {})
-        replace(@reflection.build_association(attributes))
-      end
-
+    class BelongsToAssociation < SingularAssociation #:nodoc:
       def replace(record)
-        counter_cache_name = @reflection.counter_cache_column
-
-        if record.nil?
-          if counter_cache_name && @owner.persisted?
-            @reflection.klass.decrement_counter(counter_cache_name, previous_record_id) if @owner[@reflection.primary_key_name]
-          end
-
-          @target = @owner[@reflection.primary_key_name] = nil
-        else
-          raise_on_type_mismatch(record)
+        record = check_record(record)
 
-          if counter_cache_name && @owner.persisted? && record.id != @owner[@reflection.primary_key_name]
-            @reflection.klass.increment_counter(counter_cache_name, record.id)
-            @reflection.klass.decrement_counter(counter_cache_name, @owner[@reflection.primary_key_name]) if @owner[@reflection.primary_key_name]
-          end
-
-          @target = (AssociationProxy === record ? record.target : record)
-          @owner[@reflection.primary_key_name] = record_id(record) if record.persisted?
-          @updated = true
-        end
+        update_counters(record)
+        replace_keys(record)
+        set_inverse_instance(record)
 
-        set_inverse_instance(record, @owner)
+        @updated = true if record
 
-        loaded
-        record
+        self.target = record
       end
 
       def updated?
@@ -43,50 +19,52 @@ module ActiveRecord
       end
 
       private
-        def find_target
-          find_method = if @reflection.options[:primary_key]
-                          "find_by_#{@reflection.options[:primary_key]}"
-                        else
-                          "find"
-                        end
 
-          options = @reflection.options.dup.slice(:select, :include, :readonly)
+        def update_counters(record)
+          counter_cache_name = @reflection.counter_cache_column
+
+          if counter_cache_name && @owner.persisted? && different_target?(record)
+            if record
+              record.class.increment_counter(counter_cache_name, record.id)
+            end
 
-          the_target = with_scope(:find => @scope[:find]) do
-            @reflection.klass.send(find_method,
-              @owner[@reflection.primary_key_name],
-              options
-            ) if @owner[@reflection.primary_key_name]
+            if foreign_key_present?
+              target_klass.decrement_counter(counter_cache_name, target_id)
+            end
           end
-          set_inverse_instance(the_target, @owner)
-          the_target
         end
 
-        def construct_find_scope
-          { :conditions => conditions }
+        # Checks whether record is different to the current target, without loading it
+        def different_target?(record)
+          record.nil? && @owner[@reflection.foreign_key] ||
+          record.id   != @owner[@reflection.foreign_key]
         end
 
-        def foreign_key_present
-          !@owner[@reflection.primary_key_name].nil?
+        def replace_keys(record)
+          @owner[@reflection.foreign_key] = record && record[@reflection.association_primary_key]
+        end
+
+        def foreign_key_present?
+          @owner[@reflection.foreign_key]
         end
 
         # NOTE - for now, we're only supporting inverse setting from belongs_to back onto
         # has_one associations.
-        def we_can_set_the_inverse_on_this?(record)
-          @reflection.has_inverse? && @reflection.inverse_of.macro == :has_one
+        def invertible_for?(record)
+          inverse = inverse_reflection_for(record)
+          inverse && inverse.macro == :has_one
         end
 
-        def record_id(record)
-          record.send(@reflection.options[:primary_key] || :id)
+        def target_id
+          if @reflection.options[:primary_key]
+            @owner.send(@reflection.name).try(:id)
+          else
+            @owner[@reflection.foreign_key]
+          end
         end
 
-        def previous_record_id
-          @previous_record_id ||= if @reflection.options[:primary_key]
-                                    previous_record = @owner.send(@reflection.name)
-                                    previous_record.nil? ? nil : previous_record.id
-                                  else
-                                    @owner[@reflection.primary_key_name]
-                                  end
+        def stale_state
+          @owner[@reflection.foreign_key].to_s
         end
     end
   end
diff --git a/activerecord/lib/active_record/associations/belongs_to_polymorphic_association.rb b/activerecord/lib/active_record/associations/belongs_to_polymorphic_association.rb
index a0df860623..4f67b02d00 100644
--- a/activerecord/lib/active_record/associations/belongs_to_polymorphic_association.rb
+++ b/activerecord/lib/active_record/associations/belongs_to_polymorphic_association.rb
@@ -1,77 +1,33 @@
 module ActiveRecord
   # = Active Record Belongs To Polymorphic Association
   module Associations
-    class BelongsToPolymorphicAssociation < AssociationProxy #:nodoc:
-      def replace(record)
-        if record.nil?
-          @target = @owner[@reflection.primary_key_name] = @owner[@reflection.options[:foreign_type]] = nil
-        else
-          @target = (AssociationProxy === record ? record.target : record)
-
-          @owner[@reflection.primary_key_name] = record_id(record)
-          @owner[@reflection.options[:foreign_type]] = record.class.base_class.name.to_s
-
-          @updated = true
-        end
-
-        set_inverse_instance(record, @owner)
-        loaded
-        record
-      end
-
-      def updated?
-        @updated
-      end
-
+    class BelongsToPolymorphicAssociation < BelongsToAssociation #:nodoc:
       private
 
-        # NOTE - for now, we're only supporting inverse setting from belongs_to back onto
-        # has_one associations.
-        def we_can_set_the_inverse_on_this?(record)
-          if @reflection.has_inverse?
-            inverse_association = @reflection.polymorphic_inverse_of(record.class)
-            inverse_association && inverse_association.macro == :has_one
-          else
-            false
-          end
-        end
-
-        def set_inverse_instance(record, instance)
-          return if record.nil? || !we_can_set_the_inverse_on_this?(record)
-          inverse_relationship = @reflection.polymorphic_inverse_of(record.class)
-          if inverse_relationship
-            record.send(:"set_#{inverse_relationship.name}_target", instance)
-          end
+        def replace_keys(record)
+          super
+          @owner[@reflection.foreign_type] = record && record.class.base_class.name
         end
 
-        def construct_find_scope
-          { :conditions => conditions }
+        def different_target?(record)
+          super || record.class != target_klass
         end
 
-        def find_target
-          return nil if association_class.nil?
-
-          target = association_class.send(:with_scope, :find => @scope[:find]) do
-            association_class.find(
-              @owner[@reflection.primary_key_name],
-              :select  => @reflection.options[:select],
-              :include => @reflection.options[:include]
-            )
-          end
-          set_inverse_instance(target, @owner)
-          target
+        def inverse_reflection_for(record)
+          @reflection.polymorphic_inverse_of(record.class)
         end
 
-        def foreign_key_present
-          !@owner[@reflection.primary_key_name].nil?
+        def target_klass
+          type = @owner[@reflection.foreign_type]
+          type && type.constantize
         end
 
-        def record_id(record)
-          record.send(@reflection.options[:primary_key] || :id)
+        def raise_on_type_mismatch(record)
+          # A polymorphic association cannot have a type mismatch, by definition
         end
 
-        def association_class
-          @owner[@reflection.options[:foreign_type]] ? @owner[@reflection.options[:foreign_type]].constantize : nil
+        def stale_state
+          [super, @owner[@reflection.foreign_type].to_s]
         end
     end
   end
diff --git a/activerecord/lib/active_record/associations/class_methods/join_dependency.rb b/activerecord/lib/active_record/associations/class_methods/join_dependency.rb
index a74d0a4ca8..fdd4fe8946 100644
--- a/activerecord/lib/active_record/associations/class_methods/join_dependency.rb
+++ b/activerecord/lib/active_record/associations/class_methods/join_dependency.rb
@@ -176,7 +176,7 @@ module ActiveRecord
 
             join_part = join_parts.detect { |j|
               j.reflection.name.to_s == name &&
-                j.parent_table_name    == parent.class.table_name }
+                j.parent_table_name == parent.class.table_name }
 
               raise(ConfigurationError, "No such association") unless join_part
 
@@ -201,7 +201,7 @@ module ActiveRecord
 
           macro = join_part.reflection.macro
           if macro == :has_one
-            return if record.instance_variable_defined?("@#{join_part.reflection.name}")
+            return if record.association_cache.key?(join_part.reflection.name)
             association = join_part.instantiate(row) unless row[join_part.aliased_primary_key].nil?
             set_target_and_inverse(join_part, association, record)
           else
@@ -210,9 +210,9 @@ module ActiveRecord
             case macro
             when :has_many, :has_and_belongs_to_many
               collection = record.send(join_part.reflection.name)
-              collection.loaded
+              collection.loaded!
               collection.target.push(association)
-              collection.__send__(:set_inverse_instance, association, record)
+              collection.send(:set_inverse_instance, association)
             when :belongs_to
               set_target_and_inverse(join_part, association, record)
             else
@@ -223,8 +223,9 @@ module ActiveRecord
         end
 
         def set_target_and_inverse(join_part, association, record)
-          association_proxy = record.send("set_#{join_part.reflection.name}_target", association)
-          association_proxy.__send__(:set_inverse_instance, association, record)
+          association_proxy = record.send(:association_proxy, join_part.reflection.name)
+          association_proxy.target = association
+          association_proxy.send(:set_inverse_instance, association)
         end
       end
     end
diff --git a/activerecord/lib/active_record/associations/class_methods/join_dependency/join_association.rb b/activerecord/lib/active_record/associations/class_methods/join_dependency/join_association.rb
index 694778008b..aaa475109e 100644
--- a/activerecord/lib/active_record/associations/class_methods/join_dependency/join_association.rb
+++ b/activerecord/lib/active_record/associations/class_methods/join_dependency/join_association.rb
@@ -82,12 +82,12 @@ module ActiveRecord
               connection = active_record.connection
 
               name = connection.table_alias_for "#{pluralize(reflection.name)}_#{parent_table_name}#{suffix}"
-              table_index = aliases[name] + 1
-              name = name[0, connection.table_alias_length-3] + "_#{table_index}" if table_index > 1
+              aliases[name] += 1
+              name = name[0, connection.table_alias_length-3] + "_#{aliases[name]}" if aliases[name] > 1
+            else
+              aliases[name] += 1
             end
 
-            aliases[name] += 1
-
             name
           end
 
@@ -108,6 +108,10 @@ module ActiveRecord
           end
 
           def process_conditions(conditions, table_name)
+            if conditions.respond_to?(:to_proc)
+              conditions = instance_eval(&conditions)
+            end
+
             Arel.sql(sanitize_sql(conditions, table_name))
           end
 
@@ -190,17 +194,20 @@ module ActiveRecord
             ).alias @aliased_join_table_name
 
             jt_conditions = []
-            jt_foreign_key = first_key = second_key = nil
+            first_key = second_key = nil
 
-            if through_reflection.options[:as] # has_many :through against a polymorphic join
-              as_key         = through_reflection.options[:as].to_s
-              jt_foreign_key = as_key + '_id'
-
-              jt_conditions <<
-              join_table[as_key + '_type'].
-                eq(parent.active_record.base_class.name)
+            if through_reflection.macro == :belongs_to
+              jt_primary_key = through_reflection.foreign_key
+              jt_foreign_key = through_reflection.association_primary_key
             else
-              jt_foreign_key = through_reflection.primary_key_name
+              jt_primary_key = through_reflection.active_record_primary_key
+              jt_foreign_key = through_reflection.foreign_key
+
+              if through_reflection.options[:as] # has_many :through against a polymorphic join
+                jt_conditions <<
+                join_table["#{through_reflection.options[:as]}_type"].
+                  eq(parent.active_record.base_class.name)
+              end
             end
 
             case source_reflection.macro
@@ -225,15 +232,15 @@ module ActiveRecord
                 second_key = source_reflection.association_foreign_key
 
                 jt_conditions <<
-                join_table[reflection.source_reflection.options[:foreign_type]].
+                join_table[reflection.source_reflection.foreign_type].
                   eq(reflection.options[:source_type])
               else
-                second_key = source_reflection.primary_key_name
+                second_key = source_reflection.foreign_key
               end
             end
 
             jt_conditions <<
-            parent_table[parent.primary_key].
+            parent_table[jt_primary_key].
               eq(join_table[jt_foreign_key])
 
             if through_reflection.options[:conditions]
@@ -259,7 +266,7 @@ module ActiveRecord
           end
 
           def join_belongs_to_to(relation)
-            foreign_key = options[:foreign_key] || reflection.primary_key_name
+            foreign_key = options[:foreign_key] || reflection.foreign_key
             primary_key = options[:primary_key] || reflection.klass.primary_key
 
             join_target_table(
diff --git a/activerecord/lib/active_record/associations/has_and_belongs_to_many_association.rb b/activerecord/lib/active_record/associations/has_and_belongs_to_many_association.rb
index e2ce9aefcf..b9c9919e7a 100644
--- a/activerecord/lib/active_record/associations/has_and_belongs_to_many_association.rb
+++ b/activerecord/lib/active_record/associations/has_and_belongs_to_many_association.rb
@@ -2,81 +2,48 @@ module ActiveRecord
   # = Active Record Has And Belongs To Many Association
   module Associations
     class HasAndBelongsToManyAssociation < AssociationCollection #:nodoc:
-      def create(attributes = {})
-        create_record(attributes) { |record| insert_record(record) }
-      end
-
-      def create!(attributes = {})
-        create_record(attributes) { |record| insert_record(record, true) }
-      end
-
-      def columns
-        @reflection.columns(@reflection.options[:join_table], "#{@reflection.options[:join_table]} Columns")
-      end
-
-      def reset_column_information
-        @reflection.reset_column_information
-      end
+      attr_reader :join_table
 
-      def has_primary_key?
-        @has_primary_key ||= @owner.connection.supports_primary_key? && @owner.connection.primary_key(@reflection.options[:join_table])
+      def initialize(owner, reflection)
+        @join_table = Arel::Table.new(reflection.options[:join_table])
+        super
       end
 
       protected
-        def construct_find_options!(options)
-          options[:joins]      = Arel::SqlLiteral.new(@scope[:find][:joins])
-          options[:readonly]   = finding_with_ambiguous_select?(options[:select] || @reflection.options[:select])
-          options[:select]   ||= (@reflection.options[:select] || Arel::SqlLiteral.new('*'))
-        end
 
-        def count_records
-          load_target.size
-        end
-
-        def insert_record(record, force = true, validate = true)
-          if record.new_record?
-            if force
-              record.save!
-            else
-              return false unless record.save(:validate => validate)
-            end
-          end
+        def insert_record(record, validate = true)
+          return if record.new_record? && !record.save(:validate => validate)
 
           if @reflection.options[:insert_sql]
-            @owner.connection.insert(interpolate_sql(@reflection.options[:insert_sql], record))
+            @owner.connection.insert(interpolate(@reflection.options[:insert_sql], record))
           else
-            relation   = Arel::Table.new(@reflection.options[:join_table])
-            timestamps = record_timestamp_columns(record)
-            timezone   = record.send(:current_time_from_proper_timezone) if timestamps.any?
-
-            attributes = columns.map do |column|
-              name = column.name
-              value = case name.to_s
-                when @reflection.primary_key_name.to_s
-                  @owner.id
-                when @reflection.association_foreign_key.to_s
-                  record.id
-                when *timestamps
-                  timezone
-                else
-                  @owner.send(:quote_value, record[name], column) if record.has_attribute?(name)
-              end
-              [relation[name], value] unless value.nil?
-            end
-
-            stmt = relation.compile_insert Hash[attributes]
+            stmt = join_table.compile_insert(
+              join_table[@reflection.foreign_key]             => @owner.id,
+              join_table[@reflection.association_foreign_key] => record.id
+            )
+
             @owner.connection.insert stmt.to_sql
           end
 
-          true
+          record
         end
 
-        def delete_records(records)
+        def association_scope
+          super.joins(construct_joins)
+        end
+
+      private
+
+        def count_records
+          load_target.size
+        end
+
+        def delete_records(records, method)
           if sql = @reflection.options[:delete_sql]
-            records.each { |record| @owner.connection.delete(interpolate_sql(sql, record)) }
+            records.each { |record| @owner.connection.delete(interpolate(sql, record)) }
           else
-            relation = Arel::Table.new(@reflection.options[:join_table])
-            stmt = relation.where(relation[@reflection.primary_key_name].eq(@owner.id).
+            relation = join_table
+            stmt = relation.where(relation[@reflection.foreign_key].eq(@owner.id).
               and(relation[@reflection.association_foreign_key].in(records.map { |x| x.id }.compact))
             ).compile_delete
             @owner.connection.delete stmt.to_sql
@@ -84,51 +51,25 @@ module ActiveRecord
         end
 
         def construct_joins
-          "INNER JOIN #{@owner.connection.quote_table_name @reflection.options[:join_table]} ON #{@reflection.quoted_table_name}.#{@reflection.klass.primary_key} = #{@owner.connection.quote_table_name @reflection.options[:join_table]}.#{@reflection.association_foreign_key}"
-        end
+          right = join_table
+          left  = @reflection.klass.arel_table
 
-        def construct_conditions
-          sql = "#{@owner.connection.quote_table_name @reflection.options[:join_table]}.#{@reflection.primary_key_name} = #{owner_quoted_id} "
-          sql << " AND (#{conditions})" if conditions
-          sql
-        end
+          condition = left[@reflection.klass.primary_key].eq(
+            right[@reflection.association_foreign_key])
 
-        def construct_find_scope
-          {
-            :conditions => construct_conditions,
-            :joins      => construct_joins,
-            :readonly   => false,
-            :order      => @reflection.options[:order],
-            :include    => @reflection.options[:include],
-            :limit      => @reflection.options[:limit]
-          }
+          right.create_join(right, right.create_on(condition))
         end
 
-        # Join tables with additional columns on top of the two foreign keys must be considered
-        # ambiguous unless a select clause has been explicitly defined. Otherwise you can get
-        # broken records back, if, for example, the join column also has an id column. This will
-        # then overwrite the id column of the records coming back.
-        def finding_with_ambiguous_select?(select_clause)
-          !select_clause && columns.size != 2
+        def construct_owner_conditions
+          super(join_table)
         end
 
-      private
-        def create_record(attributes, &block)
-          # Can't use Base.create because the foreign key may be a protected attribute.
-          ensure_owner_is_persisted!
-          if attributes.is_a?(Array)
-            attributes.collect { |attr| create(attr) }
-          else
-            build_record(attributes, &block)
-          end
+        def select_value
+          super || @reflection.klass.arel_table[Arel.star]
         end
 
-        def record_timestamp_columns(record)
-          if record.record_timestamps
-            record.send(:all_timestamp_attributes).map { |x| x.to_s }
-          else
-            []
-          end
+        def invertible_for?(record)
+          false
         end
     end
   end
diff --git a/activerecord/lib/active_record/associations/has_many_association.rb b/activerecord/lib/active_record/associations/has_many_association.rb
index 4ff61fff45..543b073393 100644
--- a/activerecord/lib/active_record/associations/has_many_association.rb
+++ b/activerecord/lib/active_record/associations/has_many_association.rb
@@ -7,14 +7,14 @@ module ActiveRecord
     # is provided by its child HasManyThroughAssociation.
     class HasManyAssociation < AssociationCollection #:nodoc:
       protected
-        def owner_quoted_id
-          if @reflection.options[:primary_key]
-            @owner.class.quote_value(@owner.send(@reflection.options[:primary_key]))
-          else
-            @owner.quoted_id
-          end
+
+        def insert_record(record, validate = true)
+          set_owner_attributes(record)
+          record.save(:validate => validate)
         end
 
+      private
+
         # Returns the number of records in this collection.
         #
         # If the association has a counter cache it gets that value. Otherwise
@@ -34,83 +34,69 @@ module ActiveRecord
           elsif @reflection.options[:counter_sql] || @reflection.options[:finder_sql]
             @reflection.klass.count_by_sql(custom_counter_sql)
           else
-            @reflection.klass.count(@scope[:find].slice(:conditions, :joins, :include))
+            scoped.count
           end
 
           # If there's nothing in the database and @target has no new records
           # we are certain the current target is an empty array. This is a
           # documented side-effect of the method that may avoid an extra SELECT.
-          @target ||= [] and loaded if count == 0
+          @target ||= [] and loaded! if count == 0
 
           [@reflection.options[:limit], count].compact.min
         end
 
-        def has_cached_counter?
-          @owner.attribute_present?(cached_counter_attribute_name)
-        end
-
-        def cached_counter_attribute_name
-          "#{@reflection.name}_count"
+        def has_cached_counter?(reflection = @reflection)
+          @owner.attribute_present?(cached_counter_attribute_name(reflection))
         end
 
-        def insert_record(record, force = false, validate = true)
-          set_belongs_to_association_for(record)
-          force ? record.save! : record.save(:validate => validate)
+        def cached_counter_attribute_name(reflection = @reflection)
+          "#{reflection.name}_count"
         end
 
-        # Deletes the records according to the <tt>:dependent</tt> option.
-        def delete_records(records)
-          case @reflection.options[:dependent]
-            when :destroy
-              records.each { |r| r.destroy }
-            when :delete_all
-              @reflection.klass.delete(records.map { |record| record.id })
-            else
-              relation = Arel::Table.new(@reflection.table_name)
-              stmt = relation.where(relation[@reflection.primary_key_name].eq(@owner.id).
-                  and(relation[@reflection.klass.primary_key].in(records.map { |r| r.id }))
-              ).compile_update(relation[@reflection.primary_key_name] => nil)
-              @owner.connection.update stmt.to_sql
-
-              @owner.class.update_counters(@owner.id, cached_counter_attribute_name => -records.size) if has_cached_counter?
+        def update_counter(difference, reflection = @reflection)
+          if has_cached_counter?(reflection)
+            counter = cached_counter_attribute_name(reflection)
+            @owner.class.update_counters(@owner.id, counter => difference)
+            @owner[counter] += difference
+            @owner.changed_attributes.delete(counter) # eww
           end
         end
 
-        def target_obsolete?
-          false
+        # This shit is nasty. We need to avoid the following situation:
+        #
+        #   * An associated record is deleted via record.destroy
+        #   * Hence the callbacks run, and they find a belongs_to on the record with a
+        #     :counter_cache options which points back at our @owner. So they update the
+        #     counter cache.
+        #   * In which case, we must make sure to *not* update the counter cache, or else
+        #     it will be decremented twice.
+        #
+        # Hence this method.
+        def inverse_updates_counter_cache?(reflection = @reflection)
+          counter_name = cached_counter_attribute_name(reflection)
+          reflection.klass.reflect_on_all_associations(:belongs_to).any? { |inverse_reflection|
+            inverse_reflection.counter_cache_column == counter_name
+          }
         end
 
-        def construct_conditions
-          if @reflection.options[:as]
-            sql =
-              "#{@reflection.quoted_table_name}.#{@reflection.options[:as]}_id = #{owner_quoted_id} AND " +
-              "#{@reflection.quoted_table_name}.#{@reflection.options[:as]}_type = #{@owner.class.quote_value(@owner.class.base_class.name.to_s)}"
+        # Deletes the records according to the <tt>:dependent</tt> option.
+        def delete_records(records, method)
+          if method == :destroy
+            records.each { |r| r.destroy }
+            update_counter(-records.length) unless inverse_updates_counter_cache?
           else
-            sql = "#{@reflection.quoted_table_name}.#{@reflection.primary_key_name} = #{owner_quoted_id}"
-          end
-          sql << " AND (#{conditions})" if conditions
-          sql
-        end
-
-        def construct_find_scope
-          {
-            :conditions => construct_conditions,
-            :readonly   => false,
-            :order      => @reflection.options[:order],
-            :limit      => @reflection.options[:limit],
-            :include    => @reflection.options[:include]
-          }
-        end
+            keys  = records.map { |r| r[@reflection.association_primary_key] }
+            scope = scoped.where(@reflection.association_primary_key => keys)
 
-        def construct_create_scope
-          create_scoping = {}
-          set_belongs_to_association_for(create_scoping)
-          create_scoping
+            if method == :delete_all
+              update_counter(-scope.delete_all)
+            else
+              update_counter(-scope.update_all(@reflection.foreign_key => nil))
+            end
+          end
         end
 
-        def we_can_set_the_inverse_on_this?(record)
-          @reflection.inverse_of
-        end
+        alias creation_attributes construct_owner_attributes
     end
   end
 end
diff --git a/activerecord/lib/active_record/associations/has_many_through_association.rb b/activerecord/lib/active_record/associations/has_many_through_association.rb
index 781aa7ef62..9f74d57c4d 100644
--- a/activerecord/lib/active_record/associations/has_many_through_association.rb
+++ b/activerecord/lib/active_record/associations/has_many_through_association.rb
@@ -1,82 +1,89 @@
-require "active_record/associations/through_association_scope"
 require 'active_support/core_ext/object/blank'
 
 module ActiveRecord
   # = Active Record Has Many Through Association
   module Associations
     class HasManyThroughAssociation < HasManyAssociation #:nodoc:
-      include ThroughAssociationScope
+      include ThroughAssociation
 
       alias_method :new, :build
 
-      def create!(attrs = nil)
-        create_record(attrs, true)
-      end
-
-      def create(attrs = nil)
-        create_record(attrs, false)
-      end
-
-      def destroy(*records)
-        transaction do
-          delete_records(flatten_deeper(records))
-          super
-        end
-      end
-
       # Returns the size of the collection by executing a SELECT COUNT(*) query if the collection hasn't been
       # loaded and calling collection.size if it has. If it's more likely than not that the collection does
       # have a size larger than zero, and you need to fetch that collection afterwards, it'll take one fewer
       # SELECT query if you use #length.
       def size
-        return @owner.send(:read_attribute, cached_counter_attribute_name) if has_cached_counter?
-        return @target.size if loaded?
-        return count
+        if has_cached_counter?
+          @owner.send(:read_attribute, cached_counter_attribute_name)
+        elsif loaded?
+          @target.size
+        else
+          count
+        end
+      end
+
+      def <<(*records)
+        unless @owner.new_record?
+          records.flatten.each do |record|
+            raise_on_type_mismatch(record)
+            record.save! if record.new_record?
+          end
+        end
+
+        super
       end
 
       protected
-        def create_record(attrs, force = true)
-          ensure_owner_is_persisted!
 
-          transaction do
-            object = @reflection.klass.new(attrs)
-            add_record_to_target_with_callbacks(object) {|r| insert_record(object, force) }
-            object
-          end
+        def insert_record(record, validate = true)
+          return if record.new_record? && !record.save(:validate => validate)
+
+          through_association = @owner.send(@reflection.through_reflection.name)
+          through_association.create!(construct_join_attributes(record))
+
+          update_counter(1)
+          record
         end
 
+      private
+
         def target_reflection_has_associated_record?
-          if @reflection.through_reflection.macro == :belongs_to && @owner[@reflection.through_reflection.primary_key_name].blank?
+          if @reflection.through_reflection.macro == :belongs_to && @owner[@reflection.through_reflection.foreign_key].blank?
             false
           else
             true
           end
         end
 
-        def construct_find_options!(options)
-          options[:joins]   = [construct_joins] + Array.wrap(options[:joins])
-          options[:include] = @reflection.source_reflection.options[:include] if options[:include].nil? && @reflection.source_reflection.options[:include]
+        def update_through_counter?(method)
+          case method
+          when :destroy
+            !inverse_updates_counter_cache?(@reflection.through_reflection)
+          when :nullify
+            false
+          else
+            true
+          end
         end
 
-        def insert_record(record, force = true, validate = true)
-          if record.new_record?
-            if force
-              record.save!
-            else
-              return false unless record.save(:validate => validate)
-            end
-          end
+        def delete_records(records, method)
+          through = @owner.send(:association_proxy, @reflection.through_reflection.name)
+          scope   = through.scoped.where(construct_join_attributes(*records))
 
-          through_association = @owner.send(@reflection.through_reflection.name)
-          through_association.create!(construct_join_attributes(record))
-        end
+          case method
+          when :destroy
+            count = scope.destroy_all.length
+          when :nullify
+            count = scope.update_all(@reflection.source_reflection.foreign_key => nil)
+          else
+            count = scope.delete_all
+          end
 
-        # TODO - add dependent option support
-        def delete_records(records)
-          klass = @reflection.through_reflection.klass
-          records.each do |associate|
-            klass.delete_all(construct_join_attributes(associate))
+          if @reflection.through_reflection.macro == :has_many && update_through_counter?(method)
+            update_counter(-count, @reflection.through_reflection)
           end
+
+          update_counter(-count)
         end
 
         def find_target
@@ -84,16 +91,8 @@ module ActiveRecord
           scoped.all
         end
 
-        def has_cached_counter?
-          @owner.attribute_present?(cached_counter_attribute_name)
-        end
-
-        def cached_counter_attribute_name
-          "#{@reflection.name}_count"
-        end
-
         # NOTE - not sure that we can actually cope with inverses here
-        def we_can_set_the_inverse_on_this?(record)
+        def invertible_for?(record)
           false
         end
     end
diff --git a/activerecord/lib/active_record/associations/has_one_association.rb b/activerecord/lib/active_record/associations/has_one_association.rb
index c49fd6e66a..a0828dcdea 100644
--- a/activerecord/lib/active_record/associations/has_one_association.rb
+++ b/activerecord/lib/active_record/associations/has_one_association.rb
@@ -1,135 +1,65 @@
 module ActiveRecord
   # = Active Record Belongs To Has One Association
   module Associations
-    class HasOneAssociation < AssociationProxy #:nodoc:
-      def create(attrs = {}, replace_existing = true)
-        new_record(replace_existing) do |reflection|
-          attrs = merge_with_conditions(attrs)
-          reflection.create_association(attrs)
-        end
-      end
-
-      def create!(attrs = {}, replace_existing = true)
-        new_record(replace_existing) do |reflection|
-          attrs = merge_with_conditions(attrs)
-          reflection.create_association!(attrs)
-        end
-      end
+    class HasOneAssociation < SingularAssociation #:nodoc:
+      def replace(record, save = true)
+        record = check_record(record)
+        load_target
 
-      def build(attrs = {}, replace_existing = true)
-        new_record(replace_existing) do |reflection|
-          attrs = merge_with_conditions(attrs)
-          reflection.build_association(attrs)
-        end
-      end
+        @reflection.klass.transaction do
+          if @target && @target != record
+            remove_target!(@reflection.options[:dependent])
+          end
 
-      def replace(obj, dont_save = false)
-        load_target
+          if record
+            set_inverse_instance(record)
+            set_owner_attributes(record)
 
-        unless @target.nil? || @target == obj
-          if dependent? && !dont_save
-            case @reflection.options[:dependent]
-            when :delete
-              @target.delete if @target.persisted?
-              @owner.clear_association_cache
-            when :destroy
-              @target.destroy if @target.persisted?
-              @owner.clear_association_cache
-            when :nullify
-              @target[@reflection.primary_key_name] = nil
-              @target.save if @owner.persisted? && @target.persisted?
+            if @owner.persisted? && save && !record.save
+              nullify_owner_attributes(record)
+              set_owner_attributes(@target)
+              raise RecordNotSaved, "Failed to save the new associated #{@reflection.name}."
             end
-          else
-            @target[@reflection.primary_key_name] = nil
-            @target.save if @owner.persisted? && @target.persisted?
           end
         end
 
-        if obj.nil?
-          @target = nil
-        else
-          raise_on_type_mismatch(obj)
-          set_belongs_to_association_for(obj)
-          @target = (AssociationProxy === obj ? obj.target : obj)
-        end
-
-        set_inverse_instance(obj, @owner)
-        @loaded = true
-
-        unless !@owner.persisted? || obj.nil? || dont_save
-          return (obj.save ? self : false)
-        else
-          return (obj.nil? ? nil : self)
-        end
+        self.target = record
       end
 
       protected
-        def owner_quoted_id
-          if @reflection.options[:primary_key]
-            @owner.class.quote_value(@owner.send(@reflection.options[:primary_key]))
-          else
-            @owner.quoted_id
-          end
+
+        def association_scope
+          super.order(@reflection.options[:order])
         end
 
       private
-        def find_target
-          options = @reflection.options.dup.slice(:select, :order, :include, :readonly)
-
-          the_target = with_scope(:find => @scope[:find]) do
-            @reflection.klass.find(:first, options)
-          end
-          set_inverse_instance(the_target, @owner)
-          the_target
-        end
 
-        def construct_find_scope
-          if @reflection.options[:as]
-            sql =
-              "#{@reflection.quoted_table_name}.#{@reflection.options[:as]}_id = #{owner_quoted_id} AND " +
-              "#{@reflection.quoted_table_name}.#{@reflection.options[:as]}_type = #{@owner.class.quote_value(@owner.class.base_class.name.to_s)}"
-          else
-            sql = "#{@reflection.quoted_table_name}.#{@reflection.primary_key_name} = #{owner_quoted_id}"
-          end
-          sql << " AND (#{conditions})" if conditions
-          { :conditions => sql }
-        end
+        alias creation_attributes construct_owner_attributes
 
-        def construct_create_scope
-          create_scoping = {}
-          set_belongs_to_association_for(create_scoping)
-          create_scoping
+        # The reason that the save param for replace is false, if for create (not just build),
+        # is because the setting of the foreign keys is actually handled by the scoping when
+        # the record is instantiated, and so they are set straight away and do not need to be
+        # updated within replace.
+        def set_new_record(record)
+          replace(record, false)
         end
 
-        def new_record(replace_existing)
-          # Make sure we load the target first, if we plan on replacing the existing
-          # instance. Otherwise, if the target has not previously been loaded
-          # elsewhere, the instance we create will get orphaned.
-          load_target if replace_existing
-          record = @reflection.klass.send(:with_scope, :create => @scope[:create]) do
-            yield @reflection
-          end
-
-          if replace_existing
-            replace(record, true)
+        def remove_target!(method)
+          if [:delete, :destroy].include?(method)
+            @target.send(method)
           else
-            record[@reflection.primary_key_name] = @owner.id if @owner.persisted?
-            self.target = record
-            set_inverse_instance(record, @owner)
-          end
+            nullify_owner_attributes(@target)
 
-          record
-        end
-
-        def we_can_set_the_inverse_on_this?(record)
-          inverse = @reflection.inverse_of
-          return !inverse.nil?
+            if @target.persisted? && @owner.persisted? && !@target.save
+              set_owner_attributes(@target)
+              raise RecordNotSaved, "Failed to remove the existing associated #{@reflection.name}. " +
+                                    "The record failed to save when after its foreign key was set to nil."
+            end
+          end
         end
 
-        def merge_with_conditions(attrs={})
-          attrs ||= {}
-          attrs.update(@reflection.options[:conditions]) if @reflection.options[:conditions].is_a?(Hash)
-          attrs
+        def nullify_owner_attributes(record)
+          record[@reflection.foreign_key] = nil
         end
     end
   end
diff --git a/activerecord/lib/active_record/associations/has_one_through_association.rb b/activerecord/lib/active_record/associations/has_one_through_association.rb
index e8cf73976b..69771afe50 100644
--- a/activerecord/lib/active_record/associations/has_one_through_association.rb
+++ b/activerecord/lib/active_record/associations/has_one_through_association.rb
@@ -1,40 +1,34 @@
-require "active_record/associations/through_association_scope"
-
 module ActiveRecord
   # = Active Record Has One Through Association
   module Associations
     class HasOneThroughAssociation < HasOneAssociation
-      include ThroughAssociationScope
+      include ThroughAssociation
 
-      def replace(new_value)
-        create_through_record(new_value)
-        @target = new_value
+      def replace(record)
+        create_through_record(record)
+        self.target = record
       end
 
       private
 
-      def create_through_record(new_value) #nodoc:
-        klass = @reflection.through_reflection.klass
+        def create_through_record(record)
+          through_proxy  = @owner.send(:association_proxy, @reflection.through_reflection.name)
+          through_record = through_proxy.send(:load_target)
 
-        current_object = @owner.send(@reflection.through_reflection.name)
+          if through_record && !record
+            through_record.destroy
+          elsif record
+            attributes = construct_join_attributes(record)
 
-        if current_object
-           new_value ? current_object.update_attributes(construct_join_attributes(new_value)) : current_object.destroy
-        elsif new_value
-          if @owner.new_record?
-            self.target = new_value
-            through_association = @owner.send(:association_instance_get, @reflection.through_reflection.name)
-            through_association.build(construct_join_attributes(new_value))
-          else
-            @owner.send(@reflection.through_reflection.name, klass.create(construct_join_attributes(new_value)))
+            if through_record
+              through_record.update_attributes(attributes)
+            elsif @owner.new_record?
+              through_proxy.build(attributes)
+            else
+              through_proxy.create(attributes)
+            end
           end
         end
-      end
-
-    private
-      def find_target
-        scoped.first
-      end
     end
   end
 end
diff --git a/activerecord/lib/active_record/associations/singular_association.rb b/activerecord/lib/active_record/associations/singular_association.rb
new file mode 100644
index 0000000000..7f92d9712a
--- /dev/null
+++ b/activerecord/lib/active_record/associations/singular_association.rb
@@ -0,0 +1,45 @@
+module ActiveRecord
+  module Associations
+    class SingularAssociation < AssociationProxy #:nodoc:
+      def create(attributes = {})
+        new_record(:create, attributes)
+      end
+
+      def create!(attributes = {})
+        build(attributes).tap { |record| record.save! }
+      end
+
+      def build(attributes = {})
+        new_record(:build, attributes)
+      end
+
+      private
+
+        def find_target
+          scoped.first.tap { |record| set_inverse_instance(record) }
+        end
+
+        # Implemented by subclasses
+        def replace(record)
+          raise NotImplementedError
+        end
+
+        def set_new_record(record)
+          replace(record)
+        end
+
+        def check_record(record)
+          record = record.target if AssociationProxy === record
+          raise_on_type_mismatch(record) if record
+          record
+        end
+
+        def new_record(method, attributes)
+          attributes = scoped.scope_for_create.merge(attributes || {})
+          record = @reflection.send("#{method}_association", attributes)
+          set_new_record(record)
+          record
+        end
+    end
+  end
+end
diff --git a/activerecord/lib/active_record/associations/through_association.rb b/activerecord/lib/active_record/associations/through_association.rb
new file mode 100644
index 0000000000..0550bae408
--- /dev/null
+++ b/activerecord/lib/active_record/associations/through_association.rb
@@ -0,0 +1,156 @@
+module ActiveRecord
+  # = Active Record Through Association
+  module Associations
+    module ThroughAssociation
+
+      protected
+
+        def target_scope
+          super.merge(@reflection.through_reflection.klass.scoped)
+        end
+
+        def association_scope
+          scope = super.joins(construct_joins)
+          scope = add_conditions(scope)
+          unless @reflection.options[:include]
+            scope = scope.includes(@reflection.source_reflection.options[:include])
+          end
+          scope
+        end
+
+      private
+
+        # This scope affects the creation of the associated records (not the join records). At the
+        # moment we only support creating on a :through association when the source reflection is a
+        # belongs_to. Thus it's not necessary to set a foreign key on the associated record(s), so
+        # this scope has can legitimately be empty.
+        def creation_attributes
+          { }
+        end
+
+        def aliased_through_table
+          name = @reflection.through_reflection.table_name
+
+          @reflection.table_name == name ?
+            @reflection.through_reflection.klass.arel_table.alias(name + "_join") :
+            @reflection.through_reflection.klass.arel_table
+        end
+
+        def construct_owner_conditions
+          super(aliased_through_table, @reflection.through_reflection)
+        end
+
+        def construct_joins
+          right = aliased_through_table
+          left  = @reflection.klass.arel_table
+
+          conditions = []
+
+          if @reflection.source_reflection.macro == :belongs_to
+            reflection_primary_key = @reflection.source_reflection.association_primary_key
+            source_primary_key     = @reflection.source_reflection.foreign_key
+
+            if @reflection.options[:source_type]
+              column = @reflection.source_reflection.foreign_type
+              conditions <<
+                right[column].eq(@reflection.options[:source_type])
+            end
+          else
+            reflection_primary_key = @reflection.source_reflection.foreign_key
+            source_primary_key     = @reflection.source_reflection.active_record_primary_key
+
+            if @reflection.source_reflection.options[:as]
+              column = "#{@reflection.source_reflection.options[:as]}_type"
+              conditions <<
+                left[column].eq(@reflection.through_reflection.klass.name)
+            end
+          end
+
+          conditions <<
+            left[reflection_primary_key].eq(right[source_primary_key])
+
+          right.create_join(
+            right,
+            right.create_on(right.create_and(conditions)))
+        end
+
+        # Construct attributes for :through pointing to owner and associate. This is used by the
+        # methods which create and delete records on the association.
+        #
+        # We only support indirectly modifying through associations which has a belongs_to source.
+        # This is the "has_many :tags, :through => :taggings" situation, where the join model
+        # typically has a belongs_to on both side. In other words, associations which could also
+        # be represented as has_and_belongs_to_many associations.
+        #
+        # We do not support creating/deleting records on the association where the source has
+        # some other type, because this opens up a whole can of worms, and in basically any
+        # situation it is more natural for the user to just create or modify their join records
+        # directly as required.
+        def construct_join_attributes(*records)
+          if @reflection.source_reflection.macro != :belongs_to
+            raise HasManyThroughCantAssociateThroughHasOneOrManyReflection.new(@owner, @reflection)
+          end
+
+          join_attributes = {
+            @reflection.source_reflection.foreign_key =>
+              records.map { |record|
+                record.send(@reflection.source_reflection.association_primary_key)
+              }
+          }
+
+          if @reflection.options[:source_type]
+            join_attributes[@reflection.source_reflection.foreign_type] =
+              records.map { |record| record.class.base_class.name }
+          end
+
+          if records.count == 1
+            Hash[join_attributes.map { |k, v| [k, v.first] }]
+          else
+            join_attributes
+          end
+        end
+
+        # The reason that we are operating directly on the scope here (rather than passing
+        # back some arel conditions to be added to the scope) is because scope.where([x, y])
+        # has a different meaning to scope.where(x).where(y) - the first version might
+        # perform some substitution if x is a string.
+        def add_conditions(scope)
+          unless @reflection.through_reflection.klass.descends_from_active_record?
+            scope = scope.where(@reflection.through_reflection.klass.send(:type_condition))
+          end
+
+          scope = scope.where(interpolate(@reflection.source_reflection.options[:conditions]))
+          scope.where(through_conditions)
+        end
+
+        # If there is a hash of conditions then we make sure the keys are scoped to the
+        # through table name if left ambiguous.
+        def through_conditions
+          conditions = interpolate(@reflection.through_reflection.options[:conditions])
+
+          if conditions.is_a?(Hash)
+            Hash[conditions.map { |key, value|
+              unless value.is_a?(Hash) || key.to_s.include?('.')
+                key = aliased_through_table.name + '.' + key.to_s
+              end
+
+              [key, value]
+            }]
+          else
+            conditions
+          end
+        end
+
+        def stale_state
+          if @reflection.through_reflection.macro == :belongs_to
+            @owner[@reflection.through_reflection.foreign_key].to_s
+          end
+        end
+
+        def foreign_key_present?
+          @reflection.through_reflection.macro == :belongs_to &&
+          !@owner[@reflection.through_reflection.foreign_key].nil?
+        end
+    end
+  end
+end
diff --git a/activerecord/lib/active_record/associations/through_association_scope.rb b/activerecord/lib/active_record/associations/through_association_scope.rb
deleted file mode 100644
index 5dc5b0c048..0000000000
--- a/activerecord/lib/active_record/associations/through_association_scope.rb
+++ /dev/null
@@ -1,165 +0,0 @@
-module ActiveRecord
-  # = Active Record Through Association Scope
-  module Associations
-    module ThroughAssociationScope
-
-      def scoped
-        with_scope(@scope) do
-          @reflection.klass.scoped &
-          @reflection.through_reflection.klass.scoped
-        end
-      end
-
-      protected
-
-      def construct_find_scope
-        {
-          :conditions => construct_conditions,
-          :joins      => construct_joins,
-          :include    => @reflection.options[:include] || @reflection.source_reflection.options[:include],
-          :select     => construct_select,
-          :order      => @reflection.options[:order],
-          :limit      => @reflection.options[:limit],
-          :readonly   => @reflection.options[:readonly]
-        }
-      end
-
-      def construct_create_scope
-        construct_owner_attributes(@reflection)
-      end
-
-      def aliased_through_table
-        name = @reflection.through_reflection.table_name
-
-        @reflection.table_name == name ?
-          @reflection.through_reflection.klass.arel_table.alias(name + "_join") :
-          @reflection.through_reflection.klass.arel_table
-      end
-
-      # Build SQL conditions from attributes, qualified by table name.
-      def construct_conditions
-        table = aliased_through_table
-        conditions = construct_owner_attributes(@reflection.through_reflection).map do |attr, value|
-          table[attr].eq(value)
-        end
-        conditions << Arel.sql(sql_conditions) if sql_conditions
-        table.create_and(conditions)
-      end
-
-      # Associate attributes pointing to owner
-      def construct_owner_attributes(reflection)
-        if as = reflection.options[:as]
-          { "#{as}_id"   => @owner[reflection.active_record_primary_key],
-            "#{as}_type" => @owner.class.base_class.name }
-        elsif reflection.macro == :belongs_to
-          { reflection.klass.primary_key => @owner[reflection.primary_key_name] }
-        else
-          { reflection.primary_key_name => @owner[reflection.active_record_primary_key] }
-        end
-      end
-
-      def construct_from
-        @reflection.table_name
-      end
-
-      def construct_select(custom_select = nil)
-        distinct = "DISTINCT " if @reflection.options[:uniq]
-        custom_select || @reflection.options[:select] || "#{distinct}#{@reflection.quoted_table_name}.*"
-      end
-
-      def construct_joins
-        right = aliased_through_table
-        left  = @reflection.klass.arel_table
-
-        conditions = []
-
-        if @reflection.source_reflection.macro == :belongs_to
-          reflection_primary_key = @reflection.source_reflection.options[:primary_key] ||
-                                   @reflection.klass.primary_key
-          source_primary_key     = @reflection.source_reflection.primary_key_name
-          if @reflection.options[:source_type]
-            column = @reflection.source_reflection.options[:foreign_type]
-            conditions <<
-              right[column].eq(@reflection.options[:source_type])
-          end
-        else
-          reflection_primary_key = @reflection.source_reflection.primary_key_name
-          source_primary_key     = @reflection.source_reflection.options[:primary_key] ||
-                                   @reflection.through_reflection.klass.primary_key
-          if @reflection.source_reflection.options[:as]
-            column = "#{@reflection.source_reflection.options[:as]}_type"
-            conditions <<
-              left[column].eq(@reflection.through_reflection.klass.name)
-          end
-        end
-
-        conditions <<
-          left[reflection_primary_key].eq(right[source_primary_key])
-
-        right.create_join(
-          right,
-          right.create_on(right.create_and(conditions)))
-      end
-
-      # Construct attributes for :through pointing to owner and associate.
-      def construct_join_attributes(associate)
-        # TODO: revisit this to allow it for deletion, supposing dependent option is supported
-        raise ActiveRecord::HasManyThroughCantAssociateThroughHasOneOrManyReflection.new(@owner, @reflection) if [:has_one, :has_many].include?(@reflection.source_reflection.macro)
-
-        join_attributes = construct_owner_attributes(@reflection.through_reflection).merge(@reflection.source_reflection.primary_key_name => associate.id)
-
-        if @reflection.options[:source_type]
-          join_attributes.merge!(@reflection.source_reflection.options[:foreign_type] => associate.class.base_class.name)
-        end
-
-        if @reflection.through_reflection.options[:conditions].is_a?(Hash)
-          join_attributes.merge!(@reflection.through_reflection.options[:conditions])
-        end
-
-        join_attributes
-      end
-
-      def conditions
-        @conditions = build_conditions unless defined?(@conditions)
-        @conditions
-      end
-
-      def build_conditions
-        association_conditions = @reflection.options[:conditions]
-        through_conditions = build_through_conditions
-        source_conditions = @reflection.source_reflection.options[:conditions]
-        uses_sti = !@reflection.through_reflection.klass.descends_from_active_record?
-
-        if association_conditions || through_conditions || source_conditions || uses_sti
-          all = []
-
-          [association_conditions, source_conditions].each do |conditions|
-            all << interpolate_sql(sanitize_sql(conditions)) if conditions
-          end
-
-          all << through_conditions  if through_conditions
-          all << build_sti_condition if uses_sti
-
-          all.map { |sql| "(#{sql})" } * ' AND '
-        end
-      end
-
-      def build_through_conditions
-        conditions = @reflection.through_reflection.options[:conditions]
-        if conditions.is_a?(Hash)
-          interpolate_sql(@reflection.through_reflection.klass.send(:sanitize_sql, conditions)).gsub(
-            @reflection.quoted_table_name,
-            @reflection.through_reflection.quoted_table_name)
-        elsif conditions
-          interpolate_sql(sanitize_sql(conditions))
-        end
-      end
-
-      def build_sti_condition
-        @reflection.through_reflection.klass.send(:type_condition).to_sql
-      end
-
-      alias_method :sql_conditions, :conditions
-    end
-  end
-end
diff --git a/activerecord/lib/active_record/attribute_methods/primary_key.rb b/activerecord/lib/active_record/attribute_methods/primary_key.rb
index 75ae06f5e9..fcdd31ddea 100644
--- a/activerecord/lib/active_record/attribute_methods/primary_key.rb
+++ b/activerecord/lib/active_record/attribute_methods/primary_key.rb
@@ -14,26 +14,37 @@ module ActiveRecord
         # Defines the primary key field -- can be overridden in subclasses. Overwriting will negate any effect of the
         # primary_key_prefix_type setting, though.
         def primary_key
-          reset_primary_key
+          @primary_key ||= reset_primary_key
         end
 
         def reset_primary_key #:nodoc:
-          key = get_primary_key(base_class.name)
+          key = self == base_class ? get_primary_key(base_class.name) :
+            base_class.primary_key
+
           set_primary_key(key)
           key
         end
 
         def get_primary_key(base_name) #:nodoc:
-          key = 'id'
+          return 'id' unless base_name && !base_name.blank?
+
           case primary_key_prefix_type
-            when :table_name
-              key = base_name.to_s.foreign_key(false)
-            when :table_name_with_underscore
-              key = base_name.to_s.foreign_key
+          when :table_name
+            base_name.foreign_key(false)
+          when :table_name_with_underscore
+            base_name.foreign_key
+          else
+            if ActiveRecord::Base != self && connection.table_exists?(table_name)
+              connection.primary_key(table_name)
+            else
+              'id'
+            end
           end
-          key
         end
 
+        attr_accessor :original_primary_key
+        attr_writer :primary_key
+
         # Sets the name of the primary key column to use to the given value,
         # or (if the value is nil or false) to the value returned by the given
         # block.
@@ -42,9 +53,12 @@ module ActiveRecord
         #     set_primary_key "sysid"
         #   end
         def set_primary_key(value = nil, &block)
-          define_attr_method :primary_key, value, &block
+          @primary_key ||= ''
+          self.original_primary_key = @primary_key
+          value &&= value.to_s
+          connection_pool.primary_keys[table_name] = value
+          self.primary_key = block_given? ? instance_eval(&block) : value
         end
-        alias :primary_key= :set_primary_key
       end
     end
   end
diff --git a/activerecord/lib/active_record/attribute_methods/read.rb b/activerecord/lib/active_record/attribute_methods/read.rb
index 506f6e878f..ab86d8bad1 100644
--- a/activerecord/lib/active_record/attribute_methods/read.rb
+++ b/activerecord/lib/active_record/attribute_methods/read.rb
@@ -20,7 +20,7 @@ module ActiveRecord
         # be cached. Usually caching only pays off for attributes with expensive conversion
         # methods, like time related columns (e.g. +created_at+, +updated_at+).
         def cache_attributes(*attribute_names)
-          attribute_names.each {|attr| cached_attributes << attr.to_s}
+          cached_attributes.merge attribute_names.map { |attr| attr.to_s }
         end
 
         # Returns the attributes which are cached. By default time related columns
@@ -39,7 +39,7 @@ module ActiveRecord
             if serialized_attributes.include?(attr_name)
               define_read_method_for_serialized_attribute(attr_name)
             else
-              define_read_method(attr_name.to_sym, attr_name, columns_hash[attr_name])
+              define_read_method(attr_name, attr_name, columns_hash[attr_name])
             end
 
             if attr_name == primary_key && attr_name != "id"
@@ -54,17 +54,17 @@ module ActiveRecord
 
           # Define read method for serialized attribute.
           def define_read_method_for_serialized_attribute(attr_name)
-            access_code = "@attributes_cache['#{attr_name}'] ||= unserialize_attribute('#{attr_name}')"
-            generated_attribute_methods.module_eval("def #{attr_name}; #{access_code}; end", __FILE__, __LINE__)
+            access_code = "@attributes_cache['#{attr_name}'] ||= @attributes['#{attr_name}']"
+            generated_attribute_methods.module_eval("def _#{attr_name}; #{access_code}; end; alias #{attr_name} _#{attr_name}", __FILE__, __LINE__)
           end
 
           # Define an attribute reader method.  Cope with nil column.
           def define_read_method(symbol, attr_name, column)
-            cast_code = column.type_cast_code('v') if column
-            access_code = cast_code ? "(v=@attributes['#{attr_name}']) && #{cast_code}" : "@attributes['#{attr_name}']"
+            cast_code = column.type_cast_code('v')
+            access_code = "(v=@attributes['#{attr_name}']) && #{cast_code}"
 
             unless attr_name.to_s == self.primary_key.to_s
-              access_code = access_code.insert(0, "missing_attribute('#{attr_name}', caller) unless @attributes.has_key?('#{attr_name}'); ")
+              access_code.insert(0, "missing_attribute('#{attr_name}', caller) unless @attributes.has_key?('#{attr_name}'); ")
             end
 
             if cache_attribute?(attr_name)
@@ -106,14 +106,10 @@ module ActiveRecord
 
       # Returns the unserialized object of the attribute.
       def unserialize_attribute(attr_name)
-        unserialized_object = object_from_yaml(@attributes[attr_name])
+        coder = self.class.serialized_attributes[attr_name]
+        unserialized_object = coder.load(@attributes[attr_name])
 
-        if unserialized_object.is_a?(self.class.serialized_attributes[attr_name]) || unserialized_object.nil?
-          @attributes.frozen? ? unserialized_object : @attributes[attr_name] = unserialized_object
-        else
-          raise SerializationTypeMismatch,
-            "#{attr_name} was supposed to be a #{self.class.serialized_attributes[attr_name]}, but was a #{unserialized_object.class.to_s}"
-        end
+        @attributes.frozen? ? unserialized_object : @attributes[attr_name] = unserialized_object
       end
 
       private
diff --git a/activerecord/lib/active_record/attribute_methods/time_zone_conversion.rb b/activerecord/lib/active_record/attribute_methods/time_zone_conversion.rb
index dc2785b6bf..76218d2a73 100644
--- a/activerecord/lib/active_record/attribute_methods/time_zone_conversion.rb
+++ b/activerecord/lib/active_record/attribute_methods/time_zone_conversion.rb
@@ -40,12 +40,13 @@ module ActiveRecord
           def define_method_attribute=(attr_name)
             if create_time_zone_conversion_attribute?(attr_name, columns_hash[attr_name])
               method_body, line = <<-EOV, __LINE__ + 1
-                def #{attr_name}=(time)
+                def #{attr_name}=(original_time)
+                  time = original_time.dup unless original_time.nil?
                   unless time.acts_like?(:time)
                     time = time.is_a?(String) ? Time.zone.parse(time) : time.to_time rescue time
                   end
                   time = time.in_time_zone rescue nil if time
-                  write_attribute(:#{attr_name}, time)
+                  write_attribute(:#{attr_name}, (time || original_time))
                 end
               EOV
               generated_attribute_methods.module_eval(method_body, __FILE__, line)
diff --git a/activerecord/lib/active_record/autosave_association.rb b/activerecord/lib/active_record/autosave_association.rb
index 2dd69bd12c..ec27fd2e63 100644
--- a/activerecord/lib/active_record/autosave_association.rb
+++ b/activerecord/lib/active_record/autosave_association.rb
@@ -321,27 +321,30 @@ module ActiveRecord
 
         if records = associated_records_to_validate_or_save(association, @new_record_before_save, autosave)
           begin
-            records.each do |record|
-              next if record.destroyed?
-
-              if autosave && record.marked_for_destruction?
-                association.destroy(record)
-              elsif autosave != false && (@new_record_before_save || record.new_record?)
-                if autosave
-                  saved = association.send(:insert_record, record, false, false)
-                else
-                  association.send(:insert_record, record)
-                end
-              elsif autosave
-                saved = record.save(:validate => false)
+          records.each do |record|
+            next if record.destroyed?
+
+            saved = true
+
+            if autosave && record.marked_for_destruction?
+              association.destroy(record)
+            elsif autosave != false && (@new_record_before_save || record.new_record?)
+              if autosave
+                saved = association.send(:insert_record, record, false)
+              else
+                association.send(:insert_record, record)
               end
-
-              raise ActiveRecord::Rollback if saved == false
+            elsif autosave
+              saved = record.save(:validate => false)
             end
+
+            raise ActiveRecord::Rollback unless saved
+          end
           rescue
             records.each {|x| IdentityMap.remove(x) } if IdentityMap.enabled?
             raise
           end
+
         end
 
         # reconstruct the scope now that we know the owner's id
@@ -365,8 +368,8 @@ module ActiveRecord
           association.destroy
         else
           key = reflection.options[:primary_key] ? send(reflection.options[:primary_key]) : id
-          if autosave != false && (new_record? || association.new_record? || association[reflection.primary_key_name] != key || autosave)
-            association[reflection.primary_key_name] = key
+          if autosave != false && (new_record? || association.new_record? || association[reflection.foreign_key] != key || autosave)
+            association[reflection.foreign_key] = key
             saved = association.save(:validate => !autosave)
             raise ActiveRecord::Rollback if !saved && autosave
             saved
@@ -389,7 +392,8 @@ module ActiveRecord
 
           if association.updated?
             association_id = association.send(reflection.options[:primary_key] || :id)
-            self[reflection.primary_key_name] = association_id
+            self[reflection.foreign_key] = association_id
+            association.loaded!
           end
 
           saved if autosave
diff --git a/activerecord/lib/active_record/base.rb b/activerecord/lib/active_record/base.rb
index 24c87662b8..01f5f4eccd 100644
--- a/activerecord/lib/active_record/base.rb
+++ b/activerecord/lib/active_record/base.rb
@@ -1,3 +1,8 @@
+begin
+  require 'psych'
+rescue LoadError
+end
+
 require 'yaml'
 require 'set'
 require 'active_support/benchmarkable'
@@ -244,6 +249,17 @@ module ActiveRecord #:nodoc:
   #   user = User.create(:preferences => %w( one two three ))
   #   User.find(user.id).preferences    # raises SerializationTypeMismatch
   #
+  # When you specify a class option, the default value for that attribute will be a new
+  # instance of that class.
+  #
+  #   class User < ActiveRecord::Base
+  #     serialize :preferences, OpenStruct
+  #   end
+  #
+  #   user = User.new
+  #   user.preferences.theme_color = "red"
+  #
+  #
   # == Single table inheritance
   #
   # Active Record allows inheritance by storing the name of the class in a column that by
@@ -531,7 +547,15 @@ module ActiveRecord #:nodoc:
       #     serialize :preferences
       #   end
       def serialize(attr_name, class_name = Object)
-        serialized_attributes[attr_name.to_s] = class_name
+        coder = if [:load, :dump].all? { |x| class_name.respond_to?(x) }
+                  class_name
+                else
+                  Coders::YAMLColumn.new(class_name)
+                end
+
+        # merge new serialized attribute and create new hash to ensure that each class in inheritance hierarchy
+        # has its own hash of own serialized attributes
+        self.serialized_attributes = serialized_attributes.merge(attr_name.to_s => coder)
       end
 
       # Guesses the table name (in forced lower-case) based on the name of the class in the
@@ -614,6 +638,9 @@ module ActiveRecord #:nodoc:
       def set_table_name(value = nil, &block)
         @quoted_table_name = nil
         define_attr_method :table_name, value, &block
+
+        @arel_table = Arel::Table.new(table_name, :engine => arel_engine)
+        @relation = Relation.new(self, arel_table)
       end
       alias :table_name= :set_table_name
 
@@ -657,16 +684,12 @@ module ActiveRecord #:nodoc:
 
       # Returns an array of column objects for the table associated with this class.
       def columns
-        unless defined?(@columns) && @columns
-          @columns = connection.columns(table_name, "#{name} Columns")
-          @columns.each { |column| column.primary = column.name == primary_key }
-        end
-        @columns
+        connection_pool.columns[table_name]
       end
 
       # Returns a hash of column objects for the table associated with this class.
       def columns_hash
-        @columns_hash ||= Hash[columns.map { |column| [column.name, column] }]
+        connection_pool.columns_hash[table_name]
       end
 
       # Returns an array of column names as strings.
@@ -723,8 +746,14 @@ module ActiveRecord #:nodoc:
       def reset_column_information
         connection.clear_cache!
         undefine_attribute_methods
-        @column_names = @columns = @columns_hash = @content_columns = @dynamic_methods_hash = @inheritance_column = nil
-        @arel_engine = @relation = @arel_table = nil
+        connection_pool.clear_table_cache!(table_name) if table_exists?
+
+        @column_names = @content_columns = @dynamic_methods_hash = @inheritance_column = nil
+        @arel_engine = @relation = nil
+      end
+
+      def clear_cache! # :nodoc:
+        connection_pool.clear_cache!
       end
 
       def attribute_method?(attribute)
@@ -762,7 +791,7 @@ module ActiveRecord #:nodoc:
         :true == (@finder_needs_type_condition ||= descends_from_active_record? ? :false : :true)
       end
 
-      # Returns a string like 'Post id:integer, title:string, body:text'
+      # Returns a string like 'Post(id:integer, title:string, body:text)'
       def inspect
         if self == Base
           super
@@ -827,13 +856,13 @@ module ActiveRecord #:nodoc:
       end
 
       def arel_table
-        @arel_table ||= Arel::Table.new(table_name, arel_engine)
+        Arel::Table.new(table_name, arel_engine)
       end
 
       def arel_engine
         @arel_engine ||= begin
           if self == ActiveRecord::Base
-            Arel::Table.engine
+            ActiveRecord::Base
           else
             connection_handler.connection_pools[name] ? self : superclass.arel_engine
           end
@@ -874,35 +903,51 @@ module ActiveRecord #:nodoc:
         reset_scoped_methods
       end
 
-      private
+      # Specifies how the record is loaded by +Marshal+.
+      #
+      # +_load+ sets an instance variable for each key in the hash it takes as input.
+      # Override this method if you require more complex marshalling.
+      def _load(data)
+        record = allocate
+        record.init_with(Marshal.load(data))
+        record
+      end
 
-        def relation #:nodoc:
-          @relation ||= Relation.new(self, arel_table)
-          finder_needs_type_condition? ? @relation.where(type_condition) : @relation
-        end
 
-        # Finder methods must instantiate through this method to work with the
-        # single-table inheritance model that makes it possible to create
-        # objects of different types from the same table.
-        def instantiate(record)
-          sti_class = find_sti_class(record[inheritance_column])
-          record_id = sti_class.primary_key && record[sti_class.primary_key]
+      # Finder methods must instantiate through this method to work with the
+      # single-table inheritance model that makes it possible to create
+      # objects of different types from the same table.
+      def instantiate(record)
+        sti_class = find_sti_class(record[inheritance_column])
+        record_id = sti_class.primary_key && record[sti_class.primary_key]
 
-          if ActiveRecord::IdentityMap.enabled? && record_id
-            if (column = sti_class.columns_hash[sti_class.primary_key]) && column.number?
-              record_id = record_id.to_i
-            end
-            if instance = IdentityMap.get(sti_class, record_id)
-              instance.reinit_with('attributes' => record)
-            else
-              instance = sti_class.allocate.init_with('attributes' => record)
-              IdentityMap.add(instance)
-            end
+        if ActiveRecord::IdentityMap.enabled? && record_id
+          if (column = sti_class.columns_hash[sti_class.primary_key]) && column.number?
+            record_id = record_id.to_i
+          end
+          if instance = IdentityMap.get(sti_class, record_id)
+            instance.reinit_with('attributes' => record)
           else
             instance = sti_class.allocate.init_with('attributes' => record)
+            IdentityMap.add(instance)
           end
+        else
+          instance = sti_class.allocate.init_with('attributes' => record)
+        end
+
+        instance
+      end
 
-          instance
+      private
+
+        def relation #:nodoc:
+          @relation ||= Relation.new(self, arel_table)
+
+          if finder_needs_type_condition?
+            @relation.where(type_condition).create_with(inheritance_column.to_sym => sti_name)
+          else
+            @relation
+          end
         end
 
         def find_sti_class(type_name)
@@ -932,11 +977,10 @@ module ActiveRecord #:nodoc:
         end
 
         def type_condition
-          sti_column = arel_table[inheritance_column]
-          condition = sti_column.eq(sti_name)
-          descendants.each { |subclass| condition = condition.or(sti_column.eq(subclass.sti_name)) }
+          sti_column = arel_table[inheritance_column.to_sym]
+          sti_names  = ([self] + descendants).map { |model| model.sti_name }
 
-          condition
+          sti_column.in(sti_names)
         end
 
         # Guesses the table name, but does not decorate it with prefix and suffix information.
@@ -1383,6 +1427,8 @@ MSG
       # hence you can't have attributes that aren't part of the table columns.
       def initialize(attributes = nil)
         @attributes = attributes_from_column_definition
+        @association_cache = {}
+        @aggregation_cache = {}
         @attributes_cache = {}
         @new_record = true
         @readonly = false
@@ -1392,15 +1438,32 @@ MSG
         @changed_attributes = {}
 
         ensure_proper_type
+        set_serialized_attributes
 
         populate_with_current_scope_attributes
         self.attributes = attributes unless attributes.nil?
 
         result = yield self if block_given?
-        _run_initialize_callbacks
+        run_callbacks :initialize
         result
       end
 
+      # Populate +coder+ with attributes about this record that should be
+      # serialized.  The structure of +coder+ defined in this method is
+      # guaranteed to match the structure of +coder+ passed to the +init_with+
+      # method.
+      #
+      # Example:
+      #
+      #   class Post < ActiveRecord::Base
+      #   end
+      #   coder = {}
+      #   Post.new.encode_with(coder)
+      #   coder # => { 'id' => nil, ... }
+      def encode_with(coder)
+        coder['attributes'] = attributes
+      end
+
       # Initialize an empty model object from +coder+.  +coder+ must contain
       # the attributes necessary for initializing an empty model object.  For
       # example:
@@ -1413,15 +1476,30 @@ MSG
       #   post.title # => 'hello world'
       def init_with(coder)
         @attributes = coder['attributes']
+
+        set_serialized_attributes
+
         @attributes_cache, @previously_changed, @changed_attributes = {}, {}, {}
+        @association_cache = {}
+        @aggregation_cache = {}
         @readonly = @destroyed = @marked_for_destruction = false
         @new_record = false
-        _run_find_callbacks
-        _run_initialize_callbacks
+        run_callbacks :find
+        run_callbacks :initialize
 
         self
       end
 
+      # Specifies how the record is dumped by +Marshal+.
+      #
+      # +_dump+ emits a marshalled hash which has been passed to +encode_with+. Override this
+      # method if you require more complex marshalling.
+      def _dump(level)
+        dump = {}
+        encode_with(dump)
+        Marshal.dump(dump)
+      end
+
       # Returns a String, which Action Pack uses for constructing an URL to this
       # object. The default implementation returns this record's id as a String,
       # or nil if this record's unsaved.
@@ -1511,8 +1589,10 @@ MSG
         attributes.each do |k, v|
           if k.include?("(")
             multi_parameter_attributes << [ k, v ]
+          elsif respond_to?("#{k}=")
+            send("#{k}=", v)
           else
-            respond_to?(:"#{k}=") ? send(:"#{k}=", v) : raise(UnknownAttributeError, "unknown attribute: #{k}")
+            raise(UnknownAttributeError, "unknown attribute: #{k}")
           end
         end
 
@@ -1552,7 +1632,7 @@ MSG
       # Returns true if the specified +attribute+ has been set by the user or by a database load and is neither
       # nil nor empty? (the latter only applies to objects that respond to empty?, most notably Strings).
       def attribute_present?(attribute)
-        !read_attribute(attribute).blank?
+        !_read_attribute(attribute).blank?
       end
 
       # Returns the column object for the named attribute.
@@ -1625,9 +1705,9 @@ MSG
           @changed_attributes[attr] = orig_value if field_changed?(attr, orig_value, @attributes[attr])
         end
 
-        clear_aggregation_cache
-        clear_association_cache
-        @attributes_cache   = {}
+        @aggregation_cache = {}
+        @association_cache = {}
+        @attributes_cache = {}
         @new_record  = true
 
         ensure_proper_type
@@ -1649,7 +1729,7 @@ MSG
       # Returns the contents of the record as a nicely formatted string.
       def inspect
         attributes_as_nice_string = self.class.column_names.collect { |name|
-          if has_attribute?(name) || new_record?
+          if has_attribute?(name)
             "#{name}: #{attribute_for_inspect(name)}"
           end
         }.compact.join(", ")
@@ -1673,6 +1753,13 @@ MSG
 
     private
 
+      def set_serialized_attributes
+        (@attributes.keys & self.class.serialized_attributes.keys).each do |key|
+          coder = self.class.serialized_attributes[key]
+          @attributes[key] = coder.load @attributes[key]
+        end
+      end
+
       # Sets the attribute used for single table inheritance to this class name if this is not the
       # ActiveRecord::Base descendant.
       # Considering the hierarchy Reply < Message < ActiveRecord::Base, this makes it possible to
@@ -1694,17 +1781,25 @@ MSG
       # Returns a copy of the attributes hash where all the values have been safely quoted for use in
       # an Arel insert/update method.
       def arel_attributes_values(include_primary_key = true, include_readonly_attributes = true, attribute_names = @attributes.keys)
-        attrs = {}
+        attrs      = {}
+        klass      = self.class
+        arel_table = klass.arel_table
+
         attribute_names.each do |name|
           if (column = column_for_attribute(name)) && (include_primary_key || !column.primary)
 
             if include_readonly_attributes || (!include_readonly_attributes && !self.class.readonly_attributes.include?(name))
-              value = read_attribute(name)
 
-              if !value.nil? && self.class.serialized_attributes.key?(name)
-                value = YAML.dump value
-              end
-              attrs[self.class.arel_table[name]] = value
+              value = if coder = klass.serialized_attributes[name]
+                        coder.dump @attributes[name]
+                      else
+                        # FIXME: we need @attributes to be used consistently.
+                        # If the values stored in @attributes were already type
+                        # casted, this code could be simplified
+                        read_attribute(name)
+                      end
+
+              attrs[arel_table[name]] = value
             end
           end
         end
@@ -1716,12 +1811,6 @@ MSG
         self.class.connection.quote(value, column)
       end
 
-      # Interpolate custom SQL string in instance context.
-      # Optional record argument is meant for custom insert_sql.
-      def interpolate_sql(sql, record = nil)
-        instance_eval("%@#{sql.gsub('@', '\@')}@", __FILE__, __LINE__)
-      end
-
       # Instantiates objects for all attribute classes that needs more than one constructor parameter. This is done
       # by calling new on the column type or aggregation type (through composed_of) object with these parameters.
       # So having the pairs written_on(1) = "2004", written_on(2) = "6", written_on(3) = "24", will instantiate
@@ -1828,27 +1917,20 @@ MSG
         end
       end
 
-      def object_from_yaml(string)
-        return string unless string.is_a?(String) && string =~ /^---/
-        YAML::load(string) rescue string
-      end
-
       def populate_with_current_scope_attributes
         if scope = self.class.send(:current_scoped_methods)
           create_with = scope.scope_for_create
           create_with.each { |att,value|
-            respond_to?(:"#{att}=") && send("#{att}=", value)
+            respond_to?("#{att}=") && send("#{att}=", value)
           }
         end
       end
 
       # Clear attributes and changed_attributes
       def clear_timestamp_attributes
-        %w(created_at created_on updated_at updated_on).each do |attribute_name|
-          if has_attribute?(attribute_name)
-            self[attribute_name] = nil
-            changed_attributes.delete(attribute_name)
-          end
+        all_timestamp_attributes_in_model.each do |attribute_name|
+          self[attribute_name] = nil
+          changed_attributes.delete(attribute_name)
         end
       end
   end
diff --git a/activerecord/lib/active_record/callbacks.rb b/activerecord/lib/active_record/callbacks.rb
index 47428cfd0f..ff4ce1b605 100644
--- a/activerecord/lib/active_record/callbacks.rb
+++ b/activerecord/lib/active_record/callbacks.rb
@@ -25,9 +25,13 @@ module ActiveRecord
   # Check out <tt>ActiveRecord::Transactions</tt> for more details about <tt>after_commit</tt> and
   # <tt>after_rollback</tt>.
   #
-  # That's a total of ten callbacks, which gives you immense power to react and prepare for each state in the
+  # Lastly an <tt>after_find</tt> and <tt>after_initialize</tt> callback is triggered for each object that 
+  # is found and instantiated by a finder, with <tt>after_initialize</tt> being triggered after new objects
+  # are instantiated as well.
+  #
+  # That's a total of twelve callbacks, which gives you immense power to react and prepare for each state in the
   # Active Record life cycle. The sequence for calling <tt>Base#save</tt> for an existing record is similar,
-  # except that each <tt>_on_create</tt> callback is replaced by the corresponding <tt>_on_update</tt> callback.
+  # except that each <tt>_create</tt> callback is replaced by the corresponding <tt>_update</tt> callback.
   #
   # Examples:
   #   class CreditCard < ActiveRecord::Base
@@ -185,14 +189,6 @@ module ActiveRecord
   #                    'puts "Evaluated after parents are destroyed"'
   #   end
   #
-  # == The +after_find+ and +after_initialize+ exceptions
-  #
-  # Because +after_find+ and +after_initialize+ are called for each object found and instantiated by a finder,
-  # such as <tt>Base.find(:all)</tt>, we've had to implement a simple performance constraint (50% more speed
-  # on a simple test case). Unlike all the other callbacks, +after_find+ and +after_initialize+ will only be
-  # run if an explicit implementation is defined (<tt>def after_find</tt>). In that case, all of the
-  # callback types will be called.
-  #
   # == <tt>before_validation*</tt> returning statements
   #
   # If the returning value of a +before_validation+ callback can be evaluated to +false+, the process will be
@@ -237,25 +233,25 @@ module ActiveRecord
     end
 
     def destroy #:nodoc:
-      _run_destroy_callbacks { super }
+      run_callbacks(:destroy) { super }
     end
 
     def touch(*) #:nodoc:
-      _run_touch_callbacks { super }
+      run_callbacks(:touch) { super }
     end
 
   private
 
     def create_or_update #:nodoc:
-      _run_save_callbacks { super }
+      run_callbacks(:save) { super }
     end
 
     def create #:nodoc:
-      _run_create_callbacks { super }
+      run_callbacks(:create) { super }
     end
 
     def update(*) #:nodoc:
-      _run_update_callbacks { super }
+      run_callbacks(:update) { super }
     end
   end
 end
diff --git a/activerecord/lib/active_record/coders/yaml_column.rb b/activerecord/lib/active_record/coders/yaml_column.rb
new file mode 100644
index 0000000000..fb59d9fb07
--- /dev/null
+++ b/activerecord/lib/active_record/coders/yaml_column.rb
@@ -0,0 +1,41 @@
+module ActiveRecord
+  # :stopdoc:
+  module Coders
+    class YAMLColumn
+      RESCUE_ERRORS = [ ArgumentError ]
+
+      if defined?(Psych) && defined?(Psych::SyntaxError)
+        RESCUE_ERRORS << Psych::SyntaxError
+      end
+
+      attr_accessor :object_class
+
+      def initialize(object_class = Object)
+        @object_class = object_class
+      end
+
+      def dump(obj)
+        YAML.dump obj
+      end
+
+      def load(yaml)
+        return object_class.new if object_class != Object && yaml.nil?
+        return yaml unless yaml.is_a?(String) && yaml =~ /^---/
+        begin
+          obj = YAML.load(yaml)
+
+          unless obj.is_a?(object_class) || obj.nil?
+            raise SerializationTypeMismatch,
+              "Attribute was supposed to be a #{object_class}, but was a #{obj.class}"
+          end
+          obj ||= object_class.new if object_class != Object
+
+          obj
+        rescue *RESCUE_ERRORS
+          yaml
+        end
+      end
+    end
+  end
+  # :startdoc
+end
diff --git a/activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb b/activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb
index cffa2387de..4297c26413 100644
--- a/activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb
+++ b/activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb
@@ -57,7 +57,9 @@ module ActiveRecord
     # * +wait_timeout+: number of seconds to block and wait for a connection
     #   before giving up and raising a timeout error (default 5 seconds).
     class ConnectionPool
+      attr_accessor :automatic_reconnect
       attr_reader :spec, :connections
+      attr_reader :columns, :columns_hash, :primary_keys, :tables
 
       # Creates a new ConnectionPool object. +spec+ is a ConnectionSpecification
       # object which describes database connection information (e.g. adapter,
@@ -81,6 +83,63 @@ module ActiveRecord
 
         @connections = []
         @checked_out = []
+        @automatic_reconnect = true
+        @tables = {}
+
+        @columns     = Hash.new do |h, table_name|
+          h[table_name] = with_connection do |conn|
+
+            # Fetch a list of columns
+            conn.columns(table_name, "#{table_name} Columns").tap do |columns|
+
+              # set primary key information
+              columns.each do |column|
+                column.primary = column.name == primary_keys[table_name]
+              end
+            end
+          end
+        end
+
+        @columns_hash = Hash.new do |h, table_name|
+          h[table_name] = Hash[columns[table_name].map { |col|
+            [col.name, col]
+          }]
+        end
+
+        @primary_keys = Hash.new do |h, table_name|
+          h[table_name] = with_connection do |conn|
+            table_exists?(table_name) ? conn.primary_key(table_name) : 'id'
+          end
+        end
+      end
+
+      # A cached lookup for table existence
+      def table_exists?(name)
+        return true if @tables.key? name
+
+        with_connection do |conn|
+          conn.tables.each { |table| @tables[table] = true }
+        end
+
+        @tables.key? name
+      end
+
+      # Clears out internal caches:
+      #
+      #   * columns
+      #   * columns_hash
+      #   * tables
+      def clear_cache!
+        @columns.clear
+        @columns_hash.clear
+        @tables.clear
+      end
+
+      # Clear out internal caches for table with +table_name+
+      def clear_table_cache!(table_name)
+        @columns.delete table_name
+        @columns_hash.delete table_name
+        @primary_keys.delete table_name
       end
 
       # Retrieve the connection associated with the current thread, or call
@@ -212,7 +271,7 @@ module ActiveRecord
       # calling +checkout+ on this pool.
       def checkin(conn)
         @connection_mutex.synchronize do
-          conn.send(:_run_checkin_callbacks) do
+          conn.run_callbacks :checkin do
             @checked_out.delete conn
             @queue.signal
           end
@@ -232,6 +291,8 @@ module ActiveRecord
       end
 
       def checkout_new_connection
+        raise ConnectionNotEstablished unless @automatic_reconnect
+
         c = new_connection
         @connections << c
         checkout_and_verify(c)
@@ -330,7 +391,7 @@ module ActiveRecord
         pool = @connection_pools[klass.name]
         return nil unless pool
 
-        @connection_pools.delete_if { |key, value| value == pool }
+        pool.automatic_reconnect = false
         pool.disconnect!
         pool.spec.config
       end
diff --git a/activerecord/lib/active_record/connection_adapters/abstract/database_statements.rb b/activerecord/lib/active_record/connection_adapters/abstract/database_statements.rb
index ee9a0af35c..5c1ce173c8 100644
--- a/activerecord/lib/active_record/connection_adapters/abstract/database_statements.rb
+++ b/activerecord/lib/active_record/connection_adapters/abstract/database_statements.rb
@@ -1,3 +1,5 @@
+require 'active_support/core_ext/module/deprecation'
+
 module ActiveRecord
   module ConnectionAdapters # :nodoc:
     module DatabaseStatements
@@ -229,6 +231,8 @@ module ActiveRecord
       #
       # This method *modifies* the +sql+ parameter.
       #
+      # This method is deprecated!! Stop using it!
+      #
       # ===== Examples
       #  add_limit_offset!('SELECT * FROM suppliers', {:limit => 10, :offset => 50})
       # generates
@@ -243,6 +247,7 @@ module ActiveRecord
         end
         sql
       end
+      deprecate :add_limit_offset!
 
       def default_sequence_name(table, column)
         nil
@@ -256,7 +261,15 @@ module ActiveRecord
       # Inserts the given fixture into the table. Overridden in adapters that require
       # something beyond a simple insert (eg. Oracle).
       def insert_fixture(fixture, table_name)
-        execute "INSERT INTO #{quote_table_name(table_name)} (#{fixture.key_list}) VALUES (#{fixture.value_list})", 'Fixture Insert'
+        columns = Hash[columns(table_name).map { |c| [c.name, c] }]
+
+        key_list   = []
+        value_list = fixture.map do |name, value|
+          key_list << quote_column_name(name)
+          quote(value, columns[name])
+        end
+
+        execute "INSERT INTO #{quote_table_name(table_name)} (#{key_list.join(', ')}) VALUES (#{value_list.join(', ')})", 'Fixture Insert'
       end
 
       def empty_insert_statement_value
@@ -271,6 +284,25 @@ module ActiveRecord
         "WHERE #{quoted_primary_key} IN (SELECT #{quoted_primary_key} FROM #{quoted_table_name} #{where_sql})"
       end
 
+      # Sanitizes the given LIMIT parameter in order to prevent SQL injection.
+      #
+      # The +limit+ may be anything that can evaluate to a string via #to_s. It
+      # should look like an integer, or a comma-delimited list of integers, or 
+      # an Arel SQL literal.
+      #
+      # Returns Integer and Arel::Nodes::SqlLiteral limits as is. 
+      # Returns the sanitized limit parameter, either as an integer, or as a
+      # string which contains a comma-delimited list of integers.
+      def sanitize_limit(limit)
+        if limit.is_a?(Integer) || limit.is_a?(Arel::Nodes::SqlLiteral)
+          limit
+        elsif limit.to_s =~ /,/
+          Arel.sql limit.to_s.split(',').map{ |i| Integer(i) }.join(',')
+        else
+          Integer(limit)
+        end
+      end
+
       protected
         # Returns an array of record hashes with the column names as keys and
         # column values as values.
@@ -294,21 +326,6 @@ module ActiveRecord
           update_sql(sql, name)
         end
 
-        # Sanitizes the given LIMIT parameter in order to prevent SQL injection.
-        #
-        # +limit+ may be anything that can evaluate to a string via #to_s. It
-        # should look like an integer, or a comma-delimited list of integers.
-        #
-        # Returns the sanitized limit parameter, either as an integer, or as a
-        # string which contains a comma-delimited list of integers.
-        def sanitize_limit(limit)
-          if limit.to_s =~ /,/
-            limit.to_s.split(',').map{ |i| i.to_i }.join(',')
-          else
-            limit.to_i
-          end
-        end
-
         # Send a rollback message to all records after they have been rolled back. If rollback
         # is false, only rollback records since the last save point.
         def rollback_transaction_records(rollback) #:nodoc
diff --git a/activerecord/lib/active_record/connection_adapters/abstract/query_cache.rb b/activerecord/lib/active_record/connection_adapters/abstract/query_cache.rb
index d555308485..1db397f584 100644
--- a/activerecord/lib/active_record/connection_adapters/abstract/query_cache.rb
+++ b/activerecord/lib/active_record/connection_adapters/abstract/query_cache.rb
@@ -60,7 +60,7 @@ module ActiveRecord
           result =
             if @query_cache[sql].key?(binds)
               ActiveSupport::Notifications.instrument("sql.active_record",
-                :sql => sql, :name => "CACHE", :connection_id => self.object_id)
+                :sql => sql, :name => "CACHE", :connection_id => object_id)
               @query_cache[sql][binds]
             else
               @query_cache[sql][binds] = yield
diff --git a/activerecord/lib/active_record/connection_adapters/abstract/quoting.rb b/activerecord/lib/active_record/connection_adapters/abstract/quoting.rb
index a7a12faac2..7489e88eef 100644
--- a/activerecord/lib/active_record/connection_adapters/abstract/quoting.rb
+++ b/activerecord/lib/active_record/connection_adapters/abstract/quoting.rb
@@ -33,8 +33,9 @@ module ActiveRecord
         when BigDecimal then value.to_s('F')
         when Numeric    then value.to_s
         when Date, Time then "'#{quoted_date(value)}'"
+        when Symbol     then "'#{quote_string(value.to_s)}'"
         else
-          "'#{quote_string(value.to_s)}'"
+          "'#{quote_string(value.to_yaml)}'"
         end
       end
 
diff --git a/activerecord/lib/active_record/connection_adapters/abstract/schema_definitions.rb b/activerecord/lib/active_record/connection_adapters/abstract/schema_definitions.rb
index 60ccf9edf3..7ac48c6646 100644
--- a/activerecord/lib/active_record/connection_adapters/abstract/schema_definitions.rb
+++ b/activerecord/lib/active_record/connection_adapters/abstract/schema_definitions.rb
@@ -6,264 +6,6 @@ require 'bigdecimal/util'
 
 module ActiveRecord
   module ConnectionAdapters #:nodoc:
-    # An abstract definition of a column in a table.
-    class Column
-      TRUE_VALUES = [true, 1, '1', 't', 'T', 'true', 'TRUE'].to_set
-      FALSE_VALUES = [false, 0, '0', 'f', 'F', 'false', 'FALSE'].to_set
-
-      module Format
-        ISO_DATE = /\A(\d{4})-(\d\d)-(\d\d)\z/
-        ISO_DATETIME = /\A(\d{4})-(\d\d)-(\d\d) (\d\d):(\d\d):(\d\d)(\.\d+)?\z/
-      end
-
-      attr_reader :name, :default, :type, :limit, :null, :sql_type, :precision, :scale
-      attr_accessor :primary
-
-      # Instantiates a new column in the table.
-      #
-      # +name+ is the column's name, such as <tt>supplier_id</tt> in <tt>supplier_id int(11)</tt>.
-      # +default+ is the type-casted default value, such as +new+ in <tt>sales_stage varchar(20) default 'new'</tt>.
-      # +sql_type+ is used to extract the column's length, if necessary. For example +60+ in
-      # <tt>company_name varchar(60)</tt>.
-      # It will be mapped to one of the standard Rails SQL types in the <tt>type</tt> attribute.
-      # +null+ determines if this column allows +NULL+ values.
-      def initialize(name, default, sql_type = nil, null = true)
-        @name, @sql_type, @null = name, sql_type, null
-        @limit, @precision, @scale = extract_limit(sql_type), extract_precision(sql_type), extract_scale(sql_type)
-        @type = simplified_type(sql_type)
-        @default = extract_default(default)
-
-        @primary = nil
-      end
-
-      # Returns +true+ if the column is either of type string or text.
-      def text?
-        type == :string || type == :text
-      end
-
-      # Returns +true+ if the column is either of type integer, float or decimal.
-      def number?
-        type == :integer || type == :float || type == :decimal
-      end
-
-      def has_default?
-        !default.nil?
-      end
-
-      # Returns the Ruby class that corresponds to the abstract data type.
-      def klass
-        case type
-          when :integer       then Fixnum
-          when :float         then Float
-          when :decimal       then BigDecimal
-          when :datetime      then Time
-          when :date          then Date
-          when :timestamp     then Time
-          when :time          then Time
-          when :text, :string then String
-          when :binary        then String
-          when :boolean       then Object
-        end
-      end
-
-      # Casts value (which is a String) to an appropriate instance.
-      def type_cast(value)
-        return nil if value.nil?
-        case type
-          when :string    then value
-          when :text      then value
-          when :integer   then value.to_i rescue value ? 1 : 0
-          when :float     then value.to_f
-          when :decimal   then self.class.value_to_decimal(value)
-          when :datetime  then self.class.string_to_time(value)
-          when :timestamp then self.class.string_to_time(value)
-          when :time      then self.class.string_to_dummy_time(value)
-          when :date      then self.class.string_to_date(value)
-          when :binary    then self.class.binary_to_string(value)
-          when :boolean   then self.class.value_to_boolean(value)
-          else value
-        end
-      end
-
-      def type_cast_code(var_name)
-        case type
-          when :string    then nil
-          when :text      then nil
-          when :integer   then "(#{var_name}.to_i rescue #{var_name} ? 1 : 0)"
-          when :float     then "#{var_name}.to_f"
-          when :decimal   then "#{self.class.name}.value_to_decimal(#{var_name})"
-          when :datetime  then "#{self.class.name}.string_to_time(#{var_name})"
-          when :timestamp then "#{self.class.name}.string_to_time(#{var_name})"
-          when :time      then "#{self.class.name}.string_to_dummy_time(#{var_name})"
-          when :date      then "#{self.class.name}.string_to_date(#{var_name})"
-          when :binary    then "#{self.class.name}.binary_to_string(#{var_name})"
-          when :boolean   then "#{self.class.name}.value_to_boolean(#{var_name})"
-          else nil
-        end
-      end
-
-      # Returns the human name of the column name.
-      #
-      # ===== Examples
-      #  Column.new('sales_stage', ...).human_name # => 'Sales stage'
-      def human_name
-        Base.human_attribute_name(@name)
-      end
-
-      def extract_default(default)
-        type_cast(default)
-      end
-
-      # Used to convert from Strings to BLOBs
-      def string_to_binary(value)
-        self.class.string_to_binary(value)
-      end
-
-      class << self
-        # Used to convert from Strings to BLOBs
-        def string_to_binary(value)
-          value
-        end
-
-        # Used to convert from BLOBs to Strings
-        def binary_to_string(value)
-          value
-        end
-
-        def string_to_date(string)
-          return string unless string.is_a?(String)
-          return nil if string.empty?
-
-          fast_string_to_date(string) || fallback_string_to_date(string)
-        end
-
-        def string_to_time(string)
-          return string unless string.is_a?(String)
-          return nil if string.empty?
-
-          fast_string_to_time(string) || fallback_string_to_time(string)
-        end
-
-        def string_to_dummy_time(string)
-          return string unless string.is_a?(String)
-          return nil if string.empty?
-
-          string_to_time "2000-01-01 #{string}"
-        end
-
-        # convert something to a boolean
-        def value_to_boolean(value)
-          if value.is_a?(String) && value.blank?
-            nil
-          else
-            TRUE_VALUES.include?(value)
-          end
-        end
-
-        # convert something to a BigDecimal
-        def value_to_decimal(value)
-          # Using .class is faster than .is_a? and
-          # subclasses of BigDecimal will be handled
-          # in the else clause
-          if value.class == BigDecimal
-            value
-          elsif value.respond_to?(:to_d)
-            value.to_d
-          else
-            value.to_s.to_d
-          end
-        end
-
-        protected
-          # '0.123456' -> 123456
-          # '1.123456' -> 123456
-          def microseconds(time)
-            ((time[:sec_fraction].to_f % 1) * 1_000_000).to_i
-          end
-
-          def new_date(year, mon, mday)
-            if year && year != 0
-              Date.new(year, mon, mday) rescue nil
-            end
-          end
-
-          def new_time(year, mon, mday, hour, min, sec, microsec)
-            # Treat 0000-00-00 00:00:00 as nil.
-            return nil if year.nil? || year == 0
-
-            Time.time_with_datetime_fallback(Base.default_timezone, year, mon, mday, hour, min, sec, microsec) rescue nil
-          end
-
-          def fast_string_to_date(string)
-            if string =~ Format::ISO_DATE
-              new_date $1.to_i, $2.to_i, $3.to_i
-            end
-          end
-
-          # Doesn't handle time zones.
-          def fast_string_to_time(string)
-            if string =~ Format::ISO_DATETIME
-              microsec = ($7.to_f * 1_000_000).to_i
-              new_time $1.to_i, $2.to_i, $3.to_i, $4.to_i, $5.to_i, $6.to_i, microsec
-            end
-          end
-
-          def fallback_string_to_date(string)
-            new_date(*::Date._parse(string, false).values_at(:year, :mon, :mday))
-          end
-
-          def fallback_string_to_time(string)
-            time_hash = Date._parse(string)
-            time_hash[:sec_fraction] = microseconds(time_hash)
-
-            new_time(*time_hash.values_at(:year, :mon, :mday, :hour, :min, :sec, :sec_fraction))
-          end
-      end
-
-      private
-        def extract_limit(sql_type)
-          $1.to_i if sql_type =~ /\((.*)\)/
-        end
-
-        def extract_precision(sql_type)
-          $2.to_i if sql_type =~ /^(numeric|decimal|number)\((\d+)(,\d+)?\)/i
-        end
-
-        def extract_scale(sql_type)
-          case sql_type
-            when /^(numeric|decimal|number)\((\d+)\)/i then 0
-            when /^(numeric|decimal|number)\((\d+)(,(\d+))\)/i then $4.to_i
-          end
-        end
-
-        def simplified_type(field_type)
-          case field_type
-            when /int/i
-              :integer
-            when /float|double/i
-              :float
-            when /decimal|numeric|number/i
-              extract_scale(field_type) == 0 ? :integer : :decimal
-            when /datetime/i
-              :datetime
-            when /timestamp/i
-              :timestamp
-            when /time/i
-              :time
-            when /date/i
-              :date
-            when /clob/i, /text/i
-              :text
-            when /blob/i, /binary/i
-              :binary
-            when /char/i, /string/i
-              :string
-            when /boolean/i
-              :boolean
-          end
-        end
-    end
-
     class IndexDefinition < Struct.new(:table, :name, :unique, :columns, :lengths) #:nodoc:
     end
 
diff --git a/activerecord/lib/active_record/connection_adapters/abstract/schema_statements.rb b/activerecord/lib/active_record/connection_adapters/abstract/schema_statements.rb
index 5b9c48bafa..3ec7dd02a4 100644
--- a/activerecord/lib/active_record/connection_adapters/abstract/schema_statements.rb
+++ b/activerecord/lib/active_record/connection_adapters/abstract/schema_statements.rb
@@ -176,6 +176,13 @@ module ActiveRecord
       #    # Other column alterations here
       #  end
       #
+      # The +options+ hash can include the following keys:
+      # [<tt>:bulk</tt>]
+      #   Set this to true to make this a bulk alter query, such as
+      #   ALTER TABLE `users` ADD COLUMN age INT(11), ADD COLUMN birthdate DATETIME ...
+      #
+      #   Defaults to false.
+      #
       # ===== Examples
       # ====== Add a column
       #  change_table(:suppliers) do |t|
@@ -224,8 +231,14 @@ module ActiveRecord
       #
       # See also Table for details on
       # all of the various column transformation
-      def change_table(table_name)
-        yield Table.new(table_name, self)
+      def change_table(table_name, options = {})
+        if supports_bulk_alter? && options[:bulk]
+          recorder = ActiveRecord::Migration::CommandRecorder.new(self)
+          yield Table.new(table_name, recorder)
+          bulk_change_table(table_name, recorder.commands)
+        else
+          yield Table.new(table_name, self)
+        end
       end
 
       # Renames a table.
@@ -253,10 +266,7 @@ module ActiveRecord
       #  remove_column(:suppliers, :qualification)
       #  remove_columns(:suppliers, :qualification, :experience)
       def remove_column(table_name, *column_names)
-        raise ArgumentError.new("You must specify at least one column name.  Example: remove_column(:people, :first_name)") if column_names.empty?
-        column_names.flatten.each do |column_name|
-          execute "ALTER TABLE #{quote_table_name(table_name)} DROP #{quote_column_name(column_name)}"
-        end
+        columns_for_remove(table_name, *column_names).each {|column_name| execute "ALTER TABLE #{quote_table_name(table_name)} DROP #{column_name}" }
       end
       alias :remove_columns :remove_column
 
@@ -327,25 +337,8 @@ module ActiveRecord
       #
       # Note: SQLite doesn't support index length
       def add_index(table_name, column_name, options = {})
-        column_names = Array.wrap(column_name)
-        index_name   = index_name(table_name, :column => column_names)
-
-        if Hash === options # legacy support, since this param was a string
-          index_type = options[:unique] ? "UNIQUE" : ""
-          index_name = options[:name].to_s if options.key?(:name)
-        else
-          index_type = options
-        end
-
-        if index_name.length > index_name_length
-          raise ArgumentError, "Index name '#{index_name}' on table '#{table_name}' is too long; the limit is #{index_name_length} characters"
-        end
-        if index_name_exists?(table_name, index_name, false)
-          raise ArgumentError, "Index name '#{index_name}' on table '#{table_name}' already exists"
-        end
-        quoted_column_names = quoted_columns_for_index(column_names, options).join(", ")
-
-        execute "CREATE #{index_type} INDEX #{quote_column_name(index_name)} ON #{quote_table_name(table_name)} (#{quoted_column_names})"
+        index_name, index_type, index_columns = add_index_options(table_name, column_name, options)
+        execute "CREATE #{index_type} INDEX #{quote_column_name(index_name)} ON #{quote_table_name(table_name)} (#{index_columns})"
       end
 
       # Remove the given index from the table.
@@ -359,11 +352,7 @@ module ActiveRecord
       # Remove the index named by_branch_party in the accounts table.
       #   remove_index :accounts, :name => :by_branch_party
       def remove_index(table_name, options = {})
-        index_name = index_name(table_name, options)
-        unless index_name_exists?(table_name, index_name, true)
-          raise ArgumentError, "Index name '#{index_name}' on table '#{table_name}' does not exist"
-        end
-        remove_index!(table_name, index_name)
+        remove_index!(table_name, index_name_for_remove(table_name, options))
       end
 
       def remove_index!(table_name, index_name) #:nodoc:
@@ -469,7 +458,7 @@ module ActiveRecord
       end
 
       def type_to_sql(type, limit = nil, precision = nil, scale = nil) #:nodoc:
-        if native = native_database_types[type]
+        if native = native_database_types[type.to_sym]
           column_type_sql = (native.is_a?(Hash) ? native[:name] : native).dup
 
           if type == :decimal # ignore limit, use precision and scale
@@ -537,6 +526,45 @@ module ActiveRecord
           options.include?(:default) && !(options[:null] == false && options[:default].nil?)
         end
 
+        def add_index_options(table_name, column_name, options = {})
+          column_names = Array.wrap(column_name)
+          index_name   = index_name(table_name, :column => column_names)
+
+          if Hash === options # legacy support, since this param was a string
+            index_type = options[:unique] ? "UNIQUE" : ""
+            index_name = options[:name].to_s if options.key?(:name)
+          else
+            index_type = options
+          end
+
+          if index_name.length > index_name_length
+            raise ArgumentError, "Index name '#{index_name}' on table '#{table_name}' is too long; the limit is #{index_name_length} characters"
+          end
+          if index_name_exists?(table_name, index_name, false)
+            raise ArgumentError, "Index name '#{index_name}' on table '#{table_name}' already exists"
+          end
+          index_columns = quoted_columns_for_index(column_names, options).join(", ")
+
+          [index_name, index_type, index_columns]
+        end
+
+        def index_name_for_remove(table_name, options = {})
+          index_name = index_name(table_name, options)
+
+          unless index_name_exists?(table_name, index_name, true)
+            raise ArgumentError, "Index name '#{index_name}' on table '#{table_name}' does not exist"
+          end
+
+          index_name
+        end
+
+        def columns_for_remove(table_name, *column_names)
+          column_names = column_names.flatten
+
+          raise ArgumentError.new("You must specify at least one column name.  Example: remove_column(:people, :first_name)") if column_names.blank?
+          column_names.map {|column_name| quote_column_name(column_name) }
+        end
+
       private
       def table_definition
         TableDefinition.new(self)
diff --git a/activerecord/lib/active_record/connection_adapters/abstract_adapter.rb b/activerecord/lib/active_record/connection_adapters/abstract_adapter.rb
index 0282493219..0f44baa2fe 100644
--- a/activerecord/lib/active_record/connection_adapters/abstract_adapter.rb
+++ b/activerecord/lib/active_record/connection_adapters/abstract_adapter.rb
@@ -4,6 +4,7 @@ require 'bigdecimal/util'
 require 'active_support/core_ext/benchmark'
 
 # TODO: Autoload these files
+require 'active_record/connection_adapters/column'
 require 'active_record/connection_adapters/abstract/schema_definitions'
 require 'active_record/connection_adapters/abstract/schema_statements'
 require 'active_record/connection_adapters/abstract/database_statements'
@@ -77,8 +78,12 @@ module ActiveRecord
         false
       end
 
-      # Does this adapter support savepoints? PostgreSQL and MySQL do, SQLite
-      # does not.
+      def supports_bulk_alter?
+        false
+      end
+
+      # Does this adapter support savepoints? PostgreSQL and MySQL do,
+      # SQLite < 3.6.8 does not.
       def supports_savepoints?
         false
       end
@@ -204,11 +209,13 @@ module ActiveRecord
 
       protected
 
-        def log(sql, name = "SQL")
-          @instrumenter.instrument("sql.active_record",
-            :sql => sql, :name => name, :connection_id => object_id) do
-            yield
-          end
+        def log(sql, name = "SQL", binds = [])
+          @instrumenter.instrument(
+            "sql.active_record",
+            :sql           => sql,
+            :name          => name,
+            :connection_id => object_id,
+            :binds         => binds) { yield }
         rescue Exception => e
           message = "#{e.class.name}: #{e.message}: #{sql}"
           @logger.debug message if @logger
diff --git a/activerecord/lib/active_record/connection_adapters/column.rb b/activerecord/lib/active_record/connection_adapters/column.rb
new file mode 100644
index 0000000000..4e3d8a096f
--- /dev/null
+++ b/activerecord/lib/active_record/connection_adapters/column.rb
@@ -0,0 +1,268 @@
+module ActiveRecord
+  # :stopdoc:
+  module ConnectionAdapters
+    # An abstract definition of a column in a table.
+    class Column
+      TRUE_VALUES = [true, 1, '1', 't', 'T', 'true', 'TRUE'].to_set
+      FALSE_VALUES = [false, 0, '0', 'f', 'F', 'false', 'FALSE'].to_set
+
+      module Format
+        ISO_DATE = /\A(\d{4})-(\d\d)-(\d\d)\z/
+        ISO_DATETIME = /\A(\d{4})-(\d\d)-(\d\d) (\d\d):(\d\d):(\d\d)(\.\d+)?\z/
+      end
+
+      attr_reader :name, :default, :type, :limit, :null, :sql_type, :precision, :scale
+      attr_accessor :primary, :coder
+
+      alias :encoded? :coder
+
+      # Instantiates a new column in the table.
+      #
+      # +name+ is the column's name, such as <tt>supplier_id</tt> in <tt>supplier_id int(11)</tt>.
+      # +default+ is the type-casted default value, such as +new+ in <tt>sales_stage varchar(20) default 'new'</tt>.
+      # +sql_type+ is used to extract the column's length, if necessary. For example +60+ in
+      # <tt>company_name varchar(60)</tt>.
+      # It will be mapped to one of the standard Rails SQL types in the <tt>type</tt> attribute.
+      # +null+ determines if this column allows +NULL+ values.
+      def initialize(name, default, sql_type = nil, null = true)
+        @name      = name
+        @sql_type  = sql_type
+        @null      = null
+        @limit     = extract_limit(sql_type)
+        @precision = extract_precision(sql_type)
+        @scale     = extract_scale(sql_type)
+        @type      = simplified_type(sql_type)
+        @default   = extract_default(default)
+        @primary   = nil
+        @coder     = nil
+      end
+
+      # Returns +true+ if the column is either of type string or text.
+      def text?
+        type == :string || type == :text
+      end
+
+      # Returns +true+ if the column is either of type integer, float or decimal.
+      def number?
+        type == :integer || type == :float || type == :decimal
+      end
+
+      def has_default?
+        !default.nil?
+      end
+
+      # Returns the Ruby class that corresponds to the abstract data type.
+      def klass
+        case type
+        when :integer                     then Fixnum
+        when :float                       then Float
+        when :decimal                     then BigDecimal
+        when :datetime, :timestamp, :time then Time
+        when :date                        then Date
+        when :text, :string, :binary      then String
+        when :boolean                     then Object
+        end
+      end
+
+      # Casts value (which is a String) to an appropriate instance.
+      def type_cast(value)
+        return nil if value.nil?
+        return coder.load(value) if encoded?
+
+        klass = self.class
+
+        case type
+        when :string, :text        then value
+        when :integer              then value.to_i rescue value ? 1 : 0
+        when :float                then value.to_f
+        when :decimal              then klass.value_to_decimal(value)
+        when :datetime, :timestamp then klass.string_to_time(value)
+        when :time                 then klass.string_to_dummy_time(value)
+        when :date                 then klass.string_to_date(value)
+        when :binary               then klass.binary_to_string(value)
+        when :boolean              then klass.value_to_boolean(value)
+        else value
+        end
+      end
+
+      def type_cast_code(var_name)
+        klass = self.class.name
+
+        case type
+        when :string, :text        then var_name
+        when :integer              then "(#{var_name}.to_i rescue #{var_name} ? 1 : 0)"
+        when :float                then "#{var_name}.to_f"
+        when :decimal              then "#{klass}.value_to_decimal(#{var_name})"
+        when :datetime, :timestamp then "#{klass}.string_to_time(#{var_name})"
+        when :time                 then "#{klass}.string_to_dummy_time(#{var_name})"
+        when :date                 then "#{klass}.string_to_date(#{var_name})"
+        when :binary               then "#{klass}.binary_to_string(#{var_name})"
+        when :boolean              then "#{klass}.value_to_boolean(#{var_name})"
+        else var_name
+        end
+      end
+
+      # Returns the human name of the column name.
+      #
+      # ===== Examples
+      #  Column.new('sales_stage', ...).human_name # => 'Sales stage'
+      def human_name
+        Base.human_attribute_name(@name)
+      end
+
+      def extract_default(default)
+        type_cast(default)
+      end
+
+      # Used to convert from Strings to BLOBs
+      def string_to_binary(value)
+        self.class.string_to_binary(value)
+      end
+
+      class << self
+        # Used to convert from Strings to BLOBs
+        def string_to_binary(value)
+          value
+        end
+
+        # Used to convert from BLOBs to Strings
+        def binary_to_string(value)
+          value
+        end
+
+        def string_to_date(string)
+          return string unless string.is_a?(String)
+          return nil if string.empty?
+
+          fast_string_to_date(string) || fallback_string_to_date(string)
+        end
+
+        def string_to_time(string)
+          return string unless string.is_a?(String)
+          return nil if string.empty?
+
+          fast_string_to_time(string) || fallback_string_to_time(string)
+        end
+
+        def string_to_dummy_time(string)
+          return string unless string.is_a?(String)
+          return nil if string.empty?
+
+          string_to_time "2000-01-01 #{string}"
+        end
+
+        # convert something to a boolean
+        def value_to_boolean(value)
+          if value.is_a?(String) && value.blank?
+            nil
+          else
+            TRUE_VALUES.include?(value)
+          end
+        end
+
+        # convert something to a BigDecimal
+        def value_to_decimal(value)
+          # Using .class is faster than .is_a? and
+          # subclasses of BigDecimal will be handled
+          # in the else clause
+          if value.class == BigDecimal
+            value
+          elsif value.respond_to?(:to_d)
+            value.to_d
+          else
+            value.to_s.to_d
+          end
+        end
+
+        protected
+          # '0.123456' -> 123456
+          # '1.123456' -> 123456
+          def microseconds(time)
+            ((time[:sec_fraction].to_f % 1) * 1_000_000).to_i
+          end
+
+          def new_date(year, mon, mday)
+            if year && year != 0
+              Date.new(year, mon, mday) rescue nil
+            end
+          end
+
+          def new_time(year, mon, mday, hour, min, sec, microsec)
+            # Treat 0000-00-00 00:00:00 as nil.
+            return nil if year.nil? || year == 0
+
+            Time.time_with_datetime_fallback(Base.default_timezone, year, mon, mday, hour, min, sec, microsec) rescue nil
+          end
+
+          def fast_string_to_date(string)
+            if string =~ Format::ISO_DATE
+              new_date $1.to_i, $2.to_i, $3.to_i
+            end
+          end
+
+          # Doesn't handle time zones.
+          def fast_string_to_time(string)
+            if string =~ Format::ISO_DATETIME
+              microsec = ($7.to_f * 1_000_000).to_i
+              new_time $1.to_i, $2.to_i, $3.to_i, $4.to_i, $5.to_i, $6.to_i, microsec
+            end
+          end
+
+          def fallback_string_to_date(string)
+            new_date(*::Date._parse(string, false).values_at(:year, :mon, :mday))
+          end
+
+          def fallback_string_to_time(string)
+            time_hash = Date._parse(string)
+            time_hash[:sec_fraction] = microseconds(time_hash)
+
+            new_time(*time_hash.values_at(:year, :mon, :mday, :hour, :min, :sec, :sec_fraction))
+          end
+      end
+
+      private
+        def extract_limit(sql_type)
+          $1.to_i if sql_type =~ /\((.*)\)/
+        end
+
+        def extract_precision(sql_type)
+          $2.to_i if sql_type =~ /^(numeric|decimal|number)\((\d+)(,\d+)?\)/i
+        end
+
+        def extract_scale(sql_type)
+          case sql_type
+            when /^(numeric|decimal|number)\((\d+)\)/i then 0
+            when /^(numeric|decimal|number)\((\d+)(,(\d+))\)/i then $4.to_i
+          end
+        end
+
+        def simplified_type(field_type)
+          case field_type
+          when /int/i
+            :integer
+          when /float|double/i
+            :float
+          when /decimal|numeric|number/i
+            extract_scale(field_type) == 0 ? :integer : :decimal
+          when /datetime/i
+            :datetime
+          when /timestamp/i
+            :timestamp
+          when /time/i
+            :time
+          when /date/i
+            :date
+          when /clob/i, /text/i
+            :text
+          when /blob/i, /binary/i
+            :binary
+          when /char/i, /string/i
+            :string
+          when /boolean/i
+            :boolean
+          end
+        end
+    end
+  end
+  # :startdoc:
+end
diff --git a/activerecord/lib/active_record/connection_adapters/mysql2_adapter.rb b/activerecord/lib/active_record/connection_adapters/mysql2_adapter.rb
new file mode 100644
index 0000000000..7bad511c64
--- /dev/null
+++ b/activerecord/lib/active_record/connection_adapters/mysql2_adapter.rb
@@ -0,0 +1,610 @@
+# encoding: utf-8
+
+require 'mysql2'
+
+module ActiveRecord
+  class Base
+    def self.mysql2_connection(config)
+      config[:username] = 'root' if config[:username].nil?
+
+      if Mysql2::Client.const_defined? :FOUND_ROWS
+        config[:flags] = Mysql2::Client::FOUND_ROWS
+      end
+
+      client = Mysql2::Client.new(config.symbolize_keys)
+      options = [config[:host], config[:username], config[:password], config[:database], config[:port], config[:socket], 0]
+      ConnectionAdapters::Mysql2Adapter.new(client, logger, options, config)
+    end
+  end
+
+  module ConnectionAdapters
+    class Mysql2IndexDefinition < Struct.new(:table, :name, :unique, :columns, :lengths) #:nodoc:
+    end
+
+    class Mysql2Column < Column
+      BOOL = "tinyint(1)"
+      def extract_default(default)
+        if sql_type =~ /blob/i || type == :text
+          if default.blank?
+            return null ? nil : ''
+          else
+            raise ArgumentError, "#{type} columns cannot have a default value: #{default.inspect}"
+          end
+        elsif missing_default_forged_as_empty_string?(default)
+          nil
+        else
+          super
+        end
+      end
+
+      def has_default?
+        return false if sql_type =~ /blob/i || type == :text #mysql forbids defaults on blob and text columns
+        super
+      end
+
+      private
+        def simplified_type(field_type)
+          return :boolean if Mysql2Adapter.emulate_booleans && field_type.downcase.index(BOOL)
+
+          case field_type
+          when /enum/i, /set/i then :string
+          when /year/i         then :integer
+          when /bit/i          then :binary
+          else
+            super
+          end
+        end
+
+        def extract_limit(sql_type)
+          case sql_type
+          when /blob|text/i
+            case sql_type
+            when /tiny/i
+              255
+            when /medium/i
+              16777215
+            when /long/i
+              2147483647 # mysql only allows 2^31-1, not 2^32-1, somewhat inconsistently with the tiny/medium/normal cases
+            else
+              super # we could return 65535 here, but we leave it undecorated by default
+            end
+          when /^bigint/i;    8
+          when /^int/i;       4
+          when /^mediumint/i; 3
+          when /^smallint/i;  2
+          when /^tinyint/i;   1
+          else
+            super
+          end
+        end
+
+        # MySQL misreports NOT NULL column default when none is given.
+        # We can't detect this for columns which may have a legitimate ''
+        # default (string) but we can for others (integer, datetime, boolean,
+        # and the rest).
+        #
+        # Test whether the column has default '', is not null, and is not
+        # a type allowing default ''.
+        def missing_default_forged_as_empty_string?(default)
+          type != :string && !null && default == ''
+        end
+    end
+
+    class Mysql2Adapter < AbstractAdapter
+      cattr_accessor :emulate_booleans
+      self.emulate_booleans = true
+
+      ADAPTER_NAME = 'Mysql2'
+      PRIMARY = "PRIMARY"
+
+      LOST_CONNECTION_ERROR_MESSAGES = [
+        "Server shutdown in progress",
+        "Broken pipe",
+        "Lost connection to MySQL server during query",
+        "MySQL server has gone away" ]
+
+      QUOTED_TRUE, QUOTED_FALSE = '1', '0'
+
+      NATIVE_DATABASE_TYPES = {
+        :primary_key => "int(11) DEFAULT NULL auto_increment PRIMARY KEY",
+        :string      => { :name => "varchar", :limit => 255 },
+        :text        => { :name => "text" },
+        :integer     => { :name => "int", :limit => 4 },
+        :float       => { :name => "float" },
+        :decimal     => { :name => "decimal" },
+        :datetime    => { :name => "datetime" },
+        :timestamp   => { :name => "datetime" },
+        :time        => { :name => "time" },
+        :date        => { :name => "date" },
+        :binary      => { :name => "blob" },
+        :boolean     => { :name => "tinyint", :limit => 1 }
+      }
+
+      def initialize(connection, logger, connection_options, config)
+        super(connection, logger)
+        @connection_options, @config = connection_options, config
+        @quoted_column_names, @quoted_table_names = {}, {}
+        configure_connection
+      end
+
+      def adapter_name
+        ADAPTER_NAME
+      end
+
+      def supports_migrations?
+        true
+      end
+
+      def supports_primary_key?
+        true
+      end
+
+      def supports_savepoints?
+        true
+      end
+
+      def native_database_types
+        NATIVE_DATABASE_TYPES
+      end
+
+      # QUOTING ==================================================
+
+      def quote(value, column = nil)
+        if value.kind_of?(String) && column && column.type == :binary && column.class.respond_to?(:string_to_binary)
+          s = column.class.string_to_binary(value).unpack("H*")[0]
+          "x'#{s}'"
+        elsif value.kind_of?(BigDecimal)
+          value.to_s("F")
+        else
+          super
+        end
+      end
+
+      def quote_column_name(name) #:nodoc:
+        @quoted_column_names[name] ||= "`#{name}`"
+      end
+
+      def quote_table_name(name) #:nodoc:
+        @quoted_table_names[name] ||= quote_column_name(name).gsub('.', '`.`')
+      end
+
+      def quote_string(string)
+        @connection.escape(string)
+      end
+
+      def quoted_true
+        QUOTED_TRUE
+      end
+
+      def quoted_false
+        QUOTED_FALSE
+      end
+
+      # REFERENTIAL INTEGRITY ====================================
+
+      def disable_referential_integrity(&block) #:nodoc:
+        old = select_value("SELECT @@FOREIGN_KEY_CHECKS")
+
+        begin
+          update("SET FOREIGN_KEY_CHECKS = 0")
+          yield
+        ensure
+          update("SET FOREIGN_KEY_CHECKS = #{old}")
+        end
+      end
+
+      # CONNECTION MANAGEMENT ====================================
+
+      def active?
+        return false unless @connection
+        @connection.ping
+      end
+
+      def reconnect!
+        disconnect!
+        connect
+      end
+
+      # this is set to true in 2.3, but we don't want it to be
+      def requires_reloading?
+        false
+      end
+
+      def disconnect!
+        unless @connection.nil?
+          @connection.close
+          @connection = nil
+        end
+      end
+
+      def reset!
+        disconnect!
+        connect
+      end
+
+      # DATABASE STATEMENTS ======================================
+
+      # FIXME: re-enable the following once a "better" query_cache solution is in core
+      #
+      # The overrides below perform much better than the originals in AbstractAdapter
+      # because we're able to take advantage of mysql2's lazy-loading capabilities
+      #
+      # # Returns a record hash with the column names as keys and column values
+      # # as values.
+      # def select_one(sql, name = nil)
+      #   result = execute(sql, name)
+      #   result.each(:as => :hash) do |r|
+      #     return r
+      #   end
+      # end
+      #
+      # # Returns a single value from a record
+      # def select_value(sql, name = nil)
+      #   result = execute(sql, name)
+      #   if first = result.first
+      #     first.first
+      #   end
+      # end
+      #
+      # # Returns an array of the values of the first column in a select:
+      # #   select_values("SELECT id FROM companies LIMIT 3") => [1,2,3]
+      # def select_values(sql, name = nil)
+      #   execute(sql, name).map { |row| row.first }
+      # end
+
+      # Returns an array of arrays containing the field values.
+      # Order is the same as that returned by +columns+.
+      def select_rows(sql, name = nil)
+        execute(sql, name).to_a
+      end
+
+      # Executes the SQL statement in the context of this connection.
+      def execute(sql, name = nil)
+        # make sure we carry over any changes to ActiveRecord::Base.default_timezone that have been
+        # made since we established the connection
+        @connection.query_options[:database_timezone] = ActiveRecord::Base.default_timezone
+        if name == :skip_logging
+          @connection.query(sql)
+        else
+          log(sql, name) { @connection.query(sql) }
+        end
+      rescue ActiveRecord::StatementInvalid => exception
+        if exception.message.split(":").first =~ /Packets out of order/
+          raise ActiveRecord::StatementInvalid, "'Packets out of order' error was received from the database. Please update your mysql bindings (gem install mysql) and read http://dev.mysql.com/doc/mysql/en/password-hashing.html for more information.  If you're on Windows, use the Instant Rails installer to get the updated mysql bindings."
+        else
+          raise
+        end
+      end
+
+      def insert_sql(sql, name = nil, pk = nil, id_value = nil, sequence_name = nil)
+        super
+        id_value || @connection.last_id
+      end
+      alias :create :insert_sql
+
+      def update_sql(sql, name = nil)
+        super
+        @connection.affected_rows
+      end
+
+      def begin_db_transaction
+        execute "BEGIN"
+      rescue Exception
+        # Transactions aren't supported
+      end
+
+      def commit_db_transaction
+        execute "COMMIT"
+      rescue Exception
+        # Transactions aren't supported
+      end
+
+      def rollback_db_transaction
+        execute "ROLLBACK"
+      rescue Exception
+        # Transactions aren't supported
+      end
+
+      def create_savepoint
+        execute("SAVEPOINT #{current_savepoint_name}")
+      end
+
+      def rollback_to_savepoint
+        execute("ROLLBACK TO SAVEPOINT #{current_savepoint_name}")
+      end
+
+      def release_savepoint
+        execute("RELEASE SAVEPOINT #{current_savepoint_name}")
+      end
+
+      def add_limit_offset!(sql, options)
+        limit, offset = options[:limit], options[:offset]
+        if limit && offset
+          sql << " LIMIT #{offset.to_i}, #{sanitize_limit(limit)}"
+        elsif limit
+          sql << " LIMIT #{sanitize_limit(limit)}"
+        elsif offset
+          sql << " OFFSET #{offset.to_i}"
+        end
+        sql
+      end
+      deprecate :add_limit_offset!
+
+      # SCHEMA STATEMENTS ========================================
+
+      def structure_dump
+        if supports_views?
+          sql = "SHOW FULL TABLES WHERE Table_type = 'BASE TABLE'"
+        else
+          sql = "SHOW TABLES"
+        end
+
+        select_all(sql).inject("") do |structure, table|
+          table.delete('Table_type')
+          structure += select_one("SHOW CREATE TABLE #{quote_table_name(table.to_a.first.last)}")["Create Table"] + ";\n\n"
+        end
+      end
+
+      def recreate_database(name, options = {})
+        drop_database(name)
+        create_database(name, options)
+      end
+
+      # Create a new MySQL database with optional <tt>:charset</tt> and <tt>:collation</tt>.
+      # Charset defaults to utf8.
+      #
+      # Example:
+      #   create_database 'charset_test', :charset => 'latin1', :collation => 'latin1_bin'
+      #   create_database 'matt_development'
+      #   create_database 'matt_development', :charset => :big5
+      def create_database(name, options = {})
+        if options[:collation]
+          execute "CREATE DATABASE `#{name}` DEFAULT CHARACTER SET `#{options[:charset] || 'utf8'}` COLLATE `#{options[:collation]}`"
+        else
+          execute "CREATE DATABASE `#{name}` DEFAULT CHARACTER SET `#{options[:charset] || 'utf8'}`"
+        end
+      end
+
+      def drop_database(name) #:nodoc:
+        execute "DROP DATABASE IF EXISTS `#{name}`"
+      end
+
+      def current_database
+        select_value 'SELECT DATABASE() as db'
+      end
+
+      # Returns the database character set.
+      def charset
+        show_variable 'character_set_database'
+      end
+
+      # Returns the database collation strategy.
+      def collation
+        show_variable 'collation_database'
+      end
+
+      def tables(name = nil)
+        tables = []
+        execute("SHOW TABLES", name).each do |field|
+          tables << field.first
+        end
+        tables
+      end
+
+      def drop_table(table_name, options = {})
+        super(table_name, options)
+      end
+
+      def indexes(table_name, name = nil)
+        indexes = []
+        current_index = nil
+        result = execute("SHOW KEYS FROM #{quote_table_name(table_name)}", name)
+        result.each(:symbolize_keys => true, :as => :hash) do |row|
+          if current_index != row[:Key_name]
+            next if row[:Key_name] == PRIMARY # skip the primary key
+            current_index = row[:Key_name]
+            indexes << Mysql2IndexDefinition.new(row[:Table], row[:Key_name], row[:Non_unique] == 0, [], [])
+          end
+
+          indexes.last.columns << row[:Column_name]
+          indexes.last.lengths << row[:Sub_part]
+        end
+        indexes
+      end
+
+      def columns(table_name, name = nil)
+        sql = "SHOW FIELDS FROM #{quote_table_name(table_name)}"
+        columns = []
+        result = execute(sql)
+        result.each(:symbolize_keys => true, :as => :hash) { |field|
+          columns << Mysql2Column.new(field[:Field], field[:Default], field[:Type], field[:Null] == "YES")
+        }
+        columns
+      end
+
+      def create_table(table_name, options = {})
+        super(table_name, options.reverse_merge(:options => "ENGINE=InnoDB"))
+      end
+
+      def rename_table(table_name, new_name)
+        execute "RENAME TABLE #{quote_table_name(table_name)} TO #{quote_table_name(new_name)}"
+      end
+
+      def add_column(table_name, column_name, type, options = {})
+        add_column_sql = "ALTER TABLE #{quote_table_name(table_name)} ADD #{quote_column_name(column_name)} #{type_to_sql(type, options[:limit], options[:precision], options[:scale])}"
+        add_column_options!(add_column_sql, options)
+        add_column_position!(add_column_sql, options)
+        execute(add_column_sql)
+      end
+
+      def change_column_default(table_name, column_name, default)
+        column = column_for(table_name, column_name)
+        change_column table_name, column_name, column.sql_type, :default => default
+      end
+
+      def change_column_null(table_name, column_name, null, default = nil)
+        column = column_for(table_name, column_name)
+
+        unless null || default.nil?
+          execute("UPDATE #{quote_table_name(table_name)} SET #{quote_column_name(column_name)}=#{quote(default)} WHERE #{quote_column_name(column_name)} IS NULL")
+        end
+
+        change_column table_name, column_name, column.sql_type, :null => null
+      end
+
+      def change_column(table_name, column_name, type, options = {})
+        column = column_for(table_name, column_name)
+
+        unless options_include_default?(options)
+          options[:default] = column.default
+        end
+
+        unless options.has_key?(:null)
+          options[:null] = column.null
+        end
+
+        change_column_sql = "ALTER TABLE #{quote_table_name(table_name)} CHANGE #{quote_column_name(column_name)} #{quote_column_name(column_name)} #{type_to_sql(type, options[:limit], options[:precision], options[:scale])}"
+        add_column_options!(change_column_sql, options)
+        add_column_position!(change_column_sql, options)
+        execute(change_column_sql)
+      end
+
+      def rename_column(table_name, column_name, new_column_name)
+        options = {}
+        if column = columns(table_name).find { |c| c.name == column_name.to_s }
+          options[:default] = column.default
+          options[:null] = column.null
+        else
+          raise ActiveRecordError, "No such column: #{table_name}.#{column_name}"
+        end
+        current_type = select_one("SHOW COLUMNS FROM #{quote_table_name(table_name)} LIKE '#{column_name}'")["Type"]
+        rename_column_sql = "ALTER TABLE #{quote_table_name(table_name)} CHANGE #{quote_column_name(column_name)} #{quote_column_name(new_column_name)} #{current_type}"
+        add_column_options!(rename_column_sql, options)
+        execute(rename_column_sql)
+      end
+
+      # Maps logical Rails types to MySQL-specific data types.
+      def type_to_sql(type, limit = nil, precision = nil, scale = nil)
+        return super unless type.to_s == 'integer'
+
+        case limit
+        when 1; 'tinyint'
+        when 2; 'smallint'
+        when 3; 'mediumint'
+        when nil, 4, 11; 'int(11)'  # compatibility with MySQL default
+        when 5..8; 'bigint'
+        else raise(ActiveRecordError, "No integer type has byte size #{limit}")
+        end
+      end
+
+      def add_column_position!(sql, options)
+        if options[:first]
+          sql << " FIRST"
+        elsif options[:after]
+          sql << " AFTER #{quote_column_name(options[:after])}"
+        end
+      end
+
+      def show_variable(name)
+        variables = select_all("SHOW VARIABLES LIKE '#{name}'")
+        variables.first['Value'] unless variables.empty?
+      end
+
+      def pk_and_sequence_for(table)
+        keys = []
+        result = execute("describe #{quote_table_name(table)}")
+        result.each(:symbolize_keys => true, :as => :hash) do |row|
+          keys << row[:Field] if row[:Key] == "PRI"
+        end
+        keys.length == 1 ? [keys.first, nil] : nil
+      end
+
+      # Returns just a table's primary key
+      def primary_key(table)
+        pk_and_sequence = pk_and_sequence_for(table)
+        pk_and_sequence && pk_and_sequence.first
+      end
+
+      def case_sensitive_equality_operator
+        "= BINARY"
+      end
+
+      def limited_update_conditions(where_sql, quoted_table_name, quoted_primary_key)
+        where_sql
+      end
+
+      protected
+        def quoted_columns_for_index(column_names, options = {})
+          length = options[:length] if options.is_a?(Hash)
+
+          quoted_column_names = case length
+          when Hash
+            column_names.map {|name| length[name] ? "#{quote_column_name(name)}(#{length[name]})" : quote_column_name(name) }
+          when Fixnum
+            column_names.map {|name| "#{quote_column_name(name)}(#{length})"}
+          else
+            column_names.map {|name| quote_column_name(name) }
+          end
+        end
+
+        def translate_exception(exception, message)
+          return super unless exception.respond_to?(:error_number)
+
+          case exception.error_number
+          when 1062
+            RecordNotUnique.new(message, exception)
+          when 1452
+            InvalidForeignKey.new(message, exception)
+          else
+            super
+          end
+        end
+
+      private
+        def connect
+          @connection = Mysql2::Client.new(@config)
+          configure_connection
+        end
+
+        def configure_connection
+          @connection.query_options.merge!(:as => :array)
+
+          # By default, MySQL 'where id is null' selects the last inserted id.
+          # Turn this off. http://dev.rubyonrails.org/ticket/6778
+          variable_assignments = ['SQL_AUTO_IS_NULL=0']
+          encoding = @config[:encoding]
+
+          # make sure we set the encoding
+          variable_assignments << "NAMES '#{encoding}'" if encoding
+
+          # increase timeout so mysql server doesn't disconnect us
+          wait_timeout = @config[:wait_timeout]
+          wait_timeout = 2592000 unless wait_timeout.is_a?(Fixnum)
+          variable_assignments << "@@wait_timeout = #{wait_timeout}"
+
+          execute("SET #{variable_assignments.join(', ')}", :skip_logging)
+        end
+
+        # Returns an array of record hashes with the column names as keys and
+        # column values as values.
+        def select(sql, name = nil)
+          execute(sql, name).each(:as => :hash)
+        end
+
+        def supports_views?
+          version[0] >= 5
+        end
+
+        def version
+          @version ||= @connection.info[:version].scan(/^(\d+)\.(\d+)\.(\d+)/).flatten.map { |v| v.to_i }
+        end
+
+        def column_for(table_name, column_name)
+          unless column = columns(table_name).find { |c| c.name == column_name.to_s }
+            raise "No such column: #{table_name}.#{column_name}"
+          end
+          column
+        end
+    end
+  end
+end
diff --git a/activerecord/lib/active_record/connection_adapters/mysql_adapter.rb b/activerecord/lib/active_record/connection_adapters/mysql_adapter.rb
index ce2352486b..368c5b2023 100644
--- a/activerecord/lib/active_record/connection_adapters/mysql_adapter.rb
+++ b/activerecord/lib/active_record/connection_adapters/mysql_adapter.rb
@@ -203,6 +203,10 @@ module ActiveRecord
         ADAPTER_NAME
       end
 
+      def supports_bulk_alter? #:nodoc:
+        true
+      end
+
       # Returns +true+ when the connection adapter supports prepared statement
       # caching, otherwise returns +false+
       def supports_statement_cache?
@@ -327,7 +331,7 @@ module ActiveRecord
       end
 
       def exec_query(sql, name = 'SQL', binds = [])
-        log(sql, name) do
+        log(sql, name, binds) do
           result = nil
 
           cache = {}
@@ -364,9 +368,14 @@ module ActiveRecord
         # statement API.  For those queries, we need to use this method. :'(
         log(sql, name) do
           result = @connection.query(sql)
-          cols = result.fetch_fields.map { |field| field.name }
-          rows = result.to_a
-          result.free
+          cols = []
+          rows = []
+
+          if result
+            cols = result.fetch_fields.map { |field| field.name }
+            rows = result.to_a
+            result.free
+          end
           ActiveRecord::Result.new(cols, rows)
         end
       end
@@ -400,7 +409,7 @@ module ActiveRecord
 
       def begin_db_transaction #:nodoc:
         exec_without_stmt "BEGIN"
-      rescue Exception
+      rescue Mysql::Error
         # Transactions aren't supported
       end
 
@@ -439,6 +448,7 @@ module ActiveRecord
         end
         sql
       end
+      deprecate :add_limit_offset!
 
       # SCHEMA STATEMENTS ========================================
 
@@ -527,7 +537,7 @@ module ActiveRecord
       def columns(table_name, name = nil)#:nodoc:
         sql = "SHOW FIELDS FROM #{quote_table_name(table_name)}"
         columns = []
-        result = execute(sql, :skip_logging)
+        result = execute(sql)
         result.each { |field| columns << MysqlColumn.new(field[0], field[4], field[1], field[2] == "YES") }
         result.free
         columns
@@ -541,11 +551,23 @@ module ActiveRecord
         execute "RENAME TABLE #{quote_table_name(table_name)} TO #{quote_table_name(new_name)}"
       end
 
+      def bulk_change_table(table_name, operations) #:nodoc:
+        sqls = operations.map do |command, args|
+          table, arguments = args.shift, args
+          method = :"#{command}_sql"
+
+          if respond_to?(method)
+            send(method, table, *arguments)
+          else
+            raise "Unknown method called : #{method}(#{arguments.inspect})"
+          end
+        end.flatten.join(", ")
+
+        execute("ALTER TABLE #{quote_table_name(table_name)} #{sqls}")
+      end
+
       def add_column(table_name, column_name, type, options = {})
-        add_column_sql = "ALTER TABLE #{quote_table_name(table_name)} ADD #{quote_column_name(column_name)} #{type_to_sql(type, options[:limit], options[:precision], options[:scale])}"
-        add_column_options!(add_column_sql, options)
-        add_column_position!(add_column_sql, options)
-        execute(add_column_sql)
+        execute("ALTER TABLE #{quote_table_name(table_name)} #{add_column_sql(table_name, column_name, type, options)}")
       end
 
       def change_column_default(table_name, column_name, default) #:nodoc:
@@ -564,34 +586,11 @@ module ActiveRecord
       end
 
       def change_column(table_name, column_name, type, options = {}) #:nodoc:
-        column = column_for(table_name, column_name)
-
-        unless options_include_default?(options)
-          options[:default] = column.default
-        end
-
-        unless options.has_key?(:null)
-          options[:null] = column.null
-        end
-
-        change_column_sql = "ALTER TABLE #{quote_table_name(table_name)} CHANGE #{quote_column_name(column_name)} #{quote_column_name(column_name)} #{type_to_sql(type, options[:limit], options[:precision], options[:scale])}"
-        add_column_options!(change_column_sql, options)
-        add_column_position!(change_column_sql, options)
-        execute(change_column_sql)
+        execute("ALTER TABLE #{quote_table_name(table_name)} #{change_column_sql(table_name, column_name, type, options)}")
       end
 
       def rename_column(table_name, column_name, new_column_name) #:nodoc:
-        options = {}
-        if column = columns(table_name).find { |c| c.name == column_name.to_s }
-          options[:default] = column.default
-          options[:null] = column.null
-        else
-          raise ActiveRecordError, "No such column: #{table_name}.#{column_name}"
-        end
-        current_type = select_one("SHOW COLUMNS FROM #{quote_table_name(table_name)} LIKE '#{column_name}'")["Type"]
-        rename_column_sql = "ALTER TABLE #{quote_table_name(table_name)} CHANGE #{quote_column_name(column_name)} #{quote_column_name(new_column_name)} #{current_type}"
-        add_column_options!(rename_column_sql, options)
-        execute(rename_column_sql)
+        execute("ALTER TABLE #{quote_table_name(table_name)} #{rename_column_sql(table_name, column_name, new_column_name)}")
       end
 
       # Maps logical Rails types to MySQL-specific data types.
@@ -674,6 +673,69 @@ module ActiveRecord
           end
         end
 
+        def add_column_sql(table_name, column_name, type, options = {})
+          add_column_sql = "ADD #{quote_column_name(column_name)} #{type_to_sql(type, options[:limit], options[:precision], options[:scale])}"
+          add_column_options!(add_column_sql, options)
+          add_column_position!(add_column_sql, options)
+          add_column_sql
+        end
+
+        def remove_column_sql(table_name, *column_names)
+          columns_for_remove(table_name, *column_names).map {|column_name| "DROP #{column_name}" }
+        end
+        alias :remove_columns_sql :remove_column
+
+        def change_column_sql(table_name, column_name, type, options = {})
+          column = column_for(table_name, column_name)
+
+          unless options_include_default?(options)
+            options[:default] = column.default
+          end
+
+          unless options.has_key?(:null)
+            options[:null] = column.null
+          end
+
+          change_column_sql = "CHANGE #{quote_column_name(column_name)} #{quote_column_name(column_name)} #{type_to_sql(type, options[:limit], options[:precision], options[:scale])}"
+          add_column_options!(change_column_sql, options)
+          add_column_position!(change_column_sql, options)
+          change_column_sql
+        end
+
+        def rename_column_sql(table_name, column_name, new_column_name)
+          options = {}
+
+          if column = columns(table_name).find { |c| c.name == column_name.to_s }
+            options[:default] = column.default
+            options[:null] = column.null
+          else
+            raise ActiveRecordError, "No such column: #{table_name}.#{column_name}"
+          end
+
+          current_type = select_one("SHOW COLUMNS FROM #{quote_table_name(table_name)} LIKE '#{column_name}'")["Type"]
+          rename_column_sql = "CHANGE #{quote_column_name(column_name)} #{quote_column_name(new_column_name)} #{current_type}"
+          add_column_options!(rename_column_sql, options)
+          rename_column_sql
+        end
+
+        def add_index_sql(table_name, column_name, options = {})
+          index_name, index_type, index_columns = add_index_options(table_name, column_name, options)
+          "ADD #{index_type} INDEX #{index_name} (#{index_columns})"
+        end
+
+        def remove_index_sql(table_name, options = {})
+          index_name = index_name_for_remove(table_name, options)
+          "DROP INDEX #{index_name}"
+        end
+
+        def add_timestamps_sql(table_name)
+          [add_column_sql(table_name, :created_at, :datetime), add_column_sql(table_name, :updated_at, :datetime)]
+        end
+
+        def remove_timestamps_sql(table_name)
+          [remove_column_sql(table_name, :updated_at), remove_column_sql(table_name, :created_at)]
+        end
+
       private
         def connect
           encoding = @config[:encoding]
diff --git a/activerecord/lib/active_record/connection_adapters/postgresql_adapter.rb b/activerecord/lib/active_record/connection_adapters/postgresql_adapter.rb
index a4b1aa7154..20be4a22ea 100644
--- a/activerecord/lib/active_record/connection_adapters/postgresql_adapter.rb
+++ b/activerecord/lib/active_record/connection_adapters/postgresql_adapter.rb
@@ -532,7 +532,7 @@ module ActiveRecord
       def exec_query(sql, name = 'SQL', binds = [])
         return exec_no_cache(sql, name) if binds.empty?
 
-        log(sql, name) do
+        log(sql, name, binds) do
           unless @statements.key? sql
             nextkey = "a#{@statements.length + 1}"
             @connection.prepare nextkey, sql
@@ -786,7 +786,7 @@ module ActiveRecord
       def pk_and_sequence_for(table) #:nodoc:
         # First try looking for a sequence with a dependency on the
         # given table's primary key.
-        result = query(<<-end_sql, 'PK and serial sequence')[0]
+        result = exec_query(<<-end_sql, 'PK and serial sequence').rows.first
           SELECT attr.attname, seq.relname
           FROM pg_class      seq,
                pg_attribute  attr,
@@ -1071,7 +1071,7 @@ module ActiveRecord
         #  - format_type includes the column size constraint, e.g. varchar(50)
         #  - ::regclass is a function that gives the id for a table name
         def column_definitions(table_name) #:nodoc:
-          query <<-end_sql
+          exec_query(<<-end_sql).rows
             SELECT a.attname, format_type(a.atttypid, a.atttypmod), d.adsrc, a.attnotnull
               FROM pg_attribute a LEFT JOIN pg_attrdef d
                 ON a.attrelid = d.adrelid AND a.attnum = d.adnum
diff --git a/activerecord/lib/active_record/connection_adapters/sqlite_adapter.rb b/activerecord/lib/active_record/connection_adapters/sqlite_adapter.rb
index d76fc4103e..9ee6b88ab6 100644
--- a/activerecord/lib/active_record/connection_adapters/sqlite_adapter.rb
+++ b/activerecord/lib/active_record/connection_adapters/sqlite_adapter.rb
@@ -62,6 +62,10 @@ module ActiveRecord
         sqlite_version >= '2.0.0'
       end
 
+      def supports_savepoints?
+        sqlite_version >= '3.6.8'
+      end
+
       # Returns +true+ when the connection adapter supports prepared statement
       # caching, otherwise returns +false+
       def supports_statement_cache?
@@ -144,12 +148,15 @@ module ActiveRecord
       # DATABASE STATEMENTS ======================================
 
       def exec_query(sql, name = nil, binds = [])
-        log(sql, name) do
+        log(sql, name, binds) do
 
           # Don't cache statements without bind values
           if binds.empty?
-            stmt = @connection.prepare(sql)
-            cols = stmt.columns
+            stmt    = @connection.prepare(sql)
+            cols    = stmt.columns
+            records = stmt.to_a
+            stmt.close
+            stmt = records
           else
             cache = @statements[sql] ||= {
               :stmt => @connection.prepare(sql)
@@ -189,6 +196,18 @@ module ActiveRecord
         exec_query(sql, name).rows
       end
 
+      def create_savepoint
+        execute("SAVEPOINT #{current_savepoint_name}")
+      end
+
+      def rollback_to_savepoint
+        execute("ROLLBACK TO SAVEPOINT #{current_savepoint_name}")
+      end
+
+      def release_savepoint
+        execute("RELEASE SAVEPOINT #{current_savepoint_name}")
+      end
+
       def begin_db_transaction #:nodoc:
         @connection.transaction
       end
@@ -217,6 +236,15 @@ module ActiveRecord
 
       def columns(table_name, name = nil) #:nodoc:
         table_structure(table_name).map do |field|
+          case field["dflt_value"]
+          when /^null$/i
+            field["dflt_value"] = nil
+          when /^'(.*)'$/
+            field["dflt_value"] = $1.gsub(/''/, "'")
+          when /^"(.*)"$/
+            field["dflt_value"] = $1.gsub(/""/, '"')
+          end
+
           SQLiteColumn.new(field['name'], field['dflt_value'], field['type'], field['notnull'].to_i == 0)
         end
       end
@@ -314,16 +342,11 @@ module ActiveRecord
 
       protected
         def select(sql, name = nil, binds = []) #:nodoc:
-          result = exec_query(sql, name, binds)
-          columns = result.columns.map { |column|
-            column.sub(/^"?\w+"?\./, '')
-          }
-
-          result.rows.map { |row| Hash[columns.zip(row)] }
+          exec_query(sql, name, binds).to_a
         end
 
         def table_structure(table_name)
-          structure = @connection.table_info(quote_table_name(table_name))
+          structure = exec_query("PRAGMA table_info(#{quote_table_name(table_name)})").to_hash
           raise(ActiveRecord::StatementInvalid, "Could not find table '#{table_name}'") if structure.empty?
           structure
         end
diff --git a/activerecord/lib/active_record/fixtures.rb b/activerecord/lib/active_record/fixtures.rb
index b46310f8e0..f065ef7753 100644
--- a/activerecord/lib/active_record/fixtures.rb
+++ b/activerecord/lib/active_record/fixtures.rb
@@ -1,4 +1,10 @@
 require 'erb'
+
+begin
+  require 'psych'
+rescue LoadError
+end
+
 require 'yaml'
 require 'csv'
 require 'zlib'
@@ -6,15 +12,7 @@ require 'active_support/dependencies'
 require 'active_support/core_ext/array/wrap'
 require 'active_support/core_ext/object/blank'
 require 'active_support/core_ext/logger'
-
-if RUBY_VERSION < '1.9'
-  module YAML #:nodoc:
-    class Omap #:nodoc:
-      def keys;   map { |k, v| k } end
-      def values; map { |k, v| v } end
-    end
-  end
-end
+require 'active_support/ordered_hash'
 
 if defined? ActiveRecord
   class FixtureClassNotFound < ActiveRecord::ActiveRecordError #:nodoc:
@@ -446,20 +444,24 @@ class FixturesFileNotFound < StandardError; end
 #
 # Any fixture labeled "DEFAULTS" is safely ignored.
 
-class Fixtures < (RUBY_VERSION < '1.9' ? YAML::Omap : Hash)
+class Fixtures
   MAX_ID = 2 ** 30 - 1
   DEFAULT_FILTER_RE = /\.ya?ml$/
 
-  @@all_cached_fixtures = {}
+  @@all_cached_fixtures = Hash.new { |h,k| h[k] = {} }
 
-  def self.reset_cache(connection = nil)
-    connection ||= ActiveRecord::Base.connection
-    @@all_cached_fixtures[connection.object_id] = {}
+  def self.find_table_name(table_name) # :nodoc:
+    ActiveRecord::Base.pluralize_table_names ?
+      table_name.to_s.singularize.camelize :
+      table_name.to_s.camelize
+  end
+
+  def self.reset_cache
+    @@all_cached_fixtures.clear
   end
 
   def self.cache_for_connection(connection)
-    @@all_cached_fixtures[connection.object_id] ||= {}
-    @@all_cached_fixtures[connection.object_id]
+    @@all_cached_fixtures[connection]
   end
 
   def self.fixture_is_cached?(connection, table_name)
@@ -468,11 +470,10 @@ class Fixtures < (RUBY_VERSION < '1.9' ? YAML::Omap : Hash)
 
   def self.cached_fixtures(connection, keys_to_fetch = nil)
     if keys_to_fetch
-      fixtures = cache_for_connection(connection).values_at(*keys_to_fetch)
+      cache_for_connection(connection).values_at(*keys_to_fetch)
     else
-      fixtures = cache_for_connection(connection).values
+      cache_for_connection(connection).values
     end
-    fixtures.size > 1 ? fixtures : fixtures.first
   end
 
   def self.cache_fixtures(connection, fixtures_map)
@@ -482,13 +483,11 @@ class Fixtures < (RUBY_VERSION < '1.9' ? YAML::Omap : Hash)
   def self.instantiate_fixtures(object, table_name, fixtures, load_instances = true)
     object.instance_variable_set "@#{table_name.to_s.gsub('.','_')}", fixtures
     if load_instances
-      ActiveRecord::Base.silence do
-        fixtures.each do |name, fixture|
-          begin
-            object.instance_variable_set "@#{name}", fixture.find
-          rescue FixtureClassNotFound
-            nil
-          end
+      fixtures.each do |name, fixture|
+        begin
+          object.instance_variable_set "@#{name}", fixture.find
+        rescue FixtureClassNotFound
+          nil
         end
       end
     end
@@ -505,36 +504,56 @@ class Fixtures < (RUBY_VERSION < '1.9' ? YAML::Omap : Hash)
 
   def self.create_fixtures(fixtures_directory, table_names, class_names = {})
     table_names = [table_names].flatten.map { |n| n.to_s }
-    table_names.each { |n| class_names[n.tr('/', '_').to_sym] = n.classify if n.include?('/') }
-    connection  = block_given? ? yield : ActiveRecord::Base.connection
+    table_names.each { |n|
+      class_names[n.tr('/', '_').to_sym] = n.classify if n.include?('/')
+    }
 
-    table_names_to_fetch = table_names.reject { |table_name| fixture_is_cached?(connection, table_name) }
+    # FIXME: Apparently JK uses this.
+    connection = block_given? ? yield : ActiveRecord::Base.connection
 
-    unless table_names_to_fetch.empty?
-      ActiveRecord::Base.silence do
-        connection.disable_referential_integrity do
-          fixtures_map = {}
+    files_to_read = table_names.reject { |table_name| fixture_is_cached?(connection, table_name) }
 
-          fixtures = table_names_to_fetch.map do |table_name|
-            fixtures_map[table_name] = Fixtures.new(connection, table_name.tr('/', '_'), class_names[table_name.tr('/', '_').to_sym], File.join(fixtures_directory, table_name))
-          end
+    unless files_to_read.empty?
+      connection.disable_referential_integrity do
+        fixtures_map = {}
+
+        fixture_files = files_to_read.map do |path|
+          table_name = path.tr '/', '_'
+
+          fixtures_map[path] = Fixtures.new(
+            connection,
+            table_name,
+            class_names[table_name.to_sym],
+            File.join(fixtures_directory, path))
+        end
 
-          all_loaded_fixtures.update(fixtures_map)
+        all_loaded_fixtures.update(fixtures_map)
 
-          connection.transaction(:requires_new => true) do
-            fixtures.reverse.each { |fixture| fixture.delete_existing_fixtures }
-            fixtures.each { |fixture| fixture.insert_fixtures }
+        connection.transaction(:requires_new => true) do
+          fixture_files.each do |ff|
+            conn = ff.model_class.respond_to?(:connection) ? ff.model_class.connection : connection
+            table_rows = ff.table_rows
 
-            # Cap primary key sequences to max(pk).
-            if connection.respond_to?(:reset_pk_sequence!)
-              table_names.each do |table_name|
-                connection.reset_pk_sequence!(table_name.tr('/', '_'))
+            table_rows.keys.each do |table|
+              conn.delete "DELETE FROM #{conn.quote_table_name(table)}", 'Fixture Delete'
+            end
+
+            table_rows.each do |table_name,rows|
+              rows.each do |row|
+                conn.insert_fixture(row, table_name)
               end
             end
           end
 
-          cache_fixtures(connection, fixtures_map)
+          # Cap primary key sequences to max(pk).
+          if connection.respond_to?(:reset_pk_sequence!)
+            table_names.each do |table_name|
+              connection.reset_pk_sequence!(table_name.tr('/', '_'))
+            end
+          end
         end
+
+        cache_fixtures(connection, fixtures_map)
       end
     end
     cached_fixtures(connection, table_names)
@@ -546,40 +565,60 @@ class Fixtures < (RUBY_VERSION < '1.9' ? YAML::Omap : Hash)
     Zlib.crc32(label.to_s) % MAX_ID
   end
 
-  attr_reader :table_name, :name
+  attr_reader :table_name, :name, :fixtures, :model_class
 
   def initialize(connection, table_name, class_name, fixture_path, file_filter = DEFAULT_FILTER_RE)
-    @connection, @table_name, @fixture_path, @file_filter = connection, table_name, fixture_path, file_filter
-    @name = table_name # preserve fixture base name
-    @class_name = class_name ||
-                  (ActiveRecord::Base.pluralize_table_names ? @table_name.singularize.camelize : @table_name.camelize)
-    @table_name = "#{ActiveRecord::Base.table_name_prefix}#{@table_name}#{ActiveRecord::Base.table_name_suffix}"
-    @table_name = class_name.table_name if class_name.respond_to?(:table_name)
-    @connection = class_name.connection if class_name.respond_to?(:connection)
+    @connection   = connection
+    @table_name   = table_name
+    @fixture_path = fixture_path
+    @file_filter  = file_filter
+    @name         = table_name # preserve fixture base name
+    @class_name   = class_name
+
+    @fixtures     = ActiveSupport::OrderedHash.new
+    @table_name   = "#{ActiveRecord::Base.table_name_prefix}#{@table_name}#{ActiveRecord::Base.table_name_suffix}"
+
+    # Should be an AR::Base type class
+    if class_name.is_a?(Class)
+      @table_name   = class_name.table_name
+      @connection   = class_name.connection
+      @model_class  = class_name
+    else
+      @model_class  = class_name.constantize rescue nil
+    end
+
     read_fixture_files
   end
 
-  def delete_existing_fixtures
-    @connection.delete "DELETE FROM #{@connection.quote_table_name(table_name)}", 'Fixture Delete'
+  def [](x)
+    fixtures[x]
+  end
+
+  def []=(k,v)
+    fixtures[k] = v
+  end
+
+  def each(&block)
+    fixtures.each(&block)
+  end
+
+  def size
+    fixtures.size
   end
 
-  def insert_fixtures
+  # Return a hash of rows to be inserted.  The key is the table, the value is
+  # a list of rows to insert to that table.
+  def table_rows
     now = ActiveRecord::Base.default_timezone == :utc ? Time.now.utc : Time.now
     now = now.to_s(:db)
 
     # allow a standard key to be used for doing defaults in YAML
-    if is_a?(Hash)
-      delete('DEFAULTS')
-    else
-      delete(assoc('DEFAULTS'))
-    end
+    fixtures.delete('DEFAULTS')
 
     # track any join tables we need to insert later
-    habtm_fixtures = Hash.new do |h, habtm|
-      h[habtm] = HabtmFixtures.new(@connection, habtm.options[:join_table], nil, nil)
-    end
+    rows = Hash.new { |h,table| h[table] = [] }
 
-    each do |label, fixture|
+    rows[table_name] = fixtures.map do |label, fixture|
       row = fixture.to_hash
 
       if model_class && model_class < ActiveRecord::Base
@@ -615,14 +654,9 @@ class Fixtures < (RUBY_VERSION < '1.9' ? YAML::Omap : Hash)
             fk_name = (association.options[:foreign_key] || "#{association.name}_id").to_s
 
             if association.name.to_s != fk_name && value = row.delete(association.name.to_s)
-              if association.options[:polymorphic]
-                if value.sub!(/\s*\(([^\)]*)\)\s*$/, "")
-                  target_type = $1
-                  target_type_name = (association.options[:foreign_type] || "#{association.name}_type").to_s
-
-                  # support polymorphic belongs_to as "label (Type)"
-                  row[target_type_name] = target_type
-                end
+              if association.options[:polymorphic] && value.sub!(/\s*\(([^\)]*)\)\s*$/, "")
+                # support polymorphic belongs_to as "label (Type)"
+                row[association.foreign_type] = $1
               end
 
               row[fk_name] = Fixtures.identify(value)
@@ -630,47 +664,22 @@ class Fixtures < (RUBY_VERSION < '1.9' ? YAML::Omap : Hash)
           when :has_and_belongs_to_many
             if (targets = row.delete(association.name.to_s))
               targets = targets.is_a?(Array) ? targets : targets.split(/\s*,\s*/)
-              join_fixtures = habtm_fixtures[association]
-
-              targets.each do |target|
-                join_fixtures["#{label}_#{target}"] = Fixture.new(
-                  { association.primary_key_name => row[primary_key_name],
-                    association.association_foreign_key => Fixtures.identify(target) },
-                  nil, @connection)
-              end
+              table_name = association.options[:join_table]
+              rows[table_name].concat targets.map { |target|
+                { association.foreign_key             => row[primary_key_name],
+                  association.association_foreign_key => Fixtures.identify(target) }
+              }
             end
           end
         end
       end
 
-      @connection.insert_fixture(fixture, @table_name)
-    end
-
-    # insert any HABTM join tables we discovered
-    habtm_fixtures.values.each do |fixture|
-      fixture.delete_existing_fixtures
-      fixture.insert_fixtures
+      row
     end
+    rows
   end
 
   private
-    class HabtmFixtures < ::Fixtures #:nodoc:
-      def read_fixture_files; end
-    end
-
-    def model_class
-      unless defined?(@model_class)
-        @model_class =
-          if @class_name.nil? || @class_name.is_a?(Class)
-            @class_name
-          else
-            @class_name.constantize rescue nil
-          end
-      end
-
-      @model_class
-    end
-
     def primary_key_name
       @primary_key_name ||= model_class && model_class.primary_key
     end
@@ -724,7 +733,7 @@ class Fixtures < (RUBY_VERSION < '1.9' ? YAML::Omap : Hash)
               raise Fixture::FormatError, "Bad data for #{@class_name} fixture named #{name} (nil)"
             end
 
-            self[name] = Fixture.new(data, model_class, @connection)
+            fixtures[name] = Fixture.new(data, model_class)
           end
         end
       end
@@ -737,7 +746,7 @@ class Fixtures < (RUBY_VERSION < '1.9' ? YAML::Omap : Hash)
       reader.each do |row|
         data = {}
         row.each_with_index { |cell, j| data[header[j].to_s.strip] = cell.to_s.strip }
-        self["#{@class_name.to_s.underscore}_#{i+=1}"] = Fixture.new(data, model_class, @connection)
+        fixtures["#{@class_name.to_s.underscore}_#{i+=1}"] = Fixture.new(data, model_class)
       end
     end
 
@@ -773,44 +782,30 @@ class Fixture #:nodoc:
   class FormatError < FixtureError #:nodoc:
   end
 
-  attr_reader :model_class
+  attr_reader :model_class, :fixture
 
-  def initialize(fixture, model_class, connection = ActiveRecord::Base.connection)
-    @connection = connection
-    @fixture = fixture
-    @model_class = model_class.is_a?(Class) ? model_class : model_class.constantize rescue nil
+  def initialize(fixture, model_class)
+    @fixture     = fixture
+    @model_class = model_class
   end
 
   def class_name
-    @model_class.name if @model_class
+    model_class.name if model_class
   end
 
   def each
-    @fixture.each { |item| yield item }
+    fixture.each { |item| yield item }
   end
 
   def [](key)
-    @fixture[key]
+    fixture[key]
   end
 
-  def to_hash
-    @fixture
-  end
-
-  def key_list
-    @fixture.keys.map { |column_name| @connection.quote_column_name(column_name) }.join(', ')
-  end
-
-  def value_list
-    cols = (model_class && model_class < ActiveRecord::Base) ? model_class.columns_hash : {}
-    @fixture.map do |key, value|
-      @connection.quote(value, cols[key]).gsub('[^\]\\n', "\n").gsub('[^\]\\r', "\r")
-    end.join(', ')
-  end
+  alias :to_hash :fixture
 
   def find
     if model_class
-      model_class.find(self[model_class.primary_key])
+      model_class.find(fixture[model_class.primary_key])
     else
       raise FixtureClassNotFound, "No class attached to find."
     end
@@ -837,7 +832,9 @@ module ActiveRecord
       self.use_instantiated_fixtures = false
       self.pre_loaded_fixtures = false
 
-      self.fixture_class_names = {}
+      self.fixture_class_names = Hash.new do |h, table_name|
+        h[table_name] = Fixtures.find_table_name(table_name)
+      end
     end
 
     module ClassMethods
@@ -937,7 +934,7 @@ module ActiveRecord
         if @@already_loaded_fixtures[self.class]
           @loaded_fixtures = @@already_loaded_fixtures[self.class]
         else
-          load_fixtures
+          @loaded_fixtures = load_fixtures
           @@already_loaded_fixtures[self.class] = @loaded_fixtures
         end
         ActiveRecord::Base.connection.increment_open_transactions
@@ -947,7 +944,7 @@ module ActiveRecord
       else
         Fixtures.reset_cache
         @@already_loaded_fixtures[self.class] = nil
-        load_fixtures
+        @loaded_fixtures = load_fixtures
       end
 
       # Instantiate fixtures for every test if requested.
@@ -971,15 +968,8 @@ module ActiveRecord
 
     private
       def load_fixtures
-        @loaded_fixtures = {}
         fixtures = Fixtures.create_fixtures(fixture_path, fixture_table_names, fixture_class_names)
-        unless fixtures.nil?
-          if fixtures.instance_of?(Fixtures)
-            @loaded_fixtures[fixtures.name] = fixtures
-          else
-            fixtures.each { |f| @loaded_fixtures[f.name] = f }
-          end
-        end
+        Hash[fixtures.map { |f| [f.name, f] }]
       end
 
       # for pre_loaded_fixtures, only require the classes once. huge speed improvement
diff --git a/activerecord/lib/active_record/identity_map.rb b/activerecord/lib/active_record/identity_map.rb
index 438287d3ea..6718a92e13 100644
--- a/activerecord/lib/active_record/identity_map.rb
+++ b/activerecord/lib/active_record/identity_map.rb
@@ -81,7 +81,9 @@ module ActiveRecord
         @changed_attributes.update(coder['attributes'].slice(*dirty))
         @changed_attributes.delete_if{|k,v| v.eql? @attributes[k]}
 
-        _run_find_callbacks
+        set_serialized_attributes
+
+        run_callbacks :find
 
         self
       end
diff --git a/activerecord/lib/active_record/locking/optimistic.rb b/activerecord/lib/active_record/locking/optimistic.rb
index e5065de7fb..6b2b1ebafe 100644
--- a/activerecord/lib/active_record/locking/optimistic.rb
+++ b/activerecord/lib/active_record/locking/optimistic.rb
@@ -58,6 +58,12 @@ module ActiveRecord
       end
 
       private
+        def increment_lock
+          lock_col = self.class.locking_column
+          previous_lock_value = send(lock_col).to_i
+          send(lock_col + '=', previous_lock_value + 1)
+        end
+
         def attributes_from_column_definition
           result = super
 
@@ -78,8 +84,8 @@ module ActiveRecord
           return 0 if attribute_names.empty?
 
           lock_col = self.class.locking_column
-          previous_value = send(lock_col).to_i
-          send(lock_col + '=', previous_value + 1)
+          previous_lock_value = send(lock_col).to_i
+          increment_lock
 
           attribute_names += [lock_col]
           attribute_names.uniq!
@@ -89,7 +95,7 @@ module ActiveRecord
 
             stmt = relation.where(
               relation.table[self.class.primary_key].eq(quoted_id).and(
-                relation.table[lock_col].eq(quote_value(previous_value))
+                relation.table[lock_col].eq(quote_value(previous_lock_value))
               )
             ).arel.compile_update(arel_attributes_values(false, false, attribute_names))
 
@@ -103,7 +109,7 @@ module ActiveRecord
 
           # If something went wrong, revert the version.
           rescue Exception
-            send(lock_col + '=', previous_value)
+            send(lock_col + '=', previous_lock_value)
             raise
           end
         end
diff --git a/activerecord/lib/active_record/log_subscriber.rb b/activerecord/lib/active_record/log_subscriber.rb
index c7ae12977a..afadbf03ef 100644
--- a/activerecord/lib/active_record/log_subscriber.rb
+++ b/activerecord/lib/active_record/log_subscriber.rb
@@ -22,8 +22,16 @@ module ActiveRecord
       self.class.runtime += event.duration
       return unless logger.debug?
 
-      name = '%s (%.1fms)' % [event.payload[:name], event.duration]
-      sql  = event.payload[:sql].squeeze(' ')
+      payload = event.payload
+      name    = '%s (%.1fms)' % [payload[:name], event.duration]
+      sql     = payload[:sql].squeeze(' ')
+      binds   = nil
+
+      unless (payload[:binds] || []).empty?
+        binds = "  " + payload[:binds].map { |col,v|
+          [col.name, v]
+        }.inspect
+      end
 
       if odd?
         name = color(name, CYAN, true)
@@ -32,7 +40,7 @@ module ActiveRecord
         name = color(name, MAGENTA, true)
       end
 
-      debug "  #{name}  #{sql}"
+      debug "  #{name}  #{sql}#{binds}"
     end
 
     def odd?
@@ -45,4 +53,4 @@ module ActiveRecord
   end
 end
 
-ActiveRecord::LogSubscriber.attach_to :active_record
\ No newline at end of file
+ActiveRecord::LogSubscriber.attach_to :active_record
diff --git a/activerecord/lib/active_record/migration/command_recorder.rb b/activerecord/lib/active_record/migration/command_recorder.rb
index d7e481905a..c9d57ce812 100644
--- a/activerecord/lib/active_record/migration/command_recorder.rb
+++ b/activerecord/lib/active_record/migration/command_recorder.rb
@@ -40,7 +40,7 @@ module ActiveRecord
         @commands.reverse.map { |name, args|
           method = :"invert_#{name}"
           raise IrreversibleMigration unless respond_to?(method, true)
-          __send__(method, args)
+          send(method, args)
         }
       end
 
@@ -48,12 +48,16 @@ module ActiveRecord
         super || delegate.respond_to?(*args)
       end
 
-      def send(method, *args) # :nodoc:
-        return super unless respond_to?(method)
-        record(method, args)
+      [:create_table, :rename_table, :add_column, :remove_column, :rename_index, :rename_column, :add_index, :remove_index, :add_timestamps, :remove_timestamps, :change_column, :change_column_default].each do |method|
+        class_eval <<-EOV, __FILE__, __LINE__ + 1
+          def #{method}(*args)
+            record(:"#{method}", args)
+          end
+        EOV
       end
 
       private
+
       def invert_create_table(args)
         [:drop_table, args]
       end
@@ -86,6 +90,14 @@ module ActiveRecord
       def invert_add_timestamps(args)
         [:remove_timestamps, args]
       end
+
+      # Forwards any missing method call to the \target.
+      def method_missing(method, *args, &block)
+        @delegate.send(method, *args, &block)
+      rescue NoMethodError => e
+        raise e, e.message.sub(/ for #<.*$/, " via proxy for #{@delegate}")
+      end
+
     end
   end
 end
diff --git a/activerecord/lib/active_record/nested_attributes.rb b/activerecord/lib/active_record/nested_attributes.rb
index 0a073ad6c8..52bdadcbb4 100644
--- a/activerecord/lib/active_record/nested_attributes.rb
+++ b/activerecord/lib/active_record/nested_attributes.rb
@@ -410,8 +410,20 @@ module ActiveRecord
             assign_to_or_mark_for_destruction(existing_record, attributes, options[:allow_destroy])
           end
         elsif existing_record = existing_records.detect { |record| record.id.to_s == attributes['id'].to_s }
+          unless association.loaded? || call_reject_if(association_name, attributes)
+            # Make sure we are operating on the actual object which is in the association's
+            # proxy_target array (either by finding it, or adding it if not found)
+            target_record = association.proxy_target.detect { |record| record == existing_record }
+
+            if target_record
+              existing_record = target_record
+            else
+              association.send(:add_to_target, existing_record)
+            end
+
+          end
+
           if !call_reject_if(association_name, attributes)
-            association.send(:add_record_to_target_with_callbacks, existing_record) if !association.loaded?
             assign_to_or_mark_for_destruction(existing_record, attributes, options[:allow_destroy])
           end
         else
diff --git a/activerecord/lib/active_record/persistence.rb b/activerecord/lib/active_record/persistence.rb
index 78e84c73ee..df7b22080c 100644
--- a/activerecord/lib/active_record/persistence.rb
+++ b/activerecord/lib/active_record/persistence.rb
@@ -233,6 +233,7 @@ module ActiveRecord
     def touch(name = nil)
       attributes = timestamp_attributes_for_update_in_model
       attributes << name if name
+
       unless attributes.empty?
         current_time = current_time_from_proper_timezone
         changes = {}
@@ -241,6 +242,8 @@ module ActiveRecord
           changes[column.to_s] = write_attribute(column.to_s, current_time)
         end
 
+        changes[self.class.locking_column] = increment_lock if locking_enabled?
+
         @changed_attributes.except!(*changes.keys)
         primary_key = self.class.primary_key
         self.class.update_all(changes, { primary_key => self[primary_key] }) == 1
@@ -267,11 +270,11 @@ module ActiveRecord
     # Creates a record with values matching those of the instance attributes
     # and returns its id.
     def create
-      if self.id.nil? && connection.prefetch_primary_key?(self.class.table_name)
+      if id.nil? && connection.prefetch_primary_key?(self.class.table_name)
         self.id = connection.next_sequence_value(self.class.sequence_name)
       end
 
-      attributes_values = arel_attributes_values
+      attributes_values = arel_attributes_values(!id.nil?)
 
       new_id = if attributes_values.empty?
         self.class.unscoped.insert connection.empty_insert_statement_value
@@ -292,7 +295,7 @@ module ActiveRecord
     # that instances loaded from the database would.
     def attributes_from_column_definition
       Hash[self.class.columns.map do |column|
-        [column.name, column.default] unless column.name == self.class.primary_key
+        [column.name, column.default]
       end]
     end
   end
diff --git a/activerecord/lib/active_record/railtie.rb b/activerecord/lib/active_record/railtie.rb
index a634153d6f..cace6f0cc0 100644
--- a/activerecord/lib/active_record/railtie.rb
+++ b/activerecord/lib/active_record/railtie.rb
@@ -77,6 +77,7 @@ module ActiveRecord
       ActiveSupport.on_load(:active_record) do
         ActionDispatch::Reloader.to_cleanup do
           ActiveRecord::Base.clear_reloadable_connections!
+          ActiveRecord::Base.clear_cache!
         end
       end
     end
diff --git a/activerecord/lib/active_record/railties/databases.rake b/activerecord/lib/active_record/railties/databases.rake
index a4fc18148e..ff36814684 100644
--- a/activerecord/lib/active_record/railties/databases.rake
+++ b/activerecord/lib/active_record/railties/databases.rake
@@ -193,7 +193,7 @@ db_namespace = namespace :db do
       file_list = []
       Dir.foreach(File.join(Rails.root, 'db', 'migrate')) do |file|
         # only files matching "20091231235959_some_name.rb" pattern
-        if match_data = /(\d{14})_(.+)\.rb/.match(file)
+        if match_data = /^(\d{14})_(.+)\.rb$/.match(file)
           status = db_list.delete(match_data[1]) ? 'up' : 'down'
           file_list << [status, match_data[1], match_data[2]]
         end
@@ -296,8 +296,8 @@ db_namespace = namespace :db do
       base_dir = ENV['FIXTURES_PATH'] ? File.join(Rails.root, ENV['FIXTURES_PATH']) : File.join(Rails.root, 'test', 'fixtures')
       fixtures_dir = ENV['FIXTURES_DIR'] ? File.join(base_dir, ENV['FIXTURES_DIR']) : base_dir
 
-      (ENV['FIXTURES'] ? ENV['FIXTURES'].split(/,/).map {|f| File.join(fixtures_dir, f) } : Dir["#{fixtures_dir}/**/*.{yml,csv}"]).each do |fixture_file|
-        Fixtures.create_fixtures(fixtures_dir, fixture_file[(fixtures_dir.size + 1)..-5])
+      (ENV['FIXTURES'] ? ENV['FIXTURES'].split(/,/) : Dir["#{fixtures_dir}/**/*.{yml,csv}"].map {|f| f[(fixtures_dir.size + 1)..-5] }).each do |fixture_file|
+        Fixtures.create_fixtures(fixtures_dir, fixture_file)
       end
     end
 
@@ -409,7 +409,7 @@ db_namespace = namespace :db do
         ENV['PGHOST']     = abcs["test"]["host"] if abcs["test"]["host"]
         ENV['PGPORT']     = abcs["test"]["port"].to_s if abcs["test"]["port"]
         ENV['PGPASSWORD'] = abcs["test"]["password"].to_s if abcs["test"]["password"]
-        `psql -U "#{abcs["test"]["username"]}" -f #{Rails.root}/db/#{Rails.env}_structure.sql #{abcs["test"]["database"]}`
+        `psql -U "#{abcs["test"]["username"]}" -f #{Rails.root}/db/#{Rails.env}_structure.sql #{abcs["test"]["database"]} #{abcs["test"]["template"]}`
       when "sqlite", "sqlite3"
         dbfile = abcs["test"]["database"] || abcs["test"]["dbfile"]
         `#{abcs["test"]["adapter"]} #{dbfile} < #{Rails.root}/db/#{Rails.env}_structure.sql`
diff --git a/activerecord/lib/active_record/reflection.rb b/activerecord/lib/active_record/reflection.rb
index b9caa64a0e..ceeb0ec39d 100644
--- a/activerecord/lib/active_record/reflection.rb
+++ b/activerecord/lib/active_record/reflection.rb
@@ -1,4 +1,5 @@
 require 'active_support/core_ext/class/attribute'
+require 'active_support/core_ext/module/deprecation'
 
 module ActiveRecord
   # = Active Record Reflection
@@ -196,8 +197,17 @@ module ActiveRecord
         @quoted_table_name ||= klass.quoted_table_name
       end
 
+      def foreign_key
+        @foreign_key ||= options[:foreign_key] || derive_foreign_key
+      end
+
       def primary_key_name
-        @primary_key_name ||= options[:foreign_key] || derive_primary_key_name
+        foreign_key
+      end
+      deprecate :primary_key_name => :foreign_key
+
+      def foreign_type
+        @foreign_type ||= options[:foreign_type] || "#{name}_type"
       end
 
       def primary_key_column
@@ -208,6 +218,13 @@ module ActiveRecord
         @association_foreign_key ||= options[:association_foreign_key] || class_name.foreign_key
       end
 
+      def association_primary_key
+        @association_primary_key ||=
+          options[:primary_key] ||
+          !options[:polymorphic] && klass.primary_key ||
+          'id'
+      end
+
       def active_record_primary_key
         @active_record_primary_key ||= options[:primary_key] || active_record.primary_key
       end
@@ -244,7 +261,7 @@ module ActiveRecord
         false
       end
 
-      def through_reflection_primary_key_name
+      def through_reflection_foreign_key
       end
 
       def source_reflection
@@ -291,22 +308,36 @@ module ActiveRecord
         !options[:validate].nil? ? options[:validate] : (options[:autosave] == true || macro == :has_many)
       end
 
-      def dependent_conditions(record, base_class, extra_conditions)
-        dependent_conditions = []
-        dependent_conditions << "#{primary_key_name} = #{record.send(name).send(:owner_quoted_id)}"
-        dependent_conditions << "#{options[:as]}_type = '#{base_class.name}'" if options[:as]
-        dependent_conditions << klass.send(:sanitize_sql, options[:conditions]) if options[:conditions]
-        dependent_conditions << extra_conditions if extra_conditions
-        dependent_conditions = dependent_conditions.collect {|where| "(#{where})" }.join(" AND ")
-        dependent_conditions = dependent_conditions.gsub('@', '\@')
-        dependent_conditions
-      end
-
       # Returns +true+ if +self+ is a +belongs_to+ reflection.
       def belongs_to?
         macro == :belongs_to
       end
 
+      def proxy_class
+        case macro
+        when :belongs_to
+          if options[:polymorphic]
+            Associations::BelongsToPolymorphicAssociation
+          else
+            Associations::BelongsToAssociation
+          end
+        when :has_and_belongs_to_many
+          Associations::HasAndBelongsToManyAssociation
+        when :has_many
+          if options[:through]
+            Associations::HasManyThroughAssociation
+          else
+            Associations::HasManyAssociation
+          end
+        when :has_one
+          if options[:through]
+            Associations::HasOneThroughAssociation
+          else
+            Associations::HasOneAssociation
+          end
+        end
+      end
+
       private
         def derive_class_name
           class_name = name.to_s.camelize
@@ -314,7 +345,7 @@ module ActiveRecord
           class_name
         end
 
-        def derive_primary_key_name
+        def derive_foreign_key
           if belongs_to?
             "#{name}_id"
           elsif options[:as]
@@ -368,6 +399,10 @@ module ActiveRecord
           raise HasManyThroughAssociationNotFoundError.new(active_record.name, self)
         end
 
+        if through_reflection.options[:polymorphic]
+          raise HasManyThroughAssociationPolymorphicThroughError.new(active_record.name, self)
+        end
+
         if source_reflection.nil?
           raise HasManyThroughSourceAssociationNotFoundError.new(self)
         end
@@ -377,22 +412,26 @@ module ActiveRecord
         end
 
         if source_reflection.options[:polymorphic] && options[:source_type].nil?
-          raise HasManyThroughAssociationPolymorphicError.new(active_record.name, self, source_reflection)
+          raise HasManyThroughAssociationPolymorphicSourceError.new(active_record.name, self, source_reflection)
         end
 
         unless [:belongs_to, :has_many, :has_one].include?(source_reflection.macro) && source_reflection.options[:through].nil?
           raise HasManyThroughSourceAssociationMacroError.new(self)
         end
 
+        if macro == :has_one && through_reflection.collection?
+          raise HasOneThroughCantAssociateThroughCollection.new(active_record.name, self, through_reflection)
+        end
+
         check_validity_of_inverse!
       end
 
       def through_reflection_primary_key
-        through_reflection.belongs_to? ? through_reflection.klass.primary_key : through_reflection.primary_key_name
+        through_reflection.belongs_to? ? through_reflection.klass.primary_key : through_reflection.foreign_key
       end
 
-      def through_reflection_primary_key_name
-        through_reflection.primary_key_name if through_reflection.belongs_to?
+      def through_reflection_foreign_key
+        through_reflection.foreign_key if through_reflection.belongs_to?
       end
 
       private
diff --git a/activerecord/lib/active_record/relation.rb b/activerecord/lib/active_record/relation.rb
index c6943444a4..cb684c1109 100644
--- a/activerecord/lib/active_record/relation.rb
+++ b/activerecord/lib/active_record/relation.rb
@@ -10,7 +10,9 @@ module ActiveRecord
 
     include FinderMethods, Calculations, SpawnMethods, QueryMethods, Batches
 
+    # These are explicitly delegated to improve performance (avoids method_missing)
     delegate :to_xml, :to_yaml, :length, :collect, :map, :each, :all?, :include?, :to => :to_a
+    delegate :table_name, :primary_key, :to => :klass
 
     attr_reader :table, :klass, :loaded
     attr_accessor :extensions
@@ -30,13 +32,19 @@ module ActiveRecord
     def insert(values)
       im = arel.compile_insert values
       im.into @table
-      primary_key_name = @klass.primary_key
-      primary_key_value = Hash === values ? values[primary_key_name] : nil
+
+      primary_key_value = nil
+
+      if primary_key && Hash === values
+        primary_key_value = values[values.keys.find { |k|
+          k.name == primary_key
+        }]
+      end
 
       @klass.connection.insert(
         im.to_sql,
         'SQL',
-        primary_key_name,
+        primary_key,
         primary_key_value)
     end
 
@@ -172,13 +180,19 @@ module ActiveRecord
       if conditions || options.present?
         where(conditions).apply_finder_options(options.slice(:limit, :order)).update_all(updates)
       else
+        limit = nil
+        order = []
         # Apply limit and order only if they're both present
         if @limit_value.present? == @order_values.present?
-          stmt = arel.compile_update(Arel::SqlLiteral.new(@klass.send(:sanitize_sql_for_assignment, updates)))
-          @klass.connection.update stmt.to_sql
-        else
-          except(:limit, :order).update_all(updates)
+          limit = arel.limit
+          order = arel.orders
         end
+
+        stmt = arel.compile_update(Arel.sql(@klass.send(:sanitize_sql_for_assignment, updates)))
+        stmt.take limit
+        stmt.order(*order)
+        stmt.key = table[primary_key]
+        @klass.connection.update stmt.to_sql
       end
     end
 
@@ -320,7 +334,7 @@ module ActiveRecord
     #   # Delete multiple rows
     #   Todo.delete([2,3,4])
     def delete(id_or_array)
-      where(@klass.primary_key => id_or_array).delete_all
+      where(primary_key => id_or_array).delete_all
     end
 
     def reload
@@ -336,10 +350,6 @@ module ActiveRecord
       self
     end
 
-    def primary_key
-      @primary_key ||= table[@klass.primary_key]
-    end
-
     def to_sql
       @to_sql ||= arel.to_sql
     end
@@ -373,10 +383,6 @@ module ActiveRecord
       to_a.inspect
     end
 
-    def table_name
-      @klass.table_name
-    end
-
     protected
 
     def method_missing(method, *args, &block)
diff --git a/activerecord/lib/active_record/relation/batches.rb b/activerecord/lib/active_record/relation/batches.rb
index b41e935ed5..359af9820f 100644
--- a/activerecord/lib/active_record/relation/batches.rb
+++ b/activerecord/lib/active_record/relation/batches.rb
@@ -65,7 +65,7 @@ module ActiveRecord
       batch_size = options.delete(:batch_size) || 1000
 
       relation = relation.except(:order).order(batch_order).limit(batch_size)
-      records = relation.where(primary_key.gteq(start)).all
+      records = relation.where(table[primary_key].gteq(start)).all
 
       while records.any?
         yield records
@@ -73,7 +73,7 @@ module ActiveRecord
         break if records.size < batch_size
 
         if primary_key_offset = records.last.id
-          records = relation.where(primary_key.gt(primary_key_offset)).to_a
+          records = relation.where(table[primary_key].gt(primary_key_offset)).to_a
         else
           raise "Primary key not included in the custom select clause"
         end
@@ -83,7 +83,7 @@ module ActiveRecord
     private
 
     def batch_order
-      "#{@klass.table_name}.#{@klass.primary_key} ASC"
+      "#{table_name}.#{primary_key} ASC"
     end
   end
 end
diff --git a/activerecord/lib/active_record/relation/calculations.rb b/activerecord/lib/active_record/relation/calculations.rb
index fd45bb24dd..abc4c54109 100644
--- a/activerecord/lib/active_record/relation/calculations.rb
+++ b/activerecord/lib/active_record/relation/calculations.rb
@@ -168,7 +168,7 @@ module ActiveRecord
 
         unless arel.ast.grep(Arel::Nodes::OuterJoin).empty?
           distinct = true
-          column_name = @klass.primary_key if column_name == :all
+          column_name = primary_key if column_name == :all
         end
 
         distinct = nil if column_name =~ /\s*DISTINCT\s+/i
@@ -211,7 +211,7 @@ module ActiveRecord
       group_attr      = @group_values
       association     = @klass.reflect_on_association(group_attr.first.to_sym)
       associated      = group_attr.size == 1 && association && association.macro == :belongs_to # only count belongs_to associations
-      group_fields  = Array(associated ? association.primary_key_name : group_attr)
+      group_fields  = Array(associated ? association.foreign_key : group_attr)
       group_aliases = group_fields.map { |field| column_alias_for(field) }
       group_columns = group_aliases.zip(group_fields).map { |aliaz,field|
         [aliaz, column_for(field)]
@@ -282,15 +282,11 @@ module ActiveRecord
     end
 
     def type_cast_calculated_value(value, column, operation = nil)
-      if value.is_a?(String) || value.nil?
-        case operation
-          when 'count'   then value.to_i
-          when 'sum'     then type_cast_using_column(value || '0', column)
-          when 'average' then value.try(:to_d)
-          else type_cast_using_column(value, column)
-        end
-      else
-        type_cast_using_column(value, column)
+      case operation
+        when 'count'   then value.to_i
+        when 'sum'     then type_cast_using_column(value || '0', column)
+        when 'average' then value.respond_to?(:to_d) ? value.to_d : value
+        else type_cast_using_column(value, column)
       end
     end
 
diff --git a/activerecord/lib/active_record/relation/finder_methods.rb b/activerecord/lib/active_record/relation/finder_methods.rb
index 8bc28c06cb..7f32e5538e 100644
--- a/activerecord/lib/active_record/relation/finder_methods.rb
+++ b/activerecord/lib/active_record/relation/finder_methods.rb
@@ -19,7 +19,7 @@ module ActiveRecord
     #
     # All approaches accept an options hash as their last parameter.
     #
-    # ==== Parameters
+    # ==== Options
     #
     # * <tt>:conditions</tt> - An SQL fragment like "administrator = 1", <tt>["user_name = ?", username]</tt>,
     #   or <tt>["user_name = :user_name", { :user_name => user_name }]</tt>. See conditions in the intro.
@@ -171,13 +171,13 @@ module ActiveRecord
     def exists?(id = nil)
       id = id.id if ActiveRecord::Base === id
 
-      relation = select(primary_key).limit(1)
+      relation = select("1").limit(1)
 
       case id
       when Array, Hash
         relation = relation.where(id)
       else
-        relation = relation.where(primary_key.eq(id)) if id
+        relation = relation.where(table[primary_key].eq(id)) if id
       end
 
       relation.first ? true : false
@@ -188,7 +188,8 @@ module ActiveRecord
     def find_with_associations
       including = (@eager_load_values + @includes_values).uniq
       join_dependency = ActiveRecord::Associations::ClassMethods::JoinDependency.new(@klass, including, [])
-      rows = construct_relation_for_association_find(join_dependency).to_a
+      relation = construct_relation_for_association_find(join_dependency)
+      rows = connection.select_all(relation.to_sql, 'SQL', relation.bind_values)
       join_dependency.instantiate(rows)
     rescue ThrowResult
       []
@@ -225,10 +226,10 @@ module ActiveRecord
 
     def construct_limited_ids_condition(relation)
       orders = relation.order_values
-      values = @klass.connection.distinct("#{@klass.connection.quote_table_name @klass.table_name}.#{@klass.primary_key}", orders)
+      values = @klass.connection.distinct("#{@klass.connection.quote_table_name table_name}.#{primary_key}", orders)
 
-      ids_array = relation.select(values).collect {|row| row[@klass.primary_key]}
-      ids_array.empty? ? raise(ThrowResult) : primary_key.in(ids_array)
+      ids_array = relation.select(values).collect {|row| row[primary_key]}
+      ids_array.empty? ? raise(ThrowResult) : table[primary_key].in(ids_array)
     end
 
     def find_by_attributes(match, attributes, *args)
@@ -290,24 +291,24 @@ module ActiveRecord
     def find_one(id)
       id = id.id if ActiveRecord::Base === id
 
-      column = columns_hash[primary_key.name.to_s]
+      column = columns_hash[primary_key]
 
       substitute = connection.substitute_for(column, @bind_values)
-      relation = where(primary_key.eq(substitute))
+      relation = where(table[primary_key].eq(substitute))
       relation.bind_values = [[column, id]]
       record = relation.first
 
       unless record
         conditions = arel.where_sql
         conditions = " [#{conditions}]" if conditions
-        raise RecordNotFound, "Couldn't find #{@klass.name} with #{@klass.primary_key}=#{id}#{conditions}"
+        raise RecordNotFound, "Couldn't find #{@klass.name} with #{primary_key}=#{id}#{conditions}"
       end
 
       record
     end
 
     def find_some(ids)
-      result = where(primary_key.in(ids)).all
+      result = where(table[primary_key].in(ids)).all
 
       expected_size =
         if @limit_value && ids.size > @limit_value
diff --git a/activerecord/lib/active_record/relation/predicate_builder.rb b/activerecord/lib/active_record/relation/predicate_builder.rb
index 70d84619a1..61d9974570 100644
--- a/activerecord/lib/active_record/relation/predicate_builder.rb
+++ b/activerecord/lib/active_record/relation/predicate_builder.rb
@@ -15,18 +15,18 @@ module ActiveRecord
             table = Arel::Table.new(table_name, :engine => engine)
           end
 
-          attribute = table[column] || Arel::Attribute.new(table, column)
+          attribute = table[column.to_sym]
 
           case value
           when Array, ActiveRecord::Associations::AssociationCollection, ActiveRecord::Relation
             values = value.to_a.map { |x|
-              x.respond_to?(:quoted_id) ? x.quoted_id : x
+              x.is_a?(ActiveRecord::Base) ? x.id : x
             }
             attribute.in(values)
           when Range, Arel::Relation
             attribute.in(value)
           when ActiveRecord::Base
-            attribute.eq(Arel.sql(value.quoted_id))
+            attribute.eq(value.id)
           when Class
             # FIXME: I think we need to deprecate this behavior
             attribute.eq(value.name)
diff --git a/activerecord/lib/active_record/relation/query_methods.rb b/activerecord/lib/active_record/relation/query_methods.rb
index 0ab55ae864..6a905b8588 100644
--- a/activerecord/lib/active_record/relation/query_methods.rb
+++ b/activerecord/lib/active_record/relation/query_methods.rb
@@ -128,7 +128,7 @@ module ActiveRecord
 
     def create_with(value)
       relation = clone
-      relation.create_with_value = value
+      relation.create_with_value = value && (@create_with_value || {}).merge(value)
       relation
     end
 
@@ -152,7 +152,7 @@ module ActiveRecord
       order_clause = arel.order_clauses
 
       order = order_clause.empty? ?
-        "#{@klass.table_name}.#{@klass.primary_key} DESC" :
+        "#{table_name}.#{primary_key} DESC" :
         reverse_sql_order(order_clause).join(', ')
 
       except(:order).order(Arel.sql(order))
@@ -171,7 +171,7 @@ module ActiveRecord
 
       arel.having(*@having_values.uniq.reject{|h| h.blank?}) unless @having_values.empty?
 
-      arel.take(@limit_value) if @limit_value
+      arel.take(connection.sanitize_limit(@limit_value)) if @limit_value
       arel.skip(@offset_value) if @offset_value
 
       arel.group(*@group_values.uniq.reject{|g| g.blank?}) unless @group_values.empty?
diff --git a/activerecord/lib/active_record/relation/spawn_methods.rb b/activerecord/lib/active_record/relation/spawn_methods.rb
index 5acf3ec83a..4150e36a9a 100644
--- a/activerecord/lib/active_record/relation/spawn_methods.rb
+++ b/activerecord/lib/active_record/relation/spawn_methods.rb
@@ -46,21 +46,28 @@ module ActiveRecord
 
       merged_relation.where_values = merged_wheres
 
-      (Relation::SINGLE_VALUE_METHODS - [:lock]).each do |method|
+      (Relation::SINGLE_VALUE_METHODS - [:lock, :create_with]).each do |method|
         value = r.send(:"#{method}_value")
         merged_relation.send(:"#{method}_value=", value) unless value.nil?
       end
 
       merged_relation.lock_value = r.lock_value unless merged_relation.lock_value
 
+      merged_relation = merged_relation.create_with(r.create_with_value) if r.create_with_value
+
       # Apply scope extension modules
       merged_relation.send :apply_modules, r.extensions
 
       merged_relation
     end
 
-    alias :& :merge
-
+    # Removes from the query the condition(s) specified in +skips+.
+    #
+    # Example:
+    #
+    #   Post.order('id asc').except(:order)                  # discards the order condition
+    #   Post.where('id > 10').order('id asc').except(:where) # discards the where condition but keeps the order
+    #
     def except(*skips)
       result = self.class.new(@klass, table)
 
@@ -75,6 +82,13 @@ module ActiveRecord
       result
     end
 
+    # Removes any condition from the query other than the one(s) specified in +onlies+.
+    #
+    # Example:
+    #
+    #   Post.order('id asc').only(:where)         # discards the order condition
+    #   Post.order('id asc').only(:where, :order) # uses the specified order
+    #
     def only(*onlies)
       result = self.class.new(@klass, table)
 
@@ -98,7 +112,7 @@ module ActiveRecord
 
       options.assert_valid_keys(VALID_FIND_OPTIONS)
       finders = options.dup
-      finders.delete_if { |key, value| value.nil? }
+      finders.delete_if { |key, value| value.nil? && key != :limit }
 
       ([:joins, :select, :group, :order, :having, :limit, :offset, :from, :lock, :readonly] & finders.keys).each do |finder|
         relation = relation.send(finder, finders[finder])
diff --git a/activerecord/lib/active_record/result.rb b/activerecord/lib/active_record/result.rb
index 8deff1478f..0465b21e88 100644
--- a/activerecord/lib/active_record/result.rb
+++ b/activerecord/lib/active_record/result.rb
@@ -20,10 +20,14 @@ module ActiveRecord
       hash_rows.each { |row| yield row }
     end
 
+    def to_hash
+      hash_rows
+    end
+
     private
     def hash_rows
       @hash_rows ||= @rows.map { |row|
-        ActiveSupport::OrderedHash[@columns.zip(row)]
+        Hash[@columns.zip(row)]
       }
     end
   end
diff --git a/activerecord/lib/active_record/schema_dumper.rb b/activerecord/lib/active_record/schema_dumper.rb
index e30b481fe1..a893c0ad85 100644
--- a/activerecord/lib/active_record/schema_dumper.rb
+++ b/activerecord/lib/active_record/schema_dumper.rb
@@ -83,7 +83,7 @@ HEADER
 
           # first dump primary key column
           if @connection.respond_to?(:pk_and_sequence_for)
-            pk, pk_seq = @connection.pk_and_sequence_for(table)
+            pk, _ = @connection.pk_and_sequence_for(table)
           elsif @connection.respond_to?(:primary_key)
             pk = @connection.primary_key(table)
           end
diff --git a/activerecord/lib/active_record/session_store.rb b/activerecord/lib/active_record/session_store.rb
index 3400fd6ade..7e77aefb21 100644
--- a/activerecord/lib/active_record/session_store.rb
+++ b/activerecord/lib/active_record/session_store.rb
@@ -59,10 +59,12 @@ module ActiveRecord
       end
 
       def drop_table!
+        connection_pool.clear_table_cache!(table_name)
         connection.drop_table table_name
       end
 
       def create_table!
+        connection_pool.clear_table_cache!(table_name)
         connection.create_table(table_name) do |t|
           t.string session_id_column, :limit => 255
           t.text data_column_name
diff --git a/activerecord/lib/active_record/timestamp.rb b/activerecord/lib/active_record/timestamp.rb
index 2ecbd906bd..1511c71ffc 100644
--- a/activerecord/lib/active_record/timestamp.rb
+++ b/activerecord/lib/active_record/timestamp.rb
@@ -9,24 +9,26 @@ module ActiveRecord
   #
   # Timestamping can be turned off by setting:
   #
-  #   <tt>ActiveRecord::Base.record_timestamps = false</tt>
+  #   config.active_record.record_timestamps = false
   #
   # Timestamps are in the local timezone by default but you can use UTC by setting:
   #
-  #   <tt>ActiveRecord::Base.default_timezone = :utc</tt>
+  #   config.active_record.default_timezone = :utc
   #
   # == Time Zone aware attributes
   #
   # By default, ActiveRecord::Base keeps all the datetime columns time zone aware by executing following code.
   #
-  #   ActiveRecord::Base.time_zone_aware_attributes = true
+  #   config.active_record.time_zone_aware_attributes = true
   #
   # This feature can easily be turned off by assigning value <tt>false</tt> .
   #
-  # If your attributes are time zone aware and you desire to skip time zone conversion for certain
-  # attributes then you can do following:
+  # If your attributes are time zone aware and you desire to skip time zone conversion to the current Time.zone
+  # when reading certain attributes then you can do following:
   #
-  #   Topic.skip_time_zone_conversion_for_attributes = [:written_on]
+  #   class Topic < ActiveRecord::Base
+  #     self.skip_time_zone_conversion_for_attributes = [:written_on]
+  #   end
   module Timestamp
     extend ActiveSupport::Concern
 
@@ -66,8 +68,16 @@ module ActiveRecord
       self.record_timestamps && (!partial_updates? || changed? || (attributes.keys & self.class.serialized_attributes.keys).present?)
     end
 
+    def timestamp_attributes_for_create_in_model
+      timestamp_attributes_for_create.select { |c| self.class.column_names.include?(c.to_s) }
+    end
+
     def timestamp_attributes_for_update_in_model
-      timestamp_attributes_for_update.select { |c| respond_to?(c) }
+      timestamp_attributes_for_update.select { |c| self.class.column_names.include?(c.to_s) }
+    end
+
+    def all_timestamp_attributes_in_model
+      timestamp_attributes_for_create_in_model + timestamp_attributes_for_update_in_model
     end
 
     def timestamp_attributes_for_update #:nodoc:
diff --git a/activerecord/lib/active_record/transactions.rb b/activerecord/lib/active_record/transactions.rb
index de496698f1..60d4c256c4 100644
--- a/activerecord/lib/active_record/transactions.rb
+++ b/activerecord/lib/active_record/transactions.rb
@@ -260,7 +260,7 @@ module ActiveRecord
 
     # Call the after_commit callbacks
     def committed! #:nodoc:
-      _run_commit_callbacks
+      run_callbacks :commit
     ensure
       clear_transaction_record_state
     end
@@ -268,7 +268,7 @@ module ActiveRecord
     # Call the after rollback callbacks. The restore_state argument indicates if the record
     # state should be rolled back to the beginning or just to the last savepoint.
     def rolledback!(force_restore_state = false) #:nodoc:
-      _run_rollback_callbacks
+      run_callbacks :rollback
     ensure
       restore_transaction_record_state(force_restore_state)
     end
@@ -302,8 +302,8 @@ module ActiveRecord
     # Save the new record state and id of a record so it can be restored later if a transaction fails.
     def remember_transaction_record_state #:nodoc
       @_start_transaction_state ||= {}
+      @_start_transaction_state[:id] = id if has_attribute?(self.class.primary_key)
       unless @_start_transaction_state.include?(:new_record)
-        @_start_transaction_state[:id] = id if has_attribute?(self.class.primary_key)
         @_start_transaction_state[:new_record] = @new_record
       end
       unless @_start_transaction_state.include?(:destroyed)
@@ -326,16 +326,14 @@ module ActiveRecord
         @_start_transaction_state[:level] = (@_start_transaction_state[:level] || 0) - 1
         if @_start_transaction_state[:level] < 1
           restore_state = remove_instance_variable(:@_start_transaction_state)
-          if restore_state
-            @attributes = @attributes.dup if @attributes.frozen?
-            @new_record = restore_state[:new_record]
-            @destroyed  = restore_state[:destroyed]
-            if restore_state[:id]
-              self.id = restore_state[:id]
-            else
-              @attributes.delete(self.class.primary_key)
-              @attributes_cache.delete(self.class.primary_key)
-            end
+          @attributes = @attributes.dup if @attributes.frozen?
+          @new_record = restore_state[:new_record]
+          @destroyed  = restore_state[:destroyed]
+          if restore_state.has_key?(:id)
+            self.id = restore_state[:id]
+          else
+            @attributes.delete(self.class.primary_key)
+            @attributes_cache.delete(self.class.primary_key)
           end
         end
       end
diff --git a/activerecord/lib/active_record/validations.rb b/activerecord/lib/active_record/validations.rb
index f367315b22..26c1a9db93 100644
--- a/activerecord/lib/active_record/validations.rb
+++ b/activerecord/lib/active_record/validations.rb
@@ -37,7 +37,7 @@ module ActiveRecord
       end
     end
 
-    # The validation process on save can be skipped by passing false. The regular Base#save method is
+    # The validation process on save can be skipped by passing :validate => false. The regular Base#save method is
     # replaced with this when the validations module is mixed in, which it is by default.
     def save(options={})
       perform_validations(options) ? super : false
diff --git a/activerecord/lib/active_record/validations/uniqueness.rb b/activerecord/lib/active_record/validations/uniqueness.rb
index 853808eebf..a96796f9ff 100644
--- a/activerecord/lib/active_record/validations/uniqueness.rb
+++ b/activerecord/lib/active_record/validations/uniqueness.rb
@@ -15,8 +15,10 @@ module ActiveRecord
       def validate_each(record, attribute, value)
         finder_class = find_finder_class_for(record)
 
-        if value && record.class.serialized_attributes.key?(attribute.to_s)
-          value = YAML.dump value
+        coder = record.class.serialized_attributes[attribute.to_s]
+
+        if value && coder
+          value = coder.dump value
         end
 
         sql, params = mount_sql_and_params(finder_class, record.class.quoted_table_name, attribute, value)
@@ -85,11 +87,16 @@ module ActiveRecord
       # can be named "davidhh".
       #
       #   class Person < ActiveRecord::Base
-      #     validates_uniqueness_of :user_name, :scope => :account_id
+      #     validates_uniqueness_of :user_name
       #   end
       #
-      # It can also validate whether the value of the specified attributes are unique based on multiple
-      # scope parameters.  For example, making sure that a teacher can only be on the schedule once
+      # It can also validate whether the value of the specified attributes are unique based on a scope parameter:
+      #
+      #   class Person < ActiveRecord::Base
+      #     validates_uniqueness_of :user_name, :scope => :account_id
+      #   end 
+      #
+      # Or even multiple scope parameters.  For example, making sure that a teacher can only be on the schedule once
       # per semester for a particular class.
       #
       #   class TeacherSchedule < ActiveRecord::Base