faucet-source-mysql

A MySQL query source that runs a SQL query and streams the rows out as serde_json::Value records.

Part of the faucet-stream ecosystem.

Built on sqlx with a pooled, async connection and a true row cursor (Query::fetch) — rows are decoded and handed to the sink as they arrive off the wire, so client-side memory stays bounded by batch_size no matter how large the result set is. Reach for it to pull tables, filtered extracts, or analytics queries out of MySQL / MariaDB into any faucet-stream sink, or to fan a query out per parent record in a matrix pipeline.

Feature highlights

• Streaming row cursor — Source::stream_pages drives a sqlx cursor and yields one StreamPage per batch_size rows; the sink starts writing before the query finishes draining.
• Connection pooling — a single MySqlPool is built once in new() and reused for every fetch; size it with max_connections (default 10).
• Rich type decoding — JSON, integers, floats, booleans, DATETIME/TIMESTAMP/DATE/TIME, DECIMAL (exact precision), and BLOB/BINARY (base64) all map to sensible JSON.
• Parameterised per-record queries — in a parent/child matrix run, ${parent.field} tokens in the query are substituted as safe bind parameters (? placeholders), never string-interpolated.
• TLS by default — built with tls-rustls; encrypted connections need no extra dependency.
• Credential-safe — the connection URL is masked in Debug output and stripped from the lineage dataset URI.
• batch_size: 0 sentinel — drain the whole result set into a single page for small lookup tables or load-job-style sinks.

Installation

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

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

Or via the umbrella crate:

cargo add faucet-stream --features source-mysql

The MySQL source is not in the CLI/umbrella default build — enable the source-mysql feature explicitly.

Quick start

# pipeline.yaml — faucet run pipeline.yaml
version: 1
pipeline:
  source:
    type: mysql
    config:
      connection_url: mysql://user:pass@localhost:3306/mydb
      query: SELECT id, name, email FROM users WHERE active = 1 ORDER BY id
  sink:
    type: jsonl
    config:
      path: ./users.jsonl

faucet run pipeline.yaml

Configuration reference

There is no separate auth block — credentials live in connection_url (and can be sourced from env or a secrets manager; see Config loading).

Examples

Filtered extract into Postgres

Reuses cli/examples/mysql_to_postgres.yaml:

version: 1
name: mysql_to_postgres
pipeline:
  source:
    type: mysql
    config:
      connection_url: mysql://user:pass@localhost/legacy
      query: SELECT id, name, address, created_at FROM customers ORDER BY id
      max_connections: 16
  sink:
    type: postgres
    config:
      connection_url: postgres://user:pass@localhost/modern
      table_name: customers_imported
      column_mapping:
        type: auto_map
      batch_size: 1000
      max_connections: 10

Rolling time window into stdout

version: 1
pipeline:
  source:
    type: mysql
    config:
      connection_url: ${env:MYSQL_URL}
      query: >
        SELECT id, type, payload, created_at
        FROM events
        WHERE created_at >= CURDATE() - INTERVAL 7 DAY
        ORDER BY created_at
      batch_size: 5000
  sink:
    type: stdout
    config:
      format: jsonl

Small lookup table — one large page

version: 1
pipeline:
  source:
    type: mysql
    config:
      connection_url: ${env:MYSQL_URL}
      query: SELECT code, label FROM country_codes
      batch_size: 0          # drain the whole result set into a single page
  sink:
    type: bigquery
    config:
      project_id: my-project
      dataset_id: ref
      table_id: country_codes

Per-record queries (matrix pipelines)

When a row runs once per record emitted by a parent, ${parent.field} tokens in the query are resolved per record as bind parameters — the value is never spliced into the SQL text:

version: 1
name: orders_per_customer
pipeline:
  sources:
    customers:
      type: mysql
      config:
        connection_url: ${env:MYSQL_URL}
        query: SELECT id FROM customers WHERE region = 'EU'
    orders:
      type: mysql
      config:
        connection_url: ${env:MYSQL_URL}
        query: SELECT id, total FROM orders WHERE customer_id = ${customers.id}
  sink:
    type: jsonl
    config:
      path: ./orders.jsonl
matrix:
  - id: customers
    source: { ref: customers }
  - id: orders
    parent: customers
    source: { ref: orders }

Streaming & batching

MysqlSource::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. This bounds client-side memory at O(batch_size) and lets the sink begin writing as soon as the first batch is parsed off the wire.

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.

The trait-level batch_size argument to stream_pages is ignored in favour of the config field — the config is the authoritative, user-facing knob, so a pipeline-supplied hint can never silently override an explicit config value.

Note — MySQL's wire protocol sends rows from a simple SELECT in a single response (no server-side cursor), so the streaming here bounds memory on the client side rather than asking the server to page. True server-side cursor streaming is tracked separately as a follow-up.

