- The :meth:`.Session.merge` method now tracks pending objects by

primary key before emitting an INSERT, and merges distinct objects with
duplicate primary keys together as they are encountered, which is
essentially semi-deterministic at best.   This behavior
matches what happens already with persistent objects.
fixes #3601

7 files changed, 191 insertions, 11 deletions

diff --git a/doc/build/changelog/changelog_11.rst b/doc/build/changelog/changelog_11.rst
index ea0b62a5e..0d9f997f9 100644
--- a/doc/build/changelog/changelog_11.rst
+++ b/doc/build/changelog/changelog_11.rst
@@ -22,6 +22,20 @@
     :version: 1.1.0b1
 
     .. change::
+        :tags: bug, orm
+        :tickets: 3601
+
+        The :meth:`.Session.merge` method now tracks pending objects by
+        primary key before emitting an INSERT, and merges distinct objects with
+        duplicate primary keys together as they are encountered, which is
+        essentially semi-deterministic at best.   This behavior
+        matches what happens already with persistent objects.
+
+        .. seealso::
+
+            :ref:`change_3601`
+
+    .. change::
         :tags: bug, postgresql
         :tickets: 3587
 
diff --git a/doc/build/changelog/migration_11.rst b/doc/build/changelog/migration_11.rst
index f43cfa87c..b5889c763 100644
--- a/doc/build/changelog/migration_11.rst
+++ b/doc/build/changelog/migration_11.rst
@@ -16,7 +16,7 @@ What's New in SQLAlchemy 1.1?
     some issues may be moved to later milestones in order to allow
     for a timely release.
 
-    Document last updated: November 11, 2015
+    Document last updated: December 4, 2015
 
 Introduction
 ============
@@ -291,6 +291,59 @@ time on the outside of the subquery.
 :ticket:`3582`
 
 
+.. _change_3601:
+
+Session.merge resolves pending conflicts the same as persistent
+---------------------------------------------------------------
+
+The :meth:`.Session.merge` method will now track the identities of objects given
+within a graph to maintain primary key uniqueness before emitting an INSERT.
+When duplicate objects of the same identity are encountered, non-primary-key
+attributes are **overwritten** as the objects are encountered, which is
+essentially non-deterministic.   This behavior matches that of how persistent
+objects, that is objects that are already located in the database via
+primary key, are already treated, so this behavior is more internally
+consistent.
+
+Given::
+
+    u1 = User(id=7, name='x')
+    u1.orders = [
+        Order(description='o1', address=Address(id=1, email_address='a')),
+        Order(description='o2', address=Address(id=1, email_address='b')),
+        Order(description='o3', address=Address(id=1, email_address='c'))
+    ]
+
+    sess = Session()
+    sess.merge(u1)
+
+Above, we merge a ``User`` object with three new ``Order`` objects, each referring to
+a distinct ``Address`` object, however each is given the same primary key.
+The current behavior of :meth:`.Session.merge` is to look in the identity
+map for this ``Address`` object, and use that as the target.   If the object
+is present, meaning that the database already has a row for ``Address`` with
+primary key "1", we can see that the ``email_address`` field of the ``Address``
+will be overwritten three times, in this case with the values a, b and finally
+c.
+
+However, if the ``Address`` row for primary key "1" were not present, :meth:`.Session.merge`
+would instead create three separate ``Address`` instances, and we'd then get
+a primary key conflict upon INSERT.  The new behavior is that the proposed
+primary key for these ``Address`` objects are tracked in a separate dictionary
+so that we merge the state of the three proposed ``Address`` objects onto
+one ``Address`` object to be inserted.
+
+It may have been preferable if the original case emitted some kind of warning
+that conflicting data were present in a single merge-tree, however the
+non-deterministic merging of values has been the behavior for many
+years for the persistent case; it now matches for the pending case.   A
+feature that warns for conflicting values could still be feasible for both
+cases but would add considerable performance overhead as each column value
+would have to be compared during the merge.
+
+
+:ticket:`3601`
+
 New Features and Improvements - Core
 ====================================
 
diff --git a/lib/sqlalchemy/orm/interfaces.py b/lib/sqlalchemy/orm/interfaces.py
index cd4a0116d..ed8f27332 100644
--- a/lib/sqlalchemy/orm/interfaces.py
+++ b/lib/sqlalchemy/orm/interfaces.py
@@ -234,7 +234,7 @@ class MapperProperty(_MappedAttribute, InspectionAttr, util.MemoizedSlots):
         """
 
     def merge(self, session, source_state, source_dict, dest_state,
-              dest_dict, load, _recursive):
+              dest_dict, load, _recursive, _resolve_conflict_map):
         """Merge the attribute represented by this ``MapperProperty``
         from source to destination object.
 
