Run autoflush for column attribute load operations

The "autoflush" behavior of :class:`.Query` will now trigger for nearly
all ORM level attribute load operations, including when a deferred
column is loaded as well as when an expired column is loaded.   Previously,
autoflush on load of expired or unloaded attributes was limited to
relationship-bound attributes only.   However, this led to the issue
where column-based attributes that also depended on other rows, or even
other columns in the same row, in order to express the correct value,
would show an effectively stale value when accessed as there could be
pending changes in the session left to be flushed.    Autoflush
is now disabled only in some cases where attributes are being unexpired in
the context of a history operation.

Fixes: #5226
Change-Id: Ibd965b30918cd273ae020411a704bf2bb1891f59

4 files changed, 18 insertions, 9 deletions

diff --git a/lib/sqlalchemy/orm/loading.py b/lib/sqlalchemy/orm/loading.py
index 49c71e5b2..b7ef96e1a 100644
--- a/lib/sqlalchemy/orm/loading.py
+++ b/lib/sqlalchemy/orm/loading.py
@@ -194,7 +194,12 @@ def get_from_identity(session, mapper, key, passive):
 
 
 def load_on_ident(
-    query, key, refresh_state=None, with_for_update=None, only_load_props=None
+    query,
+    key,
+    refresh_state=None,
+    with_for_update=None,
+    only_load_props=None,
+    no_autoflush=False,
 ):
     """Load the given identity key from the database."""
     if key is not None:
@@ -203,6 +208,9 @@ def load_on_ident(
     else:
         ident = identity_token = None
 
+    if no_autoflush:
+        query = query.autoflush(False)
+
     return load_on_pk_identity(
         query,
         ident,
@@ -992,7 +1000,7 @@ class PostLoad(object):
         pl.loaders[token] = (token, limit_to_mapper, loader_callable, arg, kw)
 
 
-def load_scalar_attributes(mapper, state, attribute_names):
+def load_scalar_attributes(mapper, state, attribute_names, passive):
     """initiate a column-based attribute refresh operation."""
 
     # assert mapper is _state_mapper(state)
@@ -1007,6 +1015,8 @@ def load_scalar_attributes(mapper, state, attribute_names):
 
     result = False
 
+    no_autoflush = passive & attributes.NO_AUTOFLUSH
+
     # in the case of inheritance, particularly concrete and abstract
     # concrete inheritance, the class manager might have some keys
     # of attributes on the superclass that we didn't actually map.
@@ -1031,6 +1041,7 @@ def load_scalar_attributes(mapper, state, attribute_names):
                 None,
                 only_load_props=attribute_names,
                 refresh_state=state,
+                no_autoflush=no_autoflush,
             )
 
     if result is False:
@@ -1068,6 +1079,7 @@ def load_scalar_attributes(mapper, state, attribute_names):
             identity_key,
             refresh_state=state,
             only_load_props=attribute_names,
+            no_autoflush=no_autoflush,
         )
 
     # if instance is pending, a refresh operation
diff --git a/lib/sqlalchemy/orm/query.py b/lib/sqlalchemy/orm/query.py
index 4ba82d1e8..d001ab983 100644
--- a/lib/sqlalchemy/orm/query.py
+++ b/lib/sqlalchemy/orm/query.py
@@ -3323,7 +3323,7 @@ class Query(Generative):
     def __iter__(self):
         context = self._compile_context()
         context.statement.label_style = LABEL_STYLE_TABLENAME_PLUS_COL
-        if self._autoflush and not self._populate_existing:
+        if self._autoflush:
             self.session._autoflush()
         return self._execute_and_instances(context)
 
diff --git a/lib/sqlalchemy/orm/session.py b/lib/sqlalchemy/orm/session.py
index c773aeb08..aa55fab58 100644
--- a/lib/sqlalchemy/orm/session.py
+++ b/lib/sqlalchemy/orm/session.py
@@ -1663,9 +1663,7 @@ class Session(_SessionClassMethods):
                 )
                 util.raise_(e, with_traceback=sys.exc_info()[2])
 
-    def refresh(
-        self, instance, attribute_names=None, with_for_update=None,
-    ):
+    def refresh(self, instance, attribute_names=None, with_for_update=None):
         """Expire and refresh the attributes on the given instance.
 
         A query will be issued to the database and all attributes will be
@@ -1834,7 +1832,7 @@ class Session(_SessionClassMethods):
             for o, m, st_, dct_ in cascaded:
                 self._conditional_expire(st_)
 
-    def _conditional_expire(self, state):
+    def _conditional_expire(self, state, autoflush=None):
         """Expire a state if persistent, else expunge if pending"""
 
         if state.key:
diff --git a/lib/sqlalchemy/orm/state.py b/lib/sqlalchemy/orm/state.py
index 91bf57ab9..5a885b118 100644
--- a/lib/sqlalchemy/orm/state.py
+++ b/lib/sqlalchemy/orm/state.py
@@ -570,7 +570,6 @@ class InstanceState(interfaces.InspectionAttrInfo):
 
     def _expire(self, dict_, modified_set):
         self.expired = True
-
         if self.modified:
             modified_set.discard(self)
             self.committed_state.clear()
@@ -665,7 +664,7 @@ class InstanceState(interfaces.InspectionAttrInfo):
             if not self.manager[attr].impl.load_on_unexpire
         )
 
-        self.manager.expired_attribute_loader(self, toload)
+        self.manager.expired_attribute_loader(self, toload, passive)
 
         # if the loader failed, or this
         # instance state didn't have an identity,