faucet-transform-sql

SQL-as-transform for the faucet-stream ecosystem — run DuckDB SQL over each pipeline page. The page's records are exposed as the relation batch; the query's result set replaces the page.

Reach for it when you need to filter, reshape, aggregate, or join your in-flight data with the full power of an analytical SQL engine, inline in the pipeline, without standing up a separate warehouse step. DuckDB is embedded (bundled at build time) and vectorized, so the transform is fast and self-contained — no external database, no network hop.

Feature highlights

• Full DuckDB SQL — SELECT, WHERE, GROUP BY, window functions, CTEs, json_extract, casts, string/date functions — the page is just a table named batch.
• Reference relations — pre-load static lookup data from CSV, JSONL, inline values, or a small REST endpoint (http) and JOIN against it by name. Optional reload_on_change re-reads a file when its mtime changes; http is fetched once and cached for the whole run.
• Vectorized JSON ↔ Arrow shovel — records are moved into and out of DuckDB through Arrow batches (arrow / arrow-json), not row-by-row, so throughput stays high.
• Compile-time query validation — the query is parse/bind-checked inside DuckDB and missing relation files are caught at faucet validate / start of faucet run, never mid-stream.
• Embedded engine — DuckDB is bundled (bundled feature), so there are no system dependencies and the connection is built once per transform and reused for every page.
• Tunable — optional memory_limit and threads DuckDB pragmas keep resource use predictable in high-fan-out matrix runs.

Installation

# As a library:
cargo add faucet-transform-sql

# In the umbrella crate (opt-in transform feature):
cargo add faucet-stream --features transform-sql

# In the CLI:
cargo install faucet-cli --features transform-sql

The transform-sql feature is not in the default build; it is included in full.

Quick start

The sql transform is a pipeline-level (or matrix-row-level) transform — it goes under transforms:, between a source and a sink:

# pipeline.yaml — faucet run pipeline.yaml
version: 1
pipeline:
  source:
    type: file
    config:
      path: data/users.csv
      has_header: true
  transforms:
    - type: sql
      config:
        query: "SELECT id, upper(name) AS name FROM batch WHERE active"
  sink:
    type: file
    config:
      path: ./active_users.jsonl

faucet run pipeline.yaml

Each result row becomes one output JSON record: column name → JSON key; NULL → JSON null; DuckDB STRUCT / LIST / MAP → nested JSON.

Sparse records. Arrow needs one column per key across the page, so a key some records lack is NULL for them inside DuckDB. When the query keeps batch's columns (SELECT *, SELECT * EXCLUDE (…), SELECT *, … FROM batch WHERE …) the transform remembers which keys each record did not carry and leaves them out of the output again, so an upsert sink does not overwrite those destination columns with NULL; an explicit null in the input stays null. This covers top-level keys only: a key missing inside a nested object (a struct column) comes back as null when a sibling record's object had it. A query that names its columns gets exactly those columns, null where a record had no value. To do this, a page with sparse records carries one extra column, __faucet_absent, inside DuckDB; it is removed from the output, including from a row embedded whole as a STRUCT (SELECT b FROM batch b, struct_pack(*)), but not from a row turned into text or a hash: to_json(b), b::VARCHAR, md5(b::VARCHAR) and hash(b) include "__faucet_absent" on a sparse page only, so their output differs between a dense and a sparse page holding the same data. Select the columns you need (or b itself, a STRUCT) instead. A query that batch's extra column would break, such as SELECT * FROM batch UNION ALL SELECT * FROM other, is re-run on that page without it (absent keys then come back as null). Expressions that expand every column (COLUMNS(*), UNPIVOT … ON COLUMNS(*)) still see the column on a sparse page. A record that carries a __faucet_absent key of its own turns the marking off for its page. A key that is a number in one record and a string in another becomes a string column (1 → "1").

Sandbox. The DuckDB connection has no host access once the reference relations are loaded: enable_external_access is off (no read_text, read_csv, COPY … TO, ATTACH or network reads; only the files of reload_on_change relations stay readable), extensions cannot be installed or autoloaded, and the configuration is locked, so a query cannot SET its own memory_limit or threads.

Configuration reference

Wire shape: { type: sql, config: { query, relations?, memory_limit?, threads? } }.

Reference-relation fields (RelationSpec)

Relation source types (source.type)

Reference relations

Pre-load static lookup data (CSV, JSONL, inline values, or an http endpoint) that your query can JOIN against. Relations are loaded once at compile time (when faucet validate / faucet run first reads the config) and remain resident for the lifetime of the transform.

- type: sql
  config:
    query: |
      SELECT b.id, c.country
      FROM batch b
      LEFT JOIN countries c ON b.code = c.code
    relations:
      - name: countries
        source:
          type: csv
          path: data/countries.csv
          has_header: true        # default true

Inline values need no file I/O:

relations:
  - name: tiers
    source:
      type: values
      columns: [id, label]
      rows:
        - [1, gold]
        - [2, silver]
        - [3, bronze]

