ensure composite refresh handler synced w/ mutable composite

Fixed issue where the :class:`_mutable.MutableComposite` construct could be
placed into an invalid state when the parent object was already loaded, and
then covered by a subsequent query, due to the composite properties'
refresh handler replacing the object with a new one not handled by the
mutable extension.

Fixes: #6001
Change-Id: Ieebd8e6afe6b65f8902cc12dec1efb968f5438ef
(cherry picked from commit 2f7623b6b265cd5f25f2a6022e21bc3286d397a3)

3 files changed, 71 insertions, 7 deletions

diff --git a/doc/build/changelog/unreleased_13/6001.rst b/doc/build/changelog/unreleased_13/6001.rst
new file mode 100644
index 000000000..2b6f1bc09
--- /dev/null
+++ b/doc/build/changelog/unreleased_13/6001.rst
@@ -0,0 +1,10 @@
+.. change::
+    :tags: bug, orm
+    :tickets: 6001
+
+    Fixed issue where the :class:`_mutable.MutableComposite` construct could be
+    placed into an invalid state when the parent object was already loaded, and
+    then covered by a subsequent query, due to the composite properties'
+    refresh handler replacing the object with a new one not handled by the
+    mutable extension.
+
diff --git a/lib/sqlalchemy/orm/descriptor_props.py b/lib/sqlalchemy/orm/descriptor_props.py
index 6ec9d229f..3df9d0474 100644
--- a/lib/sqlalchemy/orm/descriptor_props.py
+++ b/lib/sqlalchemy/orm/descriptor_props.py
@@ -184,6 +184,8 @@ class CompositeProperty(DescriptorProperty):
         """
         self._setup_arguments_on_columns()
 
+    _COMPOSITE_FGET = object()
+
     def _create_descriptor(self):
         """Create the Python descriptor that will serve as
         the access point on instances of the mapped class.
@@ -211,7 +213,9 @@ class CompositeProperty(DescriptorProperty):
                     state.key is not None or not _none_set.issuperset(values)
                 ):
                     dict_[self.key] = self.composite_class(*values)
-                    state.manager.dispatch.refresh(state, None, [self.key])
+                    state.manager.dispatch.refresh(
+                        state, self._COMPOSITE_FGET, [self.key]
+                    )
 
             return dict_.get(self.key, None)
 
@@ -285,16 +289,32 @@ class CompositeProperty(DescriptorProperty):
     def _setup_event_handlers(self):
         """Establish events that populate/expire the composite attribute."""
 
-        def load_handler(state, *args):
-            _load_refresh_handler(state, args, is_refresh=False)
+        def load_handler(state, context):
+            _load_refresh_handler(state, context, None, is_refresh=False)
+
+        def refresh_handler(state, context, to_load):
+            # note this corresponds to sqlalchemy.ext.mutable load_attrs()
 
-        def refresh_handler(state, *args):
-            _load_refresh_handler(state, args, is_refresh=True)
+            if not to_load or (
+                {self.key}.union(self._attribute_keys)
+            ).intersection(to_load):
+                _load_refresh_handler(state, context, to_load, is_refresh=True)
 
-        def _load_refresh_handler(state, args, is_refresh):
+        def _load_refresh_handler(state, context, to_load, is_refresh):
             dict_ = state.dict
 
-            if not is_refresh and self.key in dict_:
+            # if context indicates we are coming from the
+            # fget() handler, this already set the value; skip the
+            # handler here. (other handlers like mutablecomposite will still
+            # want to catch it)
+            # there's an insufficiency here in that the fget() handler
+            # really should not be using the refresh event and there should
+            # be some other event that mutablecomposite can subscribe
+            # towards for this.
+
+            if (
+                not is_refresh or context is self._COMPOSITE_FGET
+            ) and self.key in dict_:
                 return
 
             # if column elements aren't loaded, skip.
diff --git a/test/ext/test_mutable.py b/test/ext/test_mutable.py
index acb0ad490..6737547ac 100644
--- a/test/ext/test_mutable.py
+++ b/test/ext/test_mutable.py
@@ -22,6 +22,7 @@ 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 is_
 from sqlalchemy.testing import mock
 from sqlalchemy.testing.schema import Column
 from sqlalchemy.testing.schema import Table
@@ -1388,6 +1389,39 @@ class MutableCompositesTest(_CompositeTestBase, fixtures.MappedTest):
 
         eq_(f1.data.x, 5)
 
+    def test_dont_reset_on_attr_refresh(self):
+        sess = Session()
+        f1 = Foo(data=Point(3, 4), unrelated_data="unrelated")
+        sess.add(f1)
+        sess.flush()
+
+        f1.data.x = 5
+
+        # issue 6001, this would reload a new Point() that would be missed
+        # by the mutable composite, and tracking would be lost
+        sess.refresh(f1, ["unrelated_data"])
+
+        is_(list(f1.data._parents.keys())[0], f1)
+
+        f1.data.y = 9
+
+        sess.commit()
+
+        eq_(f1.data.x, 5)
+        eq_(f1.data.y, 9)
+
+        f1.data.x = 12
+
+        sess.refresh(f1, ["unrelated_data", "y"])
+
+        is_(list(f1.data._parents.keys())[0], f1)
+
+        f1.data.y = 15
+        sess.commit()
+
+        eq_(f1.data.x, 12)
+        eq_(f1.data.y, 15)
+
 
 class MutableCompositeCallableTest(_CompositeTestBase, fixtures.MappedTest):
     @classmethod