diff --git a/CHANGELOG.md b/CHANGELOG.md index e123fa3..e4c1938 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/). ### Fixed +- **SQLite: paging a chunked download no longer gets slower with every chunk.** The positional cursor on `cloudsync_payload_chunks` was meant to seek straight to where the previous call stopped, but it stated its resume point only inside `(db_version > ? OR (db_version = ? AND seq >= ?))`, whose two arms carry distinct parameters. SQLite does not derive a range from that, so the scan over `cloudsync_changes` ran with an upper bound only and re-read the window from the beginning on every call, discarding rows until it reached the resume point — making a full drain quadratic in the number of chunks, and long enough on a large tenant to hit a server-side deadline and never complete. An explicit `db_version >= ?` is now stated alongside the disjunction; it selects exactly the same rows and lets the scan seek. Locally, draining a 188-chunk window went from 6110 ms to 210 ms, and the per-chunk cost no longer depends on how large the window is. PostgreSQL was never affected. - **A row rewritten in one statement no longer keeps the old blocks of a shorter value.** Writing a whole row rewrites its block column from the first position, so any block past the end of the new value stayed stored. On SQLite `INSERT OR REPLACE` skips the old row's delete trigger unless `recursive_triggers` is on, so those leftovers kept their metadata and were delivered as content: replacing `AAA\nBBB\nCCC` with `ZZZ` left `ZZZ\nBBB\nCCC` on the peers and on a later local read. On PostgreSQL they carried no metadata, so peers were unaffected, but they stayed in the blocks table for the life of the row. Blocks the new value does not cover are now retired with it. - **SQLite: a payload whose commit fails no longer leaves its transaction open.** When `cloudsync_payload_apply` started the transaction itself and the commit then failed — a deferred foreign key violated at commit, or `SQLITE_BUSY` because a reader held the database — the transaction stayed open: the uncommitted rows remained visible on the connection and the next `BEGIN` failed. The failed transaction is now rolled back and the original error is returned. Changes from earlier source versions that were already committed are kept, the receive checkpoint does not move, and the rolled-back rows are no longer counted as applied, so delivering the payload again applies it. A transaction or savepoint opened by the caller is still left to the caller. - **PostgreSQL: applying a payload read from a table works at any savepoint depth.** Inside 126 or more savepoints, `SELECT cloudsync_payload_apply(payload) FROM some_table` still failed with `buffer pin ... is not owned by resource owner SubTransaction` (and a caught error at that depth could abort an assertion-enabled server): cloudsync recorded the caller's resource owner and memory context for at most 128 nesting levels, counting its own internal savepoints, and silently stopped restoring them beyond that. The fixed limit is gone; only PostgreSQL's own resource limits apply. diff --git a/Makefile b/Makefile index aeec422..4c119a1 100644 --- a/Makefile +++ b/Makefile @@ -357,6 +357,17 @@ sync-bench: $(TARGET) $(DIST_DIR)/sync_bench$(EXE) sync-bench-debug: $(MAKE) NETWORK_TRACE=1 sync-bench +# Time the positional /check drain on a locally generated window. Separate from +# the test targets because timings are machine-dependent; it needs no network. +chunk-bench: $(TARGET) $(DIST_DIR)/chunk_bench$(EXE) + @if [ -n "$(CHUNK_BENCH_ROWS)" ]; then export CHUNK_BENCH_ROWS="$(CHUNK_BENCH_ROWS)"; fi; \ + if [ -n "$(CHUNK_BENCH_ROW_BYTES)" ]; then export CHUNK_BENCH_ROW_BYTES="$(CHUNK_BENCH_ROW_BYTES)"; fi; \ + if [ -n "$(CHUNK_BENCH_TXNS)" ]; then export CHUNK_BENCH_TXNS="$(CHUNK_BENCH_TXNS)"; fi; \ + if [ -n "$(CHUNK_BENCH_REPEATS)" ]; then export CHUNK_BENCH_REPEATS="$(CHUNK_BENCH_REPEATS)"; fi; \ + if [ -n "$(CHUNK_BENCH_CHUNK_SIZE)" ]; then export CHUNK_BENCH_CHUNK_SIZE="$(CHUNK_BENCH_CHUNK_SIZE)"; fi; \ + if [ -n "$(CHUNK_BENCH_VERBOSE)" ]; then export CHUNK_BENCH_VERBOSE="$(CHUNK_BENCH_VERBOSE)"; fi; \ + ./$(DIST_DIR)/chunk_bench$(EXE) + OPENSSL_TARBALL = $(OPENSSL_DIR)/$(OPENSSL_VERSION).tar.gz $(OPENSSL_TARBALL): diff --git a/src/sqlite/cloudsync_sqlite.c b/src/sqlite/cloudsync_sqlite.c index 25cdd7c..8345dd8 100644 --- a/src/sqlite/cloudsync_sqlite.c +++ b/src/sqlite/cloudsync_sqlite.c @@ -1401,9 +1401,14 @@ static int payload_chunks_filter(sqlite3_vtab_cursor *cursor, int idxnum, const // (db_version, seq) >= (resume_dbv, resume_seq). char *sql; if (positional) { + // The redundant db_version>=? is what makes the resume a seek, and it is + // load-bearing: SQLite derives a range from a disjunction only when both arms + // compare against the same value, and these two arms carry distinct + // parameters. Without it cloudsync_changes' xBestIndex is offered no lower + // bound at all and every call replays the window from the start. sql = sqlite3_mprintf( "SELECT tbl, pk, col_name, col_value, col_version, db_version, site_id, cl, seq " - "FROM cloudsync_changes WHERE db_version<=? AND site_id%s? AND " + "FROM cloudsync_changes WHERE db_version<=? AND site_id%s? AND db_version>=? AND " "(db_version>? OR (db_version=? AND seq>=?)) ORDER BY db_version, seq ASC", site_op); } else { @@ -1421,7 +1426,8 @@ static int payload_chunks_filter(sqlite3_vtab_cursor *cursor, int idxnum, const sqlite3_bind_blob(c->src, 2, site_id, site_id_len, SQLITE_TRANSIENT); sqlite3_bind_int64(c->src, 3, resume_dbv); sqlite3_bind_int64(c->src, 4, resume_dbv); - sqlite3_bind_int64(c->src, 5, resume_seq); + sqlite3_bind_int64(c->src, 5, resume_dbv); + sqlite3_bind_int64(c->src, 6, resume_seq); } else { sqlite3_bind_int64(c->src, 1, since); sqlite3_bind_blob(c->src, 2, site_id, site_id_len, SQLITE_TRANSIENT); diff --git a/test/chunk_bench.c b/test/chunk_bench.c index 4321ac2..0bc37a1 100644 --- a/test/chunk_bench.c +++ b/test/chunk_bench.c @@ -2,28 +2,92 @@ // chunk_bench.c // cloudsync // -// Local-only benchmark for the positional /check drain: build a window of N -// chunks and time paging the whole window one chunk per call via the -// (resume_db_version, resume_seq, resume_frag_offset) cursor on -// cloudsync_payload_chunks. Reports wall time and per-chunk cost so the -// computational growth of the drain (currently O(N^2): each resume re-scans -// cloudsync_changes) can be tracked — e.g. to confirm a future indexed -// (db_version, seq) seek flattens it to O(N). +// Local-only benchmark for the positional /check drain. Two measurements, on +// one generated window: // -// Env: CHUNK_BENCH_ROWS (default 400), CHUNK_BENCH_ROW_BYTES (default 60000), -// CHUNK_BENCH_TXNS (default 1; rows split across this many db_versions), -// CHUNK_BENCH_REPEATS (default 5), CHUNK_BENCH_CHUNK_SIZE (default 262144). +// Phase 1 — drain the whole window one chunk per call through the +// (resume_db_version, resume_seq, resume_frag_offset) cursor on +// cloudsync_payload_chunks, timing every chunk. Latency that rises with the +// chunk index means each resume re-scans cloudsync_changes from the start of +// the window, making the drain O(N^2); flat latency means the resume seeks. +// +// Phase 2 — replay the same resume points straight against cloudsync_changes +// in three SQL shapes: the lower bound as payload_chunks_filter() spelled it +// before the fix, the shape it uses now, and a variant. Same rows, different +// plans, so the difference is attributable to the spelling alone. Keeping the +// old shape here is what makes this a regression check rather than a one-off. +// +// CI builds this with the other test binaries but never runs it: the timings are +// machine-dependent. Run it by hand with `make chunk-bench`. +// +// Env: CHUNK_BENCH_ROWS (default 3000), CHUNK_BENCH_ROW_BYTES (default 8000), +// CHUNK_BENCH_TXNS (default = rows, i.e. one db_version per row; 1 is the +// degenerate single-version window where no lower bound can help), +// CHUNK_BENCH_REPEATS (default 3), CHUNK_BENCH_CHUNK_SIZE (default 262144), +// CHUNK_BENCH_VERBOSE (1 = print every chunk, not just deciles). +// +// For a production-sized window, seed once and measure two builds against it: +// CHUNK_BENCH_DB (database path), CHUNK_BENCH_KEEP=1 (do not delete it), +// CHUNK_BENCH_REUSE=1 (skip seeding when it already exists), +// CHUNK_BENCH_EXT (which cloudsync to load), CHUNK_BENCH_PHASE2=0 (skip +// Phase 2, which is itself quadratic and dominates at that size). // #include #include #include #include +#include #include #include "sqlite3.h" #define DB_PATH "dist/chunk-bench.sqlite" #define EXT_PATH "./dist/cloudsync" +#define MAX_POINTS 20000 +#define BUCKETS 10 + +static const char *env_str(const char *name, const char *dflt) { + const char *v = getenv(name); + return (v && *v) ? v : dflt; +} + +static bool file_exists(const char *path) { + FILE *f = fopen(path, "rb"); + if (!f) return false; + fclose(f); + return true; +} + +// Every shape binds ?1=until, ?2=site_id, ?3/?4/?6=resume_db_version, ?5=resume_seq, +// and selects exactly the same rows. Only the spelling of the lower bound differs. + +// What payload_chunks_filter() emitted before the fix: the resume point lived +// entirely inside a disjunction whose two arms carry *distinct* parameters. +// Distinct parameters are what defeats the planner here — see below. +#define SHAPE_OLD \ + "SELECT db_version, seq FROM cloudsync_changes " \ + "WHERE db_version<=?1 AND site_id<>?2 AND (db_version>?3 OR (db_version=?4 AND seq>=?5)) " \ + "ORDER BY db_version, seq ASC LIMIT 1" + +// What it emits now: a redundant conjunct stating the lower bound where a virtual +// table's xBestIndex can see it, alongside the untouched disjunction. +#define SHAPE_FIXED \ + "SELECT db_version, seq FROM cloudsync_changes " \ + "WHERE db_version<=?1 AND site_id<>?2 AND db_version>=?6 AND (db_version>?3 OR (db_version=?4 AND seq>=?5)) " \ + "ORDER BY db_version, seq ASC LIMIT 1" + +// The same disjunction with one parameter reused across both arms. SQLite derives +// the common lower bound itself once it can see the two arms compare against the +// same value, so this needs no extra term — and it means the defect is a matter of +// parameter identity, not of disjunctions being opaque to xBestIndex. +#define SHAPE_REUSED_PARAM \ + "SELECT db_version, seq FROM cloudsync_changes " \ + "WHERE db_version<=?1 AND site_id<>?2 AND (db_version>?3 OR (db_version=?3 AND seq>=?5)) " \ + "ORDER BY db_version, seq ASC LIMIT 1" + +typedef struct { + int64_t dbv, seq, frag; +} resume_point; static double monotonic_ms(void) { struct timespec ts; @@ -40,6 +104,14 @@ static int env_int(const char *name, int dflt) { return (int)p; } +// env_int treats 0 as "unset" so that a stray empty value cannot ask for zero rows. +// Flags need the opposite, since 0 is how you turn one off. +static bool env_flag(const char *name, bool dflt) { + const char *v = getenv(name); + if (!v || !*v) return dflt; + return !(strcmp(v, "0") == 0 || strcmp(v, "false") == 0 || strcmp(v, "no") == 0); +} + static int db_exec(sqlite3 *db, const char *sql) { char *err = NULL; int rc = sqlite3_exec(db, sql, NULL, NULL, &err); @@ -50,9 +122,26 @@ static int db_exec(sqlite3 *db, const char *sql) { return rc; } -// Drain the whole window via the positional cursor, one chunk per query. Returns -// the chunk count and accumulates total payload bytes touched into *bytes. -static int drain_positional(sqlite3 *db, int *chunks_out, long long *bytes_out) { +// Print the clause cloudsync_changesvtab_best_index hands the planner for one +// shape. This is the doc's verification step 3: whether the resume lower bound +// reaches the virtual table at all is visible right here, in the idxStr. +static void print_plan(sqlite3 *db, const char *label, const char *sql) { + char *eqp = sqlite3_mprintf("EXPLAIN QUERY PLAN %s", sql); + if (!eqp) return; + sqlite3_stmt *stmt = NULL; + if (sqlite3_prepare_v2(db, eqp, -1, &stmt, NULL) == SQLITE_OK) { + while (sqlite3_step(stmt) == SQLITE_ROW) + printf(" %-9s %s\n", label, sqlite3_column_text(stmt, 3)); + } + sqlite3_finalize(stmt); + sqlite3_free(eqp); +} + +// Drain the whole window via the positional cursor, one chunk per query, timing +// each call and recording the resume point it was issued with so phase 2 can +// replay exactly the same positions. +static int drain_positional(sqlite3 *db, double *per_chunk, resume_point *points, + int *chunks_out, long long *bytes_out, int64_t *until_out) { const char *first_sql = "SELECT payload, next_db_version, next_seq, next_frag_offset, is_final, watermark_db_version " "FROM cloudsync_payload_chunks WHERE since_db_version=0 LIMIT 1;"; @@ -68,10 +157,12 @@ static int drain_positional(sqlite3 *db, int *chunks_out, long long *bytes_out) int chunks = 0; long long bytes = 0; - long long watermark = 0, rdbv = 0, rseq = 0, rfrag = 0; + int64_t watermark = 0, rdbv = 0, rseq = 0, rfrag = 0; bool is_final = false; + double t0 = monotonic_ms(); rc = sqlite3_step(first); + double dt = monotonic_ms() - t0; if (rc == SQLITE_ROW) { bytes += sqlite3_column_bytes(first, 0); rdbv = sqlite3_column_int64(first, 1); @@ -79,6 +170,7 @@ static int drain_positional(sqlite3 *db, int *chunks_out, long long *bytes_out) rfrag = sqlite3_column_int64(first, 3); is_final = sqlite3_column_int(first, 4) != 0; watermark = sqlite3_column_int64(first, 5); + if (per_chunk) per_chunk[0] = dt; chunks++; } else if (rc == SQLITE_DONE) { rc = SQLITE_OK; @@ -88,23 +180,38 @@ static int drain_positional(sqlite3 *db, int *chunks_out, long long *bytes_out) } while (!is_final) { + if (points && chunks < MAX_POINTS) { + points[chunks].dbv = rdbv; points[chunks].seq = rseq; points[chunks].frag = rfrag; + } sqlite3_reset(resume); sqlite3_bind_int64(resume, 1, watermark); sqlite3_bind_int64(resume, 2, rdbv); sqlite3_bind_int64(resume, 3, rseq); sqlite3_bind_int64(resume, 4, rfrag); + t0 = monotonic_ms(); rc = sqlite3_step(resume); - if (rc != SQLITE_ROW) { if (rc == SQLITE_DONE) rc = SQLITE_OK; break; } + dt = monotonic_ms() - t0; + // A drain that stops early looks exactly like a fast one, so never report it + // as success. SQLITE_DONE here is a failure too: the loop only runs while the + // previous chunk said it was not final, so the stream owes us another chunk. + if (rc != SQLITE_ROW) { + fprintf(stderr, "resume failed at chunk %d: %s\n", chunks, + (rc == SQLITE_DONE) ? "no chunk returned before is_final" : sqlite3_errmsg(db)); + if (rc == SQLITE_DONE) rc = SQLITE_ERROR; + goto done; + } bytes += sqlite3_column_bytes(resume, 0); rdbv = sqlite3_column_int64(resume, 1); rseq = sqlite3_column_int64(resume, 2); rfrag = sqlite3_column_int64(resume, 3); is_final = sqlite3_column_int(resume, 4) != 0; + if (per_chunk && chunks < MAX_POINTS) per_chunk[chunks] = dt; chunks++; } rc = SQLITE_OK; *chunks_out = chunks; *bytes_out = bytes; + *until_out = watermark; done: if (first) sqlite3_finalize(first); @@ -112,66 +219,207 @@ static int drain_positional(sqlite3 *db, int *chunks_out, long long *bytes_out) return rc; } +// Mean of the samples falling in decile b, so a curve over hundreds of chunks +// reads as ten numbers. Returns -1 for an empty bucket. +static double bucket_mean(const double *v, int n, int b) { + double sum = 0; int count = 0; + for (int i = 0; i < n; ++i) { + if (i * BUCKETS / n != b) continue; + sum += v[i]; count++; + } + return count ? sum / count : -1.0; +} + +static void report_curve(const char *title, const double *v, int n) { + printf("%s\n decile:", title); + for (int b = 0; b < BUCKETS; ++b) printf(" %7d%%", (b + 1) * 10); + printf("\n ms/call:"); + for (int b = 0; b < BUCKETS; ++b) { + double m = bucket_mean(v, n, b); + if (m < 0) printf(" --"); else printf(" %8.3f", m); + } + double head = bucket_mean(v, n, 0), tail = bucket_mean(v, n, BUCKETS - 1); + printf("\n last/first decile: "); + if (head > 0 && tail > 0) printf("%.2fx\n", tail / head); else printf("n/a\n"); +} + +// Time one shape at every recorded resume point. Returns the first row each query +// produced so the caller can prove the two shapes select identically. +static int probe_shape(sqlite3 *db, const char *sql, int64_t until, const void *site_id, int site_len, + const resume_point *points, int n, double *out, resume_point *rows) { + sqlite3_stmt *stmt = NULL; + int rc = sqlite3_prepare_v2(db, sql, -1, &stmt, NULL); + if (rc != SQLITE_OK) { fprintf(stderr, "prepare failed: %s\n", sqlite3_errmsg(db)); return rc; } + for (int i = 0; i < n; ++i) { + sqlite3_reset(stmt); + sqlite3_bind_int64(stmt, 1, until); + sqlite3_bind_blob(stmt, 2, site_id, site_len, SQLITE_TRANSIENT); + sqlite3_bind_int64(stmt, 3, points[i].dbv); + sqlite3_bind_int64(stmt, 4, points[i].dbv); + sqlite3_bind_int64(stmt, 5, points[i].seq); + sqlite3_bind_int64(stmt, 6, points[i].dbv); // SQLITE_RANGE where unused + double t0 = monotonic_ms(); + rc = sqlite3_step(stmt); + out[i] = monotonic_ms() - t0; + if (rc == SQLITE_ROW) { + rows[i].dbv = sqlite3_column_int64(stmt, 0); + rows[i].seq = sqlite3_column_int64(stmt, 1); + } else { + rows[i].dbv = -1; rows[i].seq = -1; + } + } + sqlite3_finalize(stmt); + return SQLITE_OK; +} + int main(void) { - int rows = env_int("CHUNK_BENCH_ROWS", 400); - int row_bytes = env_int("CHUNK_BENCH_ROW_BYTES", 60000); - int repeats = env_int("CHUNK_BENCH_REPEATS", 5); + int rows = env_int("CHUNK_BENCH_ROWS", 3000); + int row_bytes = env_int("CHUNK_BENCH_ROW_BYTES", 8000); + int repeats = env_int("CHUNK_BENCH_REPEATS", 3); int chunk_size = env_int("CHUNK_BENCH_CHUNK_SIZE", 262144); + int txns = env_int("CHUNK_BENCH_TXNS", rows); + bool verbose = env_flag("CHUNK_BENCH_VERBOSE", false); + // A large window costs minutes to seed and the unfixed arm re-reads it on every + // call, so allow one seeded database to be reused across both builds and the + // quadratic Phase 2 to be skipped. + const char *db_path = env_str("CHUNK_BENCH_DB", DB_PATH); + const char *ext_path = env_str("CHUNK_BENCH_EXT", EXT_PATH); + bool keep = env_flag("CHUNK_BENCH_KEEP", false); + bool reuse = env_flag("CHUNK_BENCH_REUSE", false) && file_exists(db_path); + bool want_phase2 = env_flag("CHUNK_BENCH_PHASE2", true); + if (txns > rows) txns = rows; - remove(DB_PATH); + if (!reuse) remove(db_path); sqlite3 *db = NULL; - if (sqlite3_open(DB_PATH, &db) != SQLITE_OK) { fprintf(stderr, "open failed\n"); return 1; } + if (sqlite3_open(db_path, &db) != SQLITE_OK) { fprintf(stderr, "open failed\n"); return 1; } if (sqlite3_enable_load_extension(db, 1) != SQLITE_OK) return 1; - if (db_exec(db, "SELECT load_extension('" EXT_PATH "');") != SQLITE_OK) return 1; + char load[512]; + snprintf(load, sizeof(load), "SELECT load_extension('%s');", ext_path); + if (db_exec(db, load) != SQLITE_OK) return 1; + printf("extension: %s database: %s%s\n", ext_path, db_path, reuse ? " (reused)" : ""); char setup[256]; - snprintf(setup, sizeof(setup), - "CREATE TABLE chunk_bench (id TEXT PRIMARY KEY, body BLOB);" - "SELECT cloudsync_init('chunk_bench');" - "SELECT cloudsync_set('payload_max_chunk_size', '%d');", chunk_size); + snprintf(setup, sizeof(setup), "SELECT cloudsync_set('payload_max_chunk_size', '%d');", chunk_size); + if (!reuse) { + char create[256]; + snprintf(create, sizeof(create), + "PRAGMA journal_mode=WAL; PRAGMA synchronous=OFF;" + "CREATE TABLE chunk_bench (id TEXT PRIMARY KEY, body BLOB);" + "SELECT cloudsync_init('chunk_bench');"); + if (db_exec(db, create) != SQLITE_OK) return 1; + } if (db_exec(db, setup) != SQLITE_OK) return 1; - // Split the rows across CHUNK_BENCH_TXNS transactions: each is one db_version, - // so TXNS=1 is the pathological single-version window and TXNS=rows is the - // many-versions case a real /check window resembles. Incompressible bodies keep - // the window many-chunked. - int txns = env_int("CHUNK_BENCH_TXNS", 1); - if (txns < 1) txns = 1; - if (txns > rows) txns = rows; - int idbase = 0; - for (int t = 0; t < txns; ++t) { - int n = rows / txns + (t < rows % txns ? 1 : 0); - if (n <= 0) continue; - char insert[256]; - snprintf(insert, sizeof(insert), - "WITH RECURSIVE c(i) AS (SELECT %d UNION ALL SELECT i+1 FROM c WHERE i < %d) " - "INSERT INTO chunk_bench(id, body) SELECT printf('row-%%06d', i), randomblob(%d) FROM c;", - idbase + 1, idbase + n, row_bytes); - if (db_exec(db, insert) != SQLITE_OK) return 1; - idbase += n; + // Each transaction is one db_version, so TXNS controls how many distinct + // db_versions the window spans. That is the axis the proposed lower bound acts + // on: with TXNS=1 every row shares one db_version and no bound on db_version can + // narrow anything. Incompressible bodies keep the window many-chunked. + if (!reuse) { + printf("seeding %d rows of %d bytes across %d transaction(s)...\n", rows, row_bytes, txns); + double seed_t0 = monotonic_ms(); + int idbase = 0; + for (int t = 0; t < txns; ++t) { + int n = rows / txns + (t < rows % txns ? 1 : 0); + if (n <= 0) continue; + char insert[256]; + snprintf(insert, sizeof(insert), + "WITH RECURSIVE c(i) AS (SELECT %d UNION ALL SELECT i+1 FROM c WHERE i < %d) " + "INSERT INTO chunk_bench(id, body) SELECT printf('row-%%06d', i), randomblob(%d) FROM c;", + idbase + 1, idbase + n, row_bytes); + if (db_exec(db, insert) != SQLITE_OK) return 1; + idbase += n; + } + printf("seeded in %.1f s\n", (monotonic_ms() - seed_t0) / 1000.0); } + double *per_chunk = calloc(MAX_POINTS, sizeof(double)); + resume_point *points = calloc(MAX_POINTS, sizeof(resume_point)); + if (!per_chunk || !points) { fprintf(stderr, "oom\n"); return 1; } + + // ---- Phase 1: the real drain, one chunk per call ---- int chunks = 0; long long bytes = 0; + int64_t until = 0; double best = 1e18, sum = 0; for (int r = 0; r < repeats; ++r) { double t0 = monotonic_ms(); - if (drain_positional(db, &chunks, &bytes) != SQLITE_OK) { fprintf(stderr, "positional drain failed\n"); return 1; } + if (drain_positional(db, per_chunk, points, &chunks, &bytes, &until) != SQLITE_OK) { + fprintf(stderr, "positional drain failed: %s\n", sqlite3_errmsg(db)); + return 1; + } double dt = monotonic_ms() - t0; sum += dt; if (dt < best) best = dt; } + int n = chunks < MAX_POINTS ? chunks : MAX_POINTS; - double mean = sum / repeats; - printf("\nPositional /check drain benchmark (local SQLite, no network)\n"); + printf("\n=== Positional /check drain (local SQLite, no network) ===\n"); printf("rows: %d row_bytes: %d txns: %d chunk_size: %d repeats: %d\n", rows, row_bytes, txns, chunk_size, repeats); - printf("chunks: %d payload_bytes: %lld\n", chunks, bytes); - printf("drain: best=%.2f ms mean=%.2f ms\n", best, mean); - if (chunks > 0) - printf("per-chunk: best=%.3f ms throughput: %.1f MB/s\n", - best / chunks, (double)bytes / 1024.0 / 1024.0 / (best / 1000.0)); + printf("chunks: %d payload_bytes: %lld until_db_version: %lld\n", chunks, bytes, (long long)until); + printf("drain: best=%.2f ms mean=%.2f ms per-chunk=%.3f ms\n", + best, sum / repeats, chunks ? best / chunks : 0.0); + if (n > 0) report_curve("\nPhase 1 - cloudsync_payload_chunks latency vs chunk index", per_chunk, n); + if (verbose) for (int i = 0; i < n; ++i) printf(" chunk %5d %8.3f ms\n", i, per_chunk[i]); + + // ---- Phase 2: the same resume points, straight at cloudsync_changes ---- + // points[0] is unset (the first chunk carries no resume point), so start at 1. + int np = (want_phase2 && n > 1) ? n - 1 : 0; + if (np > 0) { + // Exclude a site_id that cannot exist, so the filter matches the real + // query's shape without removing any row. + static const unsigned char absent_site[16] = {0}; + static const char *shape_sql[3] = { SHAPE_OLD, SHAPE_FIXED, SHAPE_REUSED_PARAM }; + static const char *shape_name[3] = { + "2a - old (disjunction, distinct parameters)", + "2b - fixed (+ redundant AND db_version>=?)", + "2c - variant (one parameter reused in both arms)", + }; + static const char *shape_tag[3] = { "old:", "fixed:", "reused:" }; + double *t[3]; + resume_point *sel[3]; + for (int s = 0; s < 3; ++s) { + t[s] = calloc(np, sizeof(double)); + sel[s] = calloc(np, sizeof(resume_point)); + if (!t[s] || !sel[s]) { fprintf(stderr, "oom\n"); return 1; } + } + + printf("\n=== Resume seek against cloudsync_changes, %d points ===\n", np); + printf("clause reaching xBestIndex (a missing db_version >= ? is the replay):\n"); + for (int s = 0; s < 3; ++s) print_plan(db, shape_tag[s], shape_sql[s]); + + for (int s = 0; s < 3; ++s) + if (probe_shape(db, shape_sql[s], until, absent_site, (int)sizeof(absent_site), + points + 1, np, t[s], sel[s]) != SQLITE_OK) return 1; + + for (int s = 0; s < 3; ++s) { + char title[128]; + snprintf(title, sizeof(title), "\nPhase %s", shape_name[s]); + report_curve(title, t[s], np); + } + + printf("\ntotal seek time:"); + for (int s = 0; s < 3; ++s) { + double total = 0; + for (int i = 0; i < np; ++i) total += t[s][i]; + printf(" %s %.2f ms", shape_tag[s], total); + } + printf("\n"); + + // The point of the redundant term is that it is logically implied, so every + // shape must select the identical row at every resume point. + for (int s = 1; s < 3; ++s) { + int mismatches = 0; + for (int i = 0; i < np; ++i) + if (sel[0][i].dbv != sel[s][i].dbv || sel[0][i].seq != sel[s][i].seq) mismatches++; + printf("rows selected by %-7s differ from the old shape at %d of %d resume points\n", + shape_tag[s], mismatches, np); + } + + for (int s = 0; s < 3; ++s) { free(t[s]); free(sel[s]); } + } + free(per_chunk); free(points); sqlite3_close(db); - remove(DB_PATH); + if (!keep) remove(db_path); return 0; }