sql_call

The sql_call filter runs a SQL query per record, then merges the result back onto the record. The query text carries $1, $2, … placeholders and the parameters list supplies the values — they are bound by the driver, never string-spliced into the SQL, so they're safe by construction.

Minimal example

filters:
  - type: sql_call
    connectionStringRef: ${REFERENCE_DATABASE_URL}
    query: SELECT tier FROM customers WHERE id = $1
    parameters:
      - record.customerId
    mergeStrategy: merge

Options

property	type	default	description
typerequired	"sql_call"	—	Module type discriminator. Must be `sql_call` 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.
cache	object	—	—
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.
mergeStrategy	string	"merge"	How to merge the call response with the input record. merge: deep-merge the response into the record. replace: overwrite the record's top-level keys with the response ones. append: nest the whole response under resultKey. mergereplaceappend
resultKey	string	—	Record key receiving the whole response with mergeStrategy: append. Required as soon as mergeStrategy is append, for the three call filters alike; ignored by the merge and replace strategies.

Query and parameters

query is a Jinja template rendered to SQL text — {% if %} blocks can build optional clauses. Dynamic values are referenced with $1, $2, … placeholders, bound to the parameters list (1-based: parameters[0] → $1). Each entry is an expr expression evaluated against the record; values bind natively typed and a missing field binds as NULL.

query: |
  SELECT plan, mrr
  FROM accounts
  WHERE id = $1
  {% if record.region %}AND region = $2{% endif %}
parameters:
  - record.account_id
  - record.region

Write $N regardless of the database — Cannectors translates placeholders per driver (kept for PostgreSQL, ? for MySQL/SQLite). A placeholder inside a skipped {% if %} branch is neither evaluated nor bound. cannectors validate rejects a $N with no matching parameter, and a parameter never referenced by the query.

cannectors validate rejects an output tag ({{ … }}) inside a SQL query — it would splice a raw, unescaped value into the SQL. Dynamic values always go through parameters. Control blocks ({% if %}) are allowed.

Connection

Same shape as the database input — connectionString or connectionStringRef. Use the ref form for production:

connectionStringRef: ${REFERENCE_DATABASE_URL}

See database input · connection for driver auto-detection and tuning.

Merge strategies

Strategy	Effect
`merge` (default)	Deep-merge the first returned row onto the record. Nested objects are merged; SQL values overwrite conflicts.
`replace`	Overlay the first returned row onto the record. Existing fields not present in the row are preserved.
`append`	Store the first returned row under `resultKey`. Requires `resultKey` to be set.

mergeStrategy: append
resultKey: enrichments

append without resultKey fails cannectors validate. http_call, soap_call and sql_call share this contract — see merge contract of the call filters.

sql_call currently consumes only the first row returned by the query. For one-to-many enrichment, aggregate the related data in SQL so the query returns one row with the desired value under a column.

Caching

sql_call has the same cache config as http_call — LRU with TTL.

cache:
  enabled: true
  maxSize: 1000
  ttlSeconds: 600
  key: "{{ record.customerId }}"

The same warning applies: without an explicit cache.key or per-record parameters, all records will hit the same cache slot.

The cache also outlives a run the same way it does for http_call: on a scheduled pipeline, ttlSeconds is how stale a lookup may get, not a per-tick setting. Reference tables tolerate a high TTL; anything that changes during the day does not.

Failures

With onError: log, a failed query does not drop the record: it passes through annotated under _errors, carrying the category and whether the failure is retryable. See Error markers.

The classification follows what the database said, not a generic default:

Failure	`category`	`retryable`
Constraint violation, syntax error	`validation`	`false`
Deadlock	`server`	`true`
Lost or refused connection, query timeout	`network`	`true`

A constraint violation will never succeed on replay, so it is recorded as functional — park the record rather than queueing it for the next cycle.

resultKey cannot be _errors — the key is reserved for the runtime's error markers and the pipeline fails validation.

Examples

examples/17-sql-call-merge-cache.yaml— merge + cacheview source ↗

