faucet-source-redshift

Amazon Redshift query source connector for the faucet-stream ecosystem.

Redshift speaks the PostgreSQL wire protocol, so this source connects through sqlx's Postgres driver (or, with credentials: { type: redshift_data_api }, through the Redshift Data API over HTTPS), runs a configurable SQL query, and streams the result rows as JSON objects with O(batch_size) memory. It supports full and incremental (bookmark-based) replication.

Configuration

¹ Not used by redshift_data_api, which addresses cluster_identifier or workgroup_name instead (the dataset URI and the derived state key name that target: redshift://workgroup:<name>/<database>).

Incremental replication

With replication.type = incremental the source reads only the rows not yet delivered, in cursor order. It wraps the query as SELECT * FROM (<query>) AS q WHERE q."<column>" >= CAST($n AS <type>) ORDER BY q."<column>", where <type> is the cursor column's type as the server describes the query (so a string bookmark compares as a TIMESTAMP, a number as an INT8, …); initial_value is the inclusive lower bound for the first run. To apply the cursor further inside the query, write ${bookmark} there with >=; it binds the same way, and a > ${bookmark} is refused at load time because it would skip rows that share the bookmark value.

Rows that share a cursor value are never skipped. The bookmark stores the last cursor value and a 128-bit fingerprint of every row already emitted at it — {"value": "...", "boundary": ["<fingerprint>", ...]}. The next read re-reads that value with >= and drops only the rows it already emitted, so a second COPY batch stamped with the same load time, or any row committed later with the last updated_at, is delivered. Above 10,000 rows at one value the set is dropped and the value is re-read in full: duplicates, never loss. Every page carries the bookmark after its last row, so a crash replays at most one page, and the bookmark never moves backwards (a run that reads nothing keeps it). A bare bookmark stored by an earlier release reads as that value with an empty boundary (its rows are re-delivered once). Rows whose cursor is NULL are never read.

Type decoding

Integers, floats, booleans, JSON and text map to JSON natively; NUMERIC is an exact decimal string, timestamps/dates/times are ISO strings, TIMETZ keeps its offset, INTERVAL is an ISO 8601 duration (P1Y2DT3H1.5S) and VARBYTE is base64, and oid / reg* values are numbers. Any other type is emitted as its text form when the server sends it as text, or when its wire form is its text (SUPER, XML, character types); a binary-encoded value of any other type (GEOMETRY's EWKB, an int[]) is refused rather than garbled. A non-NULL value the source cannot decode fails the read naming the column — cast it in the query (col::varchar) — rather than becoming null.

host: my-cluster.abc123.us-east-1.redshift.amazonaws.com
database: dev
user: admin
credentials:
  type: password
  config:
    password: ${env:REDSHIFT_PASSWORD}
query: "SELECT * FROM events"
replication:
  type: incremental
  column: updated_at
  initial_value: "2026-01-01T00:00:00Z"

Redshift Data API

With credentials: { type: redshift_data_api } the query is submitted with ExecuteStatement, polled with DescribeStatement, and its rows are read page by page with GetStatementResult, re-framed into batch_size pages (a service page is read whole). Bind values (params, context tokens, the bookmark) travel as Data API parameters :pN; a NULL or empty string is written into the SQL, since a parameter value can be neither. The typed fields render through the shared value-fidelity model: exact NUMERIC strings, timestamptz as RFC 3339 with its offset, timestamp / time at microsecond precision, VARBYTE base64. The Data API has no describe, so the incremental cursor's type and the declared columns come from a LIMIT 0 run of the query. Its round trips are metered as submit, poll and fetch; a run deadline or cancel ends the poll and cancels the statement on the service.

database: dev
credentials:
  type: redshift_data_api
  config:
    region: us-east-1
    workgroup_name: analytics
query: "SELECT * FROM events"

Testing

Redshift has no local container image, but it speaks the PostgreSQL wire protocol, so tests/integration.rs runs the read path against a Postgres container (Docker required). tests/data_api.rs runs the Data API path against a contract fake that executes each statement on that container, and an iam read against a fake GetClusterCredentials.

License: MIT OR Apache-2.0

Capabilities

A plain query source: no exactly-once, discovery or sharding. faucet list --json prints the declaration.

Timeouts, cancellation and retries

The query and every page of rows read from its cursor run through the pipeline's CallContext (#880): the run's cancel token stops a stalled read, any caller deadline ends it, and the query is metered as a query round trip. The iam credentials request (credentials) and every Data API request (submit, poll, fetch) go through it too. Rows are read a page at a time per call, so the per-row cost is unchanged; the pool's connect timeout still applies and nothing is retried.

Declared column types

The source declares the query's columns and their types (Source::declared_columns, #955) from the describe its read makes (a LIMIT 0 run through the Data API), so a sink that creates its table types it from the source rather than from the first page's JSON values. See Declared column types.