repair legacy_last_joined_entity for no onclause

Fixed regression where the :meth:`_orm.Query.filter_by` method would fail
to locate the correct source entity if the :meth:`_orm.Query.join` method
had been used targeting an entity without any kind of ON clause.

Fixes: #6092
Change-Id: I38d9099844f842f314c6673bd922467242409cdb

diff --git a/doc/build/changelog/unreleased_14/6092.rst b/doc/build/changelog/unreleased_14/6092.rst
new file mode 100644 (file)
index 0000000..76108f8
--- /dev/null
+++ b/doc/build/changelog/unreleased_14/6092.rst
@@ -0,0 +1,8 @@
+.. change::
+    :tags: bug, orm, regression
+    :tickets: 6092
+
+    Fixed regression where the :meth:`_orm.Query.filter_by` method would fail
+    to locate the correct source entity if the :meth:`_orm.Query.join` method
+    had been used targeting an entity without any kind of ON clause.
+

diff --git a/lib/sqlalchemy/orm/context.py b/lib/sqlalchemy/orm/context.py
index 9cfa0fdc6d7125deceb42fea00c73646f7362e37..45ab64593566d10c4d2f2928231a3f5aa18522e7 100644 (file)
--- a/lib/sqlalchemy/orm/context.py
+++ b/lib/sqlalchemy/orm/context.py
@@ -2136,7 +2136,7 @@ def _legacy_determine_last_joined_entity(setup_joins, entity_zero):
     if right is not None and "parententity" in right._annotations:
         right = right._annotations["parententity"].entity
 
-    if onclause is not None and right is not None:
+    if right is not None:
         last_entity = right
         insp = inspect(last_entity)
         if insp.is_clause_element or insp.is_aliased_class or insp.is_mapper:

diff --git a/test/orm/test_joins.py b/test/orm/test_joins.py
index 6dc1d5cead96e1019c5e15d2ebd17735dbc0735e..50fa861070e640ac542ba5758ad5ebdd598e4abb 100644 (file)
--- a/test/orm/test_joins.py
+++ b/test/orm/test_joins.py
@@ -1,3 +1,5 @@
+import itertools
+
 import sqlalchemy as sa
 from sqlalchemy import and_
 from sqlalchemy import desc
@@ -232,23 +234,79 @@ class JoinOnSynonymTest(_fixtures.FixtureTest, AssertsCompiledSQL):
 class JoinTest(QueryTest, AssertsCompiledSQL):
     __dialect__ = "default"
 
-    def test_filter_by_from_full_join(self):
+    @testing.combinations_list(
+        set(
+            itertools.product(
+                [
+                    "relationship",
+                    "relationship_only",
+                    "string_relationship",
+                    "string_relationship_only",
+                    "none",
+                    "explicit",
+                    "table_none",
+                    "table_explicit",
+                ],
+                [True, False],
+            )
+        ).difference(
+            [
+                ("string_relationship", False),
+                ("string_relationship_only", False),
+            ]
+        ),
+        argnames="onclause_type, use_legacy",
+    )
+    def test_filter_by_from_join(self, onclause_type, use_legacy):
         User, Address = self.classes("User", "Address")
+        (address_table,) = self.tables("addresses")
+        (user_table,) = self.tables("users")
+
+        if use_legacy:
+            sess = fixture_session()
+            q = sess.query(User)
+        else:
+            q = select(User).set_label_style(LABEL_STYLE_TABLENAME_PLUS_COL)
+
+        if onclause_type == "relationship":
+            q = q.join(Address, User.addresses)
+        elif onclause_type == "string_relationship":
+            q = q.join(Address, "addresses")
+        elif onclause_type == "relationship_only":
+            q = q.join(User.addresses)
+        elif onclause_type == "string_relationship_only":
+            q = q.join("addresses")
+        elif onclause_type == "none":
+            q = q.join(Address)
+        elif onclause_type == "explicit":
+            q = q.join(Address, User.id == Address.user_id)
+        elif onclause_type == "table_none":
+            q = q.join(address_table)
+        elif onclause_type == "table_explicit":
+            q = q.join(
+                address_table, user_table.c.id == address_table.c.user_id
+            )
+        else:
+            assert False
 
-        sess = fixture_session()
+        q2 = q.filter_by(email_address="foo")
 
-        q = (
-            sess.query(User)
-            .join(Address, User.addresses)
-            .filter_by(email_address="foo")
-        )
         self.assert_compile(
-            q,
+            q2,
             "SELECT users.id AS users_id, users.name AS users_name "
             "FROM users JOIN addresses ON users.id = addresses.user_id "
             "WHERE addresses.email_address = :email_address_1",
         )
 
+        if use_legacy:
+            q2 = q.reset_joinpoint().filter_by(name="user")
+            self.assert_compile(
+                q2,
+                "SELECT users.id AS users_id, users.name AS users_name "
+                "FROM users JOIN addresses ON users.id = addresses.user_id "
+                "WHERE users.name = :name_1",
+            )
+
     def test_invalid_kwarg_join(self):
         User = self.classes.User
         sess = fixture_session()