`sql://`¶

Driver-agnostic SQL producer + polling consumer. One component instance handles every driver the running Helix binary blank-imports — MySQL, MariaDB, Microsoft SQL Server, Oracle, PostgreSQL (via pgx/stdlib), SQLite, and anything else registered via database/sql.Register.

URI shapes¶

The component supports two URI forms:

1. CRUD shape (existing Helix convention):

sql://<table>/<operation>?driver=<name>&dsn=<dsn>

operation ∈ {query, insert, update, delete, execute}.

2. Raw-SQL shape:

sql://?driver=<name>&dsn=<dsn>&query=<urlencoded-sql>&outputType=…

The query can also be supplied per-request via the HelixSqlQuery header, or via the message body when useMessageBodyForSql=true.

Parameters¶

Common / raw-SQL form¶

Param	Type	Default	Description
`driver`	string	required	`database/sql.Register` name (mysql, pgx, sqlserver, oracle, sqlite3, …).
`dsn`	string	required	Driver-specific connection string.
`query`	string	—	Raw SQL query. Per-request override via header.
`allowNamedParameters`	boolean	`true`	Parse `:#name` placeholders into driver-appropriate positional placeholders.
`placeholder`	string	`#`	Named-parameter prefix character (the `:#name` form uses `#`).
`usePlaceholder`	boolean	`true`	When false, the SQL is sent as-is — no rewriting.
`outputType`	enum	`SelectList`	Result shape — `SelectList`, `SelectOne`, `Update`, `StreamList`.
`outputHeader`	string	—	Write the result to this header instead of the body.
`separator`	string	`,`	Split character for IN-clause CSV expansion (`SELECT … WHERE id IN (:#ids)` with a string body of `"1,2,3"` expands to three placeholders).

Producer¶

Param	Type	Default	Description
`batch`	boolean	`false`	Treat the body as `[]map[string]any` and run the SQL once per row in a single statement loop.
`batchAutoCommitDisabled`	boolean	`true`	Defer commit until end-of-batch. Rolls back the whole batch on first failure.
`noop`	boolean	`false`	Skip executing the SQL — just bind parameters and log. Dry-run validation.
`useMessageBodyForSql`	boolean	`false`	Body IS the SQL query string; params come from `HelixSqlParameters` header.
`useHeadersAsParameters`	boolean	`false`	Source named-param values from exchange headers (vs. body map). Body is fallback.
`transacted`	boolean	`false`	Wrap the producer call (or consumer batch + onConsume hooks) in a single DB transaction.
`alwaysPopulateStatement`	boolean	`true`	When true, missing `:#name` bind values cause an error. When false, missing names bind as SQL NULL (lax mode).
`readSize`	integer	`0`	Cap on rows returned per query. 0 = no cap.

Consumer (polling)¶

Param	Type	Default	Description
`delay`	duration	`500ms`	Poll interval. Accepts Go duration (`5s`) or bare ms (`500`).
`initialDelay`	duration	`1s`	Delay before the first poll.
`useFixedDelay`	boolean	`true`	Fixed-delay scheduling vs. fixed-rate.
`maxMessagesPerPoll`	integer	`0`	Cap rows processed per poll cycle. 0 = unlimited.
`repeatCount`	integer	`0`	Number of polls before the consumer stops. 0 = infinite.
`useIterator`	boolean	`true`	Emit one Exchange per row vs. one Exchange per poll batch.
`routeEmptyResultSet`	boolean	`false`	Emit an empty-body Exchange when a poll yields zero rows.
`sendEmptyMessageWhenIdle`	boolean	`false`	Emit a heartbeat Exchange (with `sql.empty=true`) when a poll yields zero rows.
`breakBatchOnConsumeFail`	boolean	`false`	Abort the batch on first failing row.
`expectedUpdateCount`	integer	`-1`	Fail the batch if an `onConsume` statement affects ≠ N rows. -1 disables.
`greedy`	boolean	`false`	Re-poll immediately when previous poll yielded rows.
`onConsume`	string	—	SQL run on each row after the processor succeeds. Typically `UPDATE … SET done=true WHERE id=:#id`.
`onConsumeFailed`	string	—	SQL run on each row after the processor fails.
`onConsumeBatchComplete`	string	—	SQL run once after the whole poll batch completes.

Legacy / Helix-only¶

Param	Type	Default	Description
`pollInterval`	duration	—	Legacy alias for `delay`. Existing routes use this name.
`pollSince`	string	—	Column used as a watermark for change-data-capture polling. The consumer establishes `max(col)` on first poll, then emits only rows newer than the watermark.

Header conventions¶

Each header is also accepted under its legacy alternate name. Read priority is the Helix form, then the alternate; writes set both names.

Header	Purpose
`HelixSqlQuery`	Per-request query override (unless `skipControlHeaders`).
`HelixSqlParameters`	Named-param map (used with `useMessageBodyForSql`).
`HelixSqlDataSource`	Per-request DSN override.
`HelixSqlRowCount`	Written on successful SELECT.
`HelixSqlUpdateCount`	Written on UPDATE/DELETE/INSERT.
`HelixSqlRetrieveGeneratedKeys`	Opt into RETURNING / generated keys.
`HelixSqlGeneratedColumns`	Column list for RETURNING (pg).
`HelixSqlGeneratedKeyRows`	`[]map[string]any` of inserted rows.
`sql.where`	Legacy — required by `update` and `delete` CRUD ops.
`sql.rowCount` / `sql.rowsAffected`	Legacy aliases (also written).

Body shape¶

Request:

CRUD insert / update: map[string]any (column → value).
CRUD query: optional {query, params} map override.
CRUD execute: {query, params} map.
CRUD batch=true insert: []map[string]any.
Raw-SQL: map[string]any for named params (unless useHeadersAsParameters); []any for positional.
useMessageBodyForSql=true: body IS the SQL string.

Response (in priority order):

outputHeader set → result on that header, body untouched.
outputType=SelectList (default): []map[string]any.
outputType=SelectOne: map[string]any (errors if ≠ 1 row).
outputType=Update: int64 rows-affected.
outputType=StreamList: <-chan map[string]any for streaming large result sets.

Examples¶

CRUD insert¶

- to: sql://users/insert?driver=mysql&dsn=root:helix@tcp(db:3306)/app

Raw SELECT with named params¶

- to: sql://?driver=pgx&dsn=postgres://app:secret@db:5432/app
  # body: {"id": 42} → :#id resolves to 42, rewritten to $1

Body:

{"id": 42}

Query (URL-encoded inline or supplied via HelixSqlQuery header):

SELECT * FROM users WHERE id = :#id

Batch insert¶

- to: sql://events/insert?driver=mysql&dsn=…&batch=true
  # body: [{"id": 1, "t": "..."}, {"id": 2, "t": "..."}, …]

Polling consumer with `onConsume` hook¶

trigger:
  uri: sql://jobs/query?driver=pgx&dsn=…&delay=500ms&onConsume=UPDATE%20jobs%20SET%20done%3Dtrue%20WHERE%20id%3D%3A%23id
  query: "SELECT id, payload FROM jobs WHERE done = false"
steps:
  - process:
      ref: handle-job

`outputType=Update` with named param¶

- to: sql://?driver=mysql&dsn=…&outputType=Update
  # body: {"delta": 5}
  # query (header HelixSqlQuery): UPDATE counters SET v = v + :#delta

sql://¶