Raise at Core / ORM concrete inh level for label overlap

Fixed regression where the :class:`.ConcreteBase` would fail to map at all
when a mapped column name overlapped with the discriminator column name,
producing an assertion error. The use case here did not function correctly
in 1.3 as the polymorphic union would produce a query that ignored the
discriminator column entirely, while emitting duplicate column warnings. As
1.4's architecture cannot easily reproduce this essentially broken behavior
of 1.3 at the ``select()`` level right now, the use case now raises an
informative error message instructing the user to use the
``.ConcreteBase._concrete_discriminator_name`` attribute to resolve the
conflict. To assist with this configuration,
``.ConcreteBase._concrete_discriminator_name`` may be placed on the base
class only where it will be automatically used by subclasses; previously
this was not the case.

Fixes: #6090
Change-Id: I8b7d01e4c9ea0dc97f30b8cd658b3505b24312a7

6 files changed, 220 insertions, 11 deletions

diff --git a/doc/build/changelog/unreleased_14/6090.rst b/doc/build/changelog/unreleased_14/6090.rst
new file mode 100644
index 000000000..c22a664d3
--- /dev/null
+++ b/doc/build/changelog/unreleased_14/6090.rst
@@ -0,0 +1,18 @@
+.. change::
+    :tags: bug, orm
+    :tickets: 6090
+
+    Fixed regression where the :class:`.ConcreteBase` would fail to map at all
+    when a mapped column name overlapped with the discriminator column name,
+    producing an assertion error. The use case here did not function correctly
+    in 1.3 as the polymorphic union would produce a query that ignored the
+    discriminator column entirely, while emitting duplicate column warnings. As
+    1.4's architecture cannot easily reproduce this essentially broken behavior
+    of 1.3 at the ``select()`` level right now, the use case now raises an
+    informative error message instructing the user to use the
+    ``.ConcreteBase._concrete_discriminator_name`` attribute to resolve the
+    conflict. To assist with this configuration,
+    ``.ConcreteBase._concrete_discriminator_name`` may be placed on the base
+    class only where it will be automatically used by subclasses; previously
+    this was not the case.
+
diff --git a/lib/sqlalchemy/ext/declarative/extensions.py b/lib/sqlalchemy/ext/declarative/extensions.py
index 5c6356d8b..fd8bed6be 100644
--- a/lib/sqlalchemy/ext/declarative/extensions.py
+++ b/lib/sqlalchemy/ext/declarative/extensions.py
@@ -15,7 +15,6 @@ from ...orm import relationships
 from ...orm.base import _mapper_or_none
 from ...orm.clsregistry import _resolver
 from ...orm.decl_base import _DeferredMapperConfig
-from ...orm.decl_base import _get_immediate_cls_attr
 from ...orm.util import polymorphic_union
 from ...schema import Table
 from ...util import OrderedDict
@@ -86,6 +85,13 @@ class ConcreteBase(object):
        attribute to :class:`_declarative.ConcreteBase` so that the
        virtual discriminator column name can be customized.
 
+    .. versionchanged:: 1.4.2 The ``_concrete_discriminator_name`` attribute
+       need only be placed on the basemost class to take correct effect for
+       all subclasses.   An explicit error message is now raised if the
+       mapped column names conflict with the discriminator name, whereas
+       in the 1.3.x series there would be some warnings and then a non-useful
+       query would be generated.
+
     .. seealso::
 
         :class:`.AbstractConcreteBase`
@@ -112,8 +118,7 @@ class ConcreteBase(object):
             return
 
         discriminator_name = (
-            _get_immediate_cls_attr(cls, "_concrete_discriminator_name")
-            or "type"
+            getattr(cls, "_concrete_discriminator_name", None) or "type"
         )
 
         mappers = list(m.self_and_descendants)
@@ -264,8 +269,7 @@ class AbstractConcreteBase(ConcreteBase):
                 mappers.append(mn)
 
         discriminator_name = (
-            _get_immediate_cls_attr(cls, "_concrete_discriminator_name")
-            or "type"
+            getattr(cls, "_concrete_discriminator_name", None) or "type"
         )
         pjoin = cls._create_polymorphic_union(mappers, discriminator_name)
 
diff --git a/lib/sqlalchemy/orm/util.py b/lib/sqlalchemy/orm/util.py
index 290f50236..37be077be 100644
--- a/lib/sqlalchemy/orm/util.py
+++ b/lib/sqlalchemy/orm/util.py
@@ -228,6 +228,15 @@ def polymorphic_union(
 
         m = {}
         for c in table.c:
+            if c.key == typecolname:
+                raise sa_exc.InvalidRequestError(
+                    "Polymorphic union can't use '%s' as the discriminator "
+                    "column due to mapped column %r; please apply the "
+                    "'typecolname' "
+                    "argument; this is available on "
+                    "ConcreteBase as '_concrete_discriminator_name'"
+                    % (typecolname, c)
+                )
             colnames.add(c.key)
             m[c.key] = c
             types[c.key] = c.type
diff --git a/lib/sqlalchemy/sql/elements.py b/lib/sqlalchemy/sql/elements.py
index 29023c9fe..26c03b57b 100644
--- a/lib/sqlalchemy/sql/elements.py
+++ b/lib/sqlalchemy/sql/elements.py
@@ -4330,12 +4330,21 @@ class Label(roles.LabeledColumnExprRole, ColumnElement):
             disallow_is_literal=True,
             name_is_truncatable=isinstance(name, _truncated_label),
         )
