sql: add the ability to execute multiple statements in sql_raw plugins

Author: rockwotj

docs/modules/components/pages/inputs/sql_raw.adoc

@@ -223,7 +223,7 @@ query: SELECT * FROM footable WHERE user_id = $1;
 
 === `args_mapping`
 
-A xref:guides:bloblang/about.adoc[Bloblang mapping] which should evaluate to an array of values matching in size to the number of columns specified.
+An optional xref:guides:bloblang/about.adoc[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`

docs/modules/components/pages/outputs/sql_raw.adoc

@@ -41,8 +41,9 @@ output:
   sql_raw:
     driver: "" # No default (required)
     dsn: clickhouse://username:password@host1:9000,host2:9000/database?dial_timeout=200ms&max_execution_time=60 # No default (required)
-    query: INSERT INTO footable (foo, bar, baz) VALUES (?, ?, ?); # No default (required)
+    query: INSERT INTO footable (foo, bar, baz) VALUES (?, ?, ?); # No default (optional)
     args_mapping: root = [ this.cat.meow, this.doc.woofs[0] ] # No default (optional)
+    queries: [] # No default (optional)
     max_in_flight: 64
     batching:
       count: 0
@@ -63,9 +64,10 @@ output:
   sql_raw:
     driver: "" # No default (required)
     dsn: clickhouse://username:password@host1:9000,host2:9000/database?dial_timeout=200ms&max_execution_time=60 # No default (required)
-    query: INSERT INTO footable (foo, bar, baz) VALUES (?, ?, ?); # No default (required)
+    query: INSERT INTO footable (foo, bar, baz) VALUES (?, ?, ?); # No default (optional)
     unsafe_dynamic_query: false
     args_mapping: root = [ this.cat.meow, this.doc.woofs[0] ] # No default (optional)
+    queries: [] # No default (optional)
     max_in_flight: 64
     init_files: [] # No default (optional)
     init_statement: | # No default (optional)
@@ -115,6 +117,35 @@ output:
       ]
 ```
 
+--
+Dynamically Creating Tables (PostgreSQL)::
++
+--
+
+Here we dynamically create output tables transactionally with inserting a record into the newly created table.
+
+```yaml
+output:
+  processors:
+    - mapping: |
+        root = this
+        # Prevent SQL injection when using unsafe_dynamic_query
+        meta table_name = "\"" + metadata("table_name").replace_all("\"", "\"\"") + "\""
+  sql_raw:
+    driver: postgres
+    dsn: postgres://localhost/postgres
+    unsafe_dynamic_query: true
+    queries:
+      - query: |
+          CREATE TABLE IF NOT EXISTS ${!metadata("table_name")} (id varchar primary key, document jsonb);
+      - query: |
+          INSERT INTO ${!metadata("table_name")} (id, document) VALUES ($1, $2)
+          ON CONFLICT (id) DO UPDATE SET document = EXCLUDED.document;
+        args_mapping: |
+          root = [ this.id, this.document.string() ]
+
+```
+
 --
 ======
 
@@ -250,6 +281,50 @@ An optional xref:guides:bloblang/about.adoc[Bloblang mapping] which should evalu
 *Type*: `string`
 
 
+```yml
+# Examples
+
+args_mapping: root = [ this.cat.meow, this.doc.woofs[0] ]
+
+args_mapping: root = [ meta("user.id") ]
+```
+
+=== `queries`
+
+A list of statements to run in addition to `query`. When specifying multiple statements, they are all executed within a transaction.
+
+
+*Type*: `array`
+
+
+=== `queries[].query`
+
+The query to execute. The style of placeholder to use depends on the driver, some drivers require question marks (`?`) whereas others expect incrementing dollar signs (`$1`, `$2`, and so on) or colons (`:1`, `:2` and so on). The style to use is outlined in this table:
+
+| Driver | Placeholder Style |
+|---|---|
+| `clickhouse` | Dollar sign |
+| `mysql` | Question mark |
+| `postgres` | Dollar sign |
+| `mssql` | Question mark |
+| `sqlite` | Question mark |
+| `oracle` | Colon |
+| `snowflake` | Question mark |
+| `trino` | Question mark |
+| `gocosmos` | Colon |
+
+
+*Type*: `string`
+
+
+=== `queries[].args_mapping`
+
+An optional xref:guides:bloblang/about.adoc[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`
+
+
 ```yml
 # Examples
 
@@ -260,7 +335,7 @@ args_mapping: root = [ meta("user.id") ]
 
 === `max_in_flight`
 
-The maximum number of inserts to run in parallel.
+The maximum number of statements to execute in parallel.
 
 
 *Type*: `int`

docs/modules/components/pages/processors/sql_raw.adoc

@@ -40,9 +40,10 @@ label: ""
 sql_raw:
   driver: "" # No default (required)
   dsn: clickhouse://username:password@host1:9000,host2:9000/database?dial_timeout=200ms&max_execution_time=60 # No default (required)
-  query: INSERT INTO footable (foo, bar, baz) VALUES (?, ?, ?); # No default (required)
+  query: INSERT INTO footable (foo, bar, baz) VALUES (?, ?, ?); # No default (optional)
   args_mapping: root = [ this.cat.meow, this.doc.woofs[0] ] # No default (optional)
-  exec_only: false
+  exec_only: false # No default (optional)
+  queries: [] # No default (optional)
 ```
 
 --
@@ -56,10 +57,11 @@ label: ""
 sql_raw:
   driver: "" # No default (required)
   dsn: clickhouse://username:password@host1:9000,host2:9000/database?dial_timeout=200ms&max_execution_time=60 # No default (required)
-  query: INSERT INTO footable (foo, bar, baz) VALUES (?, ?, ?); # No default (required)
+  query: INSERT INTO footable (foo, bar, baz) VALUES (?, ?, ?); # No default (optional)
   unsafe_dynamic_query: false
   args_mapping: root = [ this.cat.meow, this.doc.woofs[0] ] # No default (optional)
-  exec_only: false
+  exec_only: false # No default (optional)
+  queries: [] # No default (optional)
   init_files: [] # No default (optional)
   init_statement: | # No default (optional)
     CREATE TABLE IF NOT EXISTS some_table (
@@ -120,6 +122,34 @@ pipeline:
         result_map: 'root.foo_rows = this'
 ```
 
+--
+Dynamically Creating Tables (PostgreSQL)::
++
+--
+
+Here we query a database for columns of footable that share a `user_id` with the message field `user.id`. A xref:components:processors/branch.adoc[`branch` processor] is used in order to insert the resulting array into the original message at the path `foo_rows`.
+
+```yaml
+pipeline:
+  processors:
+    - mapping: |
+        root = this
+        # Prevent SQL injection when using unsafe_dynamic_query
+        meta table_name = "\"" + metadata("table_name").replace_all("\"", "\"\"") + "\""
+    - sql_raw:
+      driver: postgres
+      dsn: postgres://localhost/postgres
+      unsafe_dynamic_query: true
+      queries:
+        - query: |
+            CREATE TABLE IF NOT EXISTS ${!metadata("table_name")} (id varchar primary key, document jsonb);
+        - query: |
+            INSERT INTO ${!metadata("table_name")} (id, document) VALUES ($1, $2)
+            ON CONFLICT (id) DO UPDATE SET document = EXCLUDED.document;
+          args_mapping: |
+            root = [ this.id, this.document.string() ]
+```
+
 --
 ======
 
@@ -267,12 +297,63 @@ 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.
+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. By default this is true for the last query, and previous queries don't change the results. If set to true for any query but the last one, the subsequent `args_mappings` input is overwritten.
+
+
+*Type*: `bool`
+
+
+=== `queries`
+
+A list of statements to run in addition to `query`. When specifying multiple statements, they are all executed within a transaction. The output of the processor is always the last query that runs, unless `exec_only` is used.
+
+
+*Type*: `array`
+
+
+=== `queries[].query`
+
+The query to execute. The style of placeholder to use depends on the driver, some drivers require question marks (`?`) whereas others expect incrementing dollar signs (`$1`, `$2`, and so on) or colons (`:1`, `:2` and so on). The style to use is outlined in this table:
+
+| Driver | Placeholder Style |
+|---|---|
+| `clickhouse` | Dollar sign |
+| `mysql` | Question mark |
+| `postgres` | Dollar sign |
+| `mssql` | Question mark |
+| `sqlite` | Question mark |
+| `oracle` | Colon |
+| `snowflake` | Question mark |
+| `trino` | Question mark |
+| `gocosmos` | Colon |
+
+
+*Type*: `string`
+
+
+=== `queries[].args_mapping`
+
+An optional xref:guides:bloblang/about.adoc[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`
+
+
+```yml
+# Examples
+
+args_mapping: root = [ this.cat.meow, this.doc.woofs[0] ]
+
+args_mapping: root = [ meta("user.id") ]
+```
+
+=== `queries[].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. By default this is true for the last query, and previous queries don't change the results. If set to true for any query but the last one, the subsequent `args_mappings` input is overwritten.
 
 
 *Type*: `bool`
 
-*Default*: `false`
 
 === `init_files`
 

internal/impl/sql/conn_fields.go

@@ -23,6 +23,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/redpanda-data/benthos/v4/public/bloblang"
 	"github.com/redpanda-data/benthos/v4/public/service"
 )
 
@@ -136,6 +137,14 @@ CREATE TABLE IF NOT EXISTS some_table (
 	}
 }
 
+type rawQueryStatement struct {
+	static  string
+	dynamic *service.InterpolatedString
+
+	argsMapping *bloblang.Executor // optional
+	execOnly    bool
+}
+
 func rawQueryField() *service.ConfigField {
 	return service.NewStringField("query").
 		Description("The query to execute. The style of placeholder to use depends on the driver, some drivers require question marks (`?`) whereas others expect incrementing dollar signs (`$1`, `$2`, and so on) or colons (`:1`, `:2` and so on). The style to use is outlined in this table:" + `
@@ -154,6 +163,14 @@ func rawQueryField() *service.ConfigField {
 `)
 }
 
+func rawQueryArgsMappingField() *service.ConfigField {
+	return service.NewBloblangField("args_mapping").
+		Description("An optional xref:guides:bloblang/about.adoc[Bloblang mapping] which should evaluate to an array of values matching in size to the number of placeholder arguments in the field `query`.").
+		Example("root = [ this.cat.meow, this.doc.woofs[0] ]").
+		Example(`root = [ meta("user.id") ]`).
+		Optional()
+}
+
 type connSettings struct {
 	connMaxLifetime time.Duration
 	connMaxIdleTime time.Duration

internal/impl/sql/input_sql_raw.go

@@ -27,7 +27,7 @@ import (
 )
 
 func sqlRawInputConfig() *service.ConfigSpec {
-	spec := service.NewConfigSpec().
+	return service.NewConfigSpec().
 		Beta().
 		Categories("Services").
 		Summary("Executes a select query and creates a message for each row received.").
@@ -36,17 +36,9 @@ func sqlRawInputConfig() *service.ConfigSpec {
 		Field(dsnField).
 		Field(rawQueryField().
 			Example("SELECT * FROM footable WHERE user_id = $1;")).
-		Field(service.NewBloblangField("args_mapping").
-			Description("A xref:guides:bloblang/about.adoc[Bloblang mapping] which should evaluate to an array of values matching in size to the number of columns specified.").
-			Example("root = [ this.cat.meow, this.doc.woofs[0] ]").
-			Example(`root = [ meta("user.id") ]`).
-			Optional()).
-		Field(service.NewAutoRetryNacksToggleField())
-	for _, f := range connFields() {
-		spec = spec.Field(f)
-	}
-
-	spec = spec.
+		Field(rawQueryArgsMappingField()).
+		Field(service.NewAutoRetryNacksToggleField()).
+		Fields(connFields()...).
 		Version("4.10.0").
 		Example("Consumes an SQL table using a query as an input.",
 			`
@@ -63,7 +55,6 @@ input:
       ]
 `,
 		)
-	return spec
 }
 
 func init() {