An http relation fetches its rows from a small REST endpoint once at compile time and caches them for the whole run (every page joins against the same in-memory table — the endpoint is never re-hit):

relations:
  - name: named_lists
    source:
      type: http
      url: https://api.example.com/v1/company/named-lists
      method: GET                 # default GET; POST also supported
      headers:                    # optional static headers
        Authorization: "Bearer ${env:HIBOB_TOKEN}"
      records_path: "$.items[*]"  # JSONPath to the row array; omit if the body is already an array

records_path is a JSONPath selecting the row array in the response body; each selected element must be a JSON object (one table row). If records_path is omitted, the whole response body must be a JSON array. A non-array body without a records_path, or a path that selects non-object values, is a clear compile-time configuration error. Keep auth simple: pass a bearer token via a static headers entry (interpolated at the CLI layer with ${env:...} / ${vault:...}); the shared auth: catalog is not wired here.

reload_on_change

relations:
  - name: prices
    source:
      type: csv
      path: data/prices.csv
    reload_on_change: true   # re-read when the file's mtime changes

When true, faucet stats the file before each page and rebuilds the relation atomically if the mtime changed. Defaults to false. Ignored for values and http (both loaded once for the whole run).

A file modified less than a second ago is stat'ed twice, 100 ms apart, and is reloaded only when its mtime and size match — otherwise the old table stays for this page and the reload is retried on the next one. Replace reference files atomically (write a temp file, then rename it over the old one) so a reload never sees a half-written file.

The name batch is reserved for the page relation. Using it as a relation name is a compile-time error.

Output types

• Decimals are exact. A DECIMAL with precision up to 15 stays a JSON number; an integral DECIMAL or HUGEINT value (e.g. SUM of integers) that fits a 64-bit integer becomes a JSON integer; any wider value is an exact decimal string such as "1234567890123456789012345678.0123456789", never an f64 rounding.
• Non-finite floats (NaN, ±Infinity) have no JSON number: they become a null counted in the run summary (non_finite) and faucet_fidelity_non_finite_total; binary columns are base64.
• A field that is {} in every record of a page is passed to DuckDB as the string '{}' (DuckDB has no zero-field struct) and comes back as {}.
• Keys that differ only by letter case (a and A) fail the page: SQL identifiers are case-insensitive, so DuckDB would silently rename one. Rename one of them upstream.

Per-page semantics and batch_size: 0

This is the most important thing to understand about the SQL transform.

The transform runs once per page, not once across the whole stream. With the default batch_size of 1000, GROUP BY and window functions aggregate within a single 1000-row page — not across all pages.

# BAD: GROUP BY runs per-page, giving partial aggregates.
pipeline:
  source:
    type: file
    config:
      path: data/orders.csv
  transforms:
    - type: sql
      config:
        query: "SELECT country, SUM(amount) AS total FROM batch GROUP BY country"

To aggregate across the whole dataset, set the source's batch_size: 0 so the entire result set arrives as one page:

# CORRECT: batch_size: 0 loads the whole file as one page → global GROUP BY.
pipeline:
  source:
    type: file
    config:
      path: data/orders.csv
      batch_size: 0
  transforms:
    - type: sql
      config:
        query: "SELECT country, SUM(amount) AS total FROM batch GROUP BY country"

batch_size: 0 means "no batching" — the source emits the entire result set as a single StreamPage. All sources support it; it is appropriate for small lookup tables and for aggregating transforms like this one. Be mindful of memory: batch_size: 0 buffers the whole result set in RAM, so use it for datasets that comfortably fit in memory.

When an aggregating query receives a second page (i.e. batch_size was not set to 0), faucet emits a one-time warning:

WARN faucet::transform::sql: sql transform with aggregation received multiple pages;
aggregation is per-page — set batch_size: 0 for global aggregation

Error handling and validation

At config load time (faucet validate / start of faucet run):

• The query is parse/bind-checked inside DuckDB. Syntax errors report line and column number.
• Reference-relation files that do not exist cause an immediate error (before any page is processed).
• An http relation is fetched here: a request failure, a non-2xx status, an undecodable body, a non-array body without a records_path, or a path selecting non-object rows all cause an immediate error (before any page is processed).
• A relation named batch is rejected.

At runtime (per page):

• A query that fails mid-run aborts the pipeline immediately with FaucetError::Transform.
• Runtime query errors are not routed to the dead-letter queue — they follow the same fail-fast policy as every other built-in transform.

Empty result sets are valid: a query that matches zero rows produces zero output records for that page.

Examples

Filter and reshape (per-page, no aggregation)

transforms:
  - type: sql
    config:
      query: |
        SELECT id,
               lower(email)        AS email,
               coalesce(plan, 'free') AS plan
        FROM batch
        WHERE deleted_at IS NULL

Global aggregation joined to a reference CSV

This mirrors cli/examples/csv_to_jsonl_sql.yaml — group order data by country and join to a reference countries CSV:

version: 1
name: csv_to_jsonl_sql

