Consider merge key with (None, ) as non-persistent

Fixed bug in :meth:`.Session.merge` where objects in a collection that had
the primary key attribute set to ``None`` for a key that is  typically
autoincrementing would be considered to be a database-persisted key for
part of the internal deduplication process, causing only one object to
actually be inserted in the database.

Change-Id: I0a6e00043be0b2979cda33740e1be3b430ecf8c7
Fixes: #4056
(cherry picked from commit 5243341ed886e10a0d3f7fef8ae3d071e0ffdcf0)

3 files changed, 51 insertions, 5 deletions

diff --git a/doc/build/changelog/unreleased_11/4056.rst b/doc/build/changelog/unreleased_11/4056.rst
new file mode 100644
index 000000000..b8e02a05e
--- /dev/null
+++ b/doc/build/changelog/unreleased_11/4056.rst
@@ -0,0 +1,10 @@
+.. change::
+    :tags: bug, orm
+    :tickets: 4056
+    :versions: 1.2.0b3
+
+    Fixed bug in :meth:`.Session.merge` where objects in a collection that had
+    the primary key attribute set to ``None`` for a key that is  typically
+    autoincrementing would be considered to be a database-persisted key for
+    part of the internal deduplication process, causing only one object to
+    actually be inserted in the database.
diff --git a/lib/sqlalchemy/orm/session.py b/lib/sqlalchemy/orm/session.py
index 6ecef17f1..5488c3031 100644
--- a/lib/sqlalchemy/orm/session.py
+++ b/lib/sqlalchemy/orm/session.py
@@ -1912,7 +1912,10 @@ 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]
+            key_is_persistent = attributes.NEVER_SET not in key[1] and (
+                not _none_set.intersection(key[1]) or
+                (mapper.allow_partial_pks and not _none_set.issuperset(key[1]))
+            )
         else:
             key_is_persistent = True
 
@@ -1933,10 +1936,7 @@ class Session(_SessionClassMethods):
             self._update_impl(merged_state)
             new_instance = True
 
-        elif key_is_persistent and (
-            not _none_set.intersection(key[1]) or
-            (mapper.allow_partial_pks and
-             not _none_set.issuperset(key[1]))):
+        elif key_is_persistent:
             merged = self.query(mapper.class_).get(key[1])
         else:
             merged = None
diff --git a/test/orm/test_merge.py b/test/orm/test_merge.py
index f751d43a8..8c1390238 100644
--- a/test/orm/test_merge.py
+++ b/test/orm/test_merge.py
@@ -98,6 +98,42 @@ class MergeTest(_fixtures.FixtureTest):
                 Address(id=2, email_address='fred2'),
             ])))
 
+    def test_transient_to_pending_collection_pk_none(self):
+        User, Address, addresses, users = (self.classes.User,
+                                           self.classes.Address,
+                                           self.tables.addresses,
+                                           self.tables.users)
+
+        mapper(User, users, properties={
+            'addresses': relationship(Address, backref='user',
+                                      collection_class=OrderedSet)})
+        mapper(Address, addresses)
+        load = self.load_tracker(User)
+        self.load_tracker(Address, load)
+
+        u = User(id=None, name='fred', addresses=OrderedSet([
+            Address(id=None, email_address='fred1'),
+            Address(id=None, email_address='fred2'),
+        ]))
+        eq_(load.called, 0)
+
+        sess = create_session()
+        sess.merge(u)
+        eq_(load.called, 3)
+
+        merged_users = [e for e in sess if isinstance(e, User)]
+        eq_(len(merged_users), 1)
+        assert merged_users[0] is not u
+
+        sess.flush()
+        sess.expunge_all()
+
+        eq_(sess.query(User).one(),
+            User(name='fred', addresses=OrderedSet([
+                Address(email_address='fred1'),
+                Address(email_address='fred2'),
+            ])))
+
     def test_transient_to_persistent(self):
         User, users = self.classes.User, self.tables.users