disable all raiseload within the unit of work process.

The unit of work process now turns off all "lazy='raise'" behavior
altogether when a flush is proceeding.  While there are areas where the UOW
is sometimes loading things that aren't ultimately needed, the lazy="raise"
strategy is not helpful here as the user often does not have much control
or visibility into the flush process.

Fixes: #5984
Change-Id: I23f2e332a5faa5c7c29823c9be9434d129676a5a

6 files changed, 185 insertions, 2 deletions

diff --git a/doc/build/changelog/unreleased_14/5984.rst b/doc/build/changelog/unreleased_14/5984.rst
new file mode 100644
index 000000000..d3d48146c
--- /dev/null
+++ b/doc/build/changelog/unreleased_14/5984.rst
@@ -0,0 +1,10 @@
+.. change::
+    :tags: bug, orm
+    :tickets: 5984
+
+    The unit of work process now turns off all "lazy='raise'" behavior
+    altogether when a flush is proceeding.  While there are areas where the UOW
+    is sometimes loading things that aren't ultimately needed, the lazy="raise"
+    strategy is not helpful here as the user often does not have much control
+    or visibility into the flush process.
+
diff --git a/doc/build/orm/loading_relationships.rst b/doc/build/orm/loading_relationships.rst
index 8cc030863..770af73f0 100644
--- a/doc/build/orm/loading_relationships.rst
+++ b/doc/build/orm/loading_relationships.rst
@@ -337,6 +337,14 @@ The :func:`.raiseload` option applies only to relationship attributes.  For
 column-oriented attributes, the :func:`.defer` option supports the
 :paramref:`.orm.defer.raiseload` option which works in the same way.
 
+.. versionchanged:: 1.4.0 The "raiseload" strategies **do not take place**
+  within the unit of work flush process, as of SQLAlchemy 1.4.0. This means
+  that if the unit of work needs to load a particular attribute in order to
+  complete its work, it will perform the load. It's not always easy to prevent
+  a particular relationship load from occurring within the UOW process
+  particularly with less common kinds of relationships. The lazy="raise" case
+  is more intended for explicit attribute access within the application space.
+
 .. seealso::
 
     :ref:`wildcard_loader_strategies`
diff --git a/lib/sqlalchemy/orm/dependency.py b/lib/sqlalchemy/orm/dependency.py
index 5a329b28c..85f6b68ff 100644
--- a/lib/sqlalchemy/orm/dependency.py
+++ b/lib/sqlalchemy/orm/dependency.py
@@ -229,6 +229,11 @@ class DependencyProcessor(object):
         if not isdelete or self.passive_deletes:
             passive = attributes.PASSIVE_NO_INITIALIZE
         elif self.direction is MANYTOONE:
+            # here, we were hoping to optimize having to fetch many-to-one
+            # for history and ignore it, if there's no further cascades
+            # to take place.  however there are too many less common conditions
+            # that still take place and tests in test_relationships /
+            # test_cascade etc. will still fail.
             passive = attributes.PASSIVE_NO_FETCH_RELATED
         else:
             passive = attributes.PASSIVE_OFF
diff --git a/lib/sqlalchemy/orm/unitofwork.py b/lib/sqlalchemy/orm/unitofwork.py
index c293d90cb..1fb1c10ac 100644
--- a/lib/sqlalchemy/orm/unitofwork.py
+++ b/lib/sqlalchemy/orm/unitofwork.py
@@ -254,7 +254,9 @@ class UOWTransaction(object):
                 history = impl.get_history(
                     state,
                     state.dict,
-                    attributes.PASSIVE_OFF | attributes.LOAD_AGAINST_COMMITTED,
+                    attributes.PASSIVE_OFF
+                    | attributes.LOAD_AGAINST_COMMITTED
+                    | attributes.NO_RAISE,
                 )
                 if history and impl.uses_objects:
                     state_history = history.as_state()
@@ -266,7 +268,11 @@ class UOWTransaction(object):
             # TODO: store the history as (state, object) tuples
             # so we don't have to keep converting here
             history = impl.get_history(
-                state, state.dict, passive | attributes.LOAD_AGAINST_COMMITTED
+                state,
+                state.dict,
+                passive
+                | attributes.LOAD_AGAINST_COMMITTED
+                | attributes.NO_RAISE,
             )
             if history and impl.uses_objects:
                 state_history = history.as_state()
diff --git a/test/orm/test_cascade.py b/test/orm/test_cascade.py
index 180b479ba..879ae2993 100644
--- a/test/orm/test_cascade.py
+++ b/test/orm/test_cascade.py
@@ -20,12 +20,14 @@ from sqlalchemy.orm import relationship
 from sqlalchemy.orm import Session
 from sqlalchemy.orm import util as orm_util
 from sqlalchemy.orm.attributes import instance_state
+from sqlalchemy.orm.decl_api import declarative_base
 from sqlalchemy.testing import assert_raises
 from sqlalchemy.testing import assert_raises_message
 from sqlalchemy.testing import eq_
 from sqlalchemy.testing import fixtures
 from sqlalchemy.testing import in_
 from sqlalchemy.testing import not_in
+from sqlalchemy.testing.assertsql import CompiledSQL
 from sqlalchemy.testing.fixtures import fixture_session
 from sqlalchemy.testing.schema import Column
 from sqlalchemy.testing.schema import Table