17-sql-call-merge-cache.yaml

name: sql-call-merge-cache
version: 1.0.0
description: Enrich records with a SQL lookup and deep merge the result.
tags:
  - sql-call
  - cache
input:
  type: httpPolling
  schedule: "*/30 * * * *"
  endpoint: https://source.example.com/api/orders
  dataField: orders
filters:
  - type: sql_call
    connectionStringRef: ${REFERENCE_DATABASE_URL}
    driver: postgres
    query: |
      select segment, account_owner
      from customer_reference
      where customer_id = $1
      limit 1
    parameters:
      - record.customer.id
    mergeStrategy: merge
    cache:
      enabled: true
      maxSize: 2000
      ttlSeconds: 600
      key: "customer:{{record.customer.id}}"
output:
  type: httpRequest
  endpoint: https://destination.example.com/api/orders/enriched
  method: POST
  requestMode: batch

examples/18-sql-call-append-query-file.yaml— append + external query fileview source ↗

18-sql-call-append-query-file.yaml

name: sql-call-append-query-file
version: 1.0.0
description: Use queryFile for SQL enrichment and append the lookup result under a custom key.
tags:
  - sql-call
  - query-file
input:
  type: httpPolling
  schedule: "0 * * * *"
  endpoint: https://source.example.com/api/customers
  dataField: customers
filters:
  - type: sql_call
    connectionStringRef: ${REFERENCE_DATABASE_URL}
    driver: postgres
    queryFile: examples/assets/sql/customer_lookup.sql
    parameters:
      - record.id
    mergeStrategy: append
    resultKey: reference
    cache:
      enabled: true
      maxSize: 1000
      ttlSeconds: 300
      key: "customer:{{record.id}}"
output:
  type: httpRequest
  endpoint: https://destination.example.com/api/customers/reference
  method: POST
  requestMode: batch

examples/29-call-filters-shared-merge-contract.yaml— shared merge contract across the three call filtersview source ↗

29-call-filters-shared-merge-contract.yaml

name: call-filters-shared-merge-contract
version: 1.0.0
description: The three call filters share one merge contract - mergeStrategy append nests the whole response under resultKey.
tags:
  - http-call
  - soap-call
  - sql-call
  - append
input:
  type: httpPolling
  schedule: "*/15 * * * *"
  endpoint: https://source.example.com/api/orders
  dataField: orders
filters:
  - type: http_call
    endpoint: https://crm.example.com/api/customers
    method: GET
    keys:
      - field: customerId
        paramType: query
        paramName: id
    mergeStrategy: append
    resultKey: customer
    cache:
      enabled: true
      maxSize: 1000
      ttlSeconds: 300
      key: "customer:{{record.customerId}}"
  - type: soap_call
    endpoint: https://erp.example.com/contracts
    soapAction: urn:GetContract
    operation: GetContract
    body: |
      <m:GetContract xmlns:m="urn:contracts">
        <m:CustomerId>{{record.customerId}}</m:CustomerId>
      </m:GetContract>
    dataField: Envelope.Body.GetContractResponse.Contract
    mergeStrategy: append
    resultKey: contract
    cache:
      enabled: true
      maxSize: 1000
      ttlSeconds: 300
      key: "contract:{{record.customerId}}"
  - type: sql_call
    connectionStringRef: ${REFERENCE_DATABASE_URL}
    driver: postgres
    query: SELECT tier, discount FROM pricing WHERE customer_id = $1
    parameters:
      - record.customerId
    mergeStrategy: append
    resultKey: pricing
    cache:
      enabled: true
      maxSize: 1000
      ttlSeconds: 300
      key: "pricing:{{record.customerId}}"
output:
  type: httpRequest
  endpoint: https://destination.example.com/api/orders/enriched
  method: POST
  requestMode: batch
  headers:
    Content-Type: application/json

sql_call

Minimal example

Options

Query and parameters

Connection

Merge strategies

Caching

Failures

Examples

Cross-references

database input

http_call filter

On this page