-        # TODO: want to remove this assertion at some point.  all
-        # _make_proxy() implementations will give us back the key that
-        # is our "name" in the first place.  based on this we can
-        # safely return our "self.key" as the key here, to support a new
-        # case where the key and name are separate.
-        assert key == self.name
+
+        # there was a note here to remove this assertion, which was here
+        # to determine if we later could support a use case where
+        # the key and name of a label are separate.  But I don't know what
+        # that case was.  For now, this is an unexpected case that occurs
+        # when a label name conflicts with other columns and select()
+        # is attempting to disambiguate an explicit label, which is not what
+        # the user would want.   See issue #6090.
+        if key != self.name:
+            raise exc.InvalidRequestError(
+                "Label name %s is being renamed to an anonymous label due "
+                "to disambiguation "
+                "which is not supported right now.  Please use unique names "
+                "for explicit labels." % (self.name)
+            )
 
         e._propagate_attrs = selectable._propagate_attrs
         e._proxies.append(self)
diff --git a/test/ext/declarative/test_inheritance.py b/test/ext/declarative/test_inheritance.py
index 496b5579d..20d509ca5 100644
--- a/test/ext/declarative/test_inheritance.py
+++ b/test/ext/declarative/test_inheritance.py
@@ -1,5 +1,7 @@
+from sqlalchemy import exc as sa_exc
 from sqlalchemy import ForeignKey
 from sqlalchemy import Integer
+from sqlalchemy import select
 from sqlalchemy import String
 from sqlalchemy import testing
 from sqlalchemy.ext import declarative as decl
@@ -18,6 +20,7 @@ from sqlalchemy.testing import assert_raises_message
 from sqlalchemy.testing import eq_
 from sqlalchemy.testing import fixtures
 from sqlalchemy.testing import mock
+from sqlalchemy.testing.assertions import expect_raises_message
 from sqlalchemy.testing.fixtures import fixture_session
 from sqlalchemy.testing.schema import Column
 from sqlalchemy.testing.schema import Table
@@ -27,6 +30,8 @@ Base = None
 
 
 class DeclarativeTestBase(fixtures.TestBase, testing.AssertsExecutionResults):
+    __dialect__ = "default"
+
     def setup_test(self):
         global Base
         Base = decl.declarative_base(testing.db)
