From a862f9896dd30cc39e0b307ad5249fda7cd97f0f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Th=C3=A9ophile=20Helleboid=20-=20chtitux?= Date: Mon, 20 Apr 2026 07:15:30 +0400 Subject: [PATCH] Optimize GTFS ingestion (~40% faster) and drop bulk-load PRAGMAs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three code-level wins in the loader, measured on the ASTUCE Rouen feed (~430k stop_times rows, 34 MB uncompressed): 1. Count CSV rows via newline scan for progress totals instead of fully parsing every file twice. totalRows is now documented as an estimate. 2. Parse CSVs with papaparse header:false and bind values by pre-computed column index — no per-row { col: value } object allocation. 3. Prepare a single row-sized INSERT per table and reuse it with stmt.run(rowVals), instead of preparing a fresh multi-row INSERT per 1000-row batch. Also dropped the bulk-load PRAGMA block (synchronous=OFF, journal_mode=MEMORY, temp_store=MEMORY, cache_size=-64000, locking_mode=EXCLUSIVE) and its post-ingest reset. Benchmarked effect on sql.js is within noise (≤1%), and their removal unblocks upcoming pluggable-adapter work. Measured: - ASTUCE: 2647 → 1669 ms median (−37%) - Car Jaune: 312 → 188 ms median (−40%) See documents/gtfs-optimize-ingestion.md for the full plan. Co-Authored-By: Claude Opus 4.7 (1M context) --- CHANGELOG.md | 9 +- scripts/bench-after.ts | 45 ++++++ scripts/bench-ingest.ts | 271 ++++++++++++++++++++++++++++++++ scripts/bench-pragmas.ts | 123 +++++++++++++++ src/gtfs-sqljs.ts | 19 +-- src/loaders/csv-parser.ts | 19 +++ src/loaders/data-loader.ts | 139 ++++++++-------- tests/progress-callback.test.ts | 104 ++++++++++++ 8 files changed, 647 insertions(+), 82 deletions(-) create mode 100644 scripts/bench-after.ts create mode 100644 scripts/bench-ingest.ts create mode 100644 scripts/bench-pragmas.ts create mode 100644 tests/progress-callback.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 3f8544a..512c107 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,7 +2,14 @@ ## Upcoming release -- +### Performance + +- Ingestion is ~35-45% faster on medium-to-large feeds: ASTUCE (Rouen, ~430k stop_times rows) drops from ~2650 ms to ~1670 ms; Car Jaune from ~312 ms to ~188 ms. Wins come from parsing each CSV only once (progress totals now use a fast newline-based row-count estimate), loading rows as positional arrays instead of per-row objects, and reusing a single prepared INSERT per table instead of re-preparing a multi-row statement per 1000-row batch. +- Dropped the bulk-load PRAGMA block (`synchronous`, `journal_mode`, `temp_store`, `cache_size`, `locking_mode`) from ingestion. Benchmarked aggregate effect on sql.js is within noise (≤1%); removing them simplifies the code and unblocks upcoming pluggable-adapter work. + +### Behaviour changes + +- `ProgressInfo.totalRows` is now an estimate based on CSV line count — typically exact, but may differ by a few rows per file in edge cases (e.g. trailing blank lines). For a precise post-ingest row count, query the database directly with `COUNT(*)`. ## 0.4.1 diff --git a/scripts/bench-after.ts b/scripts/bench-after.ts new file mode 100644 index 0000000..249500d --- /dev/null +++ b/scripts/bench-after.ts @@ -0,0 +1,45 @@ +/** + * Post-refactor ingestion benchmark: end-to-end GtfsSqlJs.fromZipData on a + * real feed. No PRAGMAs, no synthetic adapter paths — exercises the real + * public API after the optimize-ingestion changes. + * + * Usage: npx tsx scripts/bench-after.ts /path/to/feed.zip + */ + +import { readFileSync } from 'node:fs'; +import { performance } from 'node:perf_hooks'; +import { GtfsSqlJs } from '../src/gtfs-sqljs.ts'; + +const ZIP_PATH = process.argv[2] ?? '/tmp/gtfs-bench/astuce.zip'; +const RUNS = 5; + +async function run() { + const zip = new Uint8Array(readFileSync(ZIP_PATH)); + // warmup + const w = await GtfsSqlJs.fromZipData(zip); + w.close(); + + const samples: number[] = []; + for (let i = 0; i < RUNS; i++) { + const t0 = performance.now(); + const gtfs = await GtfsSqlJs.fromZipData(zip); + const ms = performance.now() - t0; + gtfs.close(); + samples.push(ms); + process.stderr.write(` run ${i + 1}: ${ms.toFixed(1)} ms\n`); + } + + const sorted = [...samples].sort((a, b) => a - b); + const mean = samples.reduce((s, x) => s + x, 0) / samples.length; + const median = sorted.length % 2 + ? sorted[(sorted.length - 1) / 2] + : (sorted[sorted.length / 2 - 1] + sorted[sorted.length / 2]) / 2; + process.stdout.write(`\nfeed: ${ZIP_PATH}\n`); + process.stdout.write(`runs: ${RUNS}\n`); + process.stdout.write(`median: ${median.toFixed(1)} ms\n`); + process.stdout.write(`mean: ${mean.toFixed(1)} ms\n`); + process.stdout.write(`min: ${sorted[0].toFixed(1)} ms\n`); + process.stdout.write(`max: ${sorted[sorted.length - 1].toFixed(1)} ms\n`); +} + +run().catch((e) => { console.error(e); process.exit(1); }); diff --git a/scripts/bench-ingest.ts b/scripts/bench-ingest.ts new file mode 100644 index 0000000..295ca05 --- /dev/null +++ b/scripts/bench-ingest.ts @@ -0,0 +1,271 @@ +/** + * Ingestion-path optimization benchmark. + * + * PRAGMAs are fully disabled in every variant to isolate code-level wins. + * + * Variants tested: + * A baseline — current code (parseCSV twice, 1000-row multi-row INSERT) + * B no-preflight — parse each CSV only once, stream row count + * C no-trim — drop per-field `value.trim()` transform in papaparse + * D arrays — papaparse with `header: false`, positional access + * E big-batch — BATCH_SIZE 5000 (vs 1000) + * F prep-once — prepare single-row INSERT once per table, reuse bind/step/reset + * G combined — B+C+D+E applied together + * H combined+prep — G + F + */ + +import { readFileSync } from 'node:fs'; +import { performance } from 'node:perf_hooks'; +import initSqlJs, { type Database } from 'sql.js'; +import Papa from 'papaparse'; +import { getAllCreateTableStatements, getAllCreateIndexStatements } from '../src/schema/schema.ts'; +import { loadGTFSZip, type GTFSFiles } from '../src/loaders/zip-loader.ts'; +import { loadGTFSData } from '../src/loaders/data-loader.ts'; +import { GTFS_SCHEMA, type TableSchema } from '../src/schema/schema.ts'; +import { createRealtimeTables } from '../src/schema/gtfs-rt-schema.ts'; + +const ZIP_PATH = process.argv[2] ?? '/tmp/gtfs-bench/astuce.zip'; +const RUNS_PER_VARIANT = 5; + +type Variant = 'A' | 'B' | 'C' | 'D' | 'E' | 'F' | 'G' | 'H'; + +const VARIANT_LABELS: Record = { + A: 'A baseline', + B: 'B no-preflight', + C: 'C no-trim', + D: 'D arrays', + E: 'E big-batch(5k)', + F: 'F prep-once', + G: 'G combined (B+C+D+E)', + H: 'H G + prep-once', +}; + +// ------------------------- variant implementations ------------------------- + +// Variant A: call upstream code unchanged. +async function loadVariantA(db: Database, files: GTFSFiles) { + await loadGTFSData(db, files); +} + +// Build schema map once (shared by B–H). +const SCHEMA_BY_FILE = new Map(); +for (const s of GTFS_SCHEMA) SCHEMA_BY_FILE.set(`${s.name}.txt`, s); + +function orderedFiles(files: GTFSFiles): [string, string][] { + const priority = [ + 'agency.txt', 'feed_info.txt', 'attributions.txt', 'levels.txt', 'routes.txt', + 'calendar.txt', 'calendar_dates.txt', 'fare_attributes.txt', 'fare_rules.txt', + 'stops.txt', 'pathways.txt', 'transfers.txt', 'trips.txt', 'frequencies.txt', + 'shapes.txt', 'stop_times.txt', + ]; + const out: [string, string][] = []; + const seen = new Set(); + for (const p of priority) if (files[p]) { out.push([p, files[p]]); seen.add(p); } + for (const [k, v] of Object.entries(files)) if (!seen.has(k)) out.push([k, v]); + return out; +} + +// Variant B: single parse per file, papaparse options match baseline. +async function loadVariantB(db: Database, files: GTFSFiles, opts: { + trim?: boolean; arrays?: boolean; batchSize?: number; prepOnce?: boolean; +} = {}) { + const trim = opts.trim ?? true; + const arrays = opts.arrays ?? false; + const batchSize = opts.batchSize ?? 1000; + const prepOnce = opts.prepOnce ?? false; + + for (const [fileName, content] of orderedFiles(files)) { + const schema = SCHEMA_BY_FILE.get(fileName); + if (!schema) continue; + + let headers: string[]; + let rows: unknown[]; + if (arrays) { + const r = Papa.parse(content, { + header: false, + skipEmptyLines: true, + }); + const data = r.data; + headers = (data[0] || []).map((h) => h.trim()); + rows = data.slice(1); + } else { + const r = Papa.parse>(content, { + header: true, + skipEmptyLines: true, + transformHeader: (h) => h.trim(), + ...(trim ? { transform: (v: string) => v.trim() } : {}), + }); + headers = r.meta.fields || []; + rows = r.data; + } + if (rows.length === 0) continue; + + // Columns present in both the CSV and the schema, in CSV order. + const colIndexes: number[] = []; + const columns: string[] = []; + for (let i = 0; i < headers.length; i++) { + if (schema.columns.some((c) => c.name === headers[i])) { + colIndexes.push(i); + columns.push(headers[i]); + } + } + if (columns.length === 0) continue; + + db.run('BEGIN'); + try { + if (prepOnce) { + const insertSQL = `INSERT INTO ${schema.name} (${columns.join(',')}) VALUES (${columns.map(() => '?').join(',')})`; + const stmt = db.prepare(insertSQL); + try { + const rowVals: (string | number | null)[] = new Array(columns.length); + for (let r = 0; r < rows.length; r++) { + const row = rows[r]; + if (arrays) { + const arr = row as string[]; + for (let j = 0; j < colIndexes.length; j++) { + const v = arr[colIndexes[j]]; + rowVals[j] = v == null || v === '' ? null : (trim ? v.trim() : v); + } + } else { + const obj = row as Record; + for (let j = 0; j < columns.length; j++) { + const v = obj[columns[j]]; + rowVals[j] = v == null || v === '' ? null : v; + } + } + stmt.run(rowVals); + } + } finally { + stmt.free(); + } + } else { + // Multi-row INSERT batches; respect SQLITE_MAX_VARIABLE_NUMBER (32766). + const safeBatch = Math.max(1, Math.min(batchSize, Math.floor(30000 / columns.length))); + for (let i = 0; i < rows.length; i += safeBatch) { + const end = Math.min(i + safeBatch, rows.length); + const n = end - i; + const placeholders = new Array(n).fill(`(${columns.map(() => '?').join(',')})`).join(','); + const insertSQL = `INSERT INTO ${schema.name} (${columns.join(',')}) VALUES ${placeholders}`; + const all: (string | number | null)[] = []; + if (arrays) { + for (let r = i; r < end; r++) { + const arr = rows[r] as string[]; + for (let j = 0; j < colIndexes.length; j++) { + const v = arr[colIndexes[j]]; + all.push(v == null || v === '' ? null : (trim ? v.trim() : v)); + } + } + } else { + for (let r = i; r < end; r++) { + const obj = rows[r] as Record; + for (const col of columns) { + const v = obj[col]; + all.push(v == null || v === '' ? null : v); + } + } + } + const stmt = db.prepare(insertSQL); + try { stmt.run(all); } finally { stmt.free(); } + } + } + db.run('COMMIT'); + } catch (e) { + db.run('ROLLBACK'); + throw e; + } + } +} + +// --------------------------------- runner --------------------------------- + +async function ingest(SQL: initSqlJs.SqlJsStatic, zip: Uint8Array, variant: Variant) { + const phase: Record = {}; + const tick = (k: string, t0: number) => { phase[k] = (phase[k] || 0) + performance.now() - t0; }; + + let t = performance.now(); + const db = new SQL.Database(); + for (const s of getAllCreateTableStatements()) db.run(s); + createRealtimeTables(db); + tick('schema', t); + + t = performance.now(); + const files = await loadGTFSZip(zip); + tick('unzip', t); + + t = performance.now(); + switch (variant) { + case 'A': await loadVariantA(db, files); break; + case 'B': await loadVariantB(db, files); break; + case 'C': await loadVariantB(db, files, { trim: false }); break; + case 'D': await loadVariantB(db, files, { arrays: true, trim: false }); break; + case 'E': await loadVariantB(db, files, { batchSize: 5000 }); break; + case 'F': await loadVariantB(db, files, { prepOnce: true }); break; + case 'G': await loadVariantB(db, files, { trim: false, arrays: true, batchSize: 5000 }); break; + case 'H': await loadVariantB(db, files, { trim: false, arrays: true, batchSize: 5000, prepOnce: true }); break; + } + tick('load', t); + + t = performance.now(); + for (const s of getAllCreateIndexStatements()) db.run(s); + db.run('ANALYZE'); + tick('indexes', t); + + const total = phase.schema + phase.unzip + phase.load + phase.indexes; + db.close(); + return { total, phase }; +} + +function stats(samples: number[]) { + const sorted = [...samples].sort((a, b) => a - b); + const mean = samples.reduce((s, x) => s + x, 0) / samples.length; + const median = sorted.length % 2 + ? sorted[(sorted.length - 1) / 2] + : (sorted[sorted.length / 2 - 1] + sorted[sorted.length / 2]) / 2; + return { mean, median, min: sorted[0], max: sorted[sorted.length - 1] }; +} + +async function main() { + const SQL = await initSqlJs(); + const zip = new Uint8Array(readFileSync(ZIP_PATH)); + + process.stderr.write('warmup…\n'); + await ingest(SQL, zip, 'A'); + await ingest(SQL, zip, 'H'); + + const variants: Variant[] = ['A', 'B', 'C', 'D', 'E', 'F', 'G', 'H']; + const results: Record }> = {} as any; + + for (const v of variants) { + results[v] = { totals: [], phases: { schema: [], unzip: [], load: [], indexes: [] } }; + for (let i = 0; i < RUNS_PER_VARIANT; i++) { + const { total, phase } = await ingest(SQL, zip, v); + results[v].totals.push(total); + for (const k of Object.keys(phase)) results[v].phases[k].push(phase[k]); + process.stderr.write(` ${VARIANT_LABELS[v].padEnd(22)} run ${i + 1}: ${total.toFixed(1)} ms (load ${phase.load.toFixed(1)})\n`); + } + } + + const baseline = stats(results.A.totals).median; + + process.stdout.write('\nTotal ingestion time (median of ' + RUNS_PER_VARIANT + ' runs):\n'); + process.stdout.write('| Variant | Median | Mean | Min | Max | Δ vs A |\n'); + process.stdout.write('|------------------------|---------|---------|---------|---------|----------------|\n'); + for (const v of variants) { + const s = stats(results[v].totals); + const dMs = s.median - baseline; + const dPct = (dMs / baseline) * 100; + const delta = v === 'A' ? '—' : `${dMs >= 0 ? '+' : ''}${dMs.toFixed(0)} ms (${dPct >= 0 ? '+' : ''}${dPct.toFixed(1)}%)`; + process.stdout.write(`| ${VARIANT_LABELS[v].padEnd(22)} | ${s.median.toFixed(0).padStart(7)} | ${s.mean.toFixed(0).padStart(7)} | ${s.min.toFixed(0).padStart(7)} | ${s.max.toFixed(0).padStart(7)} | ${delta.padStart(14)} |\n`); + } + + process.stdout.write('\nPhase breakdown (median ms):\n'); + process.stdout.write('| Variant | schema | unzip | load | indexes |\n'); + process.stdout.write('|------------------------|--------|--------|--------|---------|\n'); + for (const v of variants) { + const row: Record = {}; + for (const k of Object.keys(results[v].phases)) row[k] = stats(results[v].phases[k]).median; + process.stdout.write(`| ${VARIANT_LABELS[v].padEnd(22)} | ${row.schema.toFixed(0).padStart(6)} | ${row.unzip.toFixed(0).padStart(6)} | ${row.load.toFixed(0).padStart(6)} | ${row.indexes.toFixed(0).padStart(7)} |\n`); + } +} + +main().catch((e) => { console.error(e); process.exit(1); }); diff --git a/scripts/bench-pragmas.ts b/scripts/bench-pragmas.ts new file mode 100644 index 0000000..91814e9 --- /dev/null +++ b/scripts/bench-pragmas.ts @@ -0,0 +1,123 @@ +/** + * Benchmark the individual performance impact of each bulk-load PRAGMA. + * + * For each configuration we: + * 1. fresh in-memory sql.js DB + * 2. apply a subset of the PRAGMAs used today in src/gtfs-sqljs.ts + * 3. create schema, load all CSVs, create indexes, ANALYZE + * 4. time the whole thing with performance.now() + * + * We run every config N times and report mean + median. + */ + +import { readFileSync } from 'node:fs'; +import { performance } from 'node:perf_hooks'; +import initSqlJs from 'sql.js'; +import { getAllCreateTableStatements, getAllCreateIndexStatements } from '../src/schema/schema.ts'; +import { loadGTFSZip } from '../src/loaders/zip-loader.ts'; +import { loadGTFSData } from '../src/loaders/data-loader.ts'; +import { createRealtimeTables } from '../src/schema/gtfs-rt-schema.ts'; + +const ZIP_PATH = process.argv[2] ?? '/tmp/gtfs-bench/car-jaune.zip'; +const RUNS_PER_CONFIG = 8; + +const ALL_PRAGMAS = [ + 'synchronous', + 'journal_mode', + 'temp_store', + 'cache_size', + 'locking_mode', +] as const; + +type PragmaName = typeof ALL_PRAGMAS[number]; + +const PRAGMA_SQL: Record = { + synchronous: 'PRAGMA synchronous = OFF', + journal_mode: 'PRAGMA journal_mode = MEMORY', + temp_store: 'PRAGMA temp_store = MEMORY', + cache_size: 'PRAGMA cache_size = -64000', + locking_mode: 'PRAGMA locking_mode = EXCLUSIVE', +}; + +interface Config { + label: string; + pragmas: PragmaName[]; +} + +const CONFIGS: Config[] = [ + { label: 'all-on (baseline)', pragmas: [...ALL_PRAGMAS] }, + { label: 'all-off', pragmas: [] }, + ...ALL_PRAGMAS.map((p) => ({ + label: `no-${p}`, + pragmas: ALL_PRAGMAS.filter((x) => x !== p), + })), +]; + +async function runOnce(SQL: initSqlJs.SqlJsStatic, zipBuf: Uint8Array, pragmas: PragmaName[]) { + const t0 = performance.now(); + const db = new SQL.Database(); + for (const p of pragmas) db.run(PRAGMA_SQL[p]); + for (const stmt of getAllCreateTableStatements()) db.run(stmt); + createRealtimeTables(db); + const files = await loadGTFSZip(zipBuf); + await loadGTFSData(db, files); + for (const stmt of getAllCreateIndexStatements()) db.run(stmt); + db.run('ANALYZE'); + const elapsed = performance.now() - t0; + db.close(); + return elapsed; +} + +function stats(samples: number[]) { + const sorted = [...samples].sort((a, b) => a - b); + const mean = samples.reduce((s, x) => s + x, 0) / samples.length; + const median = + sorted.length % 2 + ? sorted[(sorted.length - 1) / 2] + : (sorted[sorted.length / 2 - 1] + sorted[sorted.length / 2]) / 2; + return { mean, median, min: sorted[0], max: sorted[sorted.length - 1] }; +} + +async function main() { + const SQL = await initSqlJs(); + const zipBuf = new Uint8Array(readFileSync(ZIP_PATH)); + // Warmup (JIT + sql.js heap grow) — discarded. + process.stderr.write('warmup…\n'); + await runOnce(SQL, zipBuf, [...ALL_PRAGMAS]); + await runOnce(SQL, zipBuf, []); + + const results: Array<{ label: string; samples: number[] }> = []; + for (const cfg of CONFIGS) { + const samples: number[] = []; + for (let i = 0; i < RUNS_PER_CONFIG; i++) { + const ms = await runOnce(SQL, zipBuf, cfg.pragmas); + samples.push(ms); + process.stderr.write(` ${cfg.label} run ${i + 1}: ${ms.toFixed(1)} ms\n`); + } + results.push({ label: cfg.label, samples }); + } + + const baseline = stats(results[0].samples).median; + process.stdout.write('\n'); + process.stdout.write('| Config | Median (ms) | Mean (ms) | Min | Max | Δ vs baseline |\n'); + process.stdout.write('|-----------------------|-------------|-----------|-------|-------|---------------|\n'); + for (const r of results) { + const s = stats(r.samples); + const deltaMs = s.median - baseline; + const deltaPct = (deltaMs / baseline) * 100; + const deltaStr = + r.label === 'all-on (baseline)' + ? '—' + : `${deltaMs >= 0 ? '+' : ''}${deltaMs.toFixed(1)} ms (${deltaPct >= 0 ? '+' : ''}${deltaPct.toFixed(1)}%)`; + process.stdout.write( + `| ${r.label.padEnd(21)} | ${s.median.toFixed(1).padStart(11)} | ${s.mean + .toFixed(1) + .padStart(9)} | ${s.min.toFixed(1).padStart(5)} | ${s.max.toFixed(1).padStart(5)} | ${deltaStr.padStart(13)} |\n` + ); + } +} + +main().catch((e) => { + console.error(e); + process.exit(1); +}); diff --git a/src/gtfs-sqljs.ts b/src/gtfs-sqljs.ts index c36b66d..0128064 100644 --- a/src/gtfs-sqljs.ts +++ b/src/gtfs-sqljs.ts @@ -53,6 +53,14 @@ export interface ProgressInfo { filesCompleted: number; totalFiles: number; rowsProcessed: number; + /** + * Estimated total row count across all GTFS files being loaded. + * + * During `inserting_data` this value is a fast newline-based estimate + * (typically exact, but may differ by a few rows per file in edge cases + * such as trailing blank lines). For a precise post-ingest row count, + * query the database directly with `COUNT(*)`. + */ totalRows: number; bytesDownloaded?: number; // Bytes downloaded (used during 'downloading' phase) totalBytes?: number; // Total bytes to download (used during 'downloading' phase) @@ -362,13 +370,6 @@ export class GtfsSqlJs { // Create new database this.db = new this.SQL.Database(); - // Apply performance PRAGMAs for bulk loading - this.db.run('PRAGMA synchronous = OFF'); // Skip fsync for performance - this.db.run('PRAGMA journal_mode = MEMORY'); // Keep journal in memory - this.db.run('PRAGMA temp_store = MEMORY'); // Temp tables in memory - this.db.run('PRAGMA cache_size = -64000'); // 64MB cache - this.db.run('PRAGMA locking_mode = EXCLUSIVE'); // No locking overhead - // Create GTFS tables (without indexes) onProgress?.({ phase: 'creating_schema', @@ -461,10 +462,6 @@ export class GtfsSqlJs { this.db.run('ANALYZE'); - // Restore normal SQLite settings - this.db.run('PRAGMA synchronous = FULL'); - this.db.run('PRAGMA locking_mode = NORMAL'); - // Set RT configuration if (options.realtimeFeedUrls) { this.realtimeFeedUrls = options.realtimeFeedUrls; diff --git a/src/loaders/csv-parser.ts b/src/loaders/csv-parser.ts index 3c1bb63..7353a47 100644 --- a/src/loaders/csv-parser.ts +++ b/src/loaders/csv-parser.ts @@ -30,6 +30,25 @@ export function parseCSV(text: string): ParsedCSV { return { headers, rows }; } +/** + * Fast O(bytes) estimate of the number of data rows in a CSV string. + * + * Counts newline characters and subtracts 1 for the header row. Assumes the + * CSV has a header line, no embedded newlines in quoted fields (GTFS spec + * compliant), and may or may not have a trailing newline. The result is + * typically exact but may be ±a few rows per file in edge cases (e.g. trailing + * blank lines). Suitable for driving progress callbacks, not for precise + * bookkeeping. + */ +export function countCsvRows(csv: string): number { + let lines = 0; + for (let i = 0; i < csv.length; i++) { + if (csv.charCodeAt(i) === 10) lines++; + } + const trailing = csv.length > 0 && csv.charCodeAt(csv.length - 1) !== 10 ? 1 : 0; + return Math.max(0, lines - 1 + trailing); +} + /** * Convert parsed row to typed object with proper type conversions */ diff --git a/src/loaders/data-loader.ts b/src/loaders/data-loader.ts index b6093b0..9a8bbb1 100644 --- a/src/loaders/data-loader.ts +++ b/src/loaders/data-loader.ts @@ -3,11 +3,14 @@ */ import type { Database } from 'sql.js'; -import { parseCSV } from './csv-parser'; +import Papa from 'papaparse'; +import { countCsvRows } from './csv-parser'; import { GTFS_SCHEMA, type TableSchema } from '../schema/schema'; import type { GTFSFiles } from './zip-loader'; import type { ProgressCallback } from '../gtfs-sqljs'; +const PROGRESS_BATCH = 1000; + /** * Load GTFS files into SQLite database * @param skipFiles - Optional array of filenames to skip importing (tables will be created but remain empty) @@ -19,17 +22,13 @@ export async function loadGTFSData( skipFiles?: string[], onProgress?: ProgressCallback ): Promise { - // Map of file names to table schemas const fileToSchema: Map = new Map(); for (const schema of GTFS_SCHEMA) { - // Match file names like agency.txt to table name 'agency' fileToSchema.set(`${schema.name}.txt`, schema); } - // Normalize skipFiles to a Set for faster lookup - const skipSet = new Set(skipFiles?.map(f => f.toLowerCase()) || []); + const skipSet = new Set(skipFiles?.map((f) => f.toLowerCase()) || []); - // Define file priority order (small files first, largest last) const filePriority = [ 'agency.txt', 'feed_info.txt', @@ -49,44 +48,38 @@ export async function loadGTFSData( 'stop_times.txt', // Largest file - process last ]; - // Sort files by priority const sortedFiles: [string, string][] = []; for (const priorityFile of filePriority) { if (files[priorityFile]) { sortedFiles.push([priorityFile, files[priorityFile]]); } } - // Add any files not in priority list for (const [fileName, content] of Object.entries(files)) { if (!filePriority.includes(fileName)) { sortedFiles.push([fileName, content]); } } - // Calculate total rows for progress tracking + // Estimate total rows via newline count (O(bytes)) instead of parsing every + // file twice. totalRows may differ by a few rows from the true row count + // for files with trailing blank lines; see countCsvRows for details. let totalRows = 0; const fileRowCounts = new Map(); for (const [fileName, content] of sortedFiles) { - const schema = fileToSchema.get(fileName); - if (schema && !skipSet.has(fileName.toLowerCase())) { - const { rows } = parseCSV(content); - fileRowCounts.set(fileName, rows.length); - totalRows += rows.length; + if (fileToSchema.has(fileName) && !skipSet.has(fileName.toLowerCase())) { + const n = countCsvRows(content); + fileRowCounts.set(fileName, n); + totalRows += n; } } let rowsProcessed = 0; let filesCompleted = 0; - // Process each file for (const [fileName, content] of sortedFiles) { const schema = fileToSchema.get(fileName); - if (!schema) { - // Skip unknown files - continue; - } + if (!schema) continue; - // Skip if in skip list if (skipSet.has(fileName.toLowerCase())) { console.log(`Skipping import of ${fileName} (table ${schema.name} created but empty)`); filesCompleted++; @@ -102,7 +95,7 @@ export async function loadGTFSData( totalFiles: sortedFiles.length, rowsProcessed, totalRows, - percentComplete: 40 + Math.floor((rowsProcessed / totalRows) * 35), + percentComplete: computePercent(rowsProcessed, totalRows), message: `Loading ${fileName} (${fileRows.toLocaleString()} rows)`, }); @@ -115,7 +108,7 @@ export async function loadGTFSData( totalFiles: sortedFiles.length, rowsProcessed: currentProgress, totalRows, - percentComplete: 40 + Math.floor((currentProgress / totalRows) * 35), + percentComplete: computePercent(currentProgress, totalRows), message: `Loading ${fileName} (${processedInFile.toLocaleString()}/${fileRows.toLocaleString()} rows)`, }); }); @@ -130,14 +123,25 @@ export async function loadGTFSData( totalFiles: sortedFiles.length, rowsProcessed, totalRows, - percentComplete: 40 + Math.floor((rowsProcessed / totalRows) * 35), + percentComplete: computePercent(rowsProcessed, totalRows), message: `Completed ${fileName}`, }); } } +// Map [0, totalRows] → [40, 75] to match the legacy progress ranges reported +// by GtfsSqlJs.loadFromZipData. Clamped to 74 to avoid jumping past the +// following "creating_indexes" phase when the row-count estimate undershoots. +function computePercent(rowsProcessed: number, totalRows: number): number { + if (totalRows <= 0) return 40; + const pct = 40 + Math.floor((rowsProcessed / totalRows) * 35); + return Math.min(74, pct); +} + /** - * Load data for a single table with batch inserts and transaction + * Load data for a single table using a single prepared INSERT reused per row. + * Parses the CSV exactly once as positional arrays (no per-row object + * allocation) and binds column values by pre-computed index. */ async function loadTableData( db: Database, @@ -145,63 +149,58 @@ async function loadTableData( csvContent: string, onProgress?: (rowsProcessed: number) => void ): Promise { - const { headers, rows } = parseCSV(csvContent); - - if (rows.length === 0) { - return; - } - - // Prepare INSERT statement - const columns = headers.filter((h) => schema.columns.some((c) => c.name === h)); - if (columns.length === 0) { - return; + const parsed = Papa.parse(csvContent, { + header: false, + skipEmptyLines: true, + }); + const data = parsed.data; + if (data.length < 2) return; + + const rawHeaders = (data[0] || []).map((h) => h.trim()); + const dataRows = data.slice(1); + if (dataRows.length === 0) return; + + const colIndexes: number[] = []; + const columns: string[] = []; + for (let i = 0; i < rawHeaders.length; i++) { + if (schema.columns.some((c) => c.name === rawHeaders[i])) { + colIndexes.push(i); + columns.push(rawHeaders[i]); + } } + if (columns.length === 0) return; - const BATCH_SIZE = 1000; - let rowsProcessed = 0; + const insertSQL = `INSERT INTO ${schema.name} (${columns.join(', ')}) VALUES (${columns + .map(() => '?') + .join(', ')})`; - // Start transaction for better performance db.run('BEGIN TRANSACTION'); - try { - // Process in batches - for (let i = 0; i < rows.length; i += BATCH_SIZE) { - const batchRows = rows.slice(i, Math.min(i + BATCH_SIZE, rows.length)); - - // Build multi-row INSERT statement - const placeholders = batchRows.map(() => `(${columns.map(() => '?').join(', ')})`).join(', '); - const insertSQL = `INSERT INTO ${schema.name} (${columns.join(', ')}) VALUES ${placeholders}`; - - // Collect all values for the batch - const allValues: (string | number | null)[] = []; - for (const row of batchRows) { - for (const col of columns) { - const value = row[col]; - // Let SQLite handle type conversion - just pass empty strings as NULL - allValues.push(value === null || value === undefined || value === '' ? null : value); + const stmt = db.prepare(insertSQL); + try { + const rowVals: (string | number | null)[] = new Array(columns.length); + for (let r = 0; r < dataRows.length; r++) { + const row = dataRows[r]; + for (let j = 0; j < colIndexes.length; j++) { + const v = row[colIndexes[j]]; + if (v == null) { + rowVals[j] = null; + } else { + const trimmed = typeof v === 'string' ? v.trim() : v; + rowVals[j] = trimmed === '' ? null : trimmed; + } } - } + stmt.run(rowVals); - // Execute batch insert - const stmt = db.prepare(insertSQL); - try { - stmt.run(allValues); - } catch (error) { - console.error(`Error inserting batch into ${schema.name}:`, error); - console.error('Batch size:', batchRows.length); - throw error; - } finally { - stmt.free(); + const done = r + 1; + if (done % PROGRESS_BATCH === 0) onProgress?.(done); } - - rowsProcessed += batchRows.length; - onProgress?.(rowsProcessed); + if (dataRows.length % PROGRESS_BATCH !== 0) onProgress?.(dataRows.length); + } finally { + stmt.free(); } - - // Commit transaction db.run('COMMIT'); } catch (error) { - // Rollback on error try { db.run('ROLLBACK'); } catch (rollbackError) { diff --git a/tests/progress-callback.test.ts b/tests/progress-callback.test.ts new file mode 100644 index 0000000..98bb5b9 --- /dev/null +++ b/tests/progress-callback.test.ts @@ -0,0 +1,104 @@ +/** + * Tests for the progress callback emitted during GTFS ingestion. + * + * Guards against regressions in: + * - totalRows estimate accuracy (within a small tolerance of the real + * COUNT(*) sum after ingestion completes) + * - percentComplete bounds (stays in [0, 100]) + * - the terminal 'complete' phase being emitted exactly once + */ + +import { describe, it, expect, beforeAll, afterAll } from 'vitest'; +import { GtfsSqlJs, type ProgressInfo } from '../src/gtfs-sqljs'; +import path from 'path'; +import fs from 'fs/promises'; + +describe('Progress callback', () => { + let events: ProgressInfo[]; + let gtfs: GtfsSqlJs; + let actualRowCount: number; + + beforeAll(async () => { + const feedPath = path.join(__dirname, 'fixtures', 'sample-feed.zip'); + const zipData = await fs.readFile(feedPath); + + events = []; + gtfs = await GtfsSqlJs.fromZipData(zipData, { + onProgress: (info) => events.push({ ...info }), + }); + + const db = gtfs.getDatabase(); + const tables = [ + 'agency', + 'stops', + 'routes', + 'trips', + 'stop_times', + 'calendar', + 'calendar_dates', + 'fare_attributes', + 'fare_rules', + 'shapes', + 'frequencies', + 'transfers', + 'pathways', + 'levels', + 'feed_info', + 'attributions', + ]; + let total = 0; + for (const t of tables) { + const stmt = db.prepare(`SELECT COUNT(*) AS n FROM ${t}`); + stmt.step(); + total += (stmt.getAsObject().n as number) || 0; + stmt.free(); + } + actualRowCount = total; + }); + + afterAll(() => { + gtfs?.close(); + }); + + it('emits at least one progress event', () => { + expect(events.length).toBeGreaterThan(0); + }); + + it('reports a totalRows estimate close to the real COUNT(*) sum', () => { + // GtfsSqlJs emits one synthetic inserting_data event with totalRows=0 + // before loadGTFSData starts; ignore it and look at the loader's events. + const loaderEvents = events.filter( + (e) => e.phase === 'inserting_data' && e.totalRows > 0 + ); + expect(loaderEvents.length).toBeGreaterThan(0); + + const estimates = new Set(loaderEvents.map((e) => e.totalRows)); + expect(estimates.size).toBe(1); + + const estimate = loaderEvents[0].totalRows; + // Tolerance: up to one row of drift per GTFS file to absorb trailing + // blank lines or missing trailing newlines. The fixture has <=20 files. + expect(Math.abs(estimate - actualRowCount)).toBeLessThanOrEqual(20); + }); + + it('keeps percentComplete within [0, 100] for every event', () => { + for (const e of events) { + expect(e.percentComplete).toBeGreaterThanOrEqual(0); + expect(e.percentComplete).toBeLessThanOrEqual(100); + } + }); + + it('ends with exactly one complete event at 100%', () => { + const completeEvents = events.filter((e) => e.phase === 'complete'); + expect(completeEvents.length).toBe(1); + expect(completeEvents[0].percentComplete).toBe(100); + expect(events[events.length - 1].phase).toBe('complete'); + }); + + it('never reports rowsProcessed exceeding the estimate plus the per-file tolerance', () => { + const ingestEvents = events.filter((e) => e.phase === 'inserting_data'); + for (const e of ingestEvents) { + expect(e.rowsProcessed).toBeLessThanOrEqual(e.totalRows + 20); + } + }); +});