Fix ORM UPDATE..RETURNING cache adaptation with synchronize_session=fetch

Fixed result-metadata corruption affecting ORM-enabled UPDATE statements
that use .returning() together with synchronize_session="fetch".  In this
configuration the RETURNING clause is rendered in mapper/table column order
rather than the user-requested .returning() order; on a compiled-cache hit,
the cursor result metadata was positionally adapted against the cached
statement's user-requested column order, causing column values to be
returned under the wrong keys (e.g. row[T.a] returning T.b's value).

The mis-adaptation was frequently masked by cached row-getter functions
produced by a prior, correctly-adapted execution, so the wrong values
surfaced most reliably under concurrent execution racing on the shared
compiled cache.  The fix disables result-set adapt_to_context for this
path, mirroring the ORM INSERT and ORM SELECT load paths, since the cached
keymap already carries the correct Column objects.

ORM DELETE statements are not affected: DELETE does not pass through
crud._get_crud_params() and therefore retains the user-requested RETURNING
order; a non-regression DELETE case is included in the test.

Fixes: #13439
Change-Id: If58e93a0bb3220dc5637bc7bcf346d2177529d6d

diff --git a/doc/build/changelog/unreleased_20/13439.rst b/doc/build/changelog/unreleased_20/13439.rst
new file mode 100644 (file)
index 0000000..adfe2b9
--- /dev/null
+++ b/doc/build/changelog/unreleased_20/13439.rst
@@ -0,0 +1,11 @@
+.. change::
+    :tags: bug, orm
+    :tickets: 13439
+
+    Fixed a result-column misalignment bug in ORM-enabled UPDATE statements
+    where ``synchronize_session="fetch"`` is in use, either explicitly or
+    because the statement uses constructs such as CTEs that implicitly select
+    for it.  Columns in rows returned by ``.returning()`` could be returned
+    under incorrect keys (e.g. ``row[SomeClass.a]`` returning the value of
+    a different column), a problem most likely to manifest under concurrent
+    workloads.  ORM DELETE statements were not affected.

diff --git a/lib/sqlalchemy/orm/bulk_persistence.py b/lib/sqlalchemy/orm/bulk_persistence.py
index 11d3e8c9b53b7d31c6cf87c64bc3f3d350d5a3b6..46b30f8d778087de1bb97bebac58f3429bb0cf31 100644 (file)
--- a/lib/sqlalchemy/orm/bulk_persistence.py
+++ b/lib/sqlalchemy/orm/bulk_persistence.py
@@ -804,6 +804,22 @@ class _BulkUDCompileState(_ORMDMLState):
                 }
             )
 