pipeline:
  source:
    type: file
    config:
      path: cli/examples/data/orders.csv
      has_header: true
      batch_size: 0          # whole file as one page → global GROUP BY

  transforms:
    - type: sql
      config:
        query: |
          SELECT c.country,
                 COUNT(*)                      AS order_count,
                 SUM(CAST(o.amount AS DOUBLE)) AS total_amount
          FROM   batch o
          LEFT JOIN countries c ON o.country_code = c.code
          GROUP BY c.country
          ORDER BY c.country
        relations:
          - name: countries
            source:
              type: csv
              path: cli/examples/data/countries.csv
              has_header: true

  sink:
    type: file
    config:
      path: /tmp/faucet_sql_demo.jsonl

Run it (requires the transform-sql, source-csv, and sink-jsonl features):

faucet run cli/examples/csv_to_jsonl_sql.yaml

Output rows look like:

{"country":"Germany","order_count":1,"total_amount":3.0}
{"country":"India","order_count":1,"total_amount":7.0}
{"country":"United States","order_count":2,"total_amount":15.5}

Enrich with an inline lookup table

transforms:
  - type: sql
    config:
      query: |
        SELECT b.id, b.tier_id, t.label AS tier
        FROM batch b
        LEFT JOIN tiers t ON b.tier_id = t.id
      relations:
        - name: tiers
          source:
            type: values
            columns: [id, label]
            rows:
              - [1, gold]
              - [2, silver]
              - [3, bronze]

Working with JSON columns

DuckDB's json_extract works on string or JSON columns. If a field is a JSON string, use it directly:

SELECT json_extract(payload, '$.user.id') AS user_id FROM batch

Or cast it first:

SELECT json_extract(payload::JSON, '$.user.id') AS user_id FROM batch

Timestamp / timezone note

DuckDB's TIMESTAMP type is timezone-naive. faucet JSON timestamps are RFC 3339 strings (e.g. "2026-01-01T12:00:00Z"). To compare or cast them:

-- Parse an RFC 3339 string into a DuckDB TIMESTAMP (drops the offset)
SELECT CAST(created_at AS TIMESTAMP) AS ts FROM batch
WHERE CAST(created_at AS TIMESTAMP) > '2026-01-01'::TIMESTAMP

-- Keep the string form, compare lexicographically (safe for UTC-only data)
SELECT * FROM batch WHERE created_at > '2026-01-01T00:00:00Z'

If your timestamps include non-UTC offsets, normalise them to UTC with cast before passing to the SQL transform, or parse with strptime.

Config loading & schema

Configs load from YAML/JSON. Inspect the full JSON Schema for the transform with:

faucet schema transform sql

Library usage

Library callers build the compiled transform and attach it to any Source via a page stage and faucet_core::TransformingSource:

use faucet_transform_sql::{SqlTransform, SqlTransformConfig};
use faucet_core::stage::{compile_stage, apply_stages_to_page};

# fn run(records: Vec<serde_json::Value>) -> Result<(), faucet_core::FaucetError> {
let cfg = SqlTransformConfig {
    query: "SELECT id, upper(name) AS name FROM batch".into(),
    relations: vec![],
    memory_limit: None,
    threads: None,
};

let t = SqlTransform::compile(&cfg)?;          // parse/bind-checks the query now
let stage = compile_stage(&t.into_page_stage())?;

let output = apply_stages_to_page(records, &[stage])?;
# let _ = output;
# Ok(())
# }

To wrap a whole source, hand the same stage to faucet_core::TransformingSource::new(source, vec![stage]) and run it through Pipeline / run_stream like any other source. SqlTransform::compile is the canonical entry point; into_page_stage() produces a TransformStage::PageFn (a page-level, whole-batch stage).

How it works

1. SqlTransform::compile(&cfg) opens an in-memory DuckDB connection, applies the memory_limit / threads pragmas, materializes every reference relation, and parse/bind-checks the query — all once.
2. For each page, the records are converted to an Arrow batch and registered as the batch relation (a vectorized JSON ↔ Arrow shovel, not row-by-row).
3. The query runs; the result set is converted back from Arrow to JSON records, which replace the page.
4. With reload_on_change, a relation's file mtime is checked before each page and the relation is rebuilt and atomically swapped if it changed.

The DuckDB connection and all relations are owned by the compiled SqlTransform and reused for every page — there is no per-page connection setup.

Feature flags

This crate itself has no optional features of its own; the bundled and vtab-arrow DuckDB features are always on.

# Enable only the SQL transform
cargo add faucet-stream --features transform-sql

# Enable together with specific connectors
cargo add faucet-stream --features source-csv,sink-jsonl,transform-sql

# Enable everything
cargo add faucet-stream --features full

Troubleshooting / FAQ

See also

• SQL transform cookbook — the transforms model and worked examples.
• Transforms reference — the full transforms: grammar and layering rules.
• faucet-core — the TransformStage / TransformingSource types this transform plugs into.
• faucet-stream — the umbrella crate that exposes the transform-sql feature.
• DuckDB SQL documentation — the dialect available inside the query.

License

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