Fix joinedload + of_type() + and_() invalid SQL for subclass columns

Fixed issue where using :func:`_orm.joinedload` with
:meth:`.PropComparator.of_type` targeting a joined-table subclass combined
with :meth:`.PropComparator.and_` referencing a column on that subclass
would generate invalid SQL, where the subclass column was not adapted to
the subquery alias.  Pull request courtesy Joaquin Hui Gomez.

Fixes #13203

Closes: #13206
Pull-request: https://github.com/sqlalchemy/sqlalchemy/pull/13206
Pull-request-sha: ba55b0c3e2a8dae28a1c7d7ae646e3480a04425c

Change-Id: I78fe4672649d1d5498e3bc653e5d943ccb55dafd

diff --git a/doc/build/changelog/unreleased_20/13203.rst b/doc/build/changelog/unreleased_20/13203.rst
new file mode 100644 (file)
index 0000000..c2adc10
--- /dev/null
+++ b/doc/build/changelog/unreleased_20/13203.rst
@@ -0,0 +1,9 @@
+.. change::
+    :tags: bug, orm
+    :tickets: 13203
+
+    Fixed issue where using :func:`_orm.joinedload` with
+    :meth:`.PropComparator.of_type` targeting a joined-table subclass combined
+    with :meth:`.PropComparator.and_` referencing a column on that subclass
+    would generate invalid SQL, where the subclass column was not adapted to
+    the subquery alias.  Pull request courtesy Joaquin Hui Gomez.

diff --git a/lib/sqlalchemy/orm/relationships.py b/lib/sqlalchemy/orm/relationships.py
index 22100e9ea6051a31fbd08c98e0ad720391d65f1d..8ac91415b1c55255c7e31c47ba58b9a49ce291d6 100644 (file)
--- a/lib/sqlalchemy/orm/relationships.py
+++ b/lib/sqlalchemy/orm/relationships.py
@@ -3381,8 +3381,14 @@ class _JoinCondition:
                 )
 
                 if (
+                    # NOTE: it's not clear yet if this needs to test for
+                    # parentmapper_for_element.isa(self.prop.parent).  so far
+                    # we have not come up with a test.
                     parentmapper_for_element is not self.prop.parent
-                    and parentmapper_for_element is not self.prop.mapper
+                    and (
+                        parentmapper_for_element is None
+                        or not parentmapper_for_element.isa(self.prop.mapper)
+                    )
                     and elem not in self._secondary_lineage_set
                 ):
                     return _safe_annotate(elem, annotations)

diff --git a/test/orm/test_relationship_criteria.py b/test/orm/test_relationship_criteria.py
index cd499a37a375bd844ec78956d6274c64865da135..a71cc3f13ddc4ffd417812bba079ed54fe38ce36 100644 (file)
--- a/test/orm/test_relationship_criteria.py
+++ b/test/orm/test_relationship_criteria.py
@@ -45,6 +45,7 @@ from sqlalchemy.testing import expect_raises_message
 from sqlalchemy.testing import fixtures
 from sqlalchemy.testing.assertions import expect_raises
 from sqlalchemy.testing.assertsql import CompiledSQL
+from sqlalchemy.testing.entities import ComparableEntity
 from sqlalchemy.testing.fixtures import fixture_session
 from sqlalchemy.testing.util import resolve_lambda
 from test.orm import _fixtures
@@ -2805,3 +2806,123 @@ class SubqueryCriteriaTest(fixtures.DeclarativeMappedTest):
                 ("Red-2", ["Red-1", "Orange-2"]),
             ],
         )
