1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
|
from sqlalchemy import Column
from sqlalchemy import ForeignKey
from sqlalchemy import Integer
from sqlalchemy import join
from sqlalchemy import lateral
from sqlalchemy import String
from sqlalchemy import Table
from sqlalchemy import true
from sqlalchemy.engine import default
from sqlalchemy.sql import func
from sqlalchemy.sql import select
from sqlalchemy.testing import AssertsCompiledSQL
from sqlalchemy.testing import fixtures
class LateralTest(fixtures.TablesTest, AssertsCompiledSQL):
__dialect__ = default.DefaultDialect(supports_native_boolean=True)
run_setup_bind = None
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),
)
def test_standalone(self):
table1 = self.tables.people
subq = select([table1.c.people_id])
# alias name is not rendered because subquery is not
# in the context of a FROM clause
self.assert_compile(
lateral(subq, name="alias"),
"LATERAL (SELECT people.people_id FROM people)",
)
self.assert_compile(
subq.lateral(name="alias"),
"LATERAL (SELECT people.people_id FROM people)",
)
def test_select_from(self):
table1 = self.tables.people
subq = select([table1.c.people_id])
# in a FROM context, now you get "AS alias" and column labeling
self.assert_compile(
select([subq.lateral(name="alias")]),
"SELECT alias.people_id FROM LATERAL "
"(SELECT people.people_id AS people_id FROM people) AS alias",
)
def test_plain_join(self):
table1 = self.tables.people
table2 = self.tables.books
subq = select([table2.c.book_id]).where(
table2.c.book_owner_id == table1.c.people_id
)
# FROM books, people? isn't this wrong? No! Because
# this is only a fragment, books isn't in any other FROM clause
self.assert_compile(
join(table1, lateral(subq, name="alias"), true()),
"people JOIN LATERAL (SELECT books.book_id AS book_id "
"FROM books, people WHERE books.book_owner_id = people.people_id) "
"AS alias ON true",
)
# put it in correct context, implicit correlation works fine
self.assert_compile(
select([table1]).select_from(
join(table1, lateral(subq, name="alias"), true())
),
"SELECT people.people_id, people.age, people.name "
"FROM people JOIN LATERAL (SELECT books.book_id AS book_id "
"FROM books WHERE books.book_owner_id = people.people_id) "
"AS alias ON true",
)
# explicit correlation
subq = subq.correlate(table1)
self.assert_compile(
select([table1]).select_from(
join(table1, lateral(subq, name="alias"), true())
),
"SELECT people.people_id, people.age, people.name "
"FROM people JOIN LATERAL (SELECT books.book_id AS book_id "
"FROM books WHERE books.book_owner_id = people.people_id) "
"AS alias ON true",
)
def test_join_lateral_w_select_subquery(self):
table1 = self.tables.people
table2 = self.tables.books
subq = (
select([table2.c.book_id])
.correlate(table1)
.where(table1.c.people_id == table2.c.book_owner_id)
.lateral()
)
stmt = select([table1, subq.c.book_id]).select_from(
table1.join(subq, true())
)
self.assert_compile(
stmt,
"SELECT people.people_id, people.age, people.name, 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",
)
def test_from_function(self):
bookcases = self.tables.bookcases
srf = lateral(func.generate_series(1, bookcases.c.bookcase_shelves))
self.assert_compile(
select([bookcases]).select_from(bookcases.join(srf, true())),
"SELECT bookcases.bookcase_id, bookcases.bookcase_owner_id, "
"bookcases.bookcase_shelves, bookcases.bookcase_width "
"FROM bookcases JOIN "
"LATERAL generate_series(:generate_series_1, "
"bookcases.bookcase_shelves) AS anon_1 ON true",
)
|
