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

Author: rockwotj

Diff for: 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`

Diff for: docs/modules/components/pages/outputs/sql_raw.adoc

@@ -115,6 +115,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() ]
+
+```
+
 --
 ======
 
@@ -260,7 +289,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`

Diff for: docs/modules/components/pages/processors/sql_raw.adoc

@@ -42,7 +42,8 @@ sql_raw:
   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)
   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)
 ```
 
 --
@@ -59,7 +60,8 @@ sql_raw:
   query: INSERT INTO footable (foo, bar, baz) VALUES (?, ?, ?); # No default (required)
   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() ]
+```
+
 --
 ======
 
@@ -272,7 +302,59 @@ Whether the query result should be discarded. When set to `true` the message con
 
 *Type*: `bool`
 
-*Default*: `false`
+
+=== `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 true.
+
+
+*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.
+
+
+*Type*: `bool`
+
+*Default*: `true`
 
 === `init_files`
 

Diff for: 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

Diff for: 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() {