Warn for object replaced in identity map during flush

A warning is emitted for a condition in which the :class:`.Session` may
implicitly swap an object out of the identity map for another one with the
same primary key, detaching the old one, which can be an observed result of
load operations which occur within the :meth:`.SessionEvents.after_flush`
hook.  The warning is intended to notify the user that some special
condition has caused this to happen and that the previous object may not be
in the expected state.

Fixes: #4890
Change-Id: Ide11c6b9f21ca67ff5a96266c521d0c56fd6af8d

4 files changed, 28 insertions, 3 deletions

diff --git a/lib/sqlalchemy/orm/events.py b/lib/sqlalchemy/orm/events.py
index 6cc0e2eba..778ad3378 100644
--- a/lib/sqlalchemy/orm/events.py
+++ b/lib/sqlalchemy/orm/events.py
@@ -1463,6 +1463,15 @@ class SessionEvents(event.Events):
         'dirty', and 'deleted' lists still show pre-flush state as well
         as the history settings on instance attributes.
 
+        .. warning:: This event runs after the :class:`.Session` has emitted
+           SQL to modify the database, but **before** it has altered its
+           internal state to reflect those changes, including that newly
+           inserted objects are placed into the identity map.  ORM operations
+           emitted within this event such as loads of related items
+           may produce new identity map entries that will immediately
+           be replaced, sometimes causing confusing results.  SQLAlchemy will
+           emit a warning for this condition as of version 1.3.9.
+
         :param session: The target :class:`.Session`.
         :param flush_context: Internal :class:`.UOWTransaction` object
          which handles the details of the flush.
diff --git a/lib/sqlalchemy/orm/identity.py b/lib/sqlalchemy/orm/identity.py
index 2da5d66b3..842b7915a 100644
--- a/lib/sqlalchemy/orm/identity.py
+++ b/lib/sqlalchemy/orm/identity.py
@@ -125,10 +125,13 @@ class WeakInstanceDict(IdentityMap):
                 if existing is not state:
                     self._manage_removed_state(existing)
                 else:
-                    return
+                    return None
+        else:
+            existing = None
 
         self._dict[state.key] = state
         self._manage_incoming_state(state)
+        return existing
 
     def add(self, state):
         key = state.key
@@ -297,9 +300,12 @@ class StrongInstanceDict(IdentityMap):
                 self._manage_removed_state(existing)
             else:
                 return
+        else:
+            existing = None
 
         self._dict[state.key] = state.obj()
         self._manage_incoming_state(state)
+        return existing
 
     def add(self, state):
         if state.key in self:
diff --git a/lib/sqlalchemy/orm/persistence.py b/lib/sqlalchemy/orm/persistence.py
index 68052dfdd..58c8bcc06 100644
--- a/lib/sqlalchemy/orm/persistence.py
+++ b/lib/sqlalchemy/orm/persistence.py
@@ -1084,7 +1084,6 @@ def _emit_insert_statements(
             multiparams = [rec[2] for rec in records]
 
             c = cached_connections[connection].execute(statement, multiparams)
-
             if bookkeeping:
                 for (
                     (
diff --git a/lib/sqlalchemy/orm/session.py b/lib/sqlalchemy/orm/session.py
index c345c4281..3aa392ecf 100644
--- a/lib/sqlalchemy/orm/session.py
+++ b/lib/sqlalchemy/orm/session.py
@@ -1935,7 +1935,18 @@ class Session(_SessionClassMethods):
                 # there can be an existing state in the identity map
                 # that is replaced when the primary keys of two instances
                 # are swapped; see test/orm/test_naturalpks.py -> test_reverse
-                self.identity_map.replace(state)
+                old = self.identity_map.replace(state)
+                if (
+                    old is not None
+                    and mapper._identity_key_from_state(old) == instance_key
+                    and old.obj() is not None
+                ):
+                    util.warn(
+                        "Identity map already had an identity for %s, "
+                        "replacing it with newly flushed object.   Are there "
+                        "load operations occurring inside of an event handler "
+                        "within the flush?" % (instance_key,)
+                    )
                 state._orphaned_outside_of_session = False
 
         statelib.InstanceState._commit_all_states(