implement deferred scalarobject history load

Modified the approach used for history tracking of scalar object
relationships that are not many-to-one, i.e. one-to-one relationships that
would otherwise be one-to-many. When replacing a one-to-one value, the
"old" value that would be replaced is no longer loaded immediately, and is
instead handled during the flush process. This eliminates an historically
troublesome lazy load that otherwise often occurs when assigning to a
one-to-one attribute, and is particularly troublesome when using
"lazy='raise'" as well as asyncio use cases.

This change does cause a behavioral change within the
:meth:`_orm.AttributeEvents.set` event, which is nonetheless currently
documented, which is that the event applied to such a one-to-one attribute
will no longer receive the "old" parameter if it is unloaded and the
:paramref:`_orm.relationship.active_history` flag is not set. As is
documented in :meth:`_orm.AttributeEvents.set`, if the event handler needs
to receive the "old" value when the event fires off, the active_history
flag must be established either with the event listener or with the
relationship. This is already the behavior with other kinds of attributes
such as many-to-one and column value references.

The change additionally will defer updating a backref on the "old" value
in the less common case that the "old" value is locally present in the
session, but isn't loaded on the relationship in question, until the
next flush occurs.  If this causes an issue, again the normal
:paramref:`_orm.relationship.active_history` flag can be set to ``True``
on the relationship.

A private flag which restores the old value is retained for now,
as support within relevant test suites to exercise the old and
new behaviors together.  This is so that if the behavioral change
produces problems we have test harnesses set up to further examine these
behaviors.   The "legacy" style can go away in 2.0 or in a much later
1.4 release.

Fixes: #6708
Change-Id: Id7f72fc39dcbec9119b665e528667a9919bb73b4

13 files changed, 569 insertions, 45 deletions

diff --git a/doc/build/changelog/unreleased_14/6708.rst b/doc/build/changelog/unreleased_14/6708.rst
new file mode 100644
index 000000000..9ae09d4ad
--- /dev/null
+++ b/doc/build/changelog/unreleased_14/6708.rst
@@ -0,0 +1,30 @@
+.. change::
+    :tags: usecase, orm
+    :tickets: 6708
+
+    Modified the approach used for history tracking of scalar object
+    relationships that are not many-to-one, i.e. one-to-one relationships that
+    would otherwise be one-to-many. When replacing a one-to-one value, the
+    "old" value that would be replaced is no longer loaded immediately, and is
+    instead handled during the flush process. This eliminates an historically
+    troublesome lazy load that otherwise often occurs when assigning to a
+    one-to-one attribute, and is particularly troublesome when using
+    "lazy='raise'" as well as asyncio use cases.
+
+    This change does cause a behavioral change within the
+    :meth:`_orm.AttributeEvents.set` event, which is nonetheless currently
+    documented, which is that the event applied to such a one-to-one attribute
+    will no longer receive the "old" parameter if it is unloaded and the
+    :paramref:`_orm.relationship.active_history` flag is not set. As is
+    documented in :meth:`_orm.AttributeEvents.set`, if the event handler needs
+    to receive the "old" value when the event fires off, the active_history
+    flag must be established either with the event listener or with the
+    relationship. This is already the behavior with other kinds of attributes
+    such as many-to-one and column value references.
+
+    The change additionally will defer updating a backref on the "old" value
+    in the less common case that the "old" value is locally present in the
+    session, but isn't loaded on the relationship in question, until the
+    next flush occurs.  If this causes an issue, again the normal
+    :paramref:`_orm.relationship.active_history` flag can be set to ``True``
+    on the relationship.
diff --git a/doc/build/orm/basic_relationships.rst b/doc/build/orm/basic_relationships.rst
index 1f6ad67a6..287b3e117 100644
--- a/doc/build/orm/basic_relationships.rst
+++ b/doc/build/orm/basic_relationships.rst
@@ -223,6 +223,7 @@ in this case the ``uselist`` parameter::
         parent = relationship("Parent", backref=backref("child", uselist=False))
 
 
+
 .. _relationships_many_to_many:
 
 Many To Many
diff --git a/doc/build/orm/extensions/asyncio.rst b/doc/build/orm/extensions/asyncio.rst
index 6aca1762d..d81978f0b 100644
--- a/doc/build/orm/extensions/asyncio.rst
+++ b/doc/build/orm/extensions/asyncio.rst
@@ -80,6 +80,8 @@ cursor and provides an async/await API, such as an async iterator::
         async for row in async_result:
             print("row: %s" % (row, ))
 
+.. _asyncio_orm:
+
 
 Synopsis - ORM
 ---------------
diff --git a/lib/sqlalchemy/orm/attributes.py b/lib/sqlalchemy/orm/attributes.py
index 105a9cfd2..9ba05395a 100644
--- a/lib/sqlalchemy/orm/attributes.py
+++ b/lib/sqlalchemy/orm/attributes.py
@@ -22,6 +22,7 @@ from . import interfaces
 from .base import ATTR_EMPTY
 from .base import ATTR_WAS_SET
 from .base import CALLABLES_OK
