Connect
Self-Managed Only

sql

Deprecated

This component is deprecated and will be removed in the next major version release. Please consider moving onto alternative components.

Executes an arbitrary SQL query for each message.

Introduced in version 3.65.0.

  • Common

  • Advanced

output:
  label: ""
  sql:
    driver: "" # No default (required)
    data_source_name: "" # No default (required)
    query: "" # No default (required)
    args_mapping: "" # No default (optional)
    max_in_flight: 64
    batching:
      count: 0
      byte_size: 0
      period: ""
      check: ""
output:
  label: ""
  sql:
    driver: "" # No default (required)
    data_source_name: "" # No default (required)
    query: "" # No default (required)
    args_mapping: "" # No default (optional)
    max_in_flight: 64
    batching:
      count: 0
      byte_size: 0
      period: ""
      check: ""
      processors: [] # No default (optional)

Alternatives

For basic inserts use the sql_insert output. For more complex queries use the sql_raw output.

Fields

args_mapping

An optional Bloblang mapping that includes the same number of values in an array as the placeholder arguments in the query field.

Type: string

# Examples:
args_mapping: root = [ this.cat.meow, this.doc.woofs[0] ]

# ---

args_mapping: root = [ meta("user.id") ]

batching

Configure a batching policy.

Type: object

# Examples:
batching:
  byte_size: 5000
  count: 0
  period: 1s

# ---

batching:
  count: 10
  period: 1s

# ---

batching:
  check: this.contains("END BATCH")
  count: 0
  period: 1m

batching.byte_size

The maximum total size (in bytes) that a batch can reach before it is flushed. When the combined size of all messages in the batch reaches or exceeds this limit, the batch is immediately sent to the next stage (such as a processor or output).

Set to 0 to disable size-based batching. When disabled, messages are flushed based on other conditions (such as count or period).

Type: int

Default: 0

batching.check

A Bloblang query that returns a boolean value indicating whether a message should end a batch.

Type: string

Default: ""

# Examples:
check: this.type == "end_of_transaction"

batching.count

The number of messages at which the batch is flushed. Set to 0 to disable count-based batching.

Type: int

Default: 0

batching.period

The length of time after which an incomplete batch is flushed regardless of its size. This field accepts Go duration format strings such as 100ms, 1s, or 5s. Supported time units are ns, us, ms, s, m, and h.

Type: string

Default: ""

# Examples:
period: 1s

# ---

period: 1m

# ---

period: 500ms

batching.processors[]

A list of processors to apply to a batch as it is flushed. This allows you to aggregate and archive the batch however you see fit. All resulting messages are flushed as a single batch, so splitting the batch into smaller batches with these processors has no effect.

Type: array<processor>

# Examples:
processors:
  - archive:
      format: concatenate


# ---

processors:
  - archive:
      format: lines


# ---

processors:
  - archive:
      format: json_array

data_source_name

Data source name.

Type: string

driver

A database driver to use.

Type: string

Options: mysql, postgres, pgx, clickhouse, mssql, sqlite, oracle, snowflake, trino, gocosmos, spanner, databricks

max_in_flight

The maximum number of message batches to write in parallel.

Type: int

Default: 64

query

The query to execute.

You must include the correct placeholders for the specified database driver. Some drivers use question marks (?), whereas others expect incrementing dollar signs ($1, $2, and so on) or colons (:1, :2, and so on). The following table shows the placeholder style for each driver:

Driver Placeholder style

clickhouse

Dollar sign ($)

gocosmos

Colon (:)

mssql

Question mark (?)

mysql

Question mark (?)

oracle

Colon (:)

pgx

Dollar sign ($)

postgres

Dollar sign ($)

snowflake

Question mark (?)

spanner

Question mark (?)

sqlite

Question mark (?)

trino

Question mark (?)

Type: string

# Examples:
query: INSERT INTO footable (foo, bar, baz) VALUES (?, ?, ?);