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
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
|
#!/usr/bin/env python3
"""Queries over a bank of 100,000 accounts: Pure's registered queries beside
the same SQL on SQLite (docs/research/091-queries.md).
The bank (bank.aup) gets --branches branches, ten tellers each, --accounts
accounts spread over the branches and --transfers transfers, posted over HTTP;
SQLite (in RAM, the pgbench tables) gets the same rows. Then each query is
asked --times times on both, Pure's through GET /queries/NAME (registered
once, shown from the last save's snapshot, HTTP and JSON included), SQLite's
through its C library from Python, on pgbench's tables as they are and again
with the indexes a DBA would add for these queries (accounts by balance, and
by branch with the balance) after ANALYZE. Pure is given no index: it plans
from the query alone. The answers are checked to agree. Prints one JSON line
per query: the median and fastest time of each.
"""
import argparse
import http.client
import json
import random
import select
import shutil
import sqlite3
import statistics
import subprocess
import tempfile
import threading
import time
from pathlib import Path
HERE = Path(__file__).resolve().parent
ROOT = HERE.parents[1]
QUERIES = {
# per branch: its total and how many accounts (a GROUP BY, as a fold per row)
'per_branch': ('view per_branch = each b in branches {\n bid = b.bid\n'
' total = sum(a.balance for a in accounts where a.bid == b.bid)\n'
' accounts = count(a in accounts where a.bid == b.bid)\n}\n',
'SELECT bid, sum(abalance), count(*) FROM pgbench_accounts GROUP BY bid ORDER BY bid'),
# accounts in credit (a filter over every row)
'in_credit': ('view in_credit = count(a in accounts where a.balance > 0)\n',
'SELECT count(*) FROM pgbench_accounts WHERE abalance > 0'),
# large transfers with their account's and teller's balance (a three-way join)
'large': ('view large = each h in history where h.delta > 4900 {\n aid = h.aid\n delta = h.delta\n'
' account = accounts[h.aid].balance\n teller = tellers[h.tid].balance\n}\n',
'SELECT h.aid, h.delta, a.abalance, t.tbalance FROM pgbench_history h '
'JOIN pgbench_accounts a ON a.aid = h.aid JOIN pgbench_tellers t ON t.tid = h.tid '
'WHERE h.delta > 4900 ORDER BY h.rowid'),
# the richest accounts and how many are richer (a correlated self-join)
'ranks': ('view ranks = each a in accounts where a.balance > 4900 {\n aid = a.aid\n'
' richer = count(b in accounts where b.balance > a.balance)\n}\n',
'SELECT a.aid, (SELECT count(*) FROM pgbench_accounts b WHERE b.abalance > a.abalance) '
'FROM pgbench_accounts a WHERE a.abalance > 4900 ORDER BY a.aid'),
}
def post_all(port, path, bodies, threads=8):
failed = []
def worker(part):
connection = http.client.HTTPConnection('127.0.0.1', port, timeout=60)
for body in part:
connection.request('POST', path, body, {'Content-Type': 'application/json'})
answer = connection.getresponse()
answer.read()
if answer.status != 200:
failed.append(answer.status)
connection.close()
workers = [threading.Thread(target=worker, args=(bodies[i::threads],)) for i in range(threads)]
for w in workers:
w.start()
for w in workers:
w.join()
assert not failed, failed[:3]
def rows_of(name, data):
"""Pure's answer as SQLite's tuples."""
if name == 'in_credit':
return [(int(data),)]
items = json.loads(data)['items']
return [tuple(int(v) for v in row.values()) for row in items]
def main():
parser = argparse.ArgumentParser()
parser.add_argument('--aup', type=Path, default=ROOT.parent / 'aup' / 'build' / 'aup')
parser.add_argument('--serve', type=Path, default=ROOT / 'build' / 'pure-serve')
parser.add_argument('--accounts', type=int, default=100000)
parser.add_argument('--branches', type=int, default=10)
parser.add_argument('--transfers', type=int, default=20000)
parser.add_argument('--times', type=int, default=15)
args = parser.parse_args()
work = Path(tempfile.mkdtemp(prefix='pure-queries-', dir='/dev/shm'))
process = None
try:
module = work / 'bank.pmod'
subprocess.run([str(args.aup), 'build', str(HERE / 'bank.aup'), '-o', str(module)], check=True)
compiled = {}
for name, (view, _) in QUERIES.items():
source = work / f'{name}.aup'
source.write_text(view)
subprocess.run([str(args.aup), 'query', str(HERE / 'bank.aup'), str(source), '-o', str(work / f'{name}.pmod')],
check=True)
compiled[name] = (work / f'{name}.pmod').read_bytes()
process = subprocess.Popen(['taskset', '-c', '0-2', str(args.serve), '--module', str(module), '--store',
str(work / 'store'), '--port', '0', '--timeout-ms', '600000'], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
select.select([process.stdout], [], [], 10)
port = json.loads(process.stdout.readline())['port']
rng = random.Random(9)
tellers = 10 * args.branches
transfers = [(rng.randint(1, args.accounts), rng.randint(1, tellers), rng.randint(1, args.branches),
rng.randint(-5000, 5000)) for _ in range(args.transfers)]
post_all(port, '/inputs/branch', [json.dumps({'bid': b}).encode() for b in range(1, args.branches + 1)], 1)
post_all(port, '/inputs/teller', [json.dumps({'tid': t, 'bid': (t - 1) // 10 + 1}).encode()
for t in range(1, tellers + 1)], 1)
post_all(port, '/inputs/account', [json.dumps({'aid': a, 'bid': (a - 1) % args.branches + 1}).encode()
for a in range(1, args.accounts + 1)])
post_all(port, '/inputs/transfer', [json.dumps({'aid': a, 'tid': t, 'bid': b, 'delta': d}).encode()
for a, t, b, d in transfers], 1) # one client: history in this order
db = sqlite3.connect(str(work / 'bank.db'))
db.executescript('PRAGMA journal_mode=WAL;'
'CREATE TABLE pgbench_accounts (aid INTEGER PRIMARY KEY, bid INTEGER, abalance INTEGER, filler CHAR(84));'
'CREATE TABLE pgbench_tellers (tid INTEGER PRIMARY KEY, bid INTEGER, tbalance INTEGER, filler CHAR(84));'
'CREATE TABLE pgbench_branches (bid INTEGER PRIMARY KEY, bbalance INTEGER, filler CHAR(88));'
'CREATE TABLE pgbench_history (tid INTEGER, bid INTEGER, aid INTEGER, delta INTEGER, mtime TIMESTAMP,'
' filler CHAR(22));')
db.executemany('INSERT INTO pgbench_branches VALUES (?, 0, \'\')', [(b,) for b in range(1, args.branches + 1)])
db.executemany('INSERT INTO pgbench_tellers VALUES (?, ?, 0, \'\')',
[(t, (t - 1) // 10 + 1) for t in range(1, tellers + 1)])
db.executemany('INSERT INTO pgbench_accounts VALUES (?, ?, 0, \'\')',
[(a, (a - 1) % args.branches + 1) for a in range(1, args.accounts + 1)])
for a, t, b, d in transfers:
db.execute('UPDATE pgbench_accounts SET abalance = abalance + ? WHERE aid = ?', (d, a))
db.execute('UPDATE pgbench_tellers SET tbalance = tbalance + ? WHERE tid = ?', (d, t))
db.execute('UPDATE pgbench_branches SET bbalance = bbalance + ? WHERE bid = ?', (d, b))
db.execute('INSERT INTO pgbench_history (tid, bid, aid, delta) VALUES (?, ?, ?, ?)', (t, b, a, d))
db.commit()
client = http.client.HTTPConnection('127.0.0.1', port, timeout=120)
for name, bytes_ in compiled.items():
client.request('PUT', f'/queries/{name}', bytes_)
answer = client.getresponse()
answer.read()
assert answer.status == 200, (name, answer.status)
def timed(ask):
times, answer = [], None
for _ in range(args.times):
began = time.perf_counter()
answer = ask()
times.append(time.perf_counter() - began)
return answer, times
def pure(name):
client.request('GET', f'/queries/{name}')
answer = client.getresponse()
data = answer.read()
assert answer.status == 200, (name, answer.status)
return data
results = {}
for name, (_, sql) in QUERIES.items():
data, pure_times = timed(lambda: pure(name))
rows, sqlite_times = timed(lambda: db.execute(sql).fetchall())
assert rows_of(name, data) == [tuple(int(v) for v in row) for row in rows], name
results[name] = (len(rows), pure_times, sqlite_times)
db.executescript('CREATE INDEX accounts_balance ON pgbench_accounts (abalance);'
'CREATE INDEX accounts_branch ON pgbench_accounts (bid, abalance); ANALYZE;')
for name, (_, sql) in QUERIES.items():
rows, indexed_times = timed(lambda: db.execute(sql).fetchall())
count, pure_times, sqlite_times = results[name]
assert len(rows) == count, name
print(json.dumps({'query': name, 'rows': count, 'accounts': args.accounts,
'pure_median_ms': round(statistics.median(pure_times) * 1e3, 2),
'pure_fastest_ms': round(min(pure_times) * 1e3, 2),
'sqlite_median_ms': round(statistics.median(sqlite_times) * 1e3, 2),
'sqlite_fastest_ms': round(min(sqlite_times) * 1e3, 2),
'sqlite_indexed_median_ms': round(statistics.median(indexed_times) * 1e3, 2),
'sqlite_indexed_fastest_ms': round(min(indexed_times) * 1e3, 2)}), flush=True)
finally:
if process:
process.terminate()
process.communicate(timeout=60)
shutil.rmtree(work)
if __name__ == '__main__':
main()
|
