# faucet-sink-sqlite
[](https://crates.io/crates/faucet-sink-sqlite)
[](https://docs.rs/faucet-sink-sqlite)
SQLite sink connector for the [faucet-stream](https://github.com/PawanSikawat/faucet-stream) ecosystem.
Writes JSON records to a SQLite table using either JSON column mode (storing each record as a serialized JSON text value) or AutoMap mode (mapping top-level JSON keys directly to table columns). Uses connection pooling via `sqlx`, multi-row `INSERT` statements, and wraps each batch in a transaction (`BEGIN`/`COMMIT`) for maximum write throughput. Column discovery uses `PRAGMA table_info`.
`write_batch` accepts whatever slice the pipeline hands it. When `batch_size > 0` and the slice is larger than `batch_size`, the sink re-chunks internally and issues one multi-row INSERT (in its own transaction) per chunk; when `batch_size = 0`, the entire slice is written in a single transaction — see [Streaming and batching](#streaming-and-batching) for the tradeoffs.
## Installation
```toml
[dependencies]
faucet-sink-sqlite = "1.0"
tokio = { version = "1", features = ["full"] }
```
Or via the umbrella crate:
```toml
faucet-stream = { version = "1.0", features = ["sink-sqlite"] }
```
## Quick Start
```rust
use faucet_sink_sqlite::{SqliteSink, SqliteSinkConfig};
use faucet_core::Sink;
use serde_json::json;
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
let config = SqliteSinkConfig::new("sqlite:///tmp/app.db", "events");
let sink = SqliteSink::new(config).await?;
let records = vec![
json!({"user_id": "u123", "event": "signup"}),
json!({"user_id": "u456", "event": "login"}),
];
let rows_written = sink.write_batch(&records).await?;
println!("Wrote {rows_written} rows");
Ok(())
}
```
## Configuration
| Field | Type | Default | Description |
|-------|------|---------|-------------|
| `database_url` | `String` | *(required)* | SQLite database URL. Can be a file path (e.g. `/tmp/app.db`), a `sqlite:` URL, or `sqlite::memory:` for an in-memory database. |
| `table_name` | `String` | *(required)* | Target table name |
| `column_mapping` | `SqliteColumnMapping` | `Json { column: "data" }` | How to map JSON records to table columns (see below) |
| `batch_size` | `usize` | `1000` | Maximum number of rows per multi-row INSERT. See [Streaming and batching](#streaming-and-batching) below |
| `max_connections` | `u32` | `1` | Maximum number of connections in the connection pool. SQLite serializes writers at the file level, so one writer is the safe default — a multi-connection pool against a single file races for the write lock and risks `SQLITE_BUSY`. The pool opens connections in WAL mode with a 5s `busy_timeout`, so raising this lets extra connections read concurrently with the single writer. |
| `write_mode` | `"append"` \| `"upsert"` \| `"delete"` | `"append"` | Write semantics. Upsert/delete require `column_mapping: auto_map` and a UNIQUE/PRIMARY KEY on `key`. |
| `key` | `[String]` | `[]` | Key column(s) for upsert/delete. Must be non-empty when `write_mode` is not `append`. |
| `delete_marker` | object | absent | `upsert` only: identifies delete-flagged rows by field name + value list. |
### Streaming and batching
The SQLite sink re-chunks each incoming `StreamPage` to keep individual
multi-row INSERT statements within SQLite's per-statement parameter limits
and to amortise per-transaction overhead.
- **`batch_size > 0`** (default `1000`) — the sink slices the incoming
slice into `batch_size`-row chunks and issues one multi-row INSERT per
chunk, each wrapped in its own `BEGIN`/`COMMIT` transaction.
**Recommended value is `1000`**: large enough to amortise transaction
overhead, small enough to stay well under SQLite's default
`SQLITE_MAX_VARIABLE_NUMBER` (32766 since 3.32.0). In AutoMap mode the
sink also splits each chunk further so `rows × columns` never exceeds that
limit, so a wide table no longer fails with "too many SQL variables"
regardless of `batch_size`.
- **`batch_size = 0`** — the "no batching" sentinel. The entire upstream
`StreamPage` is written in a single multi-row INSERT inside one
transaction. Use this when the source already emits page sizes tuned for
SQLite — for example a Postgres source with `batch_size: 1000`. Pages
large enough to push the parameter count past SQLite's per-statement
limit will fail at the prepare step.
`batch_size` is purely a chunk-size knob — transaction wrapping (one
`BEGIN`/`COMMIT` per chunk), identifier quoting, and per-record error
reporting are unchanged.
### Column Mapping (`SqliteColumnMapping`)
| Variant | Description |
|---------|-------------|
| `Json { column }` | Insert each record as a serialized JSON text string in a single column. The column name defaults to `"data"` but can be overridden. Uses a multi-row INSERT wrapped in a transaction. |
| `AutoMap` | Map top-level JSON keys directly to table columns. Column names are discovered using `PRAGMA table_info(table_name)`. Only keys that match existing columns are inserted; extra keys are silently ignored. Records with no matching keys are skipped with a warning. |
### Builder Methods
```rust
use faucet_sink_sqlite::{SqliteSinkConfig, SqliteColumnMapping};
// JSON mode with custom column
let config = SqliteSinkConfig::new("sqlite:///data/app.db", "events")
.column_mapping(SqliteColumnMapping::Json { column: "payload".into() })
.with_batch_size(1000)
.max_connections(3);
// AutoMap mode
let config = SqliteSinkConfig::new("sqlite:///data/app.db", "events")
.column_mapping(SqliteColumnMapping::AutoMap)
.with_batch_size(250);
```
## Config Loading
```rust
use faucet_core::config::{load_json, load_env_file};
use faucet_sink_sqlite::SqliteSinkConfig;
// From a JSON file
let config: SqliteSinkConfig = load_json("config.json")?;
// From an .env file with a prefix
let config: SqliteSinkConfig = load_env_file(".env", "SQLITE_SINK")?;
```
### Example JSON config (JSON mode)
```json
{
"database_url": "/data/analytics.db",
"table_name": "raw_events",
"column_mapping": {
"json": {
"column": "data"
}
},
"batch_size": 1000,
"max_connections": 5
}
```
### Example JSON config (AutoMap mode)
```json
{
"database_url": "/data/analytics.db",
"table_name": "events",
"column_mapping": "auto_map",
"batch_size": 1000,
"max_connections": 3
}
```
### Example .env file
```env
SQLITE_SINK_DATABASE_URL=/data/analytics.db
SQLITE_SINK_TABLE_NAME=raw_events
SQLITE_SINK_COLUMN_MAPPING='{"json":{"column":"data"}}'
SQLITE_SINK_BATCH_SIZE=1000
SQLITE_SINK_MAX_CONNECTIONS=5
```
## Config Schema Introspection
```rust
use faucet_core::Sink;
let sink = SqliteSink::new(config).await?;
let schema = sink.config_schema();
println!("{}", serde_json::to_string_pretty(&schema)?);
```
## Pipeline Usage
```rust
use faucet_core::Pipeline;
use faucet_source_rest::{RestStream, RestStreamConfig};
use faucet_sink_sqlite::{SqliteSink, SqliteSinkConfig, SqliteColumnMapping};
let source = RestStream::new(
RestStreamConfig::new("https://api.example.com", "/v1/events")
);
let sink_config = SqliteSinkConfig::new("/data/events.db", "events")
.column_mapping(SqliteColumnMapping::AutoMap);
let sink = SqliteSink::new(sink_config).await?;
let result = Pipeline::new(source, sink).run().await?;
println!("Transferred {} records", result.records_written);
```
## Examples
### JSON mode -- store records as serialized JSON text
```sql
-- Table schema
CREATE TABLE raw_events (
id INTEGER PRIMARY KEY AUTOINCREMENT,
data TEXT NOT NULL,
created_at TEXT DEFAULT (datetime('now'))
);
```
```rust
let config = SqliteSinkConfig::new("/tmp/app.db", "raw_events")
.column_mapping(SqliteColumnMapping::Json { column: "data".into() })
.with_batch_size(1000);
let sink = SqliteSink::new(config).await?;
sink.write_batch(&records).await?;
```
### AutoMap mode -- map JSON keys to table columns
```sql
-- Table schema
CREATE TABLE events (
user_id TEXT,
event TEXT,
amount REAL,
created_at TEXT DEFAULT (datetime('now'))
);
```
```rust
let config = SqliteSinkConfig::new("/tmp/app.db", "events")
.column_mapping(SqliteColumnMapping::AutoMap)
.with_batch_size(1000);
let sink = SqliteSink::new(config).await?;
let records = vec![
json!({"user_id": "u1", "event": "purchase", "amount": 29.99}),
json!({"user_id": "u2", "event": "signup"}), // missing "amount" will be null
];
sink.write_batch(&records).await?;
```
### In-memory database for testing
```rust
let config = SqliteSinkConfig::new("sqlite::memory:", "test_table")
.column_mapping(SqliteColumnMapping::AutoMap);
let sink = SqliteSink::new(config).await?;
```
## How It Works
- A connection pool is created in `SqliteSink::new()` using `sqlx::SqlitePool` with the configured `max_connections` (default `1`). Each connection is opened in WAL journal mode (`journal_mode = WAL`) with a 5-second `busy_timeout` and `create_if_missing`, so a writer and readers can proceed concurrently and lock contention waits-and-retries instead of failing immediately with `SQLITE_BUSY`. WAL on a `sqlite::memory:` database is a harmless no-op.
- `write_batch()` slices the input into `batch_size`-row chunks (or forwards the whole slice when `batch_size = 0`). Each chunk is inserted using a single multi-row INSERT statement wrapped in a `BEGIN`/`COMMIT` transaction for write performance.
- In JSON mode, each record is serialized to a JSON string and inserted as `INSERT INTO t (col) VALUES (?), (?), ...`.
- In AutoMap mode, column names are discovered using `PRAGMA table_info(table_name)`. A multi-row INSERT is built dynamically. Column values are bound as **native SQLite types** — strings as `TEXT`, JSON numbers as `INTEGER`/`REAL`, booleans as `INTEGER` 0/1 — so column affinity and typed reads round-trip correctly. Arrays and objects (which have no scalar SQL representation) are bound as their JSON text. The INSERT column set is the **union** of record keys across the batch (in table order), so a field present only in a later record is still written; a row missing a column binds SQL `NULL`.
- All identifiers (table names, column names) are quoted using `quote_ident()` (double-quote escaping) to prevent SQL injection.
- Transaction wrapping ensures that either all rows in a batch are committed or none are, providing atomicity per batch.
## Write modes (upsert / delete)
By default the sink uses `write_mode: append` — every record is inserted as a new row. Two additional modes are available when `column_mapping: auto_map` is set and the target table has a `UNIQUE` or `PRIMARY KEY` constraint on the key column(s):
| Mode | Behaviour |
|------|-----------|
| `append` (default) | Insert every record unconditionally. |
| `upsert` | Insert-or-update by `key` (last-write-wins via `ON CONFLICT … DO UPDATE`). Optionally route delete-marked rows to `DELETE` via `delete_marker`. |
| `delete` | Delete every record by `key`. |
### Required fields
| Field | Type | Default | Description |
|-------|------|---------|-------------|
| `write_mode` | `"append"` \| `"upsert"` \| `"delete"` | `"append"` | Write semantics. |
| `key` | `[String]` | `[]` | Key column(s). Required and non-empty for `upsert` / `delete`. |
| `delete_marker` | `{ field: String, values: [String] }` | absent | `upsert` only: rows whose `field` matches one of `values` are routed to `DELETE`; all others are upserted. The marker field is stripped from upsert rows before writing. |
**Requirements:**
- `column_mapping` must be `auto_map` — key columns must be real table columns, not embedded inside a JSON blob.
- The table must have a `UNIQUE` or `PRIMARY KEY` constraint on the `key` column(s) so SQLite's `ON CONFLICT` clause can enforce uniqueness.
- Upsert/delete rows within a single batch are deduped by key (last-write-wins) before writing, so a single batch never conflicts with itself.
- A row missing or null in a key column fails. When a `dlq:` block is configured the good rows are still written and only the missing/null-key rows are routed to the DLQ per-row; without a DLQ the whole batch fails.
### YAML example
```yaml
pipeline:
source:
type: postgres-cdc
config:
connection_url: postgres://user:pass@localhost/db
slot_name: faucet_slot
publication_name: faucet_pub
sink:
type: sqlite
config:
database_url: /data/warehouse.db
table_name: users
column_mapping: auto_map
write_mode: upsert
key: [id]
delete_marker:
field: __op
values: [d, delete]
state:
type: file
config:
path: ./state
```
### Rust example
```rust
use faucet_sink_sqlite::{SqliteColumnMapping, SqliteSink, SqliteSinkConfig};
use faucet_core::{WriteMode, WriteSpec};
let config = SqliteSinkConfig {
database_url: "sqlite:///data/warehouse.db".into(),
table_name: "users".into(),
column_mapping: SqliteColumnMapping::AutoMap,
batch_size: 1000,
max_connections: 1,
write: WriteSpec {
write_mode: WriteMode::Upsert,
key: vec!["id".to_string()],
delete_marker: None,
},
};
let sink = SqliteSink::new(config).await?;
```
## Exactly-once delivery
`SqliteSink` implements `Sink::supports_idempotent_writes` (returns `true`) and the two companion hooks:
- `write_batch_idempotent(records, scope, token)` — writes `records` and UPSERTs the `token` into a `_faucet_commit_token(scope TEXT, token TEXT)` watermark table inside the **same `BEGIN`/`COMMIT` transaction**, so both either commit together or neither does.
- `last_committed_token(scope)` — reads the current watermark to let the pipeline skip already-committed pages on resume.
To use exactly-once delivery, set `delivery: exactly_once` in your pipeline config and pair this sink with one of the CDC sources (`postgres-cdc`, `mysql-cdc`, `mongodb-cdc`) plus a `state:` block. A DLQ is not permitted in exactly-once mode. All four requirements are validated at config-load time (`faucet validate`) before any run starts.
```yaml
pipeline:
source:
type: postgres-cdc
config:
connection_url: postgres://faucet:faucet@localhost:5432/appdb
slot_name: faucet_slot
publication_name: faucet_pub
sink:
type: sqlite
config:
database_url: /data/warehouse.db
table_name: change_events
column_mapping: auto_map
state:
type: file
config:
path: ./state
delivery: exactly_once
```
See the [Exactly-once delivery cookbook](https://pawansikawat.github.io/faucet-stream/cookbook/state.html#exactly-once-delivery) for full rationale and the supported source/sink set.
## Lineage dataset URI
`sqlite://<path>?table=<table>` — e.g. `sqlite:///tmp/test.db?table=events`.
## License
Licensed under MIT or Apache-2.0.