diff --git a/lib/sqlalchemy/orm/properties.py b/lib/sqlalchemy/orm/properties.py
index b1f1c61c4..0d4e1b771 100644
--- a/lib/sqlalchemy/orm/properties.py
+++ b/lib/sqlalchemy/orm/properties.py
@@ -206,7 +206,7 @@ class ColumnProperty(StrategizedProperty):
             get_committed_value(state, dict_, passive=passive)
 
     def merge(self, session, source_state, source_dict, dest_state,
-              dest_dict, load, _recursive):
+              dest_dict, load, _recursive, _resolve_conflict_map):
         if not self.instrument:
             return
         elif self.key in source_dict:
diff --git a/lib/sqlalchemy/orm/relationships.py b/lib/sqlalchemy/orm/relationships.py
index 929c923a6..1d442eff8 100644
--- a/lib/sqlalchemy/orm/relationships.py
+++ b/lib/sqlalchemy/orm/relationships.py
@@ -1430,7 +1430,7 @@ class RelationshipProperty(StrategizedProperty):
               source_dict,
               dest_state,
               dest_dict,
-              load, _recursive):
+              load, _recursive, _resolve_conflict_map):
 
         if load:
             for r in self._reverse_property:
@@ -1463,8 +1463,10 @@ class RelationshipProperty(StrategizedProperty):
                 current_state = attributes.instance_state(current)
                 current_dict = attributes.instance_dict(current)
                 _recursive[(current_state, self)] = True
-                obj = session._merge(current_state, current_dict,
-                                     load=load, _recursive=_recursive)
+                obj = session._merge(
+                    current_state, current_dict,
+                    load=load, _recursive=_recursive,
+                    _resolve_conflict_map=_resolve_conflict_map)
                 if obj is not None:
                     dest_list.append(obj)
 
@@ -1482,8 +1484,10 @@ class RelationshipProperty(StrategizedProperty):
                 current_state = attributes.instance_state(current)
                 current_dict = attributes.instance_dict(current)
                 _recursive[(current_state, self)] = True
-                obj = session._merge(current_state, current_dict,
-                                     load=load, _recursive=_recursive)
+                obj = session._merge(
+                    current_state, current_dict,
+                    load=load, _recursive=_recursive,
+                    _resolve_conflict_map=_resolve_conflict_map)
             else:
                 obj = None
 
diff --git a/lib/sqlalchemy/orm/session.py b/lib/sqlalchemy/orm/session.py
index 4272d7d78..f58e4de61 100644
--- a/lib/sqlalchemy/orm/session.py
+++ b/lib/sqlalchemy/orm/session.py
@@ -1689,6 +1689,10 @@ class Session(_SessionClassMethods):
 
         See :ref:`unitofwork_merging` for a detailed discussion of merging.
 
+        .. versionchanged:: 1.1 - :meth:`.Session.merge` will now reconcile
+           pending objects with overlapping primary keys in the same way
+           as persistent.  See :ref:`change_3601` for discussion.
+
         :param instance: Instance to be merged.
         :param load: Boolean, when False, :meth:`.merge` switches into
          a "high performance" mode which causes it to forego emitting history
@@ -1713,12 +1717,14 @@ class Session(_SessionClassMethods):
          should be "clean" as well, else this suggests a mis-use of the
          method.
 
