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

13 files changed, 162 insertions, 26 deletions

diff --git a/doc/build/changelog/unreleased_14/5226.rst b/doc/build/changelog/unreleased_14/5226.rst
new file mode 100644
index 000000000..1436d7b18
--- /dev/null
+++ b/doc/build/changelog/unreleased_14/5226.rst
@@ -0,0 +1,17 @@
+.. change::
+    :tags: bug, orm
+    :tickets: 5226
+
+    The refresh of an expired object will now trigger an autoflush if the list
+    of expired attributes include one or more attributes that were explicitly
+    expired or refreshed using the :meth:`.Session.expire` or
+    :meth:`.Session.refresh` methods.   This is an attempt to find a middle
+    ground between the normal unexpiry of attributes that can happen in many
+    cases where autoflush is not desirable, vs. the case where attributes are
+    being explicitly expired or refreshed and it is possible that these
+    attributes depend upon other pending state within the session that needs to
+    be flushed.   The two methods now also gain a new flag
+    :paramref:`.Session.expire.autoflush` and
+    :paramref:`.Session.refresh.autoflush`, defaulting to True; when set to
+    False, this will disable the autoflush that occurs on unexpire for these
+    attributes.
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,
diff --git a/test/ext/test_extendedattr.py b/test/ext/test_extendedattr.py
index df9d2f9d5..ad9bf0bc0 100644
--- a/test/ext/test_extendedattr.py
+++ b/test/ext/test_extendedattr.py
@@ -223,7 +223,7 @@ class UserDefinedExtensionTest(_ExtBase, fixtures.ORMTest):
 
             data = {"a": "this is a", "b": 12}
 
-            def loader(state, keys):
+            def loader(state, keys, passive):
                 for k in keys:
                     state.dict[k] = data[k]
                 return attributes.ATTR_WAS_SET
diff --git a/test/orm/test_attributes.py b/test/orm/test_attributes.py
index 347dd4e46..bb3399a5a 100644
--- a/test/orm/test_attributes.py
+++ b/test/orm/test_attributes.py
@@ -442,7 +442,7 @@ class AttributesTest(fixtures.ORMTest):
 
         data = {"a": "this is a", "b": 12}
 
-        def loader(state, keys):
+        def loader(state, keys, passive):
             for k in keys:
                 state.dict[k] = data[k]
             return attributes.ATTR_WAS_SET
@@ -488,7 +488,7 @@ class AttributesTest(fixtures.ORMTest):
     def test_deferred_pickleable(self):
         data = {"a": "this is a", "b": 12}
 
-        def loader(state, keys):
+        def loader(state, keys, passive):
             for k in keys:
                 state.dict[k] = data[k]
             return attributes.ATTR_WAS_SET
@@ -2242,7 +2242,7 @@ class HistoryTest(fixtures.TestBase):
         state.dict.pop("someattr", None)
         state.expired_attributes.add("someattr")
 
-        def scalar_loader(state, toload):
+        def scalar_loader(state, toload, passive):
             state.dict["someattr"] = "one"
 
         state.manager.expired_attribute_loader = scalar_loader
diff --git a/test/orm/test_cascade.py b/test/orm/test_cascade.py
index cfc3ad38f..815df3620 100644
--- a/test/orm/test_cascade.py
+++ b/test/orm/test_cascade.py
@@ -842,6 +842,17 @@ class O2OSingleParentNoFlushTest(fixtures.MappedTest):
         sess.add(u1)
         sess.commit()
 
+        # in this case, u1.address has active history set, because
+        # this operation necessarily replaces the old object which must be
+        # loaded.
+        # the set operation requires that "u1" is unexpired, because the
+        # replace operation wants to load the
+        # previous value.  The original test case for #2921 only included
+        # that the lazyload operation passed a no autoflush flag through
+        # to the operation, however in #5226 this has been enhanced to pass
+        # the no autoflush flag down through to the unexpire of the attributes
+        # as well, so that attribute unexpire can otherwise invoke autoflush.
+        assert "id" not in u1.__dict__
         a2 = Address(email_address="asdf")
         sess.add(a2)
         u1.address = a2