@@ -415,6 +420,139 @@ class ConcreteInhTest(
 
         self._roundtrip(Employee, Manager, Engineer, Boss)
 
+    def test_concrete_extension_warn_for_overlap(self):
+        class Employee(ConcreteBase, Base, fixtures.ComparableEntity):
+            __tablename__ = "employee"
+
+            employee_id = Column(
+                Integer, primary_key=True, test_needs_autoincrement=True
+            )
+            name = Column(String(50))
+            __mapper_args__ = {
+                "polymorphic_identity": "employee",
+                "concrete": True,
+            }
+
+        class Manager(Employee):
+            __tablename__ = "manager"
+            employee_id = Column(
+                Integer, primary_key=True, test_needs_autoincrement=True
+            )
+            type = Column(String(50))
+            __mapper_args__ = {
+                "polymorphic_identity": "manager",
+                "concrete": True,
+            }
+
+        with expect_raises_message(
+            sa_exc.InvalidRequestError,
+            "Polymorphic union can't use 'type' as the discriminator "
+            "column due to mapped column "
+            r"Column\('type', String\(length=50\), table=<manager>\); "
+            "please "
+            "apply the 'typecolname' argument; this is available on "
+            "ConcreteBase as '_concrete_discriminator_name'",
+        ):
+            configure_mappers()
+
+    def test_concrete_extension_warn_concrete_disc_resolves_overlap(self):
+        class Employee(ConcreteBase, Base, fixtures.ComparableEntity):
+            _concrete_discriminator_name = "_type"
+
+            __tablename__ = "employee"
+
+            employee_id = Column(
+                Integer, primary_key=True, test_needs_autoincrement=True
+            )
+            name = Column(String(50))
+            __mapper_args__ = {
+                "polymorphic_identity": "employee",
+                "concrete": True,
+            }
+
+        class Manager(Employee):
+            __tablename__ = "manager"
+            employee_id = Column(
+                Integer, primary_key=True, test_needs_autoincrement=True
+            )
+            type = Column(String(50))
+            __mapper_args__ = {
+                "polymorphic_identity": "manager",
+                "concrete": True,
+            }
+
+        configure_mappers()
+        self.assert_compile(
+            select(Employee),
+            "SELECT pjoin.employee_id, pjoin.name, pjoin._type, pjoin.type "
+            "FROM (SELECT employee.employee_id AS employee_id, "
+            "employee.name AS name, CAST(NULL AS VARCHAR(50)) AS type, "
+            "'employee' AS _type FROM employee UNION ALL "
+            "SELECT manager.employee_id AS employee_id, "
+            "CAST(NULL AS VARCHAR(50)) AS name, manager.type AS type, "
+            "'manager' AS _type FROM manager) AS pjoin",
+        )
+
+    def test_abs_concrete_extension_warn_for_overlap(self):
+        class Employee(AbstractConcreteBase, Base, fixtures.ComparableEntity):
+            name = Column(String(50))
+            __mapper_args__ = {
+                "polymorphic_identity": "employee",
+                "concrete": True,
+            }
+
+        class Manager(Employee):
+            __tablename__ = "manager"
+            employee_id = Column(
+                Integer, primary_key=True, test_needs_autoincrement=True
+            )
+            type = Column(String(50))
+            __mapper_args__ = {
+                "polymorphic_identity": "manager",
+                "concrete": True,
+            }
+
+        with expect_raises_message(
+            sa_exc.InvalidRequestError,
+            "Polymorphic union can't use 'type' as the discriminator "
+            "column due to mapped column "
+            r"Column\('type', String\(length=50\), table=<manager>\); "
+            "please "
+            "apply the 'typecolname' argument; this is available on "
+            "ConcreteBase as '_concrete_discriminator_name'",
+        ):
+            configure_mappers()
+
+    def test_abs_concrete_extension_warn_concrete_disc_resolves_overlap(self):
+        class Employee(AbstractConcreteBase, Base, fixtures.ComparableEntity):
+            _concrete_discriminator_name = "_type"
+
+            name = Column(String(50))
+            __mapper_args__ = {
+                "polymorphic_identity": "employee",
+                "concrete": True,
+            }
+
+        class Manager(Employee):
+            __tablename__ = "manager"
+            employee_id = Column(
+                Integer, primary_key=True, test_needs_autoincrement=True
+            )
+            type = Column(String(50))
+            __mapper_args__ = {
+                "polymorphic_identity": "manager",
+                "concrete": True,
+            }
+
+        configure_mappers()
+        self.assert_compile(
+            select(Employee),
+            "SELECT pjoin.name, pjoin.employee_id, pjoin.type, pjoin._type "
+            "FROM (SELECT manager.name AS name, manager.employee_id AS "
+            "employee_id, manager.type AS type, 'manager' AS _type "
+            "FROM manager) AS pjoin",
+        )
+
     def test_has_inherited_table_doesnt_consider_base(self):
         class A(Base):
             __tablename__ = "a"
diff --git a/test/sql/test_selectable.py b/test/sql/test_selectable.py
index 762146a3d..458b8f782 100644
--- a/test/sql/test_selectable.py
+++ b/test/sql/test_selectable.py
@@ -49,6 +49,7 @@ from sqlalchemy.testing import in_
 from sqlalchemy.testing import is_
 from sqlalchemy.testing import is_not
 from sqlalchemy.testing import ne_
+from sqlalchemy.testing.assertions import expect_raises_message
 
 
 metadata = MetaData()
@@ -208,6 +209,36 @@ class SelectableTest(
             checkparams={"param_1": 5},
         )
 
+    @testing.combinations((True,), (False,))
+    def test_broken_select_same_named_explicit_cols(self, use_anon):
+        # this is issue #6090.  the query is "wrong" and we dont know how
+        # to render this right now.
+        stmt = select(
+            table1.c.col1,
+            table1.c.col2,
+            literal_column("col2").label(None if use_anon else "col2"),
+        ).select_from(table1)
+
+        if use_anon:
+            self.assert_compile(
+                select(stmt.subquery()),
+                "SELECT anon_1.col1, anon_1.col2, anon_1.col2_1 FROM "
+                "(SELECT table1.col1 AS col1, table1.col2 AS col2, "
+                "col2 AS col2_1 FROM table1) AS anon_1",
+            )
+        else:
+            # the keys here are not critical as they are not what was
+            # requested anyway, maybe should raise here also.
+            eq_(stmt.selected_columns.keys(), ["col1", "col2", "col2_1"])
+            with expect_raises_message(
+                exc.InvalidRequestError,
+                "Label name col2 is being renamed to an anonymous "
+                "label due to "
+                "disambiguation which is not supported right now.  Please use "
+                "unique names for explicit labels.",
+            ):
+                select(stmt.subquery()).compile()
+
     def test_select_label_grouped_still_corresponds(self):
         label = select(table1.c.col1).label("foo")
         label2 = label.self_group()