faucet-source-mssql

Microsoft SQL Server query source for the faucet-stream ecosystem. Runs a parameterized T-SQL statement over a pooled tiberius connection, decodes each row into a typed serde_json::Value, and streams the result set back page-by-page so memory stays bounded regardless of how many rows the query returns.

Reach for it when you want to pull tables, views, or ad-hoc query results out of SQL Server (or Azure SQL) and land them in any faucet-stream sink — a file, another database, a warehouse, a queue — with one declarative config and no glue code. Built-in incremental replication lets repeat runs pick up only the rows that changed since last time.

Feature highlights

• Native row streaming — overrides Source::stream_pages to walk tiberius's QueryStream row-by-row, buffering into pages of batch_size and yielding each page as it fills. Peak memory is O(batch_size), never the full result set.
• Connection pooling — a bb8 + bb8-tiberius pool is built once in new() and reused for every query; size it with max_connections (default 10).
• Incremental / bookmark replication — track a monotonic column (e.g. updated_at) and emit only rows strictly greater than the stored bookmark on each run. The cursor pushes down to the server via the @bookmark token and is enforced client-side as a correctness backstop. If you configure incremental replication but omit the @bookmark token from the query, the run still produces correct rows (client-side filtering), but the server returns the entire table every run — so faucet validate / load warns you to add @bookmark to the WHERE clause for server-side pushdown.
• Positional bind parameters — params are sent as @P1, @P2, … bind values, and ${parent.path} matrix-context placeholders resolve to additional bind markers at runtime — no string interpolation into SQL.
• Type-aware row decoding — integers, floats, BIT, decimals (precision-preserving strings), dates/times, DATETIMEOFFSET, UNIQUEIDENTIFIER, and binary (base64) all map to sensible JSON.
• Four TLS modes — prefer (default), require, trust_server_certificate, disable — with an optional CA cert path, shared with the MSSQL sink via faucet-common-mssql.
• Per-query timeout — statement_timeout_secs bounds each query (default 300s; 0 disables).
• Credentials never logged — the Debug impl masks the connection config, and the lineage URI strips credentials.

Installation

# As a library:
cargo add faucet-source-mssql

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

This is an opt-in connector — it is not in the CLI default build. Enable it with the source-mssql feature (or the source / full aggregates).

Quick start

# pipeline.yaml — faucet run pipeline.yaml
version: 1
pipeline:
  source:
    type: mssql
    config:
      connection_url: "mssql://sa:Str0ng%40Pass@localhost:1433/sales"
      query: "SELECT id, email, updated_at FROM dbo.users"
  sink:
    type: jsonl
    config:
      path: ./users.jsonl

faucet run pipeline.yaml

URL credentials with special characters must be percent-encoded (@ → %40, : → %3A, / → %2F).

Configuration reference

Core

Batching & reliability

Replication

replication is a tagged enum ({ type: full | incremental, … }):

TLS

tls matches the YAML shape tls: { type: <mode>, ca_cert_path: <path> }:

Authentication

SQL Server authentication (username + password) only, supplied inline in the connection_url or connection_string. Windows / Integrated authentication and Azure AD / Managed Identity are out of scope in v1.

# URL form — percent-encode special characters in the password
config:
  connection_url: "mssql://sa:Str0ng%40Pass@localhost:1433/sales"

# ADO.NET connection-string form
config:
  connection_string: "Server=tcp:db.example.com,1433;Database=sales;User Id=sa;Password=${env:MSSQL_PASSWORD};"