The MySQL query source has no incremental-replication mode, so every emitted page carries bookmark: None (there is no resume or effectively-once support — see Capabilities). For change-data-capture against MySQL binlogs, use faucet-source-mysql-cdc instead.

Column types

Columns are converted to JSON in order of likelihood; an unsupported or NULL column becomes null:

Capabilities

Config loading

Load a config from a JSON file, 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_mysql::MysqlSourceConfig;

# fn example() -> Result<(), Box<dyn std::error::Error>> {
let config: MysqlSourceConfig = load_json("config.json")?;
let config: MysqlSourceConfig = load_env_file(".env", "MYSQL_SOURCE")?;
# Ok(()) }

{
  "connection_url": "mysql://analytics:password@db.example.com:3306/warehouse",
  "query": "SELECT id, name, created_at, status FROM orders WHERE created_at > '2025-01-01' ORDER BY created_at",
  "max_connections": 5,
  "batch_size": 5000
}

MYSQL_SOURCE_CONNECTION_URL=mysql://user:password@localhost:3306/mydb
MYSQL_SOURCE_QUERY=SELECT * FROM users
MYSQL_SOURCE_MAX_CONNECTIONS=10

In CLI configs the connection_url can also be drawn from env, files, or a secrets manager via ${env:VAR} / ${file:PATH} / ${vault:...} directives.

Schema introspection

Print the JSON Schema for this connector's config:

faucet schema source mysql

Or from Rust:

use faucet_core::Source;
use faucet_source_mysql::{MysqlSource, MysqlSourceConfig};

# async fn example() -> Result<(), Box<dyn std::error::Error>> {
let source = MysqlSource::new(MysqlSourceConfig::new("mysql://localhost/db", "SELECT 1")).await?;
let schema = source.config_schema();
println!("{}", serde_json::to_string_pretty(&schema)?);
# Ok(()) }

Library usage

use faucet_source_mysql::{MysqlSource, MysqlSourceConfig};
use faucet_core::Source;

# async fn example() -> Result<(), Box<dyn std::error::Error>> {
let config = MysqlSourceConfig::new(
    "mysql://user:password@localhost:3306/mydb",
    "SELECT id, name, email FROM users WHERE active = 1",
)
.with_max_connections(20)
.with_batch_size(5000);

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

// One-shot collect:
let records = source.fetch_all().await?;
for record in &records {
    println!("{record}");
}
# Ok(()) }

Wire it into a pipeline to stream straight to any sink:

use faucet_source_mysql::{MysqlSource, MysqlSourceConfig};
use faucet_core::Pipeline;

let source = MysqlSource::new(
    MysqlSourceConfig::new("mysql://localhost/production", "SELECT * FROM events"),
).await?;
let pipeline = Pipeline::new(Box::new(source), Box::new(my_sink));
let result = pipeline.run().await?;

How it works

MysqlSource::new validates batch_size, then builds one MySqlPool (MySqlPoolOptions::max_connections) and stores it on the struct — every subsequent fetch reuses the pool rather than reconnecting. stream_pages opens a sqlx cursor with Query::fetch and pulls rows incrementally via TryStreamExt::try_next, flushing a StreamPage each time the buffer reaches batch_size. Each row is turned into a JSON object keyed by column name (row_to_json), and per-column decoding (mysql_value_to_json) probes the likely sqlx types in order, falling back to Null. Connection and query failures surface as FaucetError::Config with the underlying sqlx error attached. Throughput is dominated by row width and network round-trips; benchmark with your own schema and batch_size.

Lineage dataset URI

mysql://<host>:<port>/<db>?query=<sql> (credentials stripped) — e.g. mysql://host:3306/app?query=SELECT id FROM orders.

Feature flags

This crate has no optional Cargo features of its own. Enable it in the CLI / umbrella via the source-mysql feature. TLS (tls-rustls), chrono, bigdecimal, and json sqlx features are always compiled in.

Troubleshooting / FAQ

See also

• Connector reference · Config grammar · CLI reference
• Related crates: faucet-source-mysql-cdc · faucet-sink-mysql · faucet-source-postgres

Sharded execution (cluster Mode B)

Under faucet serve --cluster, a top-level shard: { count: N } block splits this source into contiguous primary-key ranges that different cluster workers process concurrently. Opt in by naming an integer-typed key column:

shard:
  count: 8
pipeline:
  source:
    type: mysql
    config:
      connection_url: ${env:MYSQL_URL}
      query: "SELECT * FROM events"
      shard: { key: id }   # integer column to range-partition on

The coordinator computes MIN(key) / MAX(key) once and splits that range into half-open slices (`key` in the generated predicate, injection-safe). The boundary shards stay open-ended so rows inserted outside the captured range during the run are still read, and exactly one shard additionally matches key IS NULL so nullable keys are never silently dropped. Each shard keeps its own state key ({run}::{shard}), so a reassigned shard resumes where its previous owner left off.

Outside the cluster coordinator the shard config has no effect — a plain faucet run streams the whole query.

License

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