+
         """
 
         if self._warn_on_events:
             self._flush_warning("Session.merge()")
 
         _recursive = {}
+        _resolve_conflict_map = {}
 
         if load:
             # flush current contents if we expect to load data
@@ -1731,11 +1737,13 @@ class Session(_SessionClassMethods):
             return self._merge(
                 attributes.instance_state(instance),
                 attributes.instance_dict(instance),
-                load=load, _recursive=_recursive)
+                load=load, _recursive=_recursive,
+                _resolve_conflict_map=_resolve_conflict_map)
         finally:
             self.autoflush = autoflush
 
-    def _merge(self, state, state_dict, load=True, _recursive=None):
+    def _merge(self, state, state_dict, load=True, _recursive=None,
+               _resolve_conflict_map=None):
         mapper = _state_mapper(state)
         if state in _recursive:
             return _recursive[state]
@@ -1751,9 +1759,14 @@ class Session(_SessionClassMethods):
                     "all changes on mapped instances before merging with "
                     "load=False.")
             key = mapper._identity_key_from_state(state)
+            key_is_persistent = attributes.NEVER_SET not in key[1]
+        else:
+            key_is_persistent = True
 
         if key in self.identity_map:
             merged = self.identity_map[key]
+        elif key_is_persistent and key in _resolve_conflict_map:
+            merged = _resolve_conflict_map[key]
 
         elif not load:
             if state.modified:
@@ -1785,6 +1798,7 @@ class Session(_SessionClassMethods):
             merged_dict = attributes.instance_dict(merged)
 
         _recursive[state] = merged
+        _resolve_conflict_map[key] = merged
 
         # check that we didn't just pull the exact same
         # state out.
@@ -1823,7 +1837,7 @@ class Session(_SessionClassMethods):
             for prop in mapper.iterate_properties:
                 prop.merge(self, state, state_dict,
                            merged_state, merged_dict,
-                           load, _recursive)
+                           load, _recursive, _resolve_conflict_map)
 
         if not load:
             # remove any history
diff --git a/test/orm/test_merge.py b/test/orm/test_merge.py
index a52274896..dab9f4305 100644
--- a/test/orm/test_merge.py
+++ b/test/orm/test_merge.py
@@ -1102,6 +1102,101 @@ class MergeTest(_fixtures.FixtureTest):
             eq_(ustate.load_path.path, (umapper, ))
             eq_(ustate.load_options, set([opt2]))
 
+    def test_resolve_conflicts_pending_doesnt_interfere_no_ident(self):
+        User, Address, Order = (
+            self.classes.User, self.classes.Address, self.classes.Order)
+        users, addresses, orders = (
+            self.tables.users, self.tables.addresses, self.tables.orders)
+
+        mapper(User, users, properties={
+            'orders': relationship(Order)
+        })
+        mapper(Order, orders, properties={
+            'address': relationship(Address)
+        })
+        mapper(Address, addresses)
+
+        u1 = User(id=7, name='x')
+        u1.orders = [
+            Order(description='o1', address=Address(email_address='a')),
+            Order(description='o2', address=Address(email_address='b')),
+            Order(description='o3', address=Address(email_address='c'))
+        ]
+
+        sess = Session()
+        sess.merge(u1)
+        sess.flush()
+
+        eq_(
+            sess.query(Address.email_address).order_by(
+                Address.email_address).all(),
+            [('a', ), ('b', ), ('c', )]
+        )
+
+    def test_resolve_conflicts_pending(self):
+        User, Address, Order = (
+            self.classes.User, self.classes.Address, self.classes.Order)
+        users, addresses, orders = (
+            self.tables.users, self.tables.addresses, self.tables.orders)
+
+        mapper(User, users, properties={
+            'orders': relationship(Order)
+        })
+        mapper(Order, orders, properties={
+            'address': relationship(Address)
+        })
+        mapper(Address, addresses)
+
+        u1 = User(id=7, name='x')
+        u1.orders = [
+            Order(description='o1', address=Address(id=1, email_address='a')),
+            Order(description='o2', address=Address(id=1, email_address='b')),
+            Order(description='o3', address=Address(id=1, email_address='c'))
+        ]
+
+        sess = Session()
+        sess.merge(u1)
+        sess.flush()
+
+        eq_(
+            sess.query(Address).one(),
+            Address(id=1, email_address='c')
+        )
+
+    def test_resolve_conflicts_persistent(self):
+        User, Address, Order = (
+            self.classes.User, self.classes.Address, self.classes.Order)
+        users, addresses, orders = (
+            self.tables.users, self.tables.addresses, self.tables.orders)
+
+        mapper(User, users, properties={
+            'orders': relationship(Order)
+        })
+        mapper(Order, orders, properties={
+            'address': relationship(Address)
+        })
+        mapper(Address, addresses)
+
+        sess = Session()
+        sess.add(Address(id=1, email_address='z'))
+        sess.commit()
+
+        u1 = User(id=7, name='x')
+        u1.orders = [
+            Order(description='o1', address=Address(id=1, email_address='a')),
+            Order(description='o2', address=Address(id=1, email_address='b')),
+            Order(description='o3', address=Address(id=1, email_address='c'))
+        ]
+
+        sess = Session()
+        sess.merge(u1)
+        sess.flush()
+
+        eq_(
+            sess.query(Address).one(),
+            Address(id=1, email_address='c')
+        )
+
 
 class M2ONoUseGetLoadingTest(fixtures.MappedTest):
     """Merge a one-to-many.  The many-to-one on the other side is set up