+
+
+class JoinedloadOfTypeAndTest(fixtures.DeclarativeMappedTest):
+    """test #13203
+
+    joinedload + of_type() + .and_() with a subclass column should
+    adapt the column reference to the subquery alias.
+    """
+
+    @classmethod
+    def setup_classes(cls):
+        Base = cls.DeclarativeBasic
+
+        class Animal(ComparableEntity, Base):
+            __tablename__ = "animal"
+            id: Mapped[int] = mapped_column(primary_key=True)
+            type: Mapped[str] = mapped_column(String(50))
+            name: Mapped[str] = mapped_column(String(50))
+            owner_id: Mapped[int] = mapped_column(
+                ForeignKey("owner.id"), nullable=True
+            )
+            owner = relationship("Owner", back_populates="animals")
+            __mapper_args__ = {
+                "polymorphic_on": "type",
+                "polymorphic_identity": "animal",
+            }
+
+        class Dog(Animal):
+            __tablename__ = "dog"
+            id: Mapped[int] = mapped_column(
+                ForeignKey("animal.id"), primary_key=True
+            )
+            breed: Mapped[str] = mapped_column(String(50))
+            __mapper_args__ = {"polymorphic_identity": "dog"}
+
+        class Owner(ComparableEntity, Base):
+            __tablename__ = "owner"
+            id: Mapped[int] = mapped_column(primary_key=True)
+            name: Mapped[str] = mapped_column(String(50))
+            animals = relationship("Animal", back_populates="owner")
+
+    @classmethod
+    def insert_data(cls, connection):
+        Animal, Dog, Owner = cls.classes("Animal", "Dog", "Owner")
+        with Session(connection) as s:
+            o = Owner(id=1, name="Alice")
+            s.add(o)
+            s.flush()
+            s.add(Dog(id=1, name="Rex", breed="Lab", owner=o))
+            s.add(Animal(id=2, name="Generic", type="animal", owner=o))
+            s.commit()
+
+    def test_joinedload_of_type_and_subclass_col(self):
+        Animal, Dog, Owner = self.classes("Animal", "Dog", "Owner")
+
+        stmt = select(Owner).options(
+            joinedload(Owner.animals.of_type(Dog).and_(Dog.breed == "Lab"))
+        )
+
+        with self.sql_execution_asserter(testing.db) as asserter_:
+            result = fixture_session().execute(stmt).unique().scalars().all()
+            eq_(
+                result,
+                [Owner(name="Alice", animals=[Dog(name="Rex", breed="Lab")])],
+            )
+
+        asserter_.assert_(
+            CompiledSQL(
+                "SELECT owner.id, owner.name,"
+                " anon_1.animal_id, anon_1.animal_type,"
+                " anon_1.animal_name, anon_1.animal_owner_id,"
+                " anon_1.dog_id, anon_1.dog_breed"
+                " FROM owner LEFT OUTER JOIN"
+                " (SELECT animal.id AS animal_id,"
+                " animal.type AS animal_type,"
+                " animal.name AS animal_name,"
+                " animal.owner_id AS animal_owner_id,"
+                " dog.id AS dog_id, dog.breed AS dog_breed"
+                " FROM animal LEFT OUTER JOIN dog"
+                " ON animal.id = dog.id) AS anon_1"
+                " ON owner.id = anon_1.animal_owner_id"
+                " AND anon_1.dog_breed = :breed_1",
+            ),
+        )
+
+    def test_selectinload_of_type_and_subclass_col(self):
+        Animal, Dog, Owner = self.classes("Animal", "Dog", "Owner")
+
+        stmt = select(Owner).options(
+            selectinload(Owner.animals.of_type(Dog).and_(Dog.breed == "Lab"))
+        )
+
+        with self.sql_execution_asserter(testing.db) as asserter_:
+            result = fixture_session().execute(stmt).unique().scalars().all()
+            eq_(
+                result,
+                [Owner(name="Alice", animals=[Dog(name="Rex", breed="Lab")])],
+            )
+
+        asserter_.assert_(
+            CompiledSQL(
+                "SELECT owner.id, owner.name FROM owner",
+            ),
+            CompiledSQL(
+                "SELECT anon_1.animal_owner_id,"
+                " anon_1.animal_id, anon_1.animal_type,"
+                " anon_1.animal_name,"
+                " anon_1.dog_id, anon_1.dog_breed"
+                " FROM (SELECT animal.id AS animal_id,"
+                " animal.type AS animal_type,"
+                " animal.name AS animal_name,"
+                " animal.owner_id AS animal_owner_id,"
+                " dog.id AS dog_id, dog.breed AS dog_breed"
+                " FROM animal LEFT OUTER JOIN dog"
+                " ON animal.id = dog.id) AS anon_1"
+                " WHERE anon_1.animal_owner_id IN"
+                " (__[POSTCOMPILE_primary_keys])"
+                " AND anon_1.dog_breed = :breed_1",
+            ),
+        )