Run TPC-C on Pure and on SQLite, and compare the states they leave

1 files changed, 332 insertions, 0 deletions

diff --git a/benchmarks/tpcc/run.py b/benchmarks/tpcc/run.py
new file mode 100644
index 0000000..f24126f
--- /dev/null
+++ b/benchmarks/tpcc/run.py
@@ -0,0 +1,332 @@
+#!/usr/bin/env python3
+"""TPC-C on Pure and on SQLite (docs/research/092-tpcc.md).
+
+tpcc.aup is TPC-C's tables and five transactions as an aup app; data.py makes
+the initial database of clause 4.3.3.1 for --warehouses warehouses. The same
+rows go into pure-serve (in bulk, as binary rows) and into SQLite (the
+specification's tables with their composite primary keys, WITHOUT ROWID,
+WAL), both kept in --directory. Then:
+
+1. The same transactions on both: one terminal, --check transactions drawn
+   from one seed by tpcc-load (over HTTP) and by sqlite-tpcc (in process),
+   and every row of every table compared (times aside), Pure's read through
+   queries registered for the purpose.
+2. Throughput: --seconds of the whole mix, Pure's from each count of
+   --terminals connections (tpcc-load on core 3, the server on cores 0-2),
+   SQLite's from each count of --sqlite-threads threads, each with its own
+   connection, on cores 0-2, synchronous --sync (FULL syncs each commit
+   before it is answered, as Pure does).
+3. The consistency conditions of clause 3.3.2 on both after each run.
+
+Prints one JSON line a step.
+"""
+import argparse
+import http.client
+import json
+import os
+import select
+import shutil
+import sqlite3
+import subprocess
+import sys
+import tempfile
+import time
+from pathlib import Path
+
+HERE = Path(__file__).resolve().parent
+ROOT = HERE.parents[1]
+sys.path.insert(0, str(HERE))
+import data  # noqa: E402
+
+SCHEMA = '''
+CREATE TABLE warehouse (w_id INTEGER PRIMARY KEY, w_tax REAL, w_ytd REAL);
+CREATE TABLE district (d_w_id INTEGER, d_id INTEGER, d_tax REAL, d_ytd REAL, d_next_o_id INTEGER, d_next_delivery INTEGER,
+  PRIMARY KEY (d_w_id, d_id)) WITHOUT ROWID;
+CREATE TABLE customer (c_w_id INTEGER, c_d_id INTEGER, c_id INTEGER, c_discount REAL, c_credit INTEGER, c_balance REAL,
+  c_ytd_payment REAL, c_payment_cnt INTEGER, c_delivery_cnt INTEGER, c_last_order INTEGER,
+  PRIMARY KEY (c_w_id, c_d_id, c_id)) WITHOUT ROWID;
+CREATE TABLE item (i_id INTEGER PRIMARY KEY, i_price REAL);
+CREATE TABLE stock (s_w_id INTEGER, s_i_id INTEGER, s_quantity INTEGER, s_ytd INTEGER, s_order_cnt INTEGER,
+  s_remote_cnt INTEGER, PRIMARY KEY (s_w_id, s_i_id)) WITHOUT ROWID;
+CREATE TABLE orders (o_w_id INTEGER, o_d_id INTEGER, o_id INTEGER, o_c_id INTEGER, o_entry_d REAL, o_carrier_id INTEGER,
+  o_ol_cnt INTEGER, o_all_local INTEGER, PRIMARY KEY (o_w_id, o_d_id, o_id)) WITHOUT ROWID;
+CREATE TABLE new_order (no_w_id INTEGER, no_d_id INTEGER, no_o_id INTEGER,
+  PRIMARY KEY (no_w_id, no_d_id, no_o_id)) WITHOUT ROWID;
+CREATE TABLE order_line (ol_w_id INTEGER, ol_d_id INTEGER, ol_o_id INTEGER, ol_number INTEGER, ol_i_id INTEGER,
+  ol_supply_w_id INTEGER, ol_delivery_d REAL, ol_quantity INTEGER, ol_amount REAL,
+  PRIMARY KEY (ol_w_id, ol_d_id, ol_o_id, ol_number)) WITHOUT ROWID;
+CREATE TABLE history (h_c_w_id INTEGER, h_c_d_id INTEGER, h_c_id INTEGER, h_w_id INTEGER, h_d_id INTEGER, h_date REAL,
+  h_amount REAL);
+'''
+
+ORDERS = 10000000  # an order's key is its district's times this, plus its number
+
+
+def district_of(key):
+    """A district key's warehouse and district number."""
+    return (key - 1) // 10, (key - 1) % 10 + 1
+
+
+def sqlite_rows(tables):
+    """data.generate's rows as the rows of SQLite's tables."""
+    def customer(key):
+        return (*district_of((key - 1) // 3000), (key - 1) % 3000 + 1)
+
+    def order(key):
+        return (*district_of(key // ORDERS), key % ORDERS)
+
+    yield 'warehouse', tables['warehouses']
+    yield 'district', [(*district_of(k), tax, ytd, next_order, next_delivery)
+                       for k, tax, ytd, next_order, next_delivery in tables['districts']]
+    yield 'customer', [(*customer(k), discount, credit, balance, ytd, payments, deliveries, last % ORDERS)
+                       for k, discount, credit, balance, ytd, payments, deliveries, last in tables['customers']]
+    yield 'item', tables['items']
+    yield 'stock', [((k - 1) // 100000, (k - 1) % 100000 + 1, *rest) for k, *rest in tables['stock']]
+    yield 'orders', [(*order(k), (c - 1) % 3000 + 1, *rest) for k, c, *rest in tables['orders']]
+    yield 'new_order', [order(k) for k, in tables['new_orders']]
+    yield 'order_line', [(*order(k // 16), k % 16, *rest) for k, *rest in tables['lines']]
+
+
+def load_sqlite(path, tables):
+    db = sqlite3.connect(str(path))
+    db.executescript('PRAGMA journal_mode=WAL; PRAGMA synchronous=OFF;' + SCHEMA)
+    with db:
+        for table, rows in sqlite_rows(tables):
+            db.executemany(f'INSERT INTO {table} VALUES ({", ".join("?" * len(rows[0]))})', rows)
+    db.execute('PRAGMA wal_checkpoint(TRUNCATE)')
+    db.close()
+
+
+class Pure:
+    """pure-serve on cores 0-2 with the app's store in directory."""
+
+    def __init__(self, serve, module, store, threads):
+        self.process = subprocess.Popen(['taskset', '-c', '0-2', str(serve), '--module', str(module), '--store', str(store),
+                                         '--port', '0', '--threads', str(threads), '--timeout-ms', '600000'],
+                                        stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+        select.select([self.process.stdout], [], [], 30)
+        self.port = json.loads(self.process.stdout.readline())['port']
+        self.connection = http.client.HTTPConnection('127.0.0.1', self.port, timeout=600)
+
+    def call(self, method, path, body=None, kind='application/json'):
+        self.connection.request(method, path, body, {'Content-Type': kind} if body is not None else {})
+        answer = self.connection.getresponse()
+        return answer.status, answer.read()
+
+    def view(self, name):
+        status, body = self.call('GET', '/views/' + name)
+        assert status == 200, (name, status, body[:200])
+        return json.loads(body)
+
+    def load(self, tables):
+        for name, rows in tables.items():
+            if name == 'history':
+                continue
+            chunk = max(1, (1400000 // 8) // len(data.FIELDS[name]) - 1)
+            for at in range(0, len(rows), chunk):
+                status, body = self.call('POST', f'/inputs/load_{name}', data.packed(rows[at:at + chunk]),
+                                         'application/octet-stream')
+                assert status == 200, (name, status, body[:200])
+
+    def cpu_seconds(self):
+        fields = Path(f'/proc/{self.process.pid}/stat').read_text().rsplit(')', 1)[1].split()
+        return (int(fields[11]) + int(fields[12])) / os.sysconf('SC_CLK_TCK')
+
+    def resident_mib(self):
+        for line in Path(f'/proc/{self.process.pid}/status').read_text().splitlines():
+            if line.startswith('VmRSS:'):
+                return round(int(line.split()[1]) / 1024, 1)
+        return None
+
+    def stop(self):
+        self.connection.close()
+        self.process.terminate()
+        self.process.communicate(timeout=120)
+
+
+# Each table's rows, times aside, as Pure's queries and SQLite's SELECTs name them (keys packed as tpcc.aup packs them).
+DUMPS = {
+    'warehouses': ('each w in warehouses {\n  id = w.id\n  ytd = w.ytd\n}',
+                   'SELECT w_id, w_ytd FROM warehouse'),
+    'districts': ('each d in districts {\n  id = d.id\n  ytd = d.ytd\n  next_order = d.next_order\n'
+                  '  next_delivery = d.next_delivery\n}',
+                  'SELECT d_w_id * 10 + d_id, d_ytd, d_next_o_id, d_next_delivery FROM district'),
+    'customers': ('each c in customers {\n  id = c.id\n  balance = c.balance\n  ytd_payment = c.ytd_payment\n'
+                  '  payments = c.payments\n  deliveries = c.deliveries\n  last_order = c.last_order\n}',
+                  'SELECT (c_w_id * 10 + c_d_id) * 3000 + c_id, c_balance, c_ytd_payment, c_payment_cnt, c_delivery_cnt, '
+                  '(c_w_id * 10 + c_d_id) * 10000000 + c_last_order FROM customer'),
+    'stock': ('each s in stock {\n  id = s.id\n  quantity = s.quantity\n  ytd = s.ytd\n  orders = s.orders\n'
+              '  remote = s.remote\n}',
+              'SELECT s_w_id * 100000 + s_i_id, s_quantity, s_ytd, s_order_cnt, s_remote_cnt FROM stock'),
+    'orders': ('each o in orders {\n  id = o.id\n  customer = o.customer\n  carrier = o.carrier\n  lines = o.lines\n'
+               '  local = o.local\n}',
+               'SELECT (o_w_id * 10 + o_d_id) * 10000000 + o_id, (o_w_id * 10 + o_d_id) * 3000 + o_c_id, o_carrier_id, '
+               'o_ol_cnt, o_all_local FROM orders'),
+    'new_orders': ('each n in new_orders {\n  id = n.id\n}',
+                   'SELECT (no_w_id * 10 + no_d_id) * 10000000 + no_o_id FROM new_order'),
+    'lines': ('each l in lines {\n  id = l.id\n  item = l.item\n  supply = l.supply\n  quantity = l.quantity\n'
+              '  amount = l.amount\n  delivery = l.delivery\n}',
+              'SELECT ((ol_w_id * 10 + ol_d_id) * 10000000 + ol_o_id) * 16 + ol_number, ol_i_id, ol_supply_w_id, '
+              'ol_quantity, ol_amount, ol_delivery_d FROM order_line'),
+    'history': ('each h in history {\n  customer = h.customer\n  district = h.district\n  amount = h.amount\n}',
+                'SELECT (h_c_w_id * 10 + h_c_d_id) * 3000 + h_c_id, h_w_id * 10 + h_d_id, h_amount FROM history'),
+}
+
+# Clause 3.3.2.2 and 3.3.2.3, a district a row: its next order, its newest order, and its oldest and newest new orders
+# and how many (orders and new orders by their numbers in the district).
+DISTRICTS = ('each d in districts {\n  id = d.id\n  next_order = d.next_order\n'
+             '  newest = max(o.id for o in orders[d.id * 10000000 .. d.id * 10000000 + 10000000]) - d.id * 10000000\n'
+             '  newest_new = max(n.id for n in new_orders[d.id * 10000000 .. d.id * 10000000 + 10000000])'
+             ' - d.id * 10000000\n'
+             '  oldest_new = min(n.id for n in new_orders[d.id * 10000000 .. d.id * 10000000 + 10000000])'
+             ' - d.id * 10000000\n'
+             '  new_orders = count(n in new_orders[d.id * 10000000 .. d.id * 10000000 + 10000000])\n}',
+             'SELECT d_w_id * 10 + d_id, d_next_o_id, '
+             '(SELECT max(o_id) FROM orders WHERE o_w_id = d_w_id AND o_d_id = d_id), '
+             '(SELECT max(no_o_id) FROM new_order WHERE no_w_id = d_w_id AND no_d_id = d_id), '
+             '(SELECT min(no_o_id) FROM new_order WHERE no_w_id = d_w_id AND no_d_id = d_id), '
+             '(SELECT count(*) FROM new_order WHERE no_w_id = d_w_id AND no_d_id = d_id) FROM district')
+
+
+def register(pure, aup, work, name, view):
+    source = work / f'{name}.aup'
+    source.write_text(f'view {name} = {view}\n')
+    subprocess.run([str(aup), 'query', str(HERE / 'tpcc.aup'), str(source), '-o', str(work / f'{name}.pmod')], check=True)
+    status, body = pure.call('PUT', f'/queries/{name}', (work / f'{name}.pmod').read_bytes(), 'application/octet-stream')
+    assert status == 200, (name, status, body[:200])
+
+
+def asked(pure, name):
+    status, body = pure.call('GET', f'/queries/{name}')
+    assert status == 200, (name, status, body[:200])
+    return [tuple(row.values()) for row in json.loads(body)['items']]
+
+
+def same(a, b):
+    return a == b or abs(a - b) <= 1e-9 * max(1.0, abs(a), abs(b))
+
+
+def compare(table, pure_rows, sqlite_rows_):
+    """Rows that differ between the two (a table's rows in key order; history's sorted)."""
+    if table in ('lines',):  # a line's delivery time differs; whether it has one must not
+        pure_rows = [(*r[:-1], r[-1] > 0) for r in pure_rows]
+        sqlite_rows_ = [(*r[:-1], r[-1] > 0) for r in sqlite_rows_]
+    pure_rows, sqlite_rows_ = sorted(pure_rows), sorted(sqlite_rows_)
+    if len(pure_rows) != len(sqlite_rows_):
+        return [('count', len(pure_rows), len(sqlite_rows_))]
+    return [(p, s) for p, s in zip(pure_rows, sqlite_rows_) if not all(same(x, y) for x, y in zip(p, s))]
+
+
+def consistency(pure, db):
+    """Clause 3.3.2.1-4 on both: the conditions each fails (none when consistent)."""
+    failures = []
+    for system, ytd, district_ytd, districts, ordered, lines in (
+            ('pure', pure.view('warehouse_ytd'), pure.view('district_ytd'), asked(pure, 'district_check'),
+             pure.view('ordered_lines'), pure.view('line_count')),
+            ('sqlite', *db.execute('SELECT (SELECT sum(w_ytd) FROM warehouse), (SELECT sum(d_ytd) FROM district)').fetchone(),
+             db.execute(DISTRICTS[1]).fetchall(), *db.execute(
+                 'SELECT (SELECT sum(o_ol_cnt) FROM orders), (SELECT count(*) FROM order_line)').fetchone())):
+        if not same(ytd, district_ytd):
+            failures.append((system, '3.3.2.1', ytd, district_ytd))
+        for key, next_order, newest, newest_new, oldest_new, new_orders in districts:
+            if not next_order - 1 == newest == newest_new:
+                failures.append((system, '3.3.2.2', key, next_order, newest, newest_new))
+            if newest_new - oldest_new + 1 != new_orders:
+                failures.append((system, '3.3.2.3', key, oldest_new, newest_new, new_orders))
+        if ordered != lines:
+            failures.append((system, '3.3.2.4', ordered, lines))
+    return failures
+
+
+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('--directory', type=Path, default=Path('/dev/shm'), help='where the store and database live')
+    parser.add_argument('--warehouses', type=int, default=1)
+    parser.add_argument('--seed', type=int, default=1)
+    parser.add_argument('--check', type=int, default=20000, help='transactions run on both and compared (0: none)')
+    parser.add_argument('--seconds', type=float, default=20)
+    parser.add_argument('--terminals', default='16,64', help="Pure's connections, a run each")
+    parser.add_argument('--threads', default='3', help="pure-serve's --threads")
+    parser.add_argument('--sqlite-threads', default='1,4,16', help="SQLite's threads, a run each")
+    parser.add_argument('--sync', default='full', choices=('full', 'normal', 'off'))
+    args = parser.parse_args()
+    work = Path(tempfile.mkdtemp(prefix='pure-tpcc-'))
+    place = Path(tempfile.mkdtemp(prefix='pure-tpcc-', dir=args.directory))
+    pure = None
+    try:
+        module = work / 'tpcc.pmod'
+        subprocess.run([str(args.aup), 'build', str(HERE / 'tpcc.aup'), '-o', str(module)], check=True)
+        subprocess.run(['cc', '-O2', '-o', str(work / 'tpcc-load'), str(HERE / 'tpcc-load.c')], check=True)
+        subprocess.run(['cc', '-O2', '-o', str(work / 'sqlite-tpcc'), str(HERE / 'sqlite-tpcc.c'), '-lsqlite3', '-lpthread'],
+                       check=True)
+        began = time.perf_counter()
+        tables = data.generate(args.warehouses, args.seed)
+        rows = sum(len(t) for t in tables.values())
+        generated = time.perf_counter() - began
+        pure = Pure(args.serve, module, place / 'store', args.threads)
+        began = time.perf_counter()
+        pure.load(tables)
+        pure_load = time.perf_counter() - began
+        database = place / 'tpcc.db'
+        began = time.perf_counter()
+        load_sqlite(database, tables)
+        sqlite_load = time.perf_counter() - began
+        counts = {'order_count': len(tables['orders']), 'new_order_count': len(tables['new_orders']),
+                  'line_count': len(tables['lines'])}
+        assert {v: pure.view(v) for v in counts} == counts
+        print(json.dumps({'step': 'load', 'directory': str(args.directory), 'warehouses': args.warehouses, 'rows': rows,
+                          'generate_s': round(generated, 2), 'pure_load_s': round(pure_load, 2),
+                          'sqlite_load_s': round(sqlite_load, 2), 'pure_resident_mib': pure.resident_mib(),
+                          'sqlite_file_mib': round(database.stat().st_size / 2**20, 1)}), flush=True)
+        db = sqlite3.connect(str(database))
+        register(pure, args.aup, work, 'district_check', DISTRICTS[0])
+        if args.check:
+            pure_run = json.loads(subprocess.run(
+                [str(work / 'tpcc-load'), str(pure.port), '1', '3600', str(args.warehouses), str(args.seed),
+                 str(args.check)], capture_output=True, text=True, check=True).stdout)
+            sqlite_run = json.loads(subprocess.run(
+                [str(work / 'sqlite-tpcc'), str(database), str(args.warehouses), '1', '3600', args.sync, str(args.seed),
+                 str(args.check)], capture_output=True, text=True, check=True).stdout)
+            differing = {}
+            for table, (view, sql) in DUMPS.items():
+                register(pure, args.aup, work, f'dump_{table}', view)
+                found = compare(table, asked(pure, f'dump_{table}'), db.execute(sql).fetchall())
+                if found:
+                    differing[table] = [len(found), found[:3]]
+            kinds = ('new_order', 'payment', 'order_status', 'delivery', 'stock_level')
+            print(json.dumps({'step': 'same transactions', 'transactions': args.check,
+                              'pure': {k: pure_run[k]['done'] for k in kinds},
+                              'sqlite': {k: sqlite_run[k]['done'] for k in kinds},
+                              'undone_new_orders': [pure_run['undone_new_orders'], sqlite_run['undone_new_orders']],
+                              'failed': [sum(pure_run[k]['failed'] for k in kinds),
+                                         sum(sqlite_run[k]['failed'] for k in kinds)],
+                              'differing': differing, 'consistency_failures': consistency(pure, db)}), flush=True)
+        for terminals in (int(t) for t in args.terminals.split(',')):
+            used = pure.cpu_seconds()
+            result = json.loads(subprocess.run(
+                ['taskset', '-c', '3', str(work / 'tpcc-load'), str(pure.port), str(terminals), str(args.seconds),
+                 str(args.warehouses), str(args.seed + terminals)], capture_output=True, text=True, check=True).stdout)
+            used = pure.cpu_seconds() - used
+            done = sum(result[k]['done'] for k in ('new_order', 'payment', 'order_status', 'delivery', 'stock_level'))
+            print(json.dumps({'step': 'run', **result, 'server_threads': args.threads,
+                              'cpu_us_per_transaction': round(used / max(done, 1) * 1e6, 1),
+                              'resident_mib': pure.resident_mib(), 'consistency_failures': consistency(pure, db)}),
+                  flush=True)
+        for threads in (int(t) for t in args.sqlite_threads.split(',')):
+            result = json.loads(subprocess.run(
+                ['taskset', '-c', '0-2', str(work / 'sqlite-tpcc'), str(database), str(args.warehouses), str(threads),
+                 str(args.seconds), args.sync, str(args.seed + threads)], capture_output=True, text=True,
+                check=True).stdout)
+            print(json.dumps({'step': 'run', **result, 'consistency_failures': consistency(pure, db)}), flush=True)
+        db.close()
+    finally:
+        if pure:
+            pure.stop()
+        shutil.rmtree(place, ignore_errors=True)
+        shutil.rmtree(work, ignore_errors=True)
+
+
+if __name__ == '__main__':
+    main()