disable col deduping inside of Bundle

Fixed issue where attribute key names in :class:`_orm.Bundle` would not be
correct when using ORM enabled :class:`_sql.select` vs.
:class:`_orm.Query`, when the statement contained duplicate column names.

Fixed issue in typing for :class:`_orm.Bundle` where creating a nested
:class:`_orm.Bundle` structure were not allowed.

Fixes: #11347
Change-Id: I24b37c99f83068c668736caaaa06e69a6801ff50

doc/build/changelog/unreleased_20/11347.rst

@@ -0,0 +1,13 @@
+.. change::
+    :tags: bug, orm
+    :tickets: 11347
+
+    Fixed issue where attribute key names in :class:`_orm.Bundle` would not be
+    correct when using ORM enabled :class:`_sql.select` vs.
+    :class:`_orm.Query`, when the statement contained duplicate column names.
+
+.. change::
+    :tags: bug, typing
+
+    Fixed issue in typing for :class:`_orm.Bundle` where creating a nested
+    :class:`_orm.Bundle` structure were not allowed.

lib/sqlalchemy/orm/context.py

@@ -455,7 +455,7 @@ class ORMCompileState(AbstractORMCompileState):
     ) -> _LabelConventionCallable:
         if legacy:
 
-            def name(col, col_name=None):
+            def name(col, col_name=None, cancel_dedupe=False):
                 if col_name:
                     return col_name
                 else:
@@ -3154,7 +3154,9 @@ class _ORMColumnEntity(_ColumnEntity):
 
         if is_current_entities:
             self._label_name = compile_state._label_convention(
-                column, col_name=orm_key
+                column,
+                col_name=orm_key,
+                cancel_dedupe=parent_bundle is not None,
             )
         else:
             self._label_name = None

lib/sqlalchemy/sql/_typing.py

@@ -184,6 +184,7 @@ _ColumnExpressionArgument = Union[
     _HasClauseElement[_T],
     "SQLCoreOperations[_T]",
     roles.ExpressionElementRole[_T],
+    roles.TypedColumnsClauseRole[_T],
     Callable[[], "ColumnElement[_T]"],
     "LambdaElement",
 ]

lib/sqlalchemy/sql/selectable.py

@@ -4556,6 +4556,7 @@ class SelectState(util.MemoizedSlots, CompileState):
         cls, label_style: SelectLabelStyle
     ) -> _LabelConventionCallable:
         table_qualified = label_style is LABEL_STYLE_TABLENAME_PLUS_COL
+
         dedupe = label_style is not LABEL_STYLE_NONE
 
         pa = prefix_anon_map()
@@ -4564,13 +4565,14 @@ class SelectState(util.MemoizedSlots, CompileState):
         def go(
             c: Union[ColumnElement[Any], TextClause],
             col_name: Optional[str] = None,
+            cancel_dedupe: bool = False,
         ) -> Optional[str]:
             if is_text_clause(c):
                 return None
             elif TYPE_CHECKING:
                 assert is_column_element(c)
 
-            if not dedupe:
+            if not dedupe or cancel_dedupe:
                 name = c._proxy_key
                 if name is None:
                     name = "_no_label"

test/orm/test_bundle.py

@@ -159,6 +159,65 @@ class BundleTest(fixtures.MappedTest, AssertsCompiledSQL):
             select(b1.c.d1, b1.c.d2), "SELECT data.d1, data.d2 FROM data"
         )
 
+    @testing.variation("stmt_type", ["legacy", "newstyle"])
+    def test_dupe_col_name(self, stmt_type):
+        """test #11347"""
+        Data = self.classes.Data
+        sess = fixture_session()
+
+        b1 = Bundle("b1", Data.d1, Data.d3)
+
+        if stmt_type.legacy:
+            row = (
+                sess.query(Data.d1, Data.d2, b1)
+                .filter(Data.d1 == "d0d1")
+                .one()
+            )
+        elif stmt_type.newstyle:
+            row = sess.execute(
+                select(Data.d1, Data.d2, b1).filter(Data.d1 == "d0d1")
+            ).one()
+
+        eq_(row[2]._mapping, {"d1": "d0d1", "d3": "d0d3"})
+
+    @testing.variation("stmt_type", ["legacy", "newstyle"])
+    def test_dupe_col_name_nested(self, stmt_type):
+        """test #11347"""
+        Data = self.classes.Data
+        sess = fixture_session()
+
+        class DictBundle(Bundle):
+            def create_row_processor(self, query, procs, labels):
+                def proc(row):
+                    return dict(zip(labels, (proc(row) for proc in procs)))
+
+                return proc
+
+        b1 = DictBundle("b1", Data.d1, Data.d3)
+        b2 = DictBundle("b2", Data.d2, Data.d3)
+        b3 = DictBundle("b3", Data.d2, Data.d3, b1, b2)
+
+        if stmt_type.legacy:
+            row = (
+                sess.query(Data.d1, Data.d2, b3)
+                .filter(Data.d1 == "d0d1")
+                .one()
+            )
+        elif stmt_type.newstyle:
+            row = sess.execute(
+                select(Data.d1, Data.d2, b3).filter(Data.d1 == "d0d1")
+            ).one()
+
+        eq_(
+            row[2],
+            {
+                "d2": "d0d2",
+                "d3": "d0d3",
+                "b1": {"d1": "d0d1", "d3": "d0d3"},
+                "b2": {"d2": "d0d2", "d3": "d0d3"},
+            },
+        )
+
     def test_result(self):
         Data = self.classes.Data
         sess = fixture_session()

test/typing/plain_files/orm/orm_querying.py

@@ -144,3 +144,8 @@ def test_10937() -> None:
     stmt3: ScalarSelect[str] = select(A.data + B.data).scalar_subquery()
 
     select(stmt, stmt2, stmt3, stmt1)
+
+
+def test_bundles() -> None:
+    b1 = orm.Bundle("b1", A.id, A.data)
+    orm.Bundle("b2", A.id, A.data, b1)