@@ -956,6 +958,108 @@ class O2OSingleParentNoFlushTest(fixtures.MappedTest):
         u1.address = a2
 
 
+class M2OwNoUseGetCascadeTest(
+    testing.AssertsExecutionResults, fixtures.TestBase
+):
+    @testing.fixture
+    def fixture(self, metadata):
+        Base = declarative_base(metadata=metadata)
+
+        def go(lazy="select", cascade="save-update"):
+            class A(Base):
+                __tablename__ = "a"
+
+                id = Column(Integer, primary_key=True)
+                email = Column(String(50), unique=True)
+
+                bs = relationship(
+                    "B",
+                    back_populates="user",
+                    primaryjoin="A.email == B.email",
+                )
+
+            class B(Base):
+                __tablename__ = "b"
+                id = Column(Integer, primary_key=True)
+                email = Column(String(50), ForeignKey("a.email"))
+
+                user = relationship(
+                    "A",
+                    lazy=lazy,
+                    cascade=cascade,
+                    single_parent=True,
+                    back_populates="bs",
+                    primaryjoin="A.email == B.email",
+                )
+
+            Base.metadata.create_all(testing.db)
+            return A, B
+
+        yield go
+        Base.registry.dispose()
+
+    def test_cascade_deletes_user(self, fixture):
+        A, B = fixture(cascade="all, delete-orphan")
+
+        sess = fixture_session()
+
+        a1 = A(email="x")
+        b1 = B(user=a1)
+        sess.add_all([a1, b1])
+        sess.commit()
+
+        b1 = sess.execute(select(B)).scalars().first()
+
+        sess.delete(b1)
+
+        self.assert_sql_execution(
+            testing.db,
+            sess.flush,
+            # looking for other bs'
+            CompiledSQL(
+                "SELECT b.id AS b_id, b.email AS b_email "
+                "FROM b WHERE :param_1 = b.email",
+                lambda ctx: [{"param_1": "x"}],
+            ),
+            CompiledSQL(
+                "DELETE FROM b WHERE b.id = :id", lambda ctx: [{"id": 1}]
+            ),
+            CompiledSQL(
+                "DELETE FROM a WHERE a.id = :id", lambda ctx: [{"id": 1}]
+            ),
+        )
+
+    @testing.combinations(("select",), ("raise",), argnames="lazy")
+    def test_ignores_user(self, fixture, lazy):
+        A, B = fixture()
+
+        sess = fixture_session()
+
+        a1 = A(email="x")
+        b1 = B(user=a1)
+        sess.add_all([a1, b1])
+        sess.commit()
+
+        b1 = sess.execute(select(B)).scalars().first()
+
+        sess.delete(b1)
+
+        self.assert_sql_execution(
+            testing.db,
+            sess.flush,
+            # we would like it to be able to skip this SELECT but this is not
+            # implemented right now
+            CompiledSQL(
+                "SELECT a.id AS a_id, a.email AS a_email FROM a "
+                "WHERE a.email = :param_1",
+                [{"param_1": "x"}],
+            ),
+            CompiledSQL(
+                "DELETE FROM b WHERE b.id = :id", lambda ctx: [{"id": 1}]
+            ),
+        )
+
+
 class NoSaveCascadeFlushTest(_fixtures.FixtureTest):
     """Test related item not present in session, commit proceeds."""
 
diff --git a/test/orm/test_unitofworkv2.py b/test/orm/test_unitofworkv2.py
index 822b47632..0f0c74698 100644
--- a/test/orm/test_unitofworkv2.py
+++ b/test/orm/test_unitofworkv2.py
@@ -775,6 +775,56 @@ class RudimentaryFlushTest(UOWTest):
             self._assert_uow_size(sess, 6)
 
 
+class RaiseLoadIgnoredTest(
+    fixtures.DeclarativeMappedTest,
+    testing.AssertsExecutionResults,
+):
+    @classmethod
+    def setup_classes(cls):
+        Base = cls.DeclarativeBasic
+
+        class A(Base):
+            __tablename__ = "a"
+
+            id = Column(Integer, primary_key=True)
+
+            bs = relationship("B", back_populates="user", lazy="raise")
+
+        class B(Base):
+            __tablename__ = "b"
+            id = Column(Integer, primary_key=True)
+            a_id = Column(ForeignKey("a.id"))
+            user = relationship("A", back_populates="bs", lazy="raise")
+
+    def test_delete_head(self):
+        A, B = self.classes("A", "B")
+
+        sess = fixture_session()
+
+        sess.add(A(bs=[B(), B()]))
+        sess.commit()
+
+        a1 = sess.execute(select(A)).scalars().first()
+
+        sess.delete(a1)
+
+        self.assert_sql_execution(
+            testing.db,
+            sess.flush,
+            # for the flush process, lazy="raise" is ignored
+            CompiledSQL(
+                "SELECT b.id AS b_id, b.a_id AS b_a_id FROM b "
+                "WHERE :param_1 = b.a_id",
+                [{"param_1": 1}],
+            ),
+            CompiledSQL(
+                "UPDATE b SET a_id=:a_id WHERE b.id = :b_id",
+                [{"a_id": None, "b_id": 1}, {"a_id": None, "b_id": 2}],
+            ),
+            CompiledSQL("DELETE FROM a WHERE a.id = :id", [{"id": 1}]),
+        )
+
+
 class SingleCycleTest(UOWTest):
     def teardown_test(self):
         engines.testing_reaper.rollback_all()