add explicit step to set populate_existing for bulk insert

Fixed issue in new :ref:`orm_queryguide_upsert_returning` feature where the
``populate_existing`` execution option was not being propagated to the
loading option, preventing existing attributes from being refreshed
in-place.

Fixes: #9746
Change-Id: I3efcab644e2b5874c6b265d5313f353c051db629

3 files changed, 91 insertions, 1 deletions

diff --git a/doc/build/changelog/unreleased_20/9746.rst b/doc/build/changelog/unreleased_20/9746.rst
new file mode 100644
index 000000000..55d57925e
--- /dev/null
+++ b/doc/build/changelog/unreleased_20/9746.rst
@@ -0,0 +1,8 @@
+.. change::
+    :tags: bug, orm
+    :tickets: 9746
+
+    Fixed issue in new :ref:`orm_queryguide_upsert_returning` feature where the
+    ``populate_existing`` execution option was not being propagated to the
+    loading option, preventing existing attributes from being refreshed
+    in-place.
diff --git a/lib/sqlalchemy/orm/bulk_persistence.py b/lib/sqlalchemy/orm/bulk_persistence.py
index cb416d69e..257d71db4 100644
--- a/lib/sqlalchemy/orm/bulk_persistence.py
+++ b/lib/sqlalchemy/orm/bulk_persistence.py
@@ -586,6 +586,7 @@ class ORMDMLState(AbstractORMCompileState):
             load_options = execution_options.get(
                 "_sa_orm_load_options", QueryContext.default_load_options
             )
+
             querycontext = QueryContext(
                 compile_state.from_statement_ctx,
                 compile_state.select_statement,
@@ -1140,6 +1141,7 @@ class BulkORMInsert(ORMDMLState, InsertDMLState):
         _return_defaults: bool = False
         _subject_mapper: Optional[Mapper[Any]] = None
         _autoflush: bool = True
+        _populate_existing: bool = False
 
     select_statement: Optional[FromStatement] = None
 
@@ -1159,7 +1161,7 @@ class BulkORMInsert(ORMDMLState, InsertDMLState):
             execution_options,
         ) = BulkORMInsert.default_insert_options.from_execution_options(
             "_sa_orm_insert_options",
-            {"dml_strategy", "autoflush"},
+            {"dml_strategy", "autoflush", "populate_existing"},
             execution_options,
             statement._execution_options,
         )
@@ -1284,6 +1286,15 @@ class BulkORMInsert(ORMDMLState, InsertDMLState):
         if not bool(statement._returning):
             return result
 
+        if insert_options._populate_existing:
+            load_options = execution_options.get(
+                "_sa_orm_load_options", QueryContext.default_load_options
+            )
+            load_options += {"_populate_existing": True}
+            execution_options = execution_options.union(
+                {"_sa_orm_load_options": load_options}
+            )
+
         return cls._return_orm_returning(
             session,
             statement,
diff --git a/test/orm/dml/test_bulk_statements.py b/test/orm/dml/test_bulk_statements.py
index ab03b251d..af50ea045 100644
--- a/test/orm/dml/test_bulk_statements.py
+++ b/test/orm/dml/test_bulk_statements.py
@@ -302,6 +302,77 @@ class InsertStmtTest(testing.AssertsExecutionResults, fixtures.TestBase):
             else:
                 eq_(result.first(), (10, expected_qs[0]))
 
+    @testing.variation("populate_existing", [True, False])
+    @testing.requires.provisioned_upsert
+    def test_upsert_populate_existing(self, decl_base, populate_existing):
+        """test #9742"""
+
+        class Employee(ComparableEntity, decl_base):
+            __tablename__ = "employee"
+
+            uuid: Mapped[uuid.UUID] = mapped_column(primary_key=True)
+            user_name: Mapped[str] = mapped_column(nullable=False)
+
+        decl_base.metadata.create_all(testing.db)
+        s = fixture_session()
+
+        uuid1 = uuid.uuid4()
+        uuid2 = uuid.uuid4()
+        e1 = Employee(uuid=uuid1, user_name="e1 old name")
+        e2 = Employee(uuid=uuid2, user_name="e2 old name")
+        s.add_all([e1, e2])
+        s.flush()
+
+        stmt = provision.upsert(
+            config,
+            Employee,
+            (Employee,),
+            set_lambda=lambda inserted: {"user_name": inserted.user_name},
+        ).values(
+            [
+                dict(uuid=uuid1, user_name="e1 new name"),
+                dict(uuid=uuid2, user_name="e2 new name"),
+            ]
+        )
+        if populate_existing:
+            rows = s.scalars(
+                stmt, execution_options={"populate_existing": True}
+            )
+            # SPECIAL: before we actually receive the returning rows,
+            # the existing objects have not been updated yet
+            eq_(e1.user_name, "e1 old name")
+            eq_(e2.user_name, "e2 old name")
+
+            eq_(
+                set(rows),
+                {
+                    Employee(uuid=uuid1, user_name="e1 new name"),
+                    Employee(uuid=uuid2, user_name="e2 new name"),
+                },
+            )
+
+            # now they are updated
+            eq_(e1.user_name, "e1 new name")
+            eq_(e2.user_name, "e2 new name")
+        else:
+            # no populate existing
+            rows = s.scalars(stmt)
+            eq_(e1.user_name, "e1 old name")
+            eq_(e2.user_name, "e2 old name")
+            eq_(
+                set(rows),
+                {
+                    Employee(uuid=uuid1, user_name="e1 old name"),
+                    Employee(uuid=uuid2, user_name="e2 old name"),
+                },
+            )
+            eq_(e1.user_name, "e1 old name")
+            eq_(e2.user_name, "e2 old name")
+        s.commit()
+        s.expire_all()
+        eq_(e1.user_name, "e1 new name")
+        eq_(e2.user_name, "e2 new name")
+
 
 class UpdateStmtTest(fixtures.TestBase):
     __backend__ = True