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

2 files changed, 154 insertions, 0 deletions

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()