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

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.

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

mergeStrategy: append
resultKey: enrichments

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.

Examples

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

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

Cross-references