|
| 1 | +# csv2sql v5 — Architecture |
| 2 | + |
| 3 | +Csv2sql bulk-loads CSV files into a database with automatic schema |
| 4 | +inference. This document describes the v5 engine; it is kept current as |
| 5 | +stages of the [rewrite roadmap](#roadmap) land. |
| 6 | + |
| 7 | +## Design principles |
| 8 | + |
| 9 | +1. **No global mutable state.** A run's configuration is one validated, |
| 10 | + immutable `%Csv2sql.Config{}` passed down the pipeline. Nothing reads |
| 11 | + application env at runtime; nothing is globally named. |
| 12 | +2. **Structural completion.** A run is done when its pipeline fold |
| 13 | + returns — there are no completion counters, drain protocols, or |
| 14 | + busy-wait loops (the failure modes that killed the v3 and v4.0 |
| 15 | + architectures). |
| 16 | +3. **One event bus.** Every stage broadcasts typed event structs on a |
| 17 | + per-run `Phoenix.PubSub` topic. The LiveView UI, the CLI reporter, the |
| 18 | + report writer, and the test harness are all plain subscribers — there |
| 19 | + is no other progress/error channel. |
| 20 | +4. **Errors are data.** Cell-level checks are total functions; malformed |
| 21 | + input becomes structured `%Csv2sql.Error{}` values with a class |
| 22 | + (`:preflight | :run | :file | :row | :transient`) that defines the |
| 23 | + blast radius. A bad file fails alone; the run continues. |
| 24 | +5. **Adapters own dialects.** The engine speaks abstract column types |
| 25 | + (`{:varchar, n} | :text | :longtext | :integer | :bigint | :float | |
| 26 | + :boolean | :date | :datetime`); `Csv2sql.Adapter` callbacks (with |
| 27 | + overridable defaults) render SQL. No `ecto_sql`/repos — drivers |
| 28 | + (Postgrex/MyXQL) are used directly; core `ecto` provides embedded |
| 29 | + schemas + changesets for boundary data and config validation only. |
| 30 | + |
| 31 | +## Process architecture |
| 32 | + |
| 33 | +``` |
| 34 | +Csv2sql.Application (one_for_one) |
| 35 | +├── Phoenix.PubSub (Csv2sql.PubSub — the event bus) |
| 36 | +├── Registry (Csv2sql.RunRegistry — per-run process names) |
| 37 | +├── DynamicSupervisor (Csv2sql.RunsSupervisor — one child per run) |
| 38 | +├── Task.Supervisor (Csv2sql.ProbeSupervisor — UI connection probes) |
| 39 | +└── Csv2sqlWeb.Endpoint (GUI mode) |
| 40 | +
|
| 41 | +Per run (temporary, max_restarts: 0 — nothing inside a run restarts): |
| 42 | +Run.Supervisor |
| 43 | +├── Run.Server state + events + cancel; monitors the pipeline |
| 44 | +└── core (Supervisor) |
| 45 | + ├── DB pool Postgrex, owned by the run, dies with it |
| 46 | + ├── Task.Supervisor file workers |
| 47 | + └── Run.Pipeline (Task) the orchestration fold |
| 48 | +``` |
| 49 | + |
| 50 | +- **Isolation:** every process registers in `Csv2sql.RunRegistry` keyed by |
| 51 | + `run_id` (`Csv2sql.Run.server_name/1` etc.), so concurrent runs coexist |
| 52 | + and a crashed run cannot poison the next. |
| 53 | +- **Failure handling:** worker crashes become file-level errors |
| 54 | + (`async_stream_nolink`); a pipeline crash is converted to `RunFailed` by |
| 55 | + `Run.Server` (which monitors it); a server crash tears down the run |
| 56 | + tree (`max_restarts: 0`). |
| 57 | +- **Cancellation:** `Run.Server.cancel/1` stops the core supervisor |
| 58 | + (pipeline + workers die, pool closes), emits `RunCancelled`, and the |
| 59 | + tree is torn down after a grace period. |
| 60 | + |
| 61 | +## The pipeline |
| 62 | + |
| 63 | +``` |
| 64 | +Run.Pipeline.run/2 |
| 65 | + discover files (stable lexicographic order) |
| 66 | + Task.Supervisor.async_stream_nolink(files, &FilePipeline.run/2, |
| 67 | + max_concurrency: file_workers) |
| 68 | + Server.complete(results) ← returning IS completion |
| 69 | +
|
| 70 | +FilePipeline.run/2 (per file) |
| 71 | + pass 1 analyze: stream → chunks → Task.async_stream(infer) → merge |
| 72 | + (associative: AND type flags, max lengths, min/max int bounds — |
| 73 | + constant memory regardless of file size) |
| 74 | + → header repair (trim, blank→column_N, dedupe) → resolve types |
| 75 | + DDL drop (table_mode :drop) + CREATE TABLE IF NOT EXISTS |
| 76 | + pass 2 load: stream → chunks → Task.async_stream(encode + insert, |
| 77 | + max_concurrency: insert_workers) |
| 78 | + Backpressure is inherent: busy insert workers pause reading. |
| 79 | + ledger rows_read == rows_inserted + rows_rejected, per chunk, using |
| 80 | + driver-reported insert counts (never assumed) |
| 81 | +``` |
| 82 | + |
| 83 | +`mode: :analyze` stops after pass 1 — the dry-run behind schema preview. |
| 84 | + |
| 85 | +## Events |
| 86 | + |
| 87 | +All in `Csv2sql.Events`, all structs with `run_id` + `at`: |
| 88 | + |
| 89 | +`RunStarted` · `FileDiscovered` · `AnalyzeStarted` · `AnalyzeCompleted` · |
| 90 | +`SchemaCreated` · `LoadProgress` · `FileCompleted{ledger, timings}` · |
| 91 | +`FileFailed{error}` · terminal: `RunCompleted{summary}` · `RunFailed` · |
| 92 | +`RunCancelled`. |
| 93 | + |
| 94 | +Subscribe with `Events.subscribe(run_id)`; `Csv2sql.start_run(config, |
| 95 | +subscribe: true)` subscribes the caller *before* the run starts so no |
| 96 | +event can be missed. |
| 97 | + |
| 98 | +## Validation ladder |
| 99 | + |
| 100 | +1. **Ledger** (always on, free): in-run double-entry accounting, checked |
| 101 | + per chunk. Implemented. |
| 102 | +2. **Counts** (planned): post-load `COUNT(*)` vs ledger, append-aware. |
| 103 | +3. **Fingerprints** (planned, default-on toggle): per-column O(1) |
| 104 | + aggregates over post-transform values vs one aggregate query per table. |
| 105 | +4. **Deep checksum** (planned, opt-in). |
| 106 | + |
| 107 | +## Testing |
| 108 | + |
| 109 | +- `Csv2sql.EngineCase`: per-test tmp dirs, `csv!/3` fixture builder, |
| 110 | + `unique_name/1` (collision-free tables → DB tests run `async: true`), |
| 111 | + `run_import/2` (subscribed run + collected events + auto table |
| 112 | + cleanup), `assert_event/2`, `db_query!/2`. |
| 113 | +- Doctests document and pin the pure functions (TypeCheck, Inference, |
| 114 | + Resolve, Encoder, Ledger, Db URL parsing). |
| 115 | +- Integration tests drive full runs and verify against |
| 116 | + `information_schema` and real row data. |
| 117 | +- DB-backed tests are tagged (`:postgres`) and excludable: |
| 118 | + `mix test --exclude postgres`. |
| 119 | +- **Local gotcha:** this machine exports `MIX_ENV=dev`, so run tests as |
| 120 | + `MIX_ENV=test mix test`. |
| 121 | + |
| 122 | +## Roadmap |
| 123 | + |
| 124 | +Stages (from the approved v5 plan): **1. walking skeleton ✅** · |
| 125 | +2. adapter behaviour + MySQL + byte/param-aware insert engine · |
| 126 | +3. full type system (dates via TimeParse compiler, booleans, sizing) · |
| 127 | +4. error model (row isolation → errors.csv), encodings, delimiters · |
| 128 | +5. validation ladder + preflight + run_report.json · 6. CLI + config |
| 129 | +sources/persistence · 7. LiveView UI + PWA · 8. Burrito packaging + |
| 130 | +Sentry + docs. |
0 commit comments