- Fixed bug in :meth:`.Session.merge` where an object with a composite

primary key that has values for some but not all of the PK fields
would emit a SELECT statement leaking the internal NEVER_SET symbol
into the query, rather than detecting that this object does not have
a searchable primary key and no SELECT should be emitted.
fixes #3647

3 files changed, 63 insertions, 3 deletions

diff --git a/doc/build/changelog/changelog_10.rst b/doc/build/changelog/changelog_10.rst
index d607110aa..ac250a7aa 100644
--- a/doc/build/changelog/changelog_10.rst
+++ b/doc/build/changelog/changelog_10.rst
@@ -20,6 +20,16 @@
     :released:
 
     .. change::
+        :tags: bug, orm
+        :tickets: 3647
+
+        Fixed bug in :meth:`.Session.merge` where an object with a composite
+        primary key that has values for some but not all of the PK fields
+        would emit a SELECT statement leaking the internal NEVER_SET symbol
+        into the query, rather than detecting that this object does not have
+        a searchable primary key and no SELECT should be emitted.
+
+    .. change::
         :tags: bug, postgresql
         :tickets: 3644
 
diff --git a/lib/sqlalchemy/orm/session.py b/lib/sqlalchemy/orm/session.py
index 80bd902d0..48ff09b87 100644
--- a/lib/sqlalchemy/orm/session.py
+++ b/lib/sqlalchemy/orm/session.py
@@ -1768,9 +1768,10 @@ class Session(_SessionClassMethods):
             self._update_impl(merged_state)
             new_instance = True
 
-        elif not _none_set.intersection(key[1]) or \
+        elif key_is_persistent and (
+            not _none_set.intersection(key[1]) or
             (mapper.allow_partial_pks and
-             not _none_set.issuperset(key[1])):
+             not _none_set.issuperset(key[1]))):
             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 dab9f4305..f69b07fe8 100644
--- a/test/orm/test_merge.py
+++ b/test/orm/test_merge.py
@@ -6,7 +6,7 @@ from sqlalchemy import testing
 from sqlalchemy.util import OrderedSet
 from sqlalchemy.orm import mapper, relationship, create_session, \
     PropComparator, synonym, comparable_property, sessionmaker, \
-    attributes, Session, backref, configure_mappers
+    attributes, Session, backref, configure_mappers, foreign
 from sqlalchemy.orm.collections import attribute_mapped_collection
 from sqlalchemy.orm.interfaces import MapperOption
 from sqlalchemy.testing import eq_, ne_
@@ -451,6 +451,55 @@ class MergeTest(_fixtures.FixtureTest):
         eq_(u2.addresses[1].email_address, 'afafds')
         eq_(load.called, 21)
 
+    def test_dont_send_neverset_to_get(self):
+        # test issue #3647
+        CompositePk, composite_pk_table = (
+            self.classes.CompositePk, self.tables.composite_pk_table
+        )
+        mapper(CompositePk, composite_pk_table)
+        cp1 = CompositePk(j=1, k=1)
+
+        sess = Session()
+
+        rec = []
+
+        def go():
+            rec.append(sess.merge(cp1))
+        self.assert_sql_count(testing.db, go, 0)
+        rec[0].i = 5
+        sess.commit()
+        eq_(rec[0].i, 5)
+
+    def test_dont_send_neverset_to_get_w_relationship(self):
+        # test issue #3647
+        CompositePk, composite_pk_table = (
+            self.classes.CompositePk, self.tables.composite_pk_table
+        )
+        User, users = (
+            self.classes.User, self.tables.users
+        )
+        mapper(User, users, properties={
+            'elements': relationship(
+                CompositePk,
+                primaryjoin=users.c.id == foreign(composite_pk_table.c.i))
+        })
+        mapper(CompositePk, composite_pk_table)
+
+        u1 = User(id=5, name='some user')
+        cp1 = CompositePk(j=1, k=1)
+        u1.elements.append(cp1)
+        sess = Session()
+
+        rec = []
+
+        def go():
+            rec.append(sess.merge(u1))
+        self.assert_sql_count(testing.db, go, 1)
+        u2 = rec[0]
+        sess.commit()
+        eq_(u2.elements[0].i, 5)
+        eq_(u2.id, 5)
+
     def test_no_relationship_cascade(self):
         """test that merge doesn't interfere with a relationship()
            target that specifically doesn't include 'merge' cascade.