implement active_history for composites

Fixed bug where the "active history" feature was not fully
implemented for composite attributes, making it impossible to receive
events that included the "old" value.   This seems to have been the case
with older SQLAlchemy versions as well, where "active_history" would
be propagated to the underlying column-based attributes, but an event
handler listening to the composite attribute itself would not be given
the "old" value being replaced, even if the composite() were set up
with active_history=True.

Additionally, fixed a regression that's local to 2.0 which disallowed
active_history on composite from being assigned to the impl with
``attr.impl.active_history=True``.

Fixes: #9460
Change-Id: I6d7752a01c8d3fd78de7a90de10e8c52f9b3cd4e

3 files changed, 223 insertions, 2 deletions

diff --git a/doc/build/changelog/unreleased_20/9460.rst b/doc/build/changelog/unreleased_20/9460.rst
new file mode 100644
index 000000000..a2a7d9c6d
--- /dev/null
+++ b/doc/build/changelog/unreleased_20/9460.rst
@@ -0,0 +1,17 @@
+.. change::
+    :tags: bug, orm
+    :tickets: 9460
+
+    Fixed bug where the "active history" feature was not fully
+    implemented for composite attributes, making it impossible to receive
+    events that included the "old" value.   This seems to have been the case
+    with older SQLAlchemy versions as well, where "active_history" would
+    be propagated to the underlying column-based attributes, but an event
+    handler listening to the composite attribute itself would not be given
+    the "old" value being replaced, even if the composite() were set up
+    with active_history=True.
+
+    Additionally, fixed a regression that's local to 2.0 which disallowed
+    active_history on composite from being assigned to the impl with
+    ``attr.impl.active_history=True``.
+
diff --git a/lib/sqlalchemy/orm/descriptor_props.py b/lib/sqlalchemy/orm/descriptor_props.py
index fd28830d9..2e57fd0f4 100644
--- a/lib/sqlalchemy/orm/descriptor_props.py
+++ b/lib/sqlalchemy/orm/descriptor_props.py
@@ -305,7 +305,12 @@ class CompositeProperty(
             dict_ = attributes.instance_dict(instance)
             state = attributes.instance_state(instance)
             attr = state.manager[self.key]
-            previous = dict_.get(self.key, LoaderCallableStatus.NO_VALUE)
+
+            if attr.dispatch._active_history:
+                previous = fget(instance)
+            else:
+                previous = dict_.get(self.key, LoaderCallableStatus.NO_VALUE)
+
             for fn in attr.dispatch.set:
                 value = fn(state, value, previous, attr.impl)
             dict_[self.key] = value
@@ -322,7 +327,14 @@ class CompositeProperty(
         def fdel(instance: Any) -> None:
             state = attributes.instance_state(instance)
             dict_ = attributes.instance_dict(instance)
-            previous = dict_.pop(self.key, LoaderCallableStatus.NO_VALUE)
+            attr = state.manager[self.key]
+
+            if attr.dispatch._active_history:
+                previous = fget(instance)
+                dict_.pop(self.key, None)
+            else:
+                previous = dict_.pop(self.key, LoaderCallableStatus.NO_VALUE)
+
             attr = state.manager[self.key]
             attr.dispatch.remove(state, previous, attr.impl)
             for key in self._attribute_keys:
@@ -614,6 +626,10 @@ class CompositeProperty(
             self.parent, "expire", expire_handler, raw=True, propagate=True
         )
 
+        proxy_attr = self.parent.class_manager[self.key]
+        proxy_attr.impl.dispatch = proxy_attr.dispatch  # type: ignore
+        proxy_attr.impl.dispatch._active_history = self.active_history  # type: ignore  # noqa: E501
+
         # TODO: need a deserialize hook here
 
     @util.memoized_property
diff --git a/test/orm/test_composites.py b/test/orm/test_composites.py
index ddd9a45b8..ded2c25db 100644
--- a/test/orm/test_composites.py
+++ b/test/orm/test_composites.py
@@ -3,8 +3,10 @@ import operator
 import random
 
 import sqlalchemy as sa
+from sqlalchemy import event
 from sqlalchemy import ForeignKey
 from sqlalchemy import insert
+from sqlalchemy import inspect
 from sqlalchemy import Integer
 from sqlalchemy import select
 from sqlalchemy import String
@@ -14,13 +16,16 @@ from sqlalchemy.orm import aliased
 from sqlalchemy.orm import Composite
 from sqlalchemy.orm import composite
 from sqlalchemy.orm import configure_mappers
+from sqlalchemy.orm import mapped_column
 from sqlalchemy.orm import relationship
 from sqlalchemy.orm import Session
+from sqlalchemy.orm.attributes import LoaderCallableStatus
 from sqlalchemy.testing import assert_raises_message
 from sqlalchemy.testing import eq_
 from sqlalchemy.testing import expect_raises_message
 from sqlalchemy.testing import fixtures
 from sqlalchemy.testing import is_
+from sqlalchemy.testing import mock
 from sqlalchemy.testing.fixtures import fixture_session
 from sqlalchemy.testing.schema import Column
 from sqlalchemy.testing.schema import Table
@@ -851,6 +856,189 @@ class NestedTest(fixtures.MappedTest, testing.AssertsCompiledSQL):
         eq_(t1.ab, AB("a", "b", CD("c", "d")))
 
 
+class EventsEtcTest(fixtures.MappedTest):
+    @testing.fixture
+    def point_fixture(self, decl_base):
+        def go(active_history):
+            @dataclasses.dataclass
+            class Point:
+                x: int
+                y: int
+
+            class Edge(decl_base):
+                __tablename__ = "edge"
+                id = mapped_column(Integer, primary_key=True)
+
+                start = composite(
+                    Point,
+                    mapped_column("x1", Integer),
+                    mapped_column("y1", Integer),
+                    active_history=active_history,
+                )
+                end = composite(
+                    Point,
+                    mapped_column("x2", Integer, nullable=True),
+                    mapped_column("y2", Integer, nullable=True),
+                    active_history=active_history,
+                )
+
+            decl_base.metadata.create_all(testing.db)
+
+            return Point, Edge
+
+        return go
+
+    @testing.variation("active_history", [True, False])
+    @testing.variation("hist_on_mapping", [True, False])
+    def test_event_listener_no_value_to_set(
+        self, point_fixture, active_history, hist_on_mapping
+    ):
+        if hist_on_mapping:
+            config_active_history = bool(active_history)
+        else:
+            config_active_history = False
+
+        Point, Edge = point_fixture(config_active_history)
+
+        if not hist_on_mapping and active_history:
+            Edge.start.impl.active_history = True
+
+        m1 = mock.Mock()
+
+        event.listen(Edge.start, "set", m1)
+
+        e1 = Edge()
+        e1.start = Point(5, 6)
+
+        eq_(
+            m1.mock_calls,
+            [
+                mock.call(
+                    e1,
+                    Point(5, 6),
+                    LoaderCallableStatus.NO_VALUE
+                    if not active_history
+                    else None,
+                    Edge.start.impl,
+                )
+            ],
+        )
+
+        eq_(
+            inspect(e1).attrs.start.history,
+            ([Point(5, 6)], (), [Point(None, None)]),
+        )
+
+    @testing.variation("active_history", [True, False])
+    @testing.variation("hist_on_mapping", [True, False])
+    def test_event_listener_set_to_new(
+        self, point_fixture, active_history, hist_on_mapping
+    ):
+        if hist_on_mapping:
+            config_active_history = bool(active_history)
+        else:
+            config_active_history = False
+
+        Point, Edge = point_fixture(config_active_history)
+
+        if not hist_on_mapping and active_history:
+            Edge.start.impl.active_history = True
+
+        e1 = Edge()
+        e1.start = Point(5, 6)
+
+        sess = fixture_session()
+
+        sess.add(e1)
+        sess.commit()
+        assert "start" not in e1.__dict__
+
+        m1 = mock.Mock()
+
+        event.listen(Edge.start, "set", m1)
+
+        e1.start = Point(7, 8)
+
+        eq_(
+            m1.mock_calls,
+            [
+                mock.call(
+                    e1,
+                    Point(7, 8),
+                    LoaderCallableStatus.NO_VALUE
+                    if not active_history
+                    else Point(5, 6),
+                    Edge.start.impl,
+                )
+            ],
+        )
+
+        if active_history:
+            eq_(
+                inspect(e1).attrs.start.history,
+                ([Point(7, 8)], (), [Point(5, 6)]),
+            )
+        else:
+            eq_(
+                inspect(e1).attrs.start.history,
+                ([Point(7, 8)], (), [Point(None, None)]),
+            )
+
+    @testing.variation("active_history", [True, False])
+    @testing.variation("hist_on_mapping", [True, False])
+    def test_event_listener_set_to_deleted(
+        self, point_fixture, active_history, hist_on_mapping
+    ):
+        if hist_on_mapping:
+            config_active_history = bool(active_history)
+        else:
+            config_active_history = False
+
+        Point, Edge = point_fixture(config_active_history)
+
+        if not hist_on_mapping and active_history:
+            Edge.start.impl.active_history = True
+
+        e1 = Edge()
+        e1.start = Point(5, 6)
+
+        sess = fixture_session()
+
+        sess.add(e1)
+        sess.commit()
+        assert "start" not in e1.__dict__
+
+        m1 = mock.Mock()
+
+        event.listen(Edge.start, "remove", m1)
+
+        del e1.start
+
+        eq_(
+            m1.mock_calls,
+            [
+                mock.call(
+                    e1,
+                    LoaderCallableStatus.NO_VALUE
+                    if not active_history
+                    else Point(5, 6),
+                    Edge.start.impl,
+                )
+            ],
+        )
+
+        if active_history:
+            eq_(
+                inspect(e1).attrs.start.history,
+                ([Point(None, None)], (), [Point(5, 6)]),
+            )
+        else:
+            eq_(
+                inspect(e1).attrs.start.history,
+                ([Point(None, None)], (), [Point(None, None)]),
+            )
+
+
 class PrimaryKeyTest(fixtures.MappedTest):
     @classmethod
     def define_tables(cls, metadata):