fix mapper._primary_key_propkeys

Fixed issue in :meth:`_orm.Session.bulk_save` when used with persistent
objects which would fail to track the primary key of mappings where the
column name of the primary key were different than the attribute name.

Fixes: #6392
Change-Id: I9b89bf00f900b7d2287517249e4f9356f20f0bdb
(cherry picked from commit 7892cc30ad60ae450471b62f351b3beb66911892)

3 files changed, 45 insertions, 6 deletions

diff --git a/doc/build/changelog/unreleased_13/6392.rst b/doc/build/changelog/unreleased_13/6392.rst
new file mode 100644
index 000000000..ddd49ccf9
--- /dev/null
+++ b/doc/build/changelog/unreleased_13/6392.rst
@@ -0,0 +1,9 @@
+.. change::
+    :tags: bug, orm
+    :tickets: 6392
+    :versions: 1.4.12
+
+    Fixed issue in :meth:`_orm.Session.bulk_save` when used with persistent
+    objects which would fail to track the primary key of mappings where the
+    column name of the primary key were different than the attribute name.
+
diff --git a/lib/sqlalchemy/orm/mapper.py b/lib/sqlalchemy/orm/mapper.py
index c4704b484..c2cd208ea 100644
--- a/lib/sqlalchemy/orm/mapper.py
+++ b/lib/sqlalchemy/orm/mapper.py
@@ -2821,7 +2821,7 @@ class Mapper(InspectionAttr):
         return [self._columntoproperty[col] for col in self.primary_key]
 
     @_memoized_configured_property
-    def _all_pk_props(self):
+    def _all_pk_cols(self):
         collection = set()
         for table in self.tables:
             collection.update(self._pks_by_table[table])
@@ -2836,7 +2836,7 @@ class Mapper(InspectionAttr):
 
     @_memoized_configured_property
     def _primary_key_propkeys(self):
-        return {prop.key for prop in self._all_pk_props}
+        return {self._columntoproperty[col].key for col in self._all_pk_cols}
 
     def _get_state_attr_by_column(
         self, state, dict_, column, passive=attributes.PASSIVE_RETURN_NEVER_SET
diff --git a/test/orm/test_bulk.py b/test/orm/test_bulk.py
index 79de19f68..611a63adf 100644
--- a/test/orm/test_bulk.py
+++ b/test/orm/test_bulk.py
@@ -436,8 +436,15 @@ class BulkUDTestAltColKeys(BulkTest, fixtures.MappedTest):
             )
         )
 
-    def test_update_keys(self):
-        asserter = self._test_update(self.classes.PersonKeys)
+    @testing.combinations(
+        ("states",),
+        ("dicts",),
+    )
+    def test_update_keys(self, type_):
+        if type_ == "states":
+            asserter = self._test_update_states(self.classes.PersonKeys)
+        else:
+            asserter = self._test_update(self.classes.PersonKeys)
         asserter.assert_(
             CompiledSQL(
                 "UPDATE people_keys SET name=:personname "
@@ -446,9 +453,16 @@ class BulkUDTestAltColKeys(BulkTest, fixtures.MappedTest):
             )
         )
 
+    @testing.combinations(
+        ("states",),
+        ("dicts",),
+    )
     @testing.requires.updateable_autoincrement_pks
-    def test_update_attrs(self):
-        asserter = self._test_update(self.classes.PersonAttrs)
+    def test_update_attrs(self, type_):
+        if type_ == "states":
+            asserter = self._test_update_states(self.classes.PersonAttrs)
+        else:
+            asserter = self._test_update(self.classes.PersonAttrs)
         asserter.assert_(
             CompiledSQL(
                 "UPDATE people_attrs SET name=:name "
@@ -499,6 +513,22 @@ class BulkUDTestAltColKeys(BulkTest, fixtures.MappedTest):
 
         return asserter
 
+    def _test_update_states(self, person_cls):
+        Person = person_cls
+
+        s = Session()
+        s.add(Person(id=5, personname="thename"))
+        s.commit()
+
+        p = s.query(Person).get(5)
+        with self.sql_execution_asserter(testing.db) as asserter:
+            p.personname = "newname"
+            s.bulk_save_objects([p])
+
+        eq_(s.query(Person).first(), Person(id=5, personname="newname"))
+
+        return asserter
+
 
 class BulkInheritanceTest(BulkTest, fixtures.MappedTest):
     @classmethod