Further refine labeling for renamed columns

Forked from I22f6cf0f0b3360e55299cdcb2452cead2b2458ea
we are attempting to decide the case for columns mapped
under a different name.   since the .key feature of
Column seems to support this fully, see if an annotation
can be used to indicate an effective .key for a column.

The effective change is that the labeling of column expressions
in rows has been improved to retain the original name of the ORM
attribute even if used in a subquery.

References: #5933
Change-Id: If251f556f7d723f50d349f765f1690d6c679d2ef

10 files changed, 59 insertions, 41 deletions

diff --git a/lib/sqlalchemy/orm/attributes.py b/lib/sqlalchemy/orm/attributes.py
index dd354c4e0..b96b3b61e 100644
--- a/lib/sqlalchemy/orm/attributes.py
+++ b/lib/sqlalchemy/orm/attributes.py
@@ -212,7 +212,7 @@ class QueryableAttribute(
         """
 
         return self.comparator.__clause_element__()._annotate(
-            {"orm_key": self.key, "entity_namespace": self._entity_namespace}
+            {"proxy_key": self.key, "entity_namespace": self._entity_namespace}
         )
 
     @property
diff --git a/lib/sqlalchemy/orm/context.py b/lib/sqlalchemy/orm/context.py
index 621ed826c..fa192a17e 100644
--- a/lib/sqlalchemy/orm/context.py
+++ b/lib/sqlalchemy/orm/context.py
@@ -2619,12 +2619,12 @@ class _ORMColumnEntity(_ColumnEntity):
 
         _entity = parententity
 
-        # an AliasedClass won't have orm_key in the annotations for
+        # an AliasedClass won't have proxy_key in the annotations for
         # a column if it was acquired using the class' adapter directly,
         # such as using AliasedInsp._adapt_element().  this occurs
         # within internal loaders.
 
-        orm_key = annotations.get("orm_key", None)
+        orm_key = annotations.get("proxy_key", None)
         if orm_key:
             self.expr = getattr(_entity.entity, orm_key)
         else:
diff --git a/lib/sqlalchemy/orm/descriptor_props.py b/lib/sqlalchemy/orm/descriptor_props.py
index 695b1a7b4..65fe58e61 100644
--- a/lib/sqlalchemy/orm/descriptor_props.py
+++ b/lib/sqlalchemy/orm/descriptor_props.py
@@ -413,7 +413,7 @@ class CompositeProperty(DescriptorProperty):
                 {
                     "parententity": self._parententity,
                     "parentmapper": self._parententity,
-                    "orm_key": self.prop.key,
+                    "proxy_key": self.prop.key,
                 }
             )
             return CompositeProperty.CompositeBundle(self.prop, clauses)
diff --git a/lib/sqlalchemy/orm/persistence.py b/lib/sqlalchemy/orm/persistence.py
index 03de64392..f19f29daa 100644
--- a/lib/sqlalchemy/orm/persistence.py
+++ b/lib/sqlalchemy/orm/persistence.py
@@ -2018,7 +2018,7 @@ class BulkUDCompileState(CompileState):
                     elif "entity_namespace" in k._annotations:
                         k_anno = k._annotations
                         attr = _entity_namespace_key(
-                            k_anno["entity_namespace"], k_anno["orm_key"]
+                            k_anno["entity_namespace"], k_anno["proxy_key"]
                         )
                         values.extend(attr._bulk_update_tuples(v))
                     else:
diff --git a/lib/sqlalchemy/orm/properties.py b/lib/sqlalchemy/orm/properties.py
index 4d0c7528b..7823aca20 100644
--- a/lib/sqlalchemy/orm/properties.py
+++ b/lib/sqlalchemy/orm/properties.py
@@ -359,7 +359,7 @@ class ColumnProperty(StrategizedProperty):
                 "entity_namespace": pe,
                 "parententity": pe,
                 "parentmapper": pe,
-                "orm_key": self.prop.key,
+                "proxy_key": self.prop.key,
             }
 
             col = column
diff --git a/lib/sqlalchemy/orm/relationships.py b/lib/sqlalchemy/orm/relationships.py
index 1a0140914..b0a2f7b08 100644
--- a/lib/sqlalchemy/orm/relationships.py
+++ b/lib/sqlalchemy/orm/relationships.py
@@ -2693,11 +2693,11 @@ class JoinCondition(object):
         """
 
         self.primaryjoin = _deep_deannotate(
-            self.primaryjoin, values=("parententity", "orm_key")
+            self.primaryjoin, values=("parententity", "proxy_key")
         )
         if self.secondaryjoin is not None:
             self.secondaryjoin = _deep_deannotate(
-                self.secondaryjoin, values=("parententity", "orm_key")
+                self.secondaryjoin, values=("parententity", "proxy_key")
             )
 
     def _determine_joins(self):
diff --git a/lib/sqlalchemy/orm/util.py b/lib/sqlalchemy/orm/util.py
index 1bc0ceb4d..290f50236 100644
--- a/lib/sqlalchemy/orm/util.py
+++ b/lib/sqlalchemy/orm/util.py
@@ -771,7 +771,7 @@ class AliasedInsp(
             "compile_state_plugin": "orm",
         }
         if key:
-            d["orm_key"] = key
+            d["proxy_key"] = key
         return (
             self._adapter.traverse(elem)
             ._annotate(d)
diff --git a/lib/sqlalchemy/sql/elements.py b/lib/sqlalchemy/sql/elements.py
index 2bd1c3ae3..85200bf25 100644
--- a/lib/sqlalchemy/sql/elements.py
+++ b/lib/sqlalchemy/sql/elements.py
@@ -894,7 +894,9 @@ class ColumnElement(
 
     @util.memoized_property
     def _proxy_key(self):
-        if self.key:
+        if self._annotations and "proxy_key" in self._annotations:
+            return self._annotations["proxy_key"]
+        elif self.key:
             return self.key
         else:
             try:
@@ -987,7 +989,22 @@ class ColumnElement(
         expressions and function calls.
 
         """
-        return self._anon_label(getattr(self, "name", None))
+        name = getattr(self, "name", None)
+        return self._anon_label(name)
+
+    @util.memoized_property
+    def anon_key_label(self):
+        """Provides a constant 'anonymous key label' for this ColumnElement.
+
+        Compare to ``anon_label``, except that the "key" of the column,
+        if available, is used to generate the label.
+
+        This is used when a deduplicating key is placed into the columns
+        collection of a selectable.
+
+        """
+        name = getattr(self, "key", None) or getattr(self, "name", None)
+        return self._anon_label(name)
 
     @util.memoized_property
     def _dedupe_anon_label(self):
@@ -999,6 +1016,10 @@ class ColumnElement(
         return self._anon_label(getattr(self, "_label", None))
 
     @util.memoized_property
+    def _label_anon_key_label(self):
+        return self._anon_label(getattr(self, "_key_label", None))
+
+    @util.memoized_property
     def _dedupe_label_anon_label(self):
         label = getattr(self, "_label", None) or "anon"
         return self._anon_label(label + "_")
@@ -3720,9 +3741,6 @@ class Grouping(GroupedElement, ColumnElement):
     def _key_label(self):
         return self._label
 
-    def _gen_label(self, name):
-        return name
-
     @property
     def _label(self):
         return getattr(self.element, "_label", None) or self.anon_label
@@ -4345,8 +4363,9 @@ class NamedColumn(ColumnElement):
 
     @HasMemoized.memoized_attribute
     def _key_label(self):
-        if self.key != self.name:
-            return self._gen_label(self.key)
+        proxy_key = self._proxy_key
+        if proxy_key != self.name:
+            return self._gen_label(proxy_key)
         else:
             return self._label
 
@@ -4859,7 +4878,8 @@ def _corresponding_column_or_error(fromclause, column, require_embedded=False):
 class AnnotatedColumnElement(Annotated):
     def __init__(self, element, values):
         Annotated.__init__(self, element, values)
-        self.__dict__.pop("comparator", None)
+        for attr in ("comparator", "_proxy_key", "_key_label"):
+            self.__dict__.pop(attr, None)
         for attr in ("name", "key", "table"):
             if self.__dict__.get(attr, False) is None:
                 self.__dict__.pop(attr)
diff --git a/lib/sqlalchemy/sql/schema.py b/lib/sqlalchemy/sql/schema.py
index 127b12e81..6e3c9dbfb 100644
--- a/lib/sqlalchemy/sql/schema.py
+++ b/lib/sqlalchemy/sql/schema.py
@@ -1928,6 +1928,7 @@ class Column(DialectKWArgs, SchemaItem, ColumnClause):
                 if name_is_truncatable
                 else (name or self.name),
                 self.type,
+                # this may actually be ._proxy_key when the key is incoming
                 key=key if key else name if name else self.key,
                 primary_key=self.primary_key,
                 nullable=self.nullable,
diff --git a/lib/sqlalchemy/sql/selectable.py b/lib/sqlalchemy/sql/selectable.py
index 7e2c5dd3b..23fdf7e12 100644
--- a/lib/sqlalchemy/sql/selectable.py
+++ b/lib/sqlalchemy/sql/selectable.py
@@ -4078,51 +4078,42 @@ class SelectState(util.MemoizedSlots, CompileState):
 
     @classmethod
     def _column_naming_convention(cls, label_style):
-        names = set()
-        pa = []
-
         if label_style is LABEL_STYLE_NONE:
 
             def go(c, col_name=None):
-                return col_name or c._proxy_key
+                return c._proxy_key
 
         elif label_style is LABEL_STYLE_TABLENAME_PLUS_COL:
+            names = set()
+            pa = []  # late-constructed as needed, python 2 has no "nonlocal"
 
             def go(c, col_name=None):
                 # we use key_label since this name is intended for targeting
                 # within the ColumnCollection only, it's not related to SQL
                 # rendering which always uses column name for SQL label names
 
-                if col_name:
-                    name = c._gen_label(col_name)
-                else:
-                    name = c._key_label
+                name = c._key_label
 
                 if name in names:
                     if not pa:
                         pa.append(prefix_anon_map())
 
-                    name = c._label_anon_label % pa[0]
+                    name = c._label_anon_key_label % pa[0]
                 else:
                     names.add(name)
 
                 return name
 
         else:
+            names = set()
+            pa = []  # late-constructed as needed, python 2 has no "nonlocal"
 
             def go(c, col_name=None):
-                # we use key_label since this name is intended for targeting
-                # within the ColumnCollection only, it's not related to SQL
-                # rendering which always uses column name for SQL label names
-                if col_name:
-                    name = col_name
-                else:
-                    name = c._proxy_key
+                name = c._proxy_key
                 if name in names:
                     if not pa:
                         pa.append(prefix_anon_map())
-
-                    name = c.anon_label % pa[0]
+                    name = c.anon_key_label % pa[0]
                 else:
                     names.add(name)
 
@@ -5617,6 +5608,14 @@ class Select(
         return self
 
     def _generate_columns_plus_names(self, anon_for_dupe_key):
+        """Generate column names as rendered in a SELECT statement by
+        the compiler.
+
+        This is distinct from other name generators that are intended for
+        population of .c collections and similar, which may have slightly
+        different rules.
+
+        """
         cols = self._exported_columns_iterator()
 
         # when use_labels is on:
@@ -5732,19 +5731,17 @@ class Select(
                 if key is not None and key in keys_seen:
                     if pa is None:
                         pa = prefix_anon_map()
-                    key = c._label_anon_label % pa
+                    key = c._label_anon_key_label % pa
                 keys_seen.add(key)
             elif disambiguate_only:
-                key = c.key
+                key = c._proxy_key
                 if key is not None and key in keys_seen:
                     if pa is None:
                         pa = prefix_anon_map()
-                    key = c.anon_label % pa
+                    key = c.anon_key_label % pa
                 keys_seen.add(key)
             else:
-                # one of the above label styles is set for subqueries
-                # as of #5221 so this codepath is likely not called now.
-                key = None
+                key = c._proxy_key
             prox.append(
                 c._make_proxy(
                     subquery, key=key, name=name, name_is_truncatable=True