Adapt right side in join if lateral detected

Fixed bug where use of :class:`.Lateral` construct in conjunction with
:meth:`.Query.join` as well as :meth:`.Query.select_entity_from` would not
apply clause adaption to the right side of the join.   "lateral" introduces
the use case of the right side of a join being correlatable.  Previously,
adaptation of this clause wasn't considered.

Fixes: #4334
Change-Id: I3631e562092769d30069a2aa5e50a580f4661a23

4 files changed, 249 insertions, 1 deletions

diff --git a/doc/build/changelog/unreleased_12/4334.rst b/doc/build/changelog/unreleased_12/4334.rst
new file mode 100644
index 000000000..26688b180
--- /dev/null
+++ b/doc/build/changelog/unreleased_12/4334.rst
@@ -0,0 +1,13 @@
+.. change::
+    :tags: bug, orm
+    :tickets: 4334
+    :versions: 1.3.0b1
+
+    Fixed bug where use of :class:`.Lateral` construct in conjunction with
+    :meth:`.Query.join` as well as :meth:`.Query.select_entity_from` would not
+    apply clause adaption to the right side of the join.   "lateral" introduces
+    the use case of the right side of a join being correlatable.  Previously,
+    adaptation of this clause wasn't considered.   Note that in 1.2 only,
+    a selectable introduced by :meth:`.Query.subquery` is still not adapted
+    due to :ticket:`4304`; the selectable needs to be produced by the
+    :func:`.select` function to be the right side of the "lateral" join.
diff --git a/lib/sqlalchemy/orm/query.py b/lib/sqlalchemy/orm/query.py
index 627a4e01c..7e7c93527 100644
--- a/lib/sqlalchemy/orm/query.py
+++ b/lib/sqlalchemy/orm/query.py
@@ -542,6 +542,7 @@ class Query(object):
         if with_labels:
             q = q.with_labels()
         q = q.statement
+
         if reduce_columns:
             q = q.reduce_columns()
         return q.alias(name=name)
@@ -2381,6 +2382,12 @@ class Query(object):
 
         need_adapter = False
 
+        if r_info.is_clause_element and right_selectable._is_lateral:
+            # orm_only is disabled to suit the case where we have to
+            # adapt an explicit correlate(Entity) - the select() loses
+            # the ORM-ness in this case right now, ideally it would not
+            right = self._adapt_clause(right, True, False)
+
         if right_mapper and right is right_selectable:
             if not right_selectable.is_derived_from(
                     right_mapper.mapped_table):
diff --git a/lib/sqlalchemy/sql/selectable.py b/lib/sqlalchemy/sql/selectable.py
index 5b665829b..64886b326 100644
--- a/lib/sqlalchemy/sql/selectable.py
+++ b/lib/sqlalchemy/sql/selectable.py
@@ -338,6 +338,8 @@ class FromClause(Selectable):
     _is_select = False
     _is_from_container = False
 
+    _is_lateral = False
+
     _textual = False
     """a marker that allows us to easily distinguish a :class:`.TextAsFrom`
     or similar object from other kinds of :class:`.FromClause` objects."""
