Merge "use annotated entity when adding secondary" into main

3 files changed, 82 insertions, 2 deletions

diff --git a/doc/build/changelog/unreleased_14/7868.rst b/doc/build/changelog/unreleased_14/7868.rst
new file mode 100644
index 000000000..d57b22296
--- /dev/null
+++ b/doc/build/changelog/unreleased_14/7868.rst
@@ -0,0 +1,9 @@
+.. change::
+    :tags: bug, orm, regression
+    :tickets: 7868
+
+    Fixed regression in "dynamic" loader strategy where the
+    :meth:`_orm.Query.filter_by` method would not be given an appropriate
+    entity to filter from, in the case where a "secondary" table were present
+    in the relationship being queried and the mapping were against something
+    complex such as a "with polymorphic".
diff --git a/lib/sqlalchemy/orm/dynamic.py b/lib/sqlalchemy/orm/dynamic.py
index 52a6ec4b0..a3b02bb94 100644
--- a/lib/sqlalchemy/orm/dynamic.py
+++ b/lib/sqlalchemy/orm/dynamic.py
@@ -304,7 +304,10 @@ class AppenderMixin:
             # is in the FROM.  So we purposely put the mapper selectable
             # in _from_obj[0] to ensure a user-defined join() later on
             # doesn't fail, and secondary is then in _from_obj[1].
-            self._from_obj = (prop.mapper.selectable, prop.secondary)
+
+            # note also, we are using the official ORM-annotated selectable
+            # from __clause_element__(), see #7868
+            self._from_obj = (prop.mapper.__clause_element__(), prop.secondary)
 
         self._where_criteria = (
             prop._with_parent(instance, alias_secondary=False),
diff --git a/test/orm/test_dynamic.py b/test/orm/test_dynamic.py
index 2eaafb953..0e2cb34d2 100644
--- a/test/orm/test_dynamic.py
+++ b/test/orm/test_dynamic.py
@@ -1,10 +1,13 @@
 from sqlalchemy import cast
+from sqlalchemy import Column
 from sqlalchemy import desc
 from sqlalchemy import exc
+from sqlalchemy import ForeignKey
 from sqlalchemy import func
 from sqlalchemy import inspect
 from sqlalchemy import Integer
 from sqlalchemy import select
+from sqlalchemy import String
 from sqlalchemy import testing
 from sqlalchemy.orm import attributes
 from sqlalchemy.orm import backref
@@ -13,6 +16,7 @@ from sqlalchemy.orm import exc as orm_exc
 from sqlalchemy.orm import noload
 from sqlalchemy.orm import Query
 from sqlalchemy.orm import relationship
+from sqlalchemy.orm.session import make_transient_to_detached
 from sqlalchemy.testing import assert_raises
 from sqlalchemy.testing import assert_raises_message
 from sqlalchemy.testing import assert_warns_message
@@ -125,6 +129,8 @@ class _DynamicFixture:
 
 
 class DynamicTest(_DynamicFixture, _fixtures.FixtureTest, AssertsCompiledSQL):
+    __dialect__ = "default"
+
     def test_basic(self):
         User, Address = self._user_address_fixture()
         sess = fixture_session()
@@ -598,11 +604,17 @@ class DynamicTest(_DynamicFixture, _fixtures.FixtureTest, AssertsCompiledSQL):
                 )
             },
         )
-        self.mapper_registry.map_imperatively(Item, items)
+        item_mapper = self.mapper_registry.map_imperatively(Item, items)
 
         sess = fixture_session()
+
         u1 = sess.query(User).first()
 
+        dyn = u1.items
+
+        # test for #7868
+        eq_(dyn._from_obj[0]._annotations["parententity"], item_mapper)
+
         self.assert_compile(
             u1.items,
             "SELECT items.id AS items_id, "
@@ -614,6 +626,62 @@ class DynamicTest(_DynamicFixture, _fixtures.FixtureTest, AssertsCompiledSQL):
             use_default_dialect=True,
         )
 
+    def test_secondary_as_join_complex_entity(self, registry):
+        """integration test for #7868"""
+        Base = registry.generate_base()
+
+        class GrandParent(Base):
+            __tablename__ = "grandparent"
+            id = Column(Integer, primary_key=True)
+
+            grand_children = relationship(
+                "Child", secondary="parent", lazy="dynamic", viewonly=True
+            )
+
+        class Parent(Base):
+            __tablename__ = "parent"
+            id = Column(Integer, primary_key=True)
+            grand_parent_id = Column(
+                Integer, ForeignKey("grandparent.id"), nullable=False
+            )
+
+        class Child(Base):
+            __tablename__ = "child"
+            id = Column(Integer, primary_key=True)
+            type = Column(String)
+            parent_id = Column(
+                Integer, ForeignKey("parent.id"), nullable=False
+            )
+
+            __mapper_args__ = {
+                "polymorphic_on": type,
+                "polymorphic_identity": "unknown",
+                "with_polymorphic": "*",
+            }
+
+        class SubChild(Child):
+            __tablename__ = "subchild"
+            id = Column(Integer, ForeignKey("child.id"), primary_key=True)
+
+            __mapper_args__ = {
+                "polymorphic_identity": "sub",
+            }
+
+        gp = GrandParent(id=1)
+        make_transient_to_detached(gp)
+        sess = fixture_session()
+        sess.add(gp)
+        self.assert_compile(
+            gp.grand_children.filter_by(id=1),
+            "SELECT child.id AS child_id, child.type AS child_type, "
+            "child.parent_id AS child_parent_id, subchild.id AS subchild_id "
+            "FROM parent, child LEFT OUTER JOIN subchild "
+            "ON child.id = subchild.id "
+            "WHERE :param_1 = parent.grand_parent_id "
+            "AND parent.id = child.parent_id AND child.id = :id_1",
+            {"id_1": 1},
+        )
+
     def test_secondary_doesnt_interfere_w_join_to_fromlist(self):
         # tests that the "secondary" being added to the FROM
         # as part of [ticket:4349] does not prevent a subsequent join to