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.

Dataset discovery

The source supports live introspection via Source::discover(): it enumerates every base table visible to the connection (from INFORMATION_SCHEMA.COLUMNS / INFORMATION_SCHEMA.TABLES) and returns one dataset descriptor per table with:

• name — schema.table (e.g. dbo.orders), kind: table
• schema — the column shape as a JSON-Schema object (types mapped per the table below; nullable columns become ["T", "null"])
• estimated_rows — a cheap row estimate from sys.partitions (heap/clustered-index partitions only). If the sys.* views aren't readable for your principal, discovery still succeeds — the estimate is simply omitted.
• config_patch — { "query": "SELECT * FROM [schema].[table]" }, bracket-quoted (interior ] doubled), ready to deep-merge over the connection config as a matrix row

Discovery reads catalog metadata only — it never scans table data.

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.

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: mssql
    config:
      connection_url: ${env:MSSQL_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.

A sharded query must not end in a top-level ORDER BY — T-SQL forbids it inside the derived table the shard predicate wraps (ordering across concurrently-executing shards is meaningless anyway). With incremental replication, shard bounds are computed over the not-yet-synced slice (the @bookmark binding is honoured during enumeration).

License

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