+from .base import DEFERRED_HISTORY_LOAD
 from .base import INIT_OK
 from .base import instance_dict
 from .base import instance_state
@@ -768,6 +769,9 @@ class AttributeImpl(object):
         else:
             self.accepts_scalar_loader = self.default_accepts_scalar_loader
 
+        _deferred_history = kwargs.pop("_deferred_history", False)
+        self._deferred_history = _deferred_history
+
         if active_history:
             self.dispatch._active_history = True
 
@@ -786,6 +790,7 @@ class AttributeImpl(object):
         "load_on_unexpire",
         "_modified_token",
         "accepts_scalar_loader",
+        "_deferred_history",
     )
 
     def __str__(self):
@@ -918,19 +923,7 @@ class AttributeImpl(object):
                 if not passive & CALLABLES_OK:
                     return PASSIVE_NO_RESULT
 
-                if (
-                    self.accepts_scalar_loader
-                    and self.load_on_unexpire
-                    and key in state.expired_attributes
-                ):
-                    value = state._load_expired(state, passive)
-                elif key in state.callables:
-                    callable_ = state.callables[key]
-                    value = callable_(state, passive)
-                elif self.callable_:
-                    value = self.callable_(state, passive)
-                else:
-                    value = ATTR_EMPTY
+                value = self._fire_loader_callables(state, key, passive)
 
                 if value is PASSIVE_NO_RESULT or value is NO_VALUE:
                     return value
@@ -955,6 +948,21 @@ class AttributeImpl(object):
             else:
                 return self._default_value(state, dict_)
 
+    def _fire_loader_callables(self, state, key, passive):
+        if (
+            self.accepts_scalar_loader
+            and self.load_on_unexpire
+            and key in state.expired_attributes
+        ):
+            return state._load_expired(state, passive)
+        elif key in state.callables:
+            callable_ = state.callables[key]
+            return callable_(state, passive)
+        elif self.callable_:
+            return self.callable_(state, passive)
+        else:
+            return ATTR_EMPTY
+
     def append(self, state, dict_, value, initiator, passive=PASSIVE_OFF):
         self.set(state, dict_, value, initiator, passive=passive)
 
@@ -1142,15 +1150,33 @@ class ScalarObjectAttributeImpl(ScalarAttributeImpl):
 
     def get_history(self, state, dict_, passive=PASSIVE_OFF):
         if self.key in dict_:
-            return History.from_object_attribute(self, state, dict_[self.key])
+            current = dict_[self.key]
         else:
             if passive & INIT_OK:
                 passive ^= INIT_OK
             current = self.get(state, dict_, passive=passive)
             if current is PASSIVE_NO_RESULT:
                 return HISTORY_BLANK
-            else:
-                return History.from_object_attribute(self, state, current)
+
+        if not self._deferred_history:
+            return History.from_object_attribute(self, state, current)
+        else:
+            original = state.committed_state.get(self.key, _NO_HISTORY)
+            if original is PASSIVE_NO_RESULT:
+
+                loader_passive = passive | (
+                    PASSIVE_ONLY_PERSISTENT
+                    | NO_AUTOFLUSH
+                    | LOAD_AGAINST_COMMITTED
+                    | NO_RAISE
+                    | DEFERRED_HISTORY_LOAD
+                )
+                original = self._fire_loader_callables(
+                    state, self.key, loader_passive
+                )
+            return History.from_object_attribute(
+                self, state, current, original=original
+            )
 
     def get_all_pending(self, state, dict_, passive=PASSIVE_NO_INITIALIZE):
         if self.key in dict_:
@@ -1193,6 +1219,7 @@ class ScalarObjectAttributeImpl(ScalarAttributeImpl):
         pop=False,
     ):
         """Set a value on the given InstanceState."""