@@ -1329,6 +1331,7 @@ class Lateral(Alias):
     """
 
     __visit_name__ = 'lateral'
+    _is_lateral = True
 
 
 class TableSample(Alias):
@@ -3008,7 +3011,6 @@ class Select(HasPrefixes, HasSuffixes, GenerativeSelect):
         # contents, as this set is used for matching, not rendering.
         self._correlate = set(clone(f) for f in
                               self._correlate).union(self._correlate)
-
         # 4. clone other things.   The difficulty here is that Column
         # objects are not actually cloned, and refer to their original
         # .table, resulting in the wrong "from" parent after a clone
diff --git a/test/orm/test_joins.py b/test/orm/test_joins.py
index 6af8ce725..54be45233 100644
--- a/test/orm/test_joins.py
+++ b/test/orm/test_joins.py
@@ -2781,4 +2781,230 @@ class AliasFromCorrectLeftTest(
             "JOIN x ON object_1.id = x.obj_id WHERE x.name = :name_1"
         )
 
+class JoinLateralTest(fixtures.MappedTest, AssertsCompiledSQL):
+    __dialect__ = default.DefaultDialect(supports_native_boolean=True)
+
+    run_setup_bind = None
+    run_setup_mappers = 'once'
+
+    run_create_tables = None
+
+    @classmethod
+    def define_tables(cls, metadata):
+        Table('people', metadata,
+              Column('people_id', Integer, primary_key=True),
+              Column('age', Integer),
+              Column('name', String(30)))
+        Table('bookcases', metadata,
+              Column('bookcase_id', Integer, primary_key=True),
+              Column(
+                  'bookcase_owner_id',
+                  Integer, ForeignKey('people.people_id')),
+              Column('bookcase_shelves', Integer),
+              Column('bookcase_width', Integer))
+        Table('books', metadata,
+              Column('book_id', Integer, primary_key=True),
+              Column(
+                  'bookcase_id', Integer, ForeignKey('bookcases.bookcase_id')),
+              Column('book_owner_id', Integer, ForeignKey('people.people_id')),
+              Column('book_weight', Integer))
+
+    @classmethod
+    def setup_classes(cls):
+        people, bookcases, books = cls.tables('people', 'bookcases', 'books')
+
+        class Person(cls.Comparable):
+            pass
+
+        class Bookcase(cls.Comparable):
+            pass
+
+        class Book(cls.Comparable):
+            pass
+
+        mapper(Person, people)
+        mapper(Bookcase, bookcases, properties={
+            'owner': relationship(Person),
+            'books': relationship(Book)
+        })
+        mapper(Book, books)
+
+    def test_select_subquery(self):
+        Person, Book = self.classes("Person", "Book")
+
+        s = Session()
+
+        subq = s.query(Book.book_id).correlate(Person).filter(
+            Person.people_id == Book.book_owner_id
+        ).subquery().lateral()
+
+        stmt = s.query(Person, subq.c.book_id).join(
+            subq, true()
+        )
+
+        self.assert_compile(
+            stmt,
+            "SELECT people.people_id AS people_people_id, "
+            "people.age AS people_age, people.name AS people_name, "
+            "anon_1.book_id AS anon_1_book_id "
+            "FROM people JOIN LATERAL "
+            "(SELECT books.book_id AS book_id FROM books "
+            "WHERE people.people_id = books.book_owner_id) AS anon_1 ON true"
+        )
+
+    # sef == select_entity_from
+    def test_select_subquery_sef_implicit_correlate(self):
+        Person, Book = self.classes("Person", "Book")
+
+        s = Session()
+
+        stmt = s.query(Person).subquery()
+
+        subq = s.query(Book.book_id).filter(
+            Person.people_id == Book.book_owner_id
+        ).subquery().lateral()
+
+        stmt = s.query(Person, subq.c.book_id).select_entity_from(stmt).join(
+            subq, true()
+        )
+
+        self.assert_compile(
+            stmt,
+            "SELECT anon_1.people_id AS anon_1_people_id, "
+            "anon_1.age AS anon_1_age, anon_1.name AS anon_1_name, "
+            "anon_2.book_id AS anon_2_book_id "
+            "FROM "
+            "(SELECT people.people_id AS people_id, people.age AS age, "
+            "people.name AS name FROM people) AS anon_1 "
+            "JOIN LATERAL "
+            "(SELECT books.book_id AS book_id FROM books "
+            "WHERE anon_1.people_id = books.book_owner_id) AS anon_2 ON true"
+        )
+
+    def test_select_subquery_sef_implicit_correlate_coreonly(self):
+        Person, Book = self.classes("Person", "Book")
+
+        s = Session()
+
+        stmt = s.query(Person).subquery()
+
+        subq = select([Book.book_id]).where(
+            Person.people_id == Book.book_owner_id
+        ).lateral()
+
+        stmt = s.query(Person, subq.c.book_id).select_entity_from(stmt).join(
+            subq, true()
+        )
+
+        self.assert_compile(
+            stmt,
+            "SELECT anon_1.people_id AS anon_1_people_id, "
+            "anon_1.age AS anon_1_age, anon_1.name AS anon_1_name, "
+            "anon_2.book_id AS anon_2_book_id "
+            "FROM "
+            "(SELECT people.people_id AS people_id, people.age AS age, "
+            "people.name AS name FROM people) AS anon_1 "
+            "JOIN LATERAL "
+            "(SELECT books.book_id AS book_id FROM books "
+            "WHERE anon_1.people_id = books.book_owner_id) AS anon_2 ON true"
+        )
+
+    def test_select_subquery_sef_explicit_correlate_coreonly(self):
+        Person, Book = self.classes("Person", "Book")
+
+        s = Session()
+
+        stmt = s.query(Person).subquery()
+
+        subq = select([Book.book_id]).correlate(Person).where(
+            Person.people_id == Book.book_owner_id
+        ).lateral()
+
+        stmt = s.query(Person, subq.c.book_id).select_entity_from(stmt).join(
+            subq, true()
+        )
+
+        self.assert_compile(
+            stmt,
+            "SELECT anon_1.people_id AS anon_1_people_id, "
+            "anon_1.age AS anon_1_age, anon_1.name AS anon_1_name, "
+            "anon_2.book_id AS anon_2_book_id "
+            "FROM "
+            "(SELECT people.people_id AS people_id, people.age AS age, "
+            "people.name AS name FROM people) AS anon_1 "
+            "JOIN LATERAL "
+            "(SELECT books.book_id AS book_id FROM books "
+            "WHERE anon_1.people_id = books.book_owner_id) AS anon_2 ON true"
+        )
+
+    def test_select_subquery_sef_explicit_correlate(self):
+        Person, Book = self.classes("Person", "Book")
+
+        s = Session()
+
+        stmt = s.query(Person).subquery()
+
+        subq = s.query(Book.book_id).correlate(Person).filter(
+            Person.people_id == Book.book_owner_id
+        ).subquery().lateral()
+
+        stmt = s.query(Person, subq.c.book_id).select_entity_from(stmt).join(
+            subq, true()
+        )
+
+        self.assert_compile(
+            stmt,
+            "SELECT anon_1.people_id AS anon_1_people_id, "
+            "anon_1.age AS anon_1_age, anon_1.name AS anon_1_name, "
+            "anon_2.book_id AS anon_2_book_id "
+            "FROM "
+            "(SELECT people.people_id AS people_id, people.age AS age, "
+            "people.name AS name FROM people) AS anon_1 "
+            "JOIN LATERAL "
+            "(SELECT books.book_id AS book_id FROM books "
+            "WHERE anon_1.people_id = books.book_owner_id) AS anon_2 ON true"
+        )
+
+    def test_from_function(self):
+        Bookcase = self.classes.Bookcase
+
+        s = Session()
+
+        srf = lateral(func.generate_series(1, Bookcase.bookcase_shelves))
+
+        self.assert_compile(
+            s.query(Bookcase).join(srf, true()),
+            "SELECT bookcases.bookcase_id AS bookcases_bookcase_id, "
+            "bookcases.bookcase_owner_id AS bookcases_bookcase_owner_id, "
+            "bookcases.bookcase_shelves AS bookcases_bookcase_shelves, "
+            "bookcases.bookcase_width AS bookcases_bookcase_width "
+            "FROM bookcases JOIN "
+            "LATERAL generate_series(:generate_series_1, "
+            "bookcases.bookcase_shelves) AS anon_1 ON true"
+        )
+
+    def test_from_function_select_entity_from(self):
+        Bookcase = self.classes.Bookcase
+
+        s = Session()
+
+        subq = s.query(Bookcase).subquery()
+
+        srf = lateral(func.generate_series(1, Bookcase.bookcase_shelves))
+
+        self.assert_compile(
+            s.query(Bookcase).select_entity_from(subq).join(srf, true()),
+            "SELECT anon_1.bookcase_id AS anon_1_bookcase_id, "
+            "anon_1.bookcase_owner_id AS anon_1_bookcase_owner_id, "
+            "anon_1.bookcase_shelves AS anon_1_bookcase_shelves, "
+            "anon_1.bookcase_width AS anon_1_bookcase_width "
+            "FROM (SELECT bookcases.bookcase_id AS bookcase_id, "
+            "bookcases.bookcase_owner_id AS bookcase_owner_id, "
+            "bookcases.bookcase_shelves AS bookcase_shelves, "
+            "bookcases.bookcase_width AS bookcase_width FROM bookcases) "
+            "AS anon_1 "
+            "JOIN LATERAL "
+            "generate_series(:generate_series_1, anon_1.bookcase_shelves) "
+            "AS anon_2 ON true"
+        )