Keep secrets out of the YAML by referencing the environment or a secrets manager — e.g. Password=${env:MSSQL_PASSWORD} or ${vault:secret/mssql#password}.

Examples

Full table export to JSONL

version: 1
pipeline:
  source:
    type: mssql
    config:
      connection_url: "mssql://sa:Str0ng%40Pass@localhost:1433/sales"
      query: "SELECT id, email, created_at FROM dbo.users"
      batch_size: 5000
  sink:
    type: jsonl
    config:
      path: ./users.jsonl

Incremental replication with a bookmark

version: 1
name: mssql_to_jsonl
pipeline:
  source:
    type: mssql
    config:
      connection_url: "mssql://sa:Str0ng%40Pass@localhost:1433/sales"
      query: "SELECT id, email, updated_at FROM dbo.users WHERE updated_at > @bookmark ORDER BY updated_at"
      batch_size: 1000
      tls:
        type: trust_server_certificate   # self-signed dev cert; use prefer/require in prod
      replication:
        type: incremental
        column: updated_at
        initial_value: "1970-01-01T00:00:00Z"
  sink:
    type: jsonl
    config:
      path: ./out/users.jsonl
      append: true
  state:
    type: file
    config:
      path: ./.faucet-state

Parameterized query with bind values

version: 1
pipeline:
  source:
    type: mssql
    config:
      connection_url: "mssql://sa:Str0ng%40Pass@localhost:1433/sales"
      query: "SELECT id, total FROM dbo.orders WHERE region = @P1 AND total > @P2"
      params:
        - "EMEA"
        - 100
  sink:
    type: stdout
    config:
      format: jsonl

Hardened production connection

version: 1
pipeline:
  source:
    type: mssql
    config:
      connection_url: "mssql://etl_user:${env:MSSQL_PASSWORD}@sql.internal:1433/warehouse"
      query: "SELECT * FROM dbo.fact_sales"
      max_connections: 20
      statement_timeout_secs: 120
      tls:
        type: require
        ca_cert_path: /etc/ssl/certs/corp-ca.pem
  sink:
    type: parquet
    config:
      path: ./fact_sales.parquet

Streaming & batching

The source overrides Source::stream_pages: it checks out a pooled connection, runs the query, and walks tiberius's row stream. Rows are decoded and buffered until the buffer reaches batch_size, at which point a StreamPage is yielded; the final page carries the incremental bookmark (when replication is enabled) so the pipeline persists state only after every prior page has been written by the sink. With batch_size: 0 the whole result set is buffered and emitted as one page.

The batch_size argument the pipeline passes to stream_pages is informational — the config field is authoritative, so a pipeline-supplied hint never overrides an explicit value.

Resume & state

When replication.type: incremental, the source is resumable:

• state_key() returns the configured state_key, or a key derived from the connection host plus a query fingerprint (e.g. mssql:db.example.com:<hex>).
• Before fetching, the pipeline loads the stored bookmark and hands it to apply_start_bookmark(), which overrides the configured initial_value for that run.
• Each page is filtered to rows strictly greater than the bookmark on replication.column, and the running maximum is persisted on the final page after the sink confirms the batch — so a crash mid-run never advances the bookmark past unwritten rows.

Server-side pushdown: put the literal @bookmark token in your WHERE clause (e.g. WHERE updated_at > @bookmark ORDER BY updated_at). The source binds the cursor there as a parameter so SQL Server does the filtering. Backstop: whether or not @bookmark appears, the source also filters client-side, so correctness never depends on the query text — though without @bookmark the full result set is fetched and filtered in memory (fine for small tables, prefer @bookmark for large ones). @bookmark is a reserved token; don't use it as an identifier.

To use resume from the CLI, add a state: block (file, memory, redis, or postgres) to the pipeline as shown in the incremental example above.

Type mapping

Config loading & schema

Load from YAML/JSON or environment. Inspect the full JSON Schema with:

faucet schema source mssql

Library usage

use faucet_core::Source;
use faucet_source_mssql::{MssqlSource, MssqlSourceConfig};

# async fn run() -> Result<(), Box<dyn std::error::Error>> {
let cfg = MssqlSourceConfig::new(
    "mssql://sa:Str0ng%40Pass@localhost:1433/sales",
    "SELECT id, email, updated_at FROM dbo.users",
);

let source = MssqlSource::new(cfg).await?;
let rows = source.fetch_all().await?;
println!("rows: {rows:?}");
# Ok(())
# }

For incremental replication, set cfg.replication to MssqlReplication::Incremental { column, initial_value } and drive the source through a Pipeline with a StateStore so the bookmark persists across runs. The shared connection/TLS types (MssqlConnectionConfig, MssqlTls, MssqlTlsMode) are re-exported from this crate, so you don't need to depend on faucet-common-mssql directly.

How it works

1. new() validates the config, then builds the bb8 + bb8-tiberius connection pool once via faucet-common-mssql::build_pool (TLS mode applied).
2. Each query checks out a pooled connection, binds params (and the bookmark, when @bookmark is present) as typed @Pn parameters, and runs under statement_timeout_secs.
3. tiberius's QueryStream is walked row-by-row; each row is decoded by its column type into JSON and buffered into batch_size pages.
4. For incremental runs, each page is filtered against the bookmark and the running maximum is tracked; the bookmark is emitted on the final page so the pipeline persists it only after the sink confirms.
5. connector_name() reports "mssql" for metrics labels; check() probes a pooled connection for faucet doctor.

Lineage dataset URI

<connection>?query=<sql> with credentials stripped — e.g. mssql://db.example.com:1433/sales?query=SELECT id FROM orders. Emitted automatically when lineage is enabled in the CLI.

Feature flags

This crate has no optional features of its own; enable it in the CLI/umbrella via the source-mssql feature (included in the source and full aggregates, but not in the CLI default build).

Troubleshooting / FAQ

See also

• SQL Server connector docs
• State & incremental replication cookbook
• Configuration grammar reference
• faucet-common-mssql — shared connection/TLS/pool types.
• faucet-sink-mssql — the matching SQL Server sink.

License

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