diff --git a/test/orm/test_deferred.py b/test/orm/test_deferred.py
index a7957ec28..a02ee250c 100644
--- a/test/orm/test_deferred.py
+++ b/test/orm/test_deferred.py
@@ -1,6 +1,8 @@
 import sqlalchemy as sa
 from sqlalchemy import ForeignKey
+from sqlalchemy import func
 from sqlalchemy import Integer
+from sqlalchemy import select
 from sqlalchemy import testing
 from sqlalchemy import util
 from sqlalchemy.orm import aliased
@@ -2023,3 +2025,46 @@ class RaiseLoadTest(fixtures.DeclarativeMappedTest):
         eq_(a1.id, 1)
 
         assert "x" in a1.__dict__
+
+
+class AutoflushTest(fixtures.DeclarativeMappedTest):
+    @classmethod
+    def setup_classes(cls):
+        Base = cls.DeclarativeBasic
+
+        class A(Base):
+            __tablename__ = "a"
+
+            id = Column(Integer, primary_key=True)
+            bs = relationship("B")
+
+        class B(Base):
+            __tablename__ = "b"
+            id = Column(Integer, primary_key=True)
+            a_id = Column(ForeignKey("a.id"))
+
+        A.b_count = deferred(
+            select([func.count(1)]).where(A.id == B.a_id).scalar_subquery()
+        )
+
+    def test_deferred_autoflushes(self):
+        A, B = self.classes("A", "B")
+
+        s = Session()
+
+        a1 = A(id=1, bs=[B()])
+        s.add(a1)
+        s.commit()
+
+        eq_(a1.b_count, 1)
+        s.close()
+
+        a1 = s.query(A).first()
+        assert "b_count" not in a1.__dict__
+
+        b1 = B(a_id=1)
+        s.add(b1)
+
+        eq_(a1.b_count, 2)
+
+        assert b1 in s
diff --git a/test/orm/test_dynamic.py b/test/orm/test_dynamic.py
index 94ecf4ee2..1ca1bec03 100644
--- a/test/orm/test_dynamic.py
+++ b/test/orm/test_dynamic.py
@@ -2,6 +2,7 @@ from sqlalchemy import cast
 from sqlalchemy import desc
 from sqlalchemy import exc
 from sqlalchemy import func
+from sqlalchemy import inspect
 from sqlalchemy import Integer
 from sqlalchemy import select
 from sqlalchemy import testing
@@ -969,17 +970,25 @@ class HistoryTest(_DynamicFixture, _fixtures.FixtureTest):
         elif isinstance(obj, self.classes.Order):
             attrname = "items"
 
-        eq_(attributes.get_history(obj, attrname), compare)
+        sess = inspect(obj).session
 
-        if compare_passive is None:
-            compare_passive = compare
+        if sess:
+            sess.autoflush = False
+        try:
+            eq_(attributes.get_history(obj, attrname), compare)
 
-        eq_(
-            attributes.get_history(
-                obj, attrname, attributes.LOAD_AGAINST_COMMITTED
-            ),
-            compare_passive,
-        )
+            if compare_passive is None:
+                compare_passive = compare
+
+            eq_(
+                attributes.get_history(
+                    obj, attrname, attributes.LOAD_AGAINST_COMMITTED
+                ),
+                compare_passive,
+            )
+        finally:
+            if sess:
+                sess.autoflush = True
 
     def test_append_transient(self):
         u1, a1 = self._transient_fixture()
diff --git a/test/orm/test_expire.py b/test/orm/test_expire.py
index 083c2f465..127380fad 100644
--- a/test/orm/test_expire.py
+++ b/test/orm/test_expire.py
@@ -82,6 +82,24 @@ class ExpireTest(_fixtures.FixtureTest):
 
         self.assert_sql_count(testing.db, go, 0)
 
+    def test_expire_autoflush(self):
+        User, users = self.classes.User, self.tables.users
+        Address, addresses = self.classes.Address, self.tables.addresses
+
+        mapper(User, users)
+        mapper(Address, addresses, properties={"user": relationship(User)})
+
+        s = Session()
+
+        a1 = s.query(Address).get(2)
+        u1 = s.query(User).get(7)
+        a1.user = u1
+
+        s.expire(a1, ["user_id"])
+
+        # autoflushes
+        eq_(a1.user_id, 7)
+
     def test_persistence_check(self):
         users, User = self.tables.users, self.classes.User
 
@@ -1748,6 +1766,24 @@ class RefreshTest(_fixtures.FixtureTest):
             lambda: s.refresh(u),
         )
 
+    def test_refresh_autoflush(self):
+        User, users = self.classes.User, self.tables.users
+        Address, addresses = self.classes.Address, self.tables.addresses
+
+        mapper(User, users)
+        mapper(Address, addresses, properties={"user": relationship(User)})
+
+        s = Session()
+
+        a1 = s.query(Address).get(2)
+        u1 = s.query(User).get(7)
+        a1.user = u1
+
+        s.refresh(a1, ["user_id"])
+
+        # autoflushes
+        eq_(a1.user_id, 7)
+
     def test_refresh_expired(self):
         User, users = self.classes.User, self.tables.users
 
diff --git a/test/orm/test_load_on_fks.py b/test/orm/test_load_on_fks.py
index 6e9dde16b..0e8ac97e3 100644
--- a/test/orm/test_load_on_fks.py
+++ b/test/orm/test_load_on_fks.py
@@ -176,6 +176,9 @@ class LoadOnFKsTest(AssertsExecutionResults, fixtures.TestBase):
         assert c3 in p1.children
 
     def test_autoflush_on_pending(self):
+        # ensure p1.id is not expired
+        p1.id
+
         c3 = Child()
         sess.add(c3)
         c3.parent_id = p1.id
@@ -184,6 +187,9 @@ class LoadOnFKsTest(AssertsExecutionResults, fixtures.TestBase):
         assert c3.parent is None
 
     def test_autoflush_load_on_pending_on_pending(self):
+        # ensure p1.id is not expired
+        p1.id
+
         Child.parent.property.load_on_pending = True
         c3 = Child()
         sess.add(c3)
@@ -305,6 +311,10 @@ class LoadOnFKsTest(AssertsExecutionResults, fixtures.TestBase):
                 for manualflush in (False, True):
                     Child.parent.property.load_on_pending = loadonpending
                     sess.autoflush = autoflush
+
+                    # ensure p2.id not expired
+                    p2.id
+
                     c2 = Child()
                     sess.add(c2)
                     c2.parent_id = p2.id
diff --git a/test/orm/test_versioning.py b/test/orm/test_versioning.py
index c6418745d..1c540145b 100644
--- a/test/orm/test_versioning.py
+++ b/test/orm/test_versioning.py
@@ -139,11 +139,11 @@ class NullVersionIdTest(fixtures.MappedTest):
         # you should get a FlushError on update.
 
         f1.value = "f1rev2"
-        f1.version_id = None
 
         with conditional_sane_rowcount_warnings(
             update=True, only_returning=True
         ):
+            f1.version_id = None
             assert_raises_message(
                 sa.orm.exc.FlushError,
                 "Instance does not contain a non-NULL version value",
@@ -1973,10 +1973,9 @@ class VersioningMappedSelectTest(fixtures.MappedTest):
 
         s1.expire_all()
 
-        f1.value = "f2"
-        f1.version_id = 2
-
         with conditional_sane_rowcount_warnings(
             update=True, only_returning=True
         ):
+            f1.value = "f2"
+            f1.version_id = 2
             s1.flush()