+
         if self.dispatch._active_history:
             old = self.get(
                 state,
@@ -1227,7 +1254,11 @@ class ScalarObjectAttributeImpl(ScalarAttributeImpl):
         dict_[self.key] = value
 
     def fire_remove_event(self, state, dict_, value, initiator):
-        if self.trackparent and value is not None:
+        if self.trackparent and value not in (
+            None,
+            PASSIVE_NO_RESULT,
+            NO_VALUE,
+        ):
             self.sethasparent(instance_state(value), state, False)
 
         for fn in self.dispatch.remove:
@@ -1930,8 +1961,11 @@ class History(util.namedtuple("History", ["added", "unchanged", "deleted"])):
                 return cls([current], (), deleted)
 
     @classmethod
-    def from_object_attribute(cls, attribute, state, current):
-        original = state.committed_state.get(attribute.key, _NO_HISTORY)
+    def from_object_attribute(
+        cls, attribute, state, current, original=_NO_HISTORY
+    ):
+        if original is _NO_HISTORY:
+            original = state.committed_state.get(attribute.key, _NO_HISTORY)
 
         if original is _NO_HISTORY:
             if current is NO_VALUE:
diff --git a/lib/sqlalchemy/orm/base.py b/lib/sqlalchemy/orm/base.py
index 2932f1bb9..524407265 100644
--- a/lib/sqlalchemy/orm/base.py
+++ b/lib/sqlalchemy/orm/base.py
@@ -124,6 +124,12 @@ NO_RAISE = util.symbol(
     canonical=128,
 )
 
+DEFERRED_HISTORY_LOAD = util.symbol(
+    "DEFERRED_HISTORY_LOAD",
+    """indicates special load of the previous value of an attribute""",
+    canonical=256,
+)
+
 # pre-packaged sets of flags used as inputs
 PASSIVE_OFF = util.symbol(
     "PASSIVE_OFF",
diff --git a/lib/sqlalchemy/orm/relationships.py b/lib/sqlalchemy/orm/relationships.py
index 2224b4902..7b27d9fe7 100644
--- a/lib/sqlalchemy/orm/relationships.py
+++ b/lib/sqlalchemy/orm/relationships.py
@@ -151,6 +151,7 @@ class RelationshipProperty(StrategizedProperty):
         info=None,
         omit_join=None,
         sync_backref=None,
+        _legacy_inactive_history_style=False,
     ):
         """Provide a relationship between two mapped classes.
 
@@ -1014,6 +1015,8 @@ class RelationshipProperty(StrategizedProperty):
         self.distinct_target_key = distinct_target_key
         self.doc = doc
         self.active_history = active_history
+        self._legacy_inactive_history_style = _legacy_inactive_history_style
+
         self.join_depth = join_depth
         if omit_join:
             util.warn(
diff --git a/lib/sqlalchemy/orm/strategies.py b/lib/sqlalchemy/orm/strategies.py
index 2a254f8de..7d7438452 100644
--- a/lib/sqlalchemy/orm/strategies.py
+++ b/lib/sqlalchemy/orm/strategies.py
@@ -695,18 +695,27 @@ class LazyLoader(AbstractRelationshipLoader, util.MemoizedSlots):
     def init_class_attribute(self, mapper):
         self.is_class_level = True
 
-        active_history = (
-            self.parent_property.active_history
-            or self.parent_property.direction is not interfaces.MANYTOONE
-            or not self.use_get
+        _legacy_inactive_history_style = (
+            self.parent_property._legacy_inactive_history_style
         )
 
-        # MANYTOONE currently only needs the
-        # "old" value for delete-orphan
-        # cascades.  the required _SingleParentValidator
-        # will enable active_history
-        # in that case.  otherwise we don't need the
-        # "old" value during backref operations.
+        if self.parent_property.active_history:
+            active_history = True
+            _deferred_history = False
+
+        elif (
+            self.parent_property.direction is not interfaces.MANYTOONE
+            or not self.use_get
+        ):
+            if _legacy_inactive_history_style:
+                active_history = True
+                _deferred_history = False
+            else:
+                active_history = False
+                _deferred_history = True
+        else:
+            active_history = _deferred_history = False
+
         _register_attribute(
             self.parent_property,
             mapper,
@@ -714,6 +723,7 @@ class LazyLoader(AbstractRelationshipLoader, util.MemoizedSlots):
             callable_=self._load_for_state,
             typecallable=self.parent_property.collection_class,
             active_history=active_history,
+            _deferred_history=_deferred_history,
         )
 
     def _memoized_attr__simple_lazy_clause(self):
@@ -850,7 +860,10 @@ class LazyLoader(AbstractRelationshipLoader, util.MemoizedSlots):
             if _none_set.issuperset(primary_key_identity):
                 return None
 
-            if self.key in state.dict:
+            if (
+                self.key in state.dict
+                and not passive & attributes.DEFERRED_HISTORY_LOAD
+            ):
                 return attributes.ATTR_WAS_SET
 
             # look for this identity in the identity map.  Delegate to the
@@ -1016,7 +1029,10 @@ class LazyLoader(AbstractRelationshipLoader, util.MemoizedSlots):
             "_sa_orm_load_options": load_options,
         }
 
-        if self.key in state.dict:
+        if (
+            self.key in state.dict
+            and not passive & attributes.DEFERRED_HISTORY_LOAD
+        ):
             return attributes.ATTR_WAS_SET
 
         if pending:
diff --git a/test/ext/asyncio/test_session_py3k.py b/test/ext/asyncio/test_session_py3k.py
index 1f5c95054..0883cb026 100644
--- a/test/ext/asyncio/test_session_py3k.py
+++ b/test/ext/asyncio/test_session_py3k.py
@@ -1,7 +1,10 @@
+from sqlalchemy import Column
 from sqlalchemy import event
 from sqlalchemy import exc
+from sqlalchemy import ForeignKey
 from sqlalchemy import func
 from sqlalchemy import inspect
+from sqlalchemy import Integer
 from sqlalchemy import select
 from sqlalchemy import Table
 from sqlalchemy import testing
@@ -473,6 +476,93 @@ class AsyncCascadesTest(AsyncFixture):
         )
 
 
+class AsyncORMBehaviorsTest(AsyncFixture):
+    @testing.fixture
+    def one_to_one_fixture(self, registry, async_engine):
+        async def go(legacy_inactive_history_style):
+            @registry.mapped
+            class A:
+                __tablename__ = "a"
+
+                id = Column(Integer, primary_key=True)
+                b = relationship(
+                    "B",
+                    uselist=False,
+                    _legacy_inactive_history_style=(
+                        legacy_inactive_history_style
+                    ),
+                )
+
+            @registry.mapped
+            class B:
+                __tablename__ = "b"
+                id = Column(Integer, primary_key=True)
+                a_id = Column(ForeignKey("a.id"))
+
+            async with async_engine.begin() as conn:
+                await conn.run_sync(registry.metadata.create_all)
+
+            return A, B
+
+        return go
+
+    @testing.combinations(
+        (
+            "legacy_style",
+            True,
+        ),
+        (
+            "new_style",
+            False,
+        ),
+        argnames="_legacy_inactive_history_style",
+        id_="ia",
+    )
+    @async_test
+    async def test_new_style_active_history(
+        self, async_session, one_to_one_fixture, _legacy_inactive_history_style
+    ):
+
+        A, B = await one_to_one_fixture(_legacy_inactive_history_style)
+
+        a1 = A()
+        b1 = B()
+
+        a1.b = b1
+        async_session.add(a1)
+
+        await async_session.commit()
+
+        b2 = B()
+
+        if _legacy_inactive_history_style:
+            # aiomysql dialect having problems here, emitting weird
+            # pytest warnings and we might need to just skip for aiomysql
+            # here, which is also raising StatementError w/ MissingGreenlet
+            # inside of it
+            with testing.expect_raises(
+                (exc.StatementError, exc.MissingGreenlet)
+            ):
+                a1.b = b2
+        else:
+            a1.b = b2
+
+            await async_session.flush()
+
+            await async_session.refresh(b1)
+
+            eq_(
+                (
+                    await async_session.execute(
+                        select(func.count())
+                        .where(B.id == b1.id)
+                        .where(B.a_id == None)
+                    )
+                ).scalar(),
+                1,
+            )
+
+
 class AsyncEventTest(AsyncFixture):
     """The engine events all run in their normal synchronous context.
 
diff --git a/test/orm/test_backref_mutations.py b/test/orm/test_backref_mutations.py
index a6d651d22..ab31058a5 100644
--- a/test/orm/test_backref_mutations.py
+++ b/test/orm/test_backref_mutations.py
@@ -373,6 +373,18 @@ class O2MCollectionTest(_fixtures.FixtureTest):
         eq_(u1.addresses, [a1, a2, a1])
 
 
+@testing.combinations(
+    (
+        "legacy_style",
+        True,
+    ),
+    (
+        "new_style",
+        False,
+    ),
+    argnames="name, _legacy_inactive_history_style",
+    id_="sa",
+)
 class O2OScalarBackrefMoveTest(_fixtures.FixtureTest):
     run_inserts = None
 
@@ -391,7 +403,12 @@ class O2OScalarBackrefMoveTest(_fixtures.FixtureTest):
             users,
             properties={
                 "address": relationship(
-                    Address, backref=backref("user"), uselist=False
+                    Address,
+                    backref=backref("user"),
+                    uselist=False,
+                    _legacy_inactive_history_style=(
+                        cls._legacy_inactive_history_style
+                    ),
                 )
             },
         )
