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
|
from sqlalchemy.testing import fixtures
from sqlalchemy.testing import AssertsCompiledSQL, assert_raises_message
from sqlalchemy.sql import select, func
from sqlalchemy.engine import default
from sqlalchemy import exc
from sqlalchemy import Table, Integer, String, ForeignKey, Column, true
from sqlalchemy import lateral, outerjoin, join
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"
)
|
