database (output)

The database output executes a SQL query for each record (or all of them inside one transaction). Values are bound through the parameters list — never spliced into the SQL text — with optional transactional wrapping.

Minimal example

output:
  type: database
  connectionStringRef: ${WAREHOUSE_DATABASE_URL}
  query: |
    INSERT INTO orders (id, email, amount)
    VALUES ($1, $2, $3)
    ON CONFLICT (id) DO UPDATE SET
      email = EXCLUDED.email,
      amount = EXCLUDED.amount
  parameters:
    - record.id
    - record.email
    - record.amount

Options

property	type	default	description
typerequired	"database"	—	Module type discriminator. Must be `database` for this module.
id	string	—	Unique identifier within the pipeline.
name	string	—	Human-readable name.
description	string	—	—
enabled	boolean	true	Whether module is active.
tags	array<string>	—	—
onError	string	"fail"	Default error action. Case-insensitive; normalized to lowercase by the runtime.
transaction	boolean	false	Wrap operations in a transaction.
query	string	—	SQL query as a Jinja template rendered to text (control blocks {% if %}/{% for %} allowed). Dynamic values are never interpolated: use $1, $2, ... placeholders bound to the 'parameters' list.
queryFile	string	—	Path to a SQL file containing the query template (same contract as 'query').
parameters	array<string>	—	Ordered expr expressions bound to the query's $N placeholders (parameters[0] -> $1). Evaluated against record/meta/state/pagination; values bind natively typed (nil -> NULL).
connectionString	string	—	Database connection string (DSN). Avoid in production - use connectionStringRef instead.
connectionStringRef	string	—	Environment variable reference for connection string. Format: ${ENV_VAR_NAME}
driver	string	—	Database driver. Auto-detected from connection string if not specified. postgresmysqlsqlite
maxOpenConns	integer	10	Maximum number of open connections in the pool.
maxIdleConns	integer	5	Maximum number of idle connections in the pool.
connMaxLifetimeSeconds	integer	1800	Maximum lifetime of a connection in seconds.
connMaxIdleTimeSeconds	integer	300	Maximum idle time for a connection in seconds.
timeoutMs	integer	30000	Query timeout in milliseconds.

Query or query file

Provide one of query (inline) or queryFile (external). Inline is convenient for short statements; external files are clearer for anything beyond a one-liner.

queryFile: ./sql/upsert_order.sql
parameters:
  - record.id
  - record.email
  - record.amount
  - record.paid_at

sql/upsert_order.sql

INSERT INTO orders (id, email, amount, paid_at)
VALUES ($1, $2, $3, $4)
ON CONFLICT (id) DO UPDATE SET
  email     = EXCLUDED.email,
  amount    = EXCLUDED.amount,
  paid_at   = EXCLUDED.paid_at;

Write $1, $2, … placeholders regardless of the database — Cannectors translates them per driver (kept for PostgreSQL, ? for MySQL/SQLite). Each parameters entry is an expression evaluated against the record (parameters[0] → $1); values bind natively typed and a missing field binds as NULL. See Templating.

Transactional writes

For atomic batch writes — either all records persist or none do — wrap in a transaction:

output:
  type: database
  connectionStringRef: ${WAREHOUSE_DATABASE_URL}
  queryFile: ./sql/upsert_order.sql
  transaction: true

The runtime opens one transaction per output batch, executes the query once per record, then commits at the end. On any error, the transaction rolls back.

Don't combine transaction: true with onError: skip or onError: log expecting partial success. On PostgreSQL the first failing statement aborts the whole transaction (SQLSTATE 25P02); every later record then fails with current transaction is aborted, the commit fails, and nothing is persisted — not even the rows that inserted cleanly before the error. Pick one: transaction: true with onError: fail for all-or-nothing, or transaction: false with onError: skip to let good records through and drop the bad ones.

Connection

Same as the database input — see Connection for driver auto-detection and connection-pool tuning. The output reuses the same pool config (maxOpenConns, maxIdleConns, etc.).

Examples

examples/21-database-output-transaction-query-file.yaml— transactional + external query fileview source ↗

21-database-output-transaction-query-file.yaml

name: database-output-transaction-query-file
version: 1.0.0
description: Write records with a SQL query file in a transaction.
tags:
  - database-output
  - transaction
input:
  type: httpPolling
  schedule: "0 * * * *"
  endpoint: https://source.example.com/api/products
  dataField: products
filters:
  - type: mapping
    mappings:
      - source: id
        target: product_id
      - source: sku
        target: sku
      - source: price
        target: price
        transforms:
          - op: toFloat
      - source: updated_at
        target: updated_at
output:
  type: database
  connectionStringRef: ${WAREHOUSE_DATABASE_URL}
  driver: postgres
  queryFile: examples/assets/sql/upsert_product.sql
  parameters:
    - record.product_id
    - record.sku
    - record.price
    - record.updated_at
  transaction: true
  onError: fail

database (output)

Minimal example

Options

Query or query file

Transactional writes

Connection

Examples

Cross-references

database input

sql_call filter

Templating

On this page