@@ -485,9 +502,18 @@ class O2OScalarBackrefMoveTest(_fixtures.FixtureTest):
         # backref fires
         assert u1.address is a2
 
-        # stays on both sides
-        assert a1.user is u1
-        assert a2.user is u1
+        eq_(
+            a2._sa_instance_state.committed_state["user"],
+            attributes.PASSIVE_NO_RESULT,
+        )
+        if not self._legacy_inactive_history_style:
+            # autoflush during the a2.user
+            assert a1.user is None
+            assert a2.user is u1
+        else:
+            # stays on both sides
+            assert a1.user is u1
+            assert a2.user is u1
 
     def test_collection_move_commitfirst(self):
         User, Address = self.classes.User, self.classes.Address
@@ -546,6 +572,18 @@ class O2OScalarBackrefMoveTest(_fixtures.FixtureTest):
         assert a2.user is u1
 
 
+@testing.combinations(
+    (
+        "legacy_style",
+        True,
+    ),
+    (
+        "new_style",
+        False,
+    ),
+    argnames="name, _legacy_inactive_history_style",
+    id_="sa",
+)
 class O2OScalarMoveTest(_fixtures.FixtureTest):
     run_inserts = None
 
@@ -562,7 +600,15 @@ class O2OScalarMoveTest(_fixtures.FixtureTest):
         mapper(
             User,
             users,
-            properties={"address": relationship(Address, uselist=False)},
+            properties={
+                "address": relationship(
+                    Address,
+                    uselist=False,
+                    _legacy_inactive_history_style=(
+                        cls._legacy_inactive_history_style
+                    ),
+                )
+            },
         )
 
     def test_collection_move_commitfirst(self):
