Rework proxy_cache fix to restore performance

Adjustment to the fix made in I7fb134cac3604f8fe62e220fb24a0945d0a1c56f.

Fixes: #4747
Change-Id: I2f1010b0abc1faa892f5e346e58f9c4a3867622f

3 files changed, 31 insertions, 5 deletions

diff --git a/lib/sqlalchemy/sql/elements.py b/lib/sqlalchemy/sql/elements.py
index aa7b7f688..22799dab5 100644
--- a/lib/sqlalchemy/sql/elements.py
+++ b/lib/sqlalchemy/sql/elements.py
@@ -793,6 +793,18 @@ class ColumnElement(
             s.update(c.proxy_set)
         return s
 
+    def _uncached_proxy_set(self):
+        """An 'uncached' version of proxy set.
+
+        This is so that we can read annotations from the list of columns
+        without breaking the caching of the above proxy_set.
+
+        """
+        s = util.column_set([self])
+        for c in self._proxies:
+            s.update(c._uncached_proxy_set())
+        return s
+
     def shares_lineage(self, othercolumn):
         """Return True if the given :class:`.ColumnElement`
         has a common ancestor to this :class:`.ColumnElement`."""
@@ -4386,8 +4398,6 @@ class AnnotatedColumnElement(Annotated):
     def __init__(self, element, values):
         Annotated.__init__(self, element, values)
         ColumnElement.comparator._reset(self)
-        if self._proxies:
-            ColumnElement.proxy_set._reset(self)
         for attr in ("name", "key", "table"):
             if self.__dict__.get(attr, False) is None:
                 self.__dict__.pop(attr)
diff --git a/lib/sqlalchemy/sql/selectable.py b/lib/sqlalchemy/sql/selectable.py
index 2c898e835..591086a46 100644
--- a/lib/sqlalchemy/sql/selectable.py
+++ b/lib/sqlalchemy/sql/selectable.py
@@ -542,7 +542,7 @@ class FromClause(roles.FromClauseRole, Selectable):
                         operator.add,
                         [
                             sc._annotations.get("weight", 1)
-                            for sc in col.proxy_set
+                            for sc in col._uncached_proxy_set()
                             if sc.shares_lineage(column)
                         ],
                     )
@@ -550,7 +550,7 @@ class FromClause(roles.FromClauseRole, Selectable):
                         operator.add,
                         [
                             sc._annotations.get("weight", 1)
-                            for sc in c.proxy_set
+                            for sc in c._uncached_proxy_set()
                             if sc.shares_lineage(column)
                         ],
                     )
diff --git a/test/sql/test_selectable.py b/test/sql/test_selectable.py
index 1a53cd1d6..230b5423b 100644
--- a/test/sql/test_selectable.py
+++ b/test/sql/test_selectable.py
@@ -1965,7 +1965,23 @@ class AnnotationsTest(fixtures.TestBase):
         # proxy_set, as corresponding_column iterates through proxy_set
         # in this way
         d = {}
-        for col in p2.proxy_set:
+        for col in p2._uncached_proxy_set():
+            d.update(col._annotations)
+        eq_(d, {"weight": 10})
+
+    def test_proxy_set_iteration_includes_annotated_two(self):
+        from sqlalchemy.schema import Column
+
+        c1 = Column("foo", Integer)
+
+        stmt = select([c1]).alias()
+        proxy = stmt.c.foo
+        c1.proxy_set
+
+        proxy._proxies = [c1._annotate({"weight": 10})]
+
+        d = {}
+        for col in proxy._uncached_proxy_set():
             d.update(col._annotations)
         eq_(d, {"weight": 10})