run postfetch_post_update for version_id_col even if delete

Fixed issue where using :meth:`_orm.Session.delete` along with the
:paramref:`_orm.Mapper.version_id_col` feature would fail to use the
correct version identifier in the case that an additional UPDATE were
emitted against the target object as a result of the use of
:paramref:`_orm.relationship.post_update` on the object.  The issue is
similar to 🎫`10800` just fixed in version 2.0.25 for the case of
updates alone.

Fixes: #10967
Change-Id: I959e9a2cc3e750e86e8de7b12b28ee1e819ed6d8

doc/build/changelog/unreleased_20/10967.rst

@@ -0,0 +1,11 @@
+.. change::
+    :tags: bug, orm
+    :tickets: 10967
+
+    Fixed issue where using :meth:`_orm.Session.delete` along with the
+    :paramref:`_orm.Mapper.version_id_col` feature would fail to use the
+    correct version identifier in the case that an additional UPDATE were
+    emitted against the target object as a result of the use of
+    :paramref:`_orm.relationship.post_update` on the object.  The issue is
+    similar to :ticket:`10800` just fixed in version 2.0.25 for the case of
+    updates alone.

lib/sqlalchemy/orm/persistence.py

@@ -1570,16 +1570,25 @@ def _finalize_insert_update_commands(base_mapper, uowtransaction, states):
 def _postfetch_post_update(
     mapper, uowtransaction, table, state, dict_, result, params
 ):
-    if uowtransaction.is_deleted(state):
-        return
-
-    prefetch_cols = result.context.compiled.prefetch
-    postfetch_cols = result.context.compiled.postfetch
-
-    if (
+    needs_version_id = (
         mapper.version_id_col is not None
         and mapper.version_id_col in mapper._cols_by_table[table]
-    ):
+    )
+
+    if not uowtransaction.is_deleted(state):
+        # post updating after a regular INSERT or UPDATE, do a full postfetch
+        prefetch_cols = result.context.compiled.prefetch
+        postfetch_cols = result.context.compiled.postfetch
+    elif needs_version_id:
+        # post updating before a DELETE with a version_id_col, need to
+        # postfetch just version_id_col
+        prefetch_cols = postfetch_cols = ()
+    else:
+        # post updating before a DELETE without a version_id_col,
+        # don't need to postfetch
+        return
+
+    if needs_version_id:
         prefetch_cols = list(prefetch_cols) + [mapper.version_id_col]
 
     refresh_flush = bool(mapper.class_manager.dispatch.refresh_flush)

test/orm/test_versioning.py

@@ -2032,8 +2032,6 @@ class QuotedBindVersioningTest(fixtures.MappedTest):
 
 
 class PostUpdateVersioningTest(fixtures.DeclarativeMappedTest):
-    """test for #10800"""
-
     @classmethod
     def setup_classes(cls):
         Base = cls.DeclarativeBasic
@@ -2063,7 +2061,8 @@ class PostUpdateVersioningTest(fixtures.DeclarativeMappedTest):
                 "version_id_col": version_id,
             }
 
-    def test_bumped_version_id(self):
+    def test_bumped_version_id_on_update(self):
+        """test for #10800"""
         User, Parent = self.classes("User", "Parent")
 
         session = fixture_session()
@@ -2115,3 +2114,42 @@ class PostUpdateVersioningTest(fixtures.DeclarativeMappedTest):
                 ],
             ),
         )
+
+    def test_bumped_version_id_on_delete(self):
+        """test for #10967"""
+
+        User, Parent = self.classes("User", "Parent")
+
+        session = fixture_session()
+        u1 = User(id=1)
+        p1 = Parent(id=1, updated_by=u1)
+        session.add(u1)
+        session.add(p1)
+
+        session.flush()
+
+        session.delete(p1)
+
+        with self.sql_execution_asserter(testing.db) as asserter:
+            session.commit()
+
+        asserter.assert_(
+            CompiledSQL(
+                "UPDATE parent SET version_id=:version_id, "
+                "updated_by_id=:updated_by_id WHERE parent.id = :parent_id "
+                "AND parent.version_id = :parent_version_id",
+                [
+                    {
+                        "version_id": 2,
+                        "updated_by_id": None,
+                        "parent_id": 1,
+                        "parent_version_id": 1,
+                    }
+                ],
+            ),
+            CompiledSQL(
+                "DELETE FROM parent WHERE parent.id = :id AND "
+                "parent.version_id = :version_id",
+                [{"id": 1, "version_id": 2}],
+            ),
+        )