diff --git a/test/orm/test_cascade.py b/test/orm/test_cascade.py
index a7156be4a..396f843af 100644
--- a/test/orm/test_cascade.py
+++ b/test/orm/test_cascade.py
@@ -1753,6 +1753,18 @@ class NoSaveCascadeBackrefTest(_fixtures.FixtureTest):
         assert k1 not in sess
 
 
+@testing.combinations(
+    (
+        "legacy_style",
+        True,
+    ),
+    (
+        "new_style",
+        False,
+    ),
+    argnames="name, _legacy_inactive_history_style",
+    id_="sa",
+)
 class M2OCascadeDeleteOrphanTestOne(fixtures.MappedTest):
     @classmethod
     def define_tables(cls, metadata):
@@ -1824,6 +1836,7 @@ class M2OCascadeDeleteOrphanTestOne(fixtures.MappedTest):
             prefs,
             properties=dict(extra=relationship(Extra, cascade="all, delete")),
         )
+
         mapper(
             User,
             users,
@@ -1834,7 +1847,13 @@ class M2OCascadeDeleteOrphanTestOne(fixtures.MappedTest):
                     cascade="all, delete-orphan",
                     single_parent=True,
                 ),
-                foo=relationship(Foo),
+                foo=relationship(
+                    Foo,
+                    active_history=False,
+                    _legacy_inactive_history_style=(
+                        cls._legacy_inactive_history_style
+                    ),
+                ),
             ),
         )  # straight m2o
         mapper(Foo, foo)
@@ -1884,7 +1903,7 @@ class M2OCascadeDeleteOrphanTestOne(fixtures.MappedTest):
         )
 
     def test_cascade_on_deleted(self):
-        """test a bug introduced by r6711"""
+        """test a bug introduced by #6711"""
 
         Foo, User = self.classes.Foo, self.classes.User
 
@@ -1899,6 +1918,7 @@ class M2OCascadeDeleteOrphanTestOne(fixtures.MappedTest):
         # the error condition relies upon
         # these things being true
         assert User.foo.dispatch._active_history is False
+
         eq_(attributes.get_history(u1, "foo"), ([None], (), ()))
 
         sess.add(u1)
diff --git a/test/orm/test_cycles.py b/test/orm/test_cycles.py
index ed11b89c9..697009c2a 100644
--- a/test/orm/test_cycles.py
+++ b/test/orm/test_cycles.py
@@ -733,6 +733,18 @@ class BiDirectionalOneToManyTest2(fixtures.MappedTest):
         sess.flush()
 
 
+@testing.combinations(
+    (
+        "legacy_style",
+        True,
+    ),
+    (
+        "new_style",
+        False,
+    ),
+    argnames="name, _legacy_inactive_history_style",
+    id_="sa",
+)
 class OneToManyManyToOneTest(fixtures.MappedTest):
     """
 
@@ -804,11 +816,17 @@ class OneToManyManyToOneTest(fixtures.MappedTest):
                     Ball,
                     primaryjoin=ball.c.person_id == person.c.id,
                     remote_side=ball.c.person_id,
+                    _legacy_inactive_history_style=(
+                        self._legacy_inactive_history_style
+                    ),
                 ),
                 favorite=relationship(
                     Ball,
                     primaryjoin=person.c.favorite_ball_id == ball.c.id,
                     remote_side=ball.c.id,
+                    _legacy_inactive_history_style=(
+                        self._legacy_inactive_history_style
+                    ),
                 ),
             ),
         )
@@ -837,6 +855,9 @@ class OneToManyManyToOneTest(fixtures.MappedTest):
                     Ball,
                     primaryjoin=person.c.favorite_ball_id == ball.c.id,
                     post_update=True,
+                    _legacy_inactive_history_style=(
+                        self._legacy_inactive_history_style
+                    ),
                 )
             ),
         )
@@ -884,12 +905,18 @@ class OneToManyManyToOneTest(fixtures.MappedTest):
                     remote_side=ball.c.person_id,
                     post_update=False,
                     cascade="all, delete-orphan",
+                    _legacy_inactive_history_style=(
+                        self._legacy_inactive_history_style
+                    ),
                 ),
                 favorite=relationship(
                     Ball,
                     primaryjoin=person.c.favorite_ball_id == ball.c.id,
                     remote_side=person.c.favorite_ball_id,
                     post_update=True,
+                    _legacy_inactive_history_style=(
+                        self._legacy_inactive_history_style
+                    ),
                 ),
             ),
         )
@@ -989,12 +1016,24 @@ class OneToManyManyToOneTest(fixtures.MappedTest):
                     primaryjoin=ball.c.person_id == person.c.id,
                     remote_side=ball.c.person_id,
                     post_update=True,
-                    backref=backref("person", post_update=True),
+                    backref=backref(
+                        "person",
+                        post_update=True,
+                        _legacy_inactive_history_style=(
+                            self._legacy_inactive_history_style
+                        ),
+                    ),
+                    _legacy_inactive_history_style=(
+                        self._legacy_inactive_history_style
+                    ),
                 ),
                 favorite=relationship(
                     Ball,
                     primaryjoin=person.c.favorite_ball_id == ball.c.id,
                     remote_side=person.c.favorite_ball_id,
+                    _legacy_inactive_history_style=(
+                        self._legacy_inactive_history_style
+                    ),
                 ),
             ),
         )
@@ -1044,11 +1083,17 @@ class OneToManyManyToOneTest(fixtures.MappedTest):
                     cascade="all, delete-orphan",
                     post_update=True,
                     backref="person",
+                    _legacy_inactive_history_style=(
+                        self._legacy_inactive_history_style
+                    ),
                 ),
                 favorite=relationship(
                     Ball,
                     primaryjoin=person.c.favorite_ball_id == ball.c.id,
                     remote_side=person.c.favorite_ball_id,
+                    _legacy_inactive_history_style=(
+                        self._legacy_inactive_history_style
+                    ),
                 ),
             ),
         )
@@ -1169,13 +1214,17 @@ class OneToManyManyToOneTest(fixtures.MappedTest):
                     Person,
                     post_update=True,
                     primaryjoin=person.c.id == ball.c.person_id,
+                    _legacy_inactive_history_style=(
+                        self._legacy_inactive_history_style
+                    ),
                 )
             },
         )
         mapper(Person, person)
 
         sess = fixture_session(autocommit=False, expire_on_commit=True)
-        sess.add(Ball(person=Person()))
+        p1 = Person()
+        sess.add(Ball(person=p1))
         sess.commit()
         b1 = sess.query(Ball).first()
 
diff --git a/test/orm/test_onetoone.py b/test/orm/test_onetoone.py
index ae9f9b3a1..83213839f 100644
--- a/test/orm/test_onetoone.py
+++ b/test/orm/test_onetoone.py
@@ -1,6 +1,7 @@
 from sqlalchemy import ForeignKey
 from sqlalchemy import Integer
 from sqlalchemy import String
+from sqlalchemy import testing
 from sqlalchemy.orm import mapper
 from sqlalchemy.orm import relationship
 from sqlalchemy.testing import fixtures
@@ -42,7 +43,13 @@ class O2OTest(fixtures.MappedTest):
         class Port(cls.Basic):
             pass
 
-    def test_basic(self):
+    @testing.combinations(
+        (True, False),
+        (False, False),
+        (False, True),
+        argnames="_legacy_inactive_history_style, active_history",
+    )
+    def test_basic(self, _legacy_inactive_history_style, active_history):
         Port, port, jack, Jack = (
             self.classes.Port,
             self.tables.port,
@@ -55,7 +62,15 @@ class O2OTest(fixtures.MappedTest):
             Jack,
             jack,
             properties=dict(
-                port=relationship(Port, backref="jack", uselist=False)
+                port=relationship(
+                    Port,
+                    backref="jack",
+                    uselist=False,
+                    active_history=active_history,
+                    _legacy_inactive_history_style=(
+                        _legacy_inactive_history_style
+                    ),
+                )
             ),
         )
 
@@ -85,8 +100,91 @@ class O2OTest(fixtures.MappedTest):
         p = session.query(Port).get(pid)
 
         j.port = None
-        self.assert_(p.jack is None)
-        session.flush()
+
+        if not active_history and not _legacy_inactive_history_style:
+            session.flush()
+            self.assert_(p.jack is None)
+        else:
+            self.assert_(p.jack is None)
+            session.flush()
 
         session.delete(j)
         session.flush()
+
+    @testing.combinations(
+        (True,), (False,), argnames="_legacy_inactive_history_style"
+    )
+    def test_simple_replace(self, _legacy_inactive_history_style):
+        Port, port, jack, Jack = (
+            self.classes.Port,
+            self.tables.port,
+            self.tables.jack,
+            self.classes.Jack,
+        )
+
+        mapper(Port, port)
+        mapper(
+            Jack,
+            jack,
+            properties=dict(
+                port=relationship(
+                    Port,
+                    uselist=False,
+                    _legacy_inactive_history_style=(
+                        _legacy_inactive_history_style
+                    ),
+                )
+            ),
+        )
+
+        s = fixture_session()
+
+        p1 = Port(name="p1")
+        j1 = Jack(number="j1", port=p1)
+
+        s.add(j1)
+        s.commit()
+
+        j1.port = Port(name="p2")
+        s.commit()
+
+        assert s.query(Port).filter_by(name="p1").one().jack_id is None
+
+    @testing.combinations(
+        (True,), (False,), argnames="_legacy_inactive_history_style"
+    )
+    def test_simple_del(self, _legacy_inactive_history_style):
+        Port, port, jack, Jack = (
+            self.classes.Port,
+            self.tables.port,
+            self.tables.jack,
+            self.classes.Jack,
+        )
+
+        mapper(Port, port)
+        mapper(
+            Jack,
+            jack,
+            properties=dict(
+                port=relationship(
+                    Port,
+                    uselist=False,
+                    _legacy_inactive_history_style=(
+                        _legacy_inactive_history_style
+                    ),
+                )
+            ),
+        )
+
+        s = fixture_session()
+
+        p1 = Port(name="p1")
+        j1 = Jack(number="j1", port=p1)
+
+        s.add(j1)
+        s.commit()
+
+        del j1.port
+        s.commit()
+
+        assert s.query(Port).filter_by(name="p1").one().jack_id is None
diff --git a/test/orm/test_relationships.py b/test/orm/test_relationships.py
index 8f622d705..95c5a196f 100644
--- a/test/orm/test_relationships.py
+++ b/test/orm/test_relationships.py
@@ -5559,6 +5559,7 @@ class InactiveHistoryNoRaiseTest(_fixtures.FixtureTest):
             backref=False,
             delete=False,
             active_history=False,
+            legacy_inactive_history_style=True,
         ),
         dict(
             detached=True,
@@ -5566,6 +5567,7 @@ class InactiveHistoryNoRaiseTest(_fixtures.FixtureTest):
             backref=False,
             delete=False,
             active_history=False,
+            legacy_inactive_history_style=True,
         ),
         dict(
             detached=False,
@@ -5573,6 +5575,7 @@ class InactiveHistoryNoRaiseTest(_fixtures.FixtureTest):
             backref=False,
             delete=False,
             active_history=False,
+            legacy_inactive_history_style=True,
         ),
         dict(
             detached=True,
@@ -5580,6 +5583,7 @@ class InactiveHistoryNoRaiseTest(_fixtures.FixtureTest):
             backref=False,
             delete=False,
             active_history=False,
+            legacy_inactive_history_style=True,
         ),
         dict(
             detached=False,
@@ -5587,6 +5591,7 @@ class InactiveHistoryNoRaiseTest(_fixtures.FixtureTest):
             backref=True,
             delete=False,
             active_history=False,
+            legacy_inactive_history_style=True,
         ),
         dict(
             detached=True,
@@ -5594,6 +5599,7 @@ class InactiveHistoryNoRaiseTest(_fixtures.FixtureTest):
             backref=True,
             delete=False,
             active_history=False,
+            legacy_inactive_history_style=True,
         ),
         dict(
             detached=False,
@@ -5601,6 +5607,7 @@ class InactiveHistoryNoRaiseTest(_fixtures.FixtureTest):
             backref=True,
             delete=False,
             active_history=False,
+            legacy_inactive_history_style=True,
         ),
         dict(
             detached=True,
@@ -5608,6 +5615,72 @@ class InactiveHistoryNoRaiseTest(_fixtures.FixtureTest):
             backref=True,
             delete=False,
             active_history=False,
+            legacy_inactive_history_style=True,
+        ),
+        #####
+        dict(
+            detached=False,
+            raiseload=False,
+            backref=False,
+            delete=False,
+            active_history=False,
+            legacy_inactive_history_style=False,
+        ),
+        dict(
+            detached=True,
+            raiseload=False,
+            backref=False,
+            delete=False,
+            active_history=False,
+            legacy_inactive_history_style=False,
+        ),
+        dict(
+            detached=False,
+            raiseload=True,
+            backref=False,
+            delete=False,
+            active_history=False,
+            legacy_inactive_history_style=False,
+        ),
+        dict(
+            detached=True,
+            raiseload=True,
+            backref=False,
+            delete=False,
+            active_history=False,
+            legacy_inactive_history_style=False,
+        ),
+        dict(
+            detached=False,
+            raiseload=False,
+            backref=True,
+            delete=False,
+            active_history=False,
+            legacy_inactive_history_style=False,
+        ),
+        dict(
+            detached=True,
+            raiseload=False,
+            backref=True,
+            delete=False,
+            active_history=False,
+            legacy_inactive_history_style=False,
+        ),
+        dict(
+            detached=False,
+            raiseload=True,
+            backref=True,
+            delete=False,
+            active_history=False,
+            legacy_inactive_history_style=False,
+        ),
+        dict(
+            detached=True,
+            raiseload=True,
+            backref=True,
+            delete=False,
+            active_history=False,
+            legacy_inactive_history_style=False,
         ),
         dict(
             detached=False,
@@ -5615,6 +5688,7 @@ class InactiveHistoryNoRaiseTest(_fixtures.FixtureTest):
             backref=False,
             delete=False,
             active_history=True,
+            legacy_inactive_history_style=True,
         ),
         dict(
             detached=True,
@@ -5622,6 +5696,7 @@ class InactiveHistoryNoRaiseTest(_fixtures.FixtureTest):
             backref=False,
             delete=False,
             active_history=True,
+            legacy_inactive_history_style=True,
         ),
         dict(
             detached=False,
@@ -5629,6 +5704,7 @@ class InactiveHistoryNoRaiseTest(_fixtures.FixtureTest):
             backref=False,
             delete=False,
             active_history=True,
+            legacy_inactive_history_style=True,
         ),
         dict(
             detached=True,
@@ -5636,6 +5712,7 @@ class InactiveHistoryNoRaiseTest(_fixtures.FixtureTest):
             backref=False,
             delete=False,
             active_history=True,
+            legacy_inactive_history_style=True,
         ),
         dict(
             detached=False,
@@ -5643,6 +5720,7 @@ class InactiveHistoryNoRaiseTest(_fixtures.FixtureTest):
             backref=True,
             delete=False,
             active_history=True,
+            legacy_inactive_history_style=True,
         ),
         dict(
             detached=True,
@@ -5650,6 +5728,7 @@ class InactiveHistoryNoRaiseTest(_fixtures.FixtureTest):
             backref=True,
             delete=False,
             active_history=True,
+            legacy_inactive_history_style=True,
         ),
         dict(
             detached=False,
@@ -5657,6 +5736,7 @@ class InactiveHistoryNoRaiseTest(_fixtures.FixtureTest):
             backref=True,
             delete=False,
             active_history=True,
+            legacy_inactive_history_style=True,
         ),
         dict(
             detached=True,
@@ -5664,13 +5744,16 @@ class InactiveHistoryNoRaiseTest(_fixtures.FixtureTest):
             backref=True,
             delete=False,
             active_history=True,
+            legacy_inactive_history_style=True,
         ),
+        ####
         dict(
             detached=False,
             raiseload=False,
             backref=False,
             delete=True,
             active_history=False,
+            legacy_inactive_history_style=True,
         ),
         dict(
             detached=True,
@@ -5678,6 +5761,7 @@ class InactiveHistoryNoRaiseTest(_fixtures.FixtureTest):
             backref=False,
             delete=True,
             active_history=False,
+            legacy_inactive_history_style=True,
         ),
         dict(
             detached=False,
@@ -5685,6 +5769,7 @@ class InactiveHistoryNoRaiseTest(_fixtures.FixtureTest):
             backref=False,
             delete=True,
             active_history=False,
+            legacy_inactive_history_style=True,
         ),
         dict(
             detached=True,
@@ -5692,7 +5777,42 @@ class InactiveHistoryNoRaiseTest(_fixtures.FixtureTest):
             backref=False,
             delete=True,
             active_history=False,
+            legacy_inactive_history_style=True,
         ),
+        ###
+        dict(
+            detached=False,
+            raiseload=False,
+            backref=False,
+            delete=True,
+            active_history=False,
+            legacy_inactive_history_style=False,
+        ),
+        dict(
+            detached=True,
+            raiseload=False,
+            backref=False,
+            delete=True,
+            active_history=False,
+            legacy_inactive_history_style=False,
+        ),
+        dict(
+            detached=False,
+            raiseload=True,
+            backref=False,
+            delete=True,
+            active_history=False,
+            legacy_inactive_history_style=False,
+        ),
+        dict(
+            detached=True,
+            raiseload=True,
+            backref=False,
+            delete=True,
+            active_history=False,
+            legacy_inactive_history_style=False,
+        ),
+        #
         dict(
             detached=False,
             raiseload=False,
@@ -5722,7 +5842,15 @@ class InactiveHistoryNoRaiseTest(_fixtures.FixtureTest):
             active_history=True,
         ),
     )
-    def test_m2o(self, detached, raiseload, backref, active_history, delete):
+    def test_m2o(
+        self,
+        detached,
+        raiseload,
+        backref,
+        active_history,
+        delete,
+        legacy_inactive_history_style,
+    ):
 
         if delete:
             assert not backref, "delete and backref are mutually exclusive"
@@ -5739,6 +5867,7 @@ class InactiveHistoryNoRaiseTest(_fixtures.FixtureTest):
             opts["active_history"] = True
         if raiseload:
             opts["lazy"] = "raise"
+        opts["_legacy_inactive_history_style"] = legacy_inactive_history_style
 
         mapper(
             Address,