Merge "dont add non-server-side cols to returning for versioning" into main

2 files changed, 27 insertions, 6 deletions

diff --git a/lib/sqlalchemy/orm/mapper.py b/lib/sqlalchemy/orm/mapper.py
index a3b209e4a..bb7e470ff 100644
--- a/lib/sqlalchemy/orm/mapper.py
+++ b/lib/sqlalchemy/orm/mapper.py
@@ -83,6 +83,7 @@ from ..sql import util as sql_util
 from ..sql import visitors
 from ..sql.cache_key import MemoizedHasCacheKey
 from ..sql.elements import KeyedColumnElement
+from ..sql.schema import Column
 from ..sql.schema import Table
 from ..sql.selectable import LABEL_STYLE_TABLENAME_PLUS_COL
 from ..util import HasMemoized
@@ -112,7 +113,6 @@ if TYPE_CHECKING:
     from ..sql.base import ReadOnlyColumnCollection
     from ..sql.elements import ColumnClause
     from ..sql.elements import ColumnElement
-    from ..sql.schema import Column
     from ..sql.selectable import FromClause
     from ..util import OrderedSet
 
@@ -2523,6 +2523,24 @@ class Mapper(
         return from_obj
 
     @HasMemoized.memoized_attribute
+    def _version_id_has_server_side_value(self) -> bool:
+        vid_col = self.version_id_col
+
+        if vid_col is None:
+            return False
+
+        elif not isinstance(vid_col, Column):
+            return True
+        else:
+            return vid_col.server_default is not None or (
+                vid_col.default is not None
+                and (
+                    not vid_col.default.is_scalar
+                    and not vid_col.default.is_callable
+                )
+            )
+
+    @HasMemoized.memoized_attribute
     def _single_table_criterion(self):
         if self.single and self.inherits and self.polymorphic_on is not None:
             return self.polymorphic_on._annotate(
diff --git a/lib/sqlalchemy/orm/persistence.py b/lib/sqlalchemy/orm/persistence.py
index cc7e321b4..b8368001b 100644
--- a/lib/sqlalchemy/orm/persistence.py
+++ b/lib/sqlalchemy/orm/persistence.py
@@ -801,7 +801,7 @@ def _emit_update_statements(
             )
             return_defaults = True
 
-        if mapper.version_id_col is not None:
+        if mapper._version_id_has_server_side_value:
             statement = statement.return_defaults(mapper.version_id_col)
             return_defaults = True
 
@@ -1268,13 +1268,16 @@ def _emit_post_update_statements(
 
         stmt = table.update().where(clauses)
 
-        if mapper.version_id_col is not None:
-            stmt = stmt.return_defaults(mapper.version_id_col)
-
         return stmt
 
     statement = base_mapper._memo(("post_update", table), update_stmt)
 
+    if mapper._version_id_has_server_side_value:
+        statement = statement.return_defaults(mapper.version_id_col)
+        return_defaults = True
+    else:
+        return_defaults = False
+
     # execute each UPDATE in the order according to the original
     # list of states to guarantee row access order, but
     # also group them into common (connection, cols) sets
@@ -1290,7 +1293,7 @@ def _emit_post_update_statements(
 
         assert_singlerow = (
             connection.dialect.supports_sane_rowcount
-            if mapper.version_id_col is None
+            if not return_defaults
             else connection.dialect.supports_sane_rowcount_returning
         )
         assert_multirow = (