+            # disable result-level adapt_to_context.  ORM UPDATE/DELETE with
+            # RETURNING uses a "two level" statement: the invoked ORM statement
+            # holds the user's returning() columns while the cached Core
+            # statement returns primary key + supplemental columns, so the two
+            # are not positionally aligned.  adapt_to_context() would remap
+            # result keys positionally between them and mislabel columns on a
+            # cache hit; the ORM instead interprets rows via its own
+            # from_statement context, so the Core adaptation would interfere
+            # here.
+            if not execution_options:
+                execution_options = context._orm_load_exec_options
+            else:
+                execution_options = execution_options.union(
+                    context._orm_load_exec_options
+                )
+
         return (
             statement,
             util.immutabledict(execution_options).union(

diff --git a/test/orm/dml/test_orm_upd_del_assorted.py b/test/orm/dml/test_orm_upd_del_assorted.py
index cd1bcf00467c0d9056f43d39181a2676751a89d0..7cc5b8ff7026bc67f0a17fc2d903489fb4928c5e 100644 (file)
--- a/test/orm/dml/test_orm_upd_del_assorted.py
+++ b/test/orm/dml/test_orm_upd_del_assorted.py
@@ -15,6 +15,7 @@ from sqlalchemy import select
 from sqlalchemy import String
 from sqlalchemy import testing
 from sqlalchemy import update
+from sqlalchemy.engine.interfaces import CacheStats
 from sqlalchemy.orm import Mapped
 from sqlalchemy.orm import mapped_column
 from sqlalchemy.orm import relationship
@@ -439,6 +440,120 @@ class PGIssue11849Test(fixtures.DeclarativeMappedTest):
         eq_(obj.test_field, {"test1": 1, "test2": "2", "test3": {"test4": 4}})
 
 
+class CacheAdaptReturningOrderTest(fixtures.DeclarativeMappedTest):
+    """test for #13439"""
+
+    __sparse_driver_backend__ = True
+
+    @classmethod
+    def setup_classes(cls):
+        Base = cls.DeclarativeBasic
+
+        class T(Base):
+            __tablename__ = "t"
+
+            # declared column order: id, b, a   (deliberately != the
+            # .returning() order below, which is id, a, b)
+            id: Mapped[int] = mapped_column(Integer, primary_key=True)
+            b: Mapped[int] = mapped_column(Integer)
+            a: Mapped[int] = mapped_column(Integer)
+
+    @classmethod
+    def insert_data(cls, connection):
+        t = cls.classes.T.__table__
+        connection.execute(
+            t.insert(),
+            [dict(id=i, a=i * 10, b=i * 100) for i in range(1, 5)],
+        )
+
+    @testing.variation(
+        "dml",
+        [
+            ("update", testing.requires.update_returning),
+            ("delete", testing.requires.delete_returning),
+        ],
+    )
+    def test_returning_keymap_stable_across_cache_hit(
+        self, dml: testing.Variation, connection
+    ):
+        T = self.classes.T
+
+        # returning order id, a, b deliberately != mapper order (id, b, a)
+        def stmt(ident):
+            if dml.update:
+                return (
+                    update(T)
+                    .where(T.id == ident)
+                    .values(b=999)
+                    .returning(T.id, T.a, T.b)
+                )
+            elif dml.delete:
+                return delete(T).where(T.id == ident).returning(T.id, T.a, T.b)
+            else:
+                dml.fail()
+
+        opts = {"synchronize_session": "fetch"}
+
+        # execute twice with different parameter values: the first call
+        # populates the compiled cache, the second is a cache hit (same
+        # statement shape, new statement object).  assert both the cache
+        # status and the cursor keymap on each call.
+        for n, ident in enumerate((1, 2), 1):
+            with Session(connection) as sess:
+                result = sess.execute(stmt(ident), execution_options=opts)
+
+                is_hit = n > 1
+                eq_(
+                    result.raw.context.cache_hit,
+                    (
+                        CacheStats.CACHE_HIT
+                        if is_hit
+                        else CacheStats.CACHE_MISS
+                    ),
+                    f"execution {n} expected cache "
+                    f"{'HIT' if is_hit else 'MISS'}",
+                )
+
+                # the cursor-level result metadata must map each requested
+                # returning column to the same physical index whether it
+                # is looked up by Column object or by string name; a
+                # divergence means Column-object row access (row[T.a])
+                # would read a different column's value than string access
+                # (row["a"]).  the cursor keymap is keyed by the raw
+                # ``Column`` objects of the mapped table, so use those
+                # rather than the mapped ``InstrumentedAttribute``s.
+                raw_meta = result.raw._metadata
+                tc = T.__table__.c
+                for col, name in [
+                    (tc.id, "id"),
+                    (tc.a, "a"),
+                    (tc.b, "b"),
+                ]:
+                    by_obj = raw_meta._index_for_key(col, False)
+                    by_name = raw_meta._index_for_key(name, False)
+                    eq_(
+                        by_obj,
+                        by_name,
+                        f"execution {n} (cache "
+                        f"{'HIT' if is_hit else 'MISS'}): column "
+                        f"{name!r} resolves to index {by_obj} by Column "
+                        f"object but {by_name} by string name",
+                    )
+
+                # end-to-end: values must be correct for both string and
+                # Column-object access.  for UPDATE, b was set to 999;
+                # for DELETE, RETURNING yields the original row value.
+                expected_b = 999 if dml.update else ident * 100
+                row = result.mappings().first()
+                assert row
+                eq_(row["id"], ident)
+                eq_(row[T.id], ident)
+                eq_(row["a"], ident * 10)
+                eq_(row[T.a], ident * 10)
+                eq_(row["b"], expected_b)
+                eq_(row[T.b], expected_b)
+
+
 class _FilterByDMLSuite(fixtures.MappedTest, AssertsCompiledSQL):
     """Base test suite for filter_by() on ORM DML statements.