diff options
Diffstat (limited to 'test/sql')
| -rw-r--r-- | test/sql/test_from_linter.py | 103 | ||||
| -rw-r--r-- | test/sql/test_lateral.py | 62 |
2 files changed, 165 insertions, 0 deletions
diff --git a/test/sql/test_from_linter.py b/test/sql/test_from_linter.py index b0bcee18e..9e0ededec 100644 --- a/test/sql/test_from_linter.py +++ b/test/sql/test_from_linter.py @@ -1,6 +1,7 @@ from sqlalchemy import Integer from sqlalchemy import select from sqlalchemy import sql +from sqlalchemy import testing from sqlalchemy import true from sqlalchemy.testing import config from sqlalchemy.testing import engines @@ -58,6 +59,108 @@ class TestFindUnmatchingFroms(fixtures.TablesTest): assert start == self.b assert froms == {self.a} + @testing.combinations(("lateral",), ("cartesian",), ("join",)) + def test_lateral_subqueries(self, control): + """ + :: + + test=> create table a (id integer); + CREATE TABLE + test=> create table b (id integer); + CREATE TABLE + test=> insert into a(id) values (1), (2), (3); + INSERT 0 3 + test=> insert into b(id) values (1), (2), (3); + INSERT 0 3 + + test=> select * from (select id from a) as a1, + lateral (select id from b where id=a1.id) as b1; + id | id + ----+---- + 1 | 1 + 2 | 2 + 3 | 3 + (3 rows) + + """ + p1 = select(self.a).subquery() + + p2 = select(self.b).where(self.b.c.col_b == p1.c.col_a).subquery() + + if control == "lateral": + p2 = p2.lateral() + + query = select(p1, p2) + + if control == "join": + query = query.join_from(p1, p2, p1.c.col_a == p2.c.col_b) + + froms, start = find_unmatching_froms(query, p1) + + if control == "cartesian": + assert start is p1 + assert froms == {p2} + else: + assert start is None + assert froms is None + + froms, start = find_unmatching_froms(query, p2) + + if control == "cartesian": + assert start is p2 + assert froms == {p1} + else: + assert start is None + assert froms is None + + def test_lateral_subqueries_w_joins(self): + p1 = select(self.a).subquery() + p2 = ( + select(self.b) + .where(self.b.c.col_b == p1.c.col_a) + .subquery() + .lateral() + ) + p3 = ( + select(self.c) + .where(self.c.c.col_c == p1.c.col_a) + .subquery() + .lateral() + ) + + query = select(p1, p2, p3).join_from(p1, p2, true()).join(p3, true()) + + for p in (p1, p2, p3): + froms, start = find_unmatching_froms(query, p) + assert start is None + assert froms is None + + def test_lateral_subqueries_ok_do_we_still_find_cartesians(self): + p1 = select(self.a).subquery() + + p3 = select(self.a).subquery() + + p2 = select(self.b).where(self.b.c.col_b == p3.c.col_a).subquery() + + p2 = p2.lateral() + + query = select(p1, p2, p3) + + froms, start = find_unmatching_froms(query, p1) + + assert start is p1 + assert froms == {p2, p3} + + froms, start = find_unmatching_froms(query, p2) + + assert start is p2 + assert froms == {p1} + + froms, start = find_unmatching_froms(query, p3) + + assert start is p3 + assert froms == {p1} + def test_count_non_eq_comparison_operators(self): query = select(self.a).where(self.a.c.col_a > self.b.c.col_b) froms, start = find_unmatching_froms(query, self.a) diff --git a/test/sql/test_lateral.py b/test/sql/test_lateral.py index a80ad7083..6723be850 100644 --- a/test/sql/test_lateral.py +++ b/test/sql/test_lateral.py @@ -7,6 +7,7 @@ from sqlalchemy import lateral from sqlalchemy import String from sqlalchemy import Table from sqlalchemy import table +from sqlalchemy import testing from sqlalchemy import text from sqlalchemy import true from sqlalchemy.engine import default @@ -21,6 +22,8 @@ from sqlalchemy.testing import fixtures class LateralTest(fixtures.TablesTest, AssertsCompiledSQL): __dialect__ = default.DefaultDialect(supports_native_boolean=True) + assert_from_linting = True + run_setup_bind = None run_create_tables = None @@ -234,6 +237,65 @@ class LateralTest(fixtures.TablesTest, AssertsCompiledSQL): "WHERE people.people_id = books.book_owner_id) AS anon_1 ON true", ) + @testing.combinations((True,), (False,)) + def test_join_lateral_subquery_twolevel(self, use_twolevel): + people, books, bookcases = self.tables("people", "books", "bookcases") + + p1 = select( + books.c.book_id, books.c.bookcase_id, books.c.book_owner_id + ).subquery() + p2 = ( + select(bookcases.c.bookcase_id, bookcases.c.bookcase_owner_id) + .where(bookcases.c.bookcase_id == p1.c.bookcase_id) + .subquery() + .lateral() + ) + p3 = ( + select(people.c.people_id) + .where(p1.c.book_owner_id == people.c.people_id) + .subquery() + .lateral() + ) + + onelevel = ( + select(p1.c.book_id, p2.c.bookcase_id) + .select_from(p1) + .join(p2, true()) + ) + + if use_twolevel: + twolevel = onelevel.add_columns(p3.c.people_id).join(p3, true()) + + self.assert_compile( + twolevel, + "SELECT anon_1.book_id, anon_2.bookcase_id, anon_3.people_id " + "FROM (SELECT books.book_id AS book_id, books.bookcase_id AS " + "bookcase_id, books.book_owner_id AS book_owner_id " + "FROM books) " + "AS anon_1 JOIN LATERAL (SELECT bookcases.bookcase_id AS " + "bookcase_id, " + "bookcases.bookcase_owner_id AS bookcase_owner_id " + "FROM bookcases " + "WHERE bookcases.bookcase_id = anon_1.bookcase_id) " + "AS anon_2 ON true JOIN LATERAL " + "(SELECT people.people_id AS people_id FROM people " + "WHERE anon_1.book_owner_id = people.people_id) AS anon_3 " + "ON true", + ) + else: + self.assert_compile( + onelevel, + "SELECT anon_1.book_id, anon_2.bookcase_id FROM " + "(SELECT books.book_id AS book_id, books.bookcase_id " + "AS bookcase_id, books.book_owner_id AS book_owner_id " + "FROM books) AS anon_1 JOIN LATERAL " + "(SELECT bookcases.bookcase_id AS bookcase_id, " + "bookcases.bookcase_owner_id AS bookcase_owner_id " + "FROM bookcases " + "WHERE bookcases.bookcase_id = anon_1.bookcase_id) AS anon_2 " + "ON true", + ) + def test_join_lateral_w_select_implicit_subquery(self): table1 = self.tables.people table2 = self.tables.books |