diff --git a/README.md b/README.md index 47e5b53..b6fcd30 100644 --- a/README.md +++ b/README.md @@ -645,6 +645,30 @@ Conditions are disclosed in the output (rows, batch/transactions, source WAL byt process-wide (it includes the Node baseline), so the meaningful claim is that it stays **flat** as the database grows — see the `SCALE` mode, which shows WAL bytes and peak RSS at two sizes. +[`bench/bench-paged-restore.ts`](bench/bench-paged-restore.ts) compares a full restore with a +[paged restore](#restore-page-by-page) as the database grows. The replica is read through a simulated +network (`LATENCY_MS` per request, one link of `BANDWIDTH_MBPS`), and a worker thread reads a random row +as soon as the paged restore resolves: + +```bash +pnpm test:paged-vfs # builds the paged VFS +pnpm bench:paged +SIZES=16,64 LATENCY_MS=5 BANDWIDTH_MBPS=500 pnpm bench:paged +``` + +On an Apple M4 Max, Node 24.18, with 20 ms per request and 100 MB/s: + +| Database | Full restore | Paged: open | Paged: first response | Paged: every page | +| --- | --- | --- | --- | --- | +| 16 MiB | 1.31 s | 298 ms | 335 ms | 510 ms | +| 64 MiB | 4.30 s | 318 ms | 373 ms | 951 ms | +| 257 MiB | 16.03 s | 343 ms | 370 ms | 2.78 s | +| 1027 MiB | 64.02 s | 447 ms | 528 ms | 10.09 s | + +The first response of a paged restore barely grows with the database: before it resolves, it reads the +page indexes of the restore plan (a few bytes per page) and the schema, and a query then waits only for +the pages it reads. Most of the time to open is listing the ten levels of the replica, one request each. + ### Compaction & retention Over time `sync()` writes one level-0 LTX file per transaction. To keep the replica bounded and diff --git a/bench/bench-paged-restore.ts b/bench/bench-paged-restore.ts new file mode 100644 index 0000000..f85f372 --- /dev/null +++ b/bench/bench-paged-restore.ts @@ -0,0 +1,188 @@ +/** + * Time to first response of a paged restore, against a full restore, as the + * database grows. + * + * For each size, a database of incompressible rows (about one per page) is + * replicated to a file replica, which is then read through a simulated + * network: each request waits LATENCY_MS, and the bytes share one link of + * BANDWIDTH_MBPS, as reads from S3 would. Then the database is restored: + * + * - full: restore() of the replica's own path, until it resolves, after + * which a query can run; + * - paged: restorePaged(), until it resolves (open), until a query from a + * worker thread returns a random row (first response), and until + * every page is local (filled). + * + * Tunables (env): + * SIZES database sizes, in MiB (default 16,64,256,1024) + * LATENCY_MS latency of each request, in ms (default 20) + * BANDWIDTH_MBPS bandwidth of the link, in MB/s (default 100) + * + * Run (the paged VFS must be built, as pnpm test:paged-vfs does): + * pnpm bench:paged + * SIZES=16,64 LATENCY_MS=5 pnpm bench:paged + */ + +import { DatabaseSync } from 'node:sqlite'; +import { mkdtempSync } from 'node:fs'; +import { rm, stat } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { Readable } from 'node:stream'; +import { setTimeout as sleep } from 'node:timers/promises'; +import { Worker } from 'node:worker_threads'; + +import { databaseUri } from '@platformatic/sqlite-js-stream-paged-vfs'; + +import { + FileReplicaClient, + Replica, + type LTXFilesOptions, + type OpenLTXFileOptions, + type ReplicaClient, +} from '../src/index.ts'; + +const SIZES = (process.env.SIZES ?? '16,64,256,1024').split(',').map(Number); +const LATENCY_MS = Number(process.env.LATENCY_MS ?? 20); +const BANDWIDTH_MBPS = Number(process.env.BANDWIDTH_MBPS ?? 100); + +/** + * Reads `client` through a simulated network: every request (a read or a + * listing) waits `latencyMs`, and the bytes read share one link of + * `megabytesPerSecond`, whatever the number of concurrent reads. + */ +function network(client: ReplicaClient, latencyMs: number, megabytesPerSecond: number): ReplicaClient { + const bytesPerMs = (megabytesPerSecond * 1e6) / 1000; + let linkFree = 0; + const transfer = async (bytes: number): Promise => { + const now = performance.now(); + linkFree = Math.max(now, linkFree) + bytes / bytesPerMs; + if (linkFree - now >= 1) await sleep(linkFree - now); + }; + + return new Proxy(client, { + get(target, prop) { + if (prop === 'openLtxFile') { + return async (opts: OpenLTXFileOptions) => { + await sleep(latencyMs); + const stream = await target.openLtxFile(opts); + return Readable.from( + (async function* () { + for await (const chunk of stream) { + await transfer(chunk.length); + yield chunk; + } + })(), + ); + }; + } + if (prop === 'ltxFiles') { + return (opts: LTXFilesOptions) => + (async function* () { + await sleep(latencyMs); + yield* target.ltxFiles(opts); + })(); + } + const value = Reflect.get(target, prop, target); + return typeof value === 'function' ? value.bind(target) : value; + }, + }); +} + +/** Replicates a database of about `mib` MiB, then changes a few rows twice. Returns its number of rows. */ +async function replicate(dir: string, client: ReplicaClient, mib: number): Promise { + const path = join(dir, 'source.db'); + const rows = mib * 256; // one row of 3,000 random bytes per 4 KiB page + const db = new DatabaseSync(path); + db.exec('PRAGMA journal_mode=WAL; PRAGMA wal_autocheckpoint=0; PRAGMA synchronous=OFF'); + db.exec('CREATE TABLE t (id INTEGER PRIMARY KEY, v BLOB)'); + db.exec(`WITH RECURSIVE n(i) AS (SELECT 1 UNION ALL SELECT i + 1 FROM n WHERE i < ${rows}) INSERT INTO t SELECT i, randomblob(3000) FROM n`); + // The first sync reads the database file rather than a WAL of the same size + db.exec('PRAGMA wal_checkpoint(TRUNCATE)'); + + const replica = new Replica(path, client); + await replica.sync(); + for (let i = 0; i < 2; i++) { + db.exec(`UPDATE t SET v = randomblob(3000) WHERE id % 1000 = ${i}`); + await replica.sync(); + } + replica.close(); + db.close(); + return rows; +} + +/** A worker thread which opens the database and reads one row when told, as a Durable Object host would. */ +function queryWorker(): { ready: Promise; query: (uri: string, id: number) => Promise; close: () => Promise } { + const worker = new Worker( + ` + const { parentPort } = require('node:worker_threads') + const { DatabaseSync } = require('node:sqlite') + parentPort.on('message', ({ uri, id }) => { + const db = new DatabaseSync(uri) + const row = db.prepare('SELECT length(v) AS n FROM t WHERE id = ?').get(id) + db.close() + parentPort.postMessage(row) + }) + `, + { eval: true }, + ); + return { + ready: new Promise((resolve) => worker.once('online', () => resolve())), + query: (uri, id) => + new Promise((resolve, reject) => { + worker.once('message', (row: { n: number } | undefined) => (row?.n === 3000 ? resolve() : reject(new Error(`unexpected row ${JSON.stringify(row)}`)))); + worker.postMessage({ uri, id }); + }), + close: () => worker.terminate(), + }; +} + +function seconds(ms: number): string { + return ms < 1000 ? `${Math.round(ms)} ms` : `${(ms / 1000).toFixed(2)} s`; +} + +console.log(`Paged restore benchmark: Node ${process.version}, ${LATENCY_MS} ms per request, ${BANDWIDTH_MBPS} MB/s`); +console.log(''); +console.log('| Database | Full restore | Paged: open | Paged: first response | Paged: every page |'); +console.log('| --- | --- | --- | --- | --- |'); + +for (const mib of SIZES) { + const dir = mkdtempSync(join(tmpdir(), 'bench-paged-')); + try { + const backing = new FileReplicaClient(join(dir, 'replica')); + const rows = await replicate(dir, backing, mib); + const client = network(backing, LATENCY_MS, BANDWIDTH_MBPS); + + // Full restore + const fullPath = join(dir, 'full.db'); + const full = new Replica(fullPath, client); + let start = performance.now(); + await full.restore({ outputPath: fullPath }); + const fullMs = performance.now() - start; + full.close(); + const size = (await stat(fullPath)).size; + await rm(fullPath, { force: true }); + + // Paged restore, queried from another thread as soon as it resolves + const pagedPath = join(dir, 'paged.db'); + const paged = new Replica(pagedPath, client); + const worker = queryWorker(); + await worker.ready; + start = performance.now(); + const result = await paged.restorePaged(); + if (!result.paged) throw new Error('the paged VFS cannot be used: build it with pnpm test:paged-vfs'); + const openMs = performance.now() - start; + await worker.query(databaseUri(pagedPath), 1 + Math.floor(Math.random() * rows)); + const firstMs = performance.now() - start; + await result.filled; + const filledMs = performance.now() - start; + await worker.close(); + paged.close(); + + console.log( + `| ${(size / 1024 / 1024).toFixed(0)} MiB | ${seconds(fullMs)} | ${seconds(openMs)} | ${seconds(firstMs)} | ${seconds(filledMs)} |`, + ); + } finally { + await rm(dir, { recursive: true, force: true }); + } +} diff --git a/package.json b/package.json index e99bdad..aa9d1d2 100644 --- a/package.json +++ b/package.json @@ -52,7 +52,8 @@ "test:all": "pnpm test && pnpm -r --filter '!@platformatic/sqlite-js-stream-paged-vfs' test", "test:paged-vfs": "pnpm --filter @platformatic/sqlite-js-stream-paged-vfs build:native && pnpm --filter @platformatic/sqlite-js-stream-paged-vfs build && pnpm --filter @platformatic/sqlite-js-stream-paged-vfs test && node --test test/integration/paged-restore.test.ts", "typecheck:all": "pnpm typecheck", - "bench": "node bench/bench-sync-restore.ts" + "bench": "node bench/bench-sync-restore.ts", + "bench:paged": "node bench/bench-paged-restore.ts" }, "dependencies": { "lz4js": "^0.2.0" diff --git a/test/integration/paged-restore.test.ts b/test/integration/paged-restore.test.ts index 30d5f3e..a2acdd9 100644 --- a/test/integration/paged-restore.test.ts +++ b/test/integration/paged-restore.test.ts @@ -7,6 +7,7 @@ import { once } from 'node:events'; import { existsSync } from 'node:fs'; import { readFile } from 'node:fs/promises'; import { join } from 'node:path'; +import { Readable } from 'node:stream'; import { Decoder, @@ -41,11 +42,11 @@ const Rows = 2000; const Tables = 80; /** - * A database of about 2,000 pages, whose schema spans several pages and has + * A database of about `rows` pages, whose schema spans several pages and has * records larger than a page (so overflow pages), replicated as a level-9 * snapshot followed by level-0 files which change some of its pages. */ -async function replicate(t: TestContext, client: ReplicaClient): Promise { +async function replicate(t: TestContext, client: ReplicaClient, rows = Rows): Promise { const path = join(makeTempDir('paged-src-'), 'src.db'); const src = new DatabaseSync(path); t.after(() => src.close()); @@ -56,7 +57,7 @@ async function replicate(t: TestContext, client: ReplicaClient): Promise `column_with_a_long_name_${i} TEXT`).join(', '); src.exec(`CREATE TABLE wide (${columns})`); - src.exec(`WITH RECURSIVE n(i) AS (SELECT 1 UNION ALL SELECT i + 1 FROM n WHERE i < ${Rows}) INSERT INTO t SELECT i, randomblob(3000) FROM n`); + src.exec(`WITH RECURSIVE n(i) AS (SELECT 1 UNION ALL SELECT i + 1 FROM n WHERE i < ${rows}) INSERT INTO t SELECT i, randomblob(3000) FROM n`); src.exec('COMMIT'); const replica = new Replica(path, client); @@ -78,28 +79,40 @@ async function replicate(t: TestContext, client: ReplicaClient): Promise void; reads: OpenLTXFileOptions[] } { +function gated( + client: ReplicaClient, + minSize = 256 * 1024, +): { client: ReplicaClient; release: () => void; reads: OpenLTXFileOptions[]; bytes: () => number } { let release!: () => void; const open = new Promise((resolve) => { release = resolve; }); const reads: OpenLTXFileOptions[] = []; + let bytes = 0; const proxy = new Proxy(client, { get(target, prop) { if (prop === 'openLtxFile') { return async (opts: OpenLTXFileOptions) => { reads.push(opts); if (opts.size >= minSize) await open; - return target.openLtxFile(opts); + const stream = await target.openLtxFile(opts); + return Readable.from( + (async function* () { + for await (const chunk of stream) { + bytes += chunk.length; + yield chunk; + } + })(), + ); }; } const value = Reflect.get(target, prop, target); return typeof value === 'function' ? value.bind(target) : value; }, }); - return { client: proxy, release, reads }; + return { client: proxy, release, reads, bytes: () => bytes }; } interface SqlWorker { @@ -251,6 +264,62 @@ test('replication goes on while a paged restore fills the database', { skip }, a await db.close(); }); +test('once every page is restored, nothing is read from the replica', { skip }, async (t) => { + const backing = new FileReplicaClient(makeTempDir('paged-repl-')); + const src = await replicate(t, backing); + const { client, release, reads } = gated(backing); + release(); + + const path = join(makeTempDir('paged-dst-'), 'db.db'); + const replica = new Replica(path, client); + t.after(() => replica.close()); + const result = await replica.restorePaged(); + await result.filled; + reads.length = 0; + + // Queries read every page, writes are replicated, and a snapshot reads the database again + const db = sqlWorker(t, path, true); + assert.deepEqual(await db.all('PRAGMA integrity_check'), [{ integrity_check: 'ok' }]); + assert.deepEqual( + await db.all('SELECT count(*) AS n, sum(length(v)) AS size FROM t'), + plain(src.prepare('SELECT count(*) AS n, sum(length(v)) AS size FROM t').all()), + ); + await db.exec("INSERT INTO t (v) VALUES (x'02')"); + assert.equal((await replica.sync())?.maxTXID, 7n); + assert.equal((await replica.snapshot())?.maxTXID, 7n); + await db.close(); + + assert.deepEqual(reads, []); +}); + +test('the reads before a paged restore resolves barely grow with the database (acceptance)', { skip }, async (t) => { + // What is read before the database can be opened: the headers of two files, the page indexes of the plan (a few + // bytes per page), and the schema with its read-ahead. Queries then read only the pages they need, so the time to + // the first response does not grow with the size of the database the way a full restore does. + const measure = async (rows: number): Promise<{ size: number; before: number; reads: number }> => { + const backing = new FileReplicaClient(makeTempDir('paged-repl-')); + await replicate(t, backing, rows); + const { client, release, bytes, reads } = gated(backing); + t.after(release); + + const path = join(makeTempDir('paged-dst-'), 'db.db'); + const replica = new Replica(path, client); + t.after(() => replica.close()); + const result = await replica.restorePaged(); + const before = { before: bytes(), reads: reads.length }; + release(); + await result.filled; + return { size: (await readFile(path)).length, ...before }; + }; + + const small = await measure(2000); + const large = await measure(8000); + assert.ok(large.size > 3.5 * small.size, `the large database (${large.size} bytes) is 4 times the small one (${small.size})`); + assert.ok(large.before < 2 * small.before, `read before opening: ${large.before} bytes for ${large.size}, ${small.before} for ${small.size}`); + assert.ok(large.before < large.size / 50, `read before opening: ${large.before} bytes, under 2% of ${large.size}`); + t.diagnostic(`read before opening: ${small.before} of ${small.size} bytes, ${large.before} of ${large.size} bytes`); +}); + test('closing the replica stops a paged restore, and the database is restored again', { skip }, async (t) => { const backing = new FileReplicaClient(makeTempDir('paged-repl-')); await replicate(t, backing);