sql_raw

Runs an arbitrary SQL query against a database and (optionally) returns the result as an array of objects, one for each row returned.

Introduced in version 3.65.0.

Common

# Common config fields, showing default values
label: ""
sql_raw:
  driver: ""
  dsn: ""
  query: ""
  args_mapping: ""
  exec_only: false

Advanced

# All config fields, showing default values
label: ""
sql_raw:
  driver: ""
  dsn: ""
  query: ""
  unsafe_dynamic_query: false
  args_mapping: ""
  exec_only: false

If the query fails to execute then the message will remain unchanged and the error can be caught using error handling methods outlined here.

Examples

Table Insert (MySQL)

The following example inserts rows into the table footable with the columns foo, bar and baz populated with values extracted from messages.

pipeline:
  processors:
    - sql_raw:
        driver: mysql
        dsn: foouser:foopassword@tcp(localhost:3306)/foodb
        query: "INSERT INTO footable (foo, bar, baz) VALUES (?, ?, ?);"
        args_mapping: '[ document.foo, document.bar, meta("kafka_topic") ]'
        exec_only: true

Table Query (PostgreSQL)

Here we query a database for columns of footable that share a user_id with the message field user.id. A branch processor is used in order to insert the resulting array into the original message at the path foo_rows.

pipeline:
  processors:
    - branch:
        processors:
          - sql_raw:
              driver: postgres
              dsn: postgres://foouser:foopass@localhost:5432/testdb?sslmode=disable
              query: "SELECT * FROM footable WHERE user_id = $1;"
              args_mapping: '[ this.user.id ]'
        result_map: 'root.foo_rows = this'

Fields

driver

A database driver to use.

Type: string
Options: mysql, postgres, clickhouse, mssql.

dsn

A Data Source Name to identify the target database.

Drivers

The following is a list of supported drivers and their respective DSN formats:

Driver	Data Source Name Format
clickhouse	tcp://[netloc][:port][?param1=value1&...&paramN=valueN]
mysql	[username[:password]@][protocol[(address)]]/dbname[?param1=value1&...&paramN=valueN]
postgres	postgres://[user[:password]@][netloc][:port][/dbname][?param1=value1&...]
mssql	sqlserver://[user[:password]@][netloc][:port][?database=dbname&param1=value1&...]

Please note that the postgres driver enforces SSL by default, you can override this with the parameter sslmode=disable if required.

Type: string

# Examples

dsn: tcp://host1:9000?username=user&password=qwerty&database=clicks&read_timeout=10&write_timeout=20&alt_hosts=host2:9000,host3:9000

dsn: foouser:foopassword@tcp(localhost:3306)/foodb

dsn: postgres://foouser:foopass@localhost:5432/foodb?sslmode=disable

query

The query to execute.

Type: string

# Examples

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

unsafe_dynamic_query

Whether to enable interpolation functions in the query. Great care should be made to ensure your queries are defended against injection attacks.

Type: bool
Default: false

args_mapping

An optional Bloblang mapping which should evaluate to an array of values matching in size to the number of placeholder arguments in the field query.

Type: string

# Examples

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

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

exec_only

Whether the query result should be discarded. When set to true the message contents will remain unchanged, which is useful in cases where you are executing inserts, updates, etc.

Type: bool
Default: false