faucet-source-sqlite

A SQLite query source for the faucet-stream ecosystem. Executes a SQL statement against a SQLite database (a file or an in-memory database), decodes each row into a typed serde_json::Value, and streams the result set back page-by-page via a sqlx row cursor so memory stays bounded no matter how large the table is.

SQLite is an in-process, file-based engine — there is no server, no network wire, and no auth. That makes this connector the simplest, lowest-overhead way to pull rows out of a local database and land them in any faucet-stream sink (a file, another database, a warehouse, a queue) with one declarative config and no glue code. It's also ideal for CI and locked-down environments where you don't want to stand up a server.

Feature highlights

• Native cursor streaming — overrides Source::stream_pages to drive a sqlx row cursor (Query::fetch) without buffering the whole result. Rows are accumulated into a batch_size buffer and yielded as a StreamPage as soon as it fills, so peak client-side memory is O(batch_size) and the sink can start writing while the rest of the table is still being read off disk.
• batch_size: 0 "no batching" sentinel — drain the entire cursor into one page for small lookup tables or for sinks that prefer one large request to many small ones.
• Dynamic-type aware decoding — SQLite's storage classes (INTEGER, REAL, TEXT, BLOB, NULL) are probed in order of specificity; TEXT that parses as JSON is returned as a native JSON value, BLOBs are base64-encoded so binary survives the round-trip.
• Connection pooling — a sqlx::SqlitePool is built once in new() and reused for every query; pool size is configurable.
• Safe parameter binding — {field} placeholders in the query are bound from the matrix / parent-record context as positional ? parameters, so values are never string-interpolated into SQL (no injection).
• File or in-memory — point at sqlite:data.db, an absolute path, or sqlite::memory:.

Installation

# As a library:
cargo add faucet-source-sqlite
cargo add tokio --features full

# In the CLI (opt-in connector feature):
cargo install faucet-cli --features source-sqlite

Or via the umbrella crate:

cargo add faucet-stream --features source-sqlite

Quick start

# pipeline.yaml — faucet run pipeline.yaml
version: 1
name: sqlite_to_jsonl
pipeline:
  source:
    type: sqlite
    config:
      database_url: sqlite:./app.db
      query: SELECT * FROM events ORDER BY ts
  sink:
    type: jsonl
    config:
      path: events.jsonl

faucet run pipeline.yaml

Configuration reference

All fields live under pipeline.source.config.

There is no auth, TLS, or credential configuration — SQLite is a local in-process engine.

Examples

Read a table into CSV (no network required)

version: 1
name: sqlite_to_csv
pipeline:
  source:
    type: sqlite
    config:
      database_url: sqlite://./data/app.db
      query: SELECT id, email, created_at FROM users ORDER BY id
  sink:
    type: csv
    config:
      path: ./out/users.csv
      delimiter: 44        # ','
      write_headers: true

Project a JSON column out of a row

json_extract is a SQLite built-in; the extracted TEXT is returned as a native JSON value when it parses.

source:
  type: sqlite
  config:
    database_url: sqlite:/var/data/app.db
    query: |
      SELECT id, name, created_at,
             json_extract(metadata, '$.tags') AS tags
      FROM items
      WHERE active = 1
    batch_size: 5000

Whole-table export with no re-chunking

source:
  type: sqlite
  config:
    database_url: sqlite:analytics.db
    query: SELECT * FROM daily_snapshot
    batch_size: 0          # emit the entire result set as one page

Per-parent query in a matrix fan-out

A child matrix row can substitute values from each parent record into the query with {field} placeholders. Each value is bound as a positional parameter, so injection is impossible.

matrix:
  - id: tenants
    source:
      ref: tenant_list
  - id: rows
    parent: tenants
    source:
      type: sqlite
      config:
        database_url: sqlite:tenants.db
        query: SELECT * FROM records WHERE tenant_id = {tenants.id}

Streaming & batching

SqliteSource::stream_pages drives a sqlx row cursor (Query::fetch) without buffering the full result. Rows are accumulated into a batch_size buffer and yielded as a StreamPage once the buffer fills; the trailing partial page (if any) is yielded after the cursor drains. The pipeline writes each page to the sink as it arrives, so memory is bounded at O(batch_size) on both sides.

batch_size = 0 is the "no batching" sentinel — the cursor is drained completely and the entire result set is emitted in a single StreamPage. Use it for small lookup tables, or for downstream sinks (SQL COPY, BigQuery load jobs, Snowflake stage uploads) that prefer one large request to many small ones. Values larger than MAX_BATCH_SIZE (1,000,000) are rejected by faucet_core::validate_batch_size at construction time.

Note — SQLite is an in-process, file-based engine: there is no server-side cursor concept and no network wire to worry about. The streaming implementation bounds client-side memory and lets the sink begin writing as soon as the first batch is parsed off disk, rather than waiting for the whole result set to materialise in a Vec.

The batch_size argument passed to stream_pages by the pipeline is informational — the source always uses its own config.batch_size so a pipeline-supplied hint cannot silently override an explicit config value.

No incremental / resume support

This is a one-shot query source. It does not implement bookmark-based resume, incremental replication, or exactly-once delivery: every emitted page carries bookmark: None, and each run re-executes the full query. For incremental loads, encode the watermark directly in the query (e.g. WHERE updated_at > '2026-01-01') and drive it from a matrix context or a ${now.*} token. If you need change-data-capture, use a CDC source (postgres-cdc / mysql-cdc / mongodb-cdc) instead.

Supported column types

SQLite has dynamic typing — values are stored as INTEGER, REAL, TEXT, BLOB, or NULL. The source probes each column value in order of specificity:

SQLite has no native datetime / UUID / decimal types — those are stored as TEXT/INTEGER/REAL and surface accordingly. A non-finite f64 (NaN / infinity) cannot be represented in JSON and decodes to null.

Config loading

Load from YAML/JSON, environment variables, or a .env file via the helpers in faucet_core::config:

use faucet_core::config::{load_json, load_env_file};
use faucet_source_sqlite::SqliteSourceConfig;

let config: SqliteSourceConfig = load_json("config.json")?;
let config: SqliteSourceConfig = load_env_file(".env", "SQLITE_SOURCE")?;

Example JSON config

{
  "database_url": "sqlite:/var/data/app.db",
  "query": "SELECT id, name, created_at FROM items WHERE active = 1",
  "max_connections": 5,
  "batch_size": 5000
}

Example .env file

SQLITE_SOURCE_DATABASE_URL=sqlite:data.db
SQLITE_SOURCE_QUERY=SELECT * FROM events
SQLITE_SOURCE_MAX_CONNECTIONS=10
SQLITE_SOURCE_BATCH_SIZE=1000

Schema introspection

Inspect the full JSON Schema for this connector's config with:

faucet schema source sqlite

Programmatically, every Source exposes config_schema():

use faucet_core::Source;

let source = SqliteSource::new(config).await?;
let schema = source.config_schema();
println!("{}", serde_json::to_string_pretty(&schema)?);

Library usage

use faucet_core::Source;
use faucet_source_sqlite::{SqliteSource, SqliteSourceConfig};

# async fn run() -> Result<(), Box<dyn std::error::Error>> {
let config = SqliteSourceConfig::new(
    "sqlite:data.db",
    "SELECT id, name, score FROM students ORDER BY score DESC",
)
.with_max_connections(5)
.with_batch_size(2000);

let source = SqliteSource::new(config).await?;

// Convenience: buffer the whole result set.
let records = source.fetch_all().await?;
for record in &records {
    println!("{record}");
}
# Ok(())
# }

To wire it into a pipeline with a sink, build a faucet_core::Pipeline (or call run_stream) over this source and any Sink — the pipeline drives stream_pages and writes each page as it arrives.

How it works

1. new() validates batch_size and builds a sqlx::SqlitePool (SqlitePoolOptions::max_connections) once, reusing it for every query. A connection failure surfaces as FaucetError::Config.
2. stream_pages resolves any {field} context placeholders into positional ? bind markers, binds the values by JSON type (string / integer / float / bool / null — integers up to u64::MAX bind exactly, with no f64 precision loss), and opens a streaming cursor with Query::fetch.
3. Rows are decoded one at a time (row_to_json), buffered to batch_size, and yielded as StreamPages; the final partial buffer is flushed at the end.
4. Each column value is decoded by probing storage classes in specificity order (see Supported column types).

Lineage dataset URI

sqlite://<path>?query=<sql> — e.g. sqlite:///var/db/app.db?query=SELECT id FROM events. The sqlite:// / sqlite: scheme prefix is stripped from database_url before the path is rendered.

Feature flags

This crate has no optional features of its own. Enable it in the CLI / umbrella crate via the source-sqlite feature. It is not in the CLI default build — opt in with cargo install faucet-cli --features source-sqlite.

Troubleshooting / FAQ

See also

• Connector reference & capability matrix
• CLI & config-file grammar
• State & resume cookbook
• faucet-sink-sqlite — the matching SQLite sink
• faucet-source-postgres · faucet-source-mysql — server-based SQL sources

License

Licensed under either of Apache License, Version 2.0 or MIT license at your option.