sql: add pgx driver support to missing sql connectors (#4064)

josephwoodward · web-flow · commit b1b639be78b3 · 2026-03-05T13:21:57.000Z
diff --git a/docs/modules/components/pages/caches/sql.adoc b/docs/modules/components/pages/caches/sql.adoc
@@ -114,6 +114,7 @@ A database <<drivers, driver>> to use.
 Options:
 `mysql`
 , `postgres`
+, `pgx`
 , `clickhouse`
 , `mssql`
 , `sqlite`
diff --git a/docs/modules/components/pages/inputs/sql_raw.adoc b/docs/modules/components/pages/inputs/sql_raw.adoc
@@ -119,6 +119,7 @@ A database <<drivers, driver>> to use.
 Options:
 `mysql`
 , `postgres`
+, `pgx`
 , `clickhouse`
 , `mssql`
 , `sqlite`
diff --git a/docs/modules/components/pages/inputs/sql_select.adoc b/docs/modules/components/pages/inputs/sql_select.adoc
@@ -127,6 +127,7 @@ A database <<drivers, driver>> to use.
 Options:
 `mysql`
 , `postgres`
+, `pgx`
 , `clickhouse`
 , `mssql`
 , `sqlite`
diff --git a/docs/modules/components/pages/outputs/sql_insert.adoc b/docs/modules/components/pages/outputs/sql_insert.adoc
@@ -136,6 +136,7 @@ A database <<drivers, driver>> to use.
 Options:
 `mysql`
 , `postgres`
+, `pgx`
 , `clickhouse`
 , `mssql`
 , `sqlite`
diff --git a/docs/modules/components/pages/outputs/sql_raw.adoc b/docs/modules/components/pages/outputs/sql_raw.adoc
@@ -162,6 +162,7 @@ A database <<drivers, driver>> to use.
 Options:
 `mysql`
 , `postgres`
+, `pgx`
 , `clickhouse`
 , `mssql`
 , `sqlite`
diff --git a/docs/modules/components/pages/processors/archive.adoc b/docs/modules/components/pages/processors/archive.adoc
@@ -39,6 +39,8 @@ The resulting archived message adopts the metadata of the _first_ message part o
 
 The functionality of this processor depends on being applied across messages that are batched. You can find out more about batching xref:configuration:batching.adoc[in this doc].
 
+To reverse this process use the xref:components:processors/unarchive.adoc[`unarchive` processor] followed by a xref:components:processors/split.adoc[`split` processor] to process each message individually.
+
 == Fields
 
 === `format`
diff --git a/docs/modules/components/pages/processors/group_by.adoc b/docs/modules/components/pages/processors/group_by.adoc
@@ -35,6 +35,8 @@ Once the groups are established a list of processors are applied to their respec
 
 The functionality of this processor depends on being applied across messages that are batched. You can find out more about batching xref:configuration:batching.adoc[in this doc].
 
+To further divide each group into individual messages, follow this processor with a xref:components:processors/split.adoc[`split` processor].
+
 == Fields
 
 === `[].check`
diff --git a/docs/modules/components/pages/processors/group_by_value.adoc b/docs/modules/components/pages/processors/group_by_value.adoc
@@ -36,6 +36,8 @@ This allows you to group messages using arbitrary fields within their content or
 
 The functionality of this processor depends on being applied across messages that are batched. You can find out more about batching xref:configuration:batching.adoc[in this doc].
 
+To further divide each group into individual messages, follow this processor with a xref:components:processors/split.adoc[`split` processor].
+
 == Fields
 
 === `value`
diff --git a/docs/modules/components/pages/processors/sql_insert.adoc b/docs/modules/components/pages/processors/sql_insert.adoc
@@ -124,6 +124,7 @@ A database <<drivers, driver>> to use.
 Options:
 `mysql`
 , `postgres`
+, `pgx`
 , `clickhouse`
 , `mssql`
 , `sqlite`
diff --git a/docs/modules/components/pages/processors/sql_raw.adoc b/docs/modules/components/pages/processors/sql_raw.adoc
@@ -166,6 +166,7 @@ A database <<drivers, driver>> to use.
 Options:
 `mysql`
 , `postgres`
+, `pgx`
 , `clickhouse`
 , `mssql`
 , `sqlite`
diff --git a/docs/modules/components/pages/processors/sql_select.adoc b/docs/modules/components/pages/processors/sql_select.adoc
@@ -127,6 +127,7 @@ A database <<drivers, driver>> to use.
 Options:
 `mysql`
 , `postgres`
+, `pgx`
 , `clickhouse`
 , `mssql`
 , `sqlite`
diff --git a/docs/modules/components/pages/processors/unarchive.adoc b/docs/modules/components/pages/processors/unarchive.adoc
@@ -34,6 +34,8 @@ unarchive:
 
 When a message is unarchived the new messages replace the original message in the batch. Messages that are selected but fail to unarchive (invalid format) will remain unchanged in the message batch but will be flagged as having failed, allowing you to xref:configuration:error_handling.adoc[error handle them].
 
+Unarchived messages are kept in the same batch. To process each unarchived message individually, follow this processor with a xref:components:processors/split.adoc[`split` processor].
+
 == Metadata
 
 The metadata found on the messages handled by this processor will be copied into the resulting messages. For the unarchive formats that contain file information (tar, zip), a metadata field is also added to each message called `archive_filename` with the extracted filename.
diff --git a/internal/impl/sql/conn_fields.go b/internal/impl/sql/conn_fields.go
@@ -27,7 +27,7 @@ import (
 	"github.com/redpanda-data/benthos/v4/public/service"
 )
 
-var driverField = service.NewStringEnumField("driver", "mysql", "postgres", "clickhouse", "mssql", "sqlite", "oracle", "snowflake", "trino", "gocosmos", "spanner", "databricks").
+var driverField = service.NewStringEnumField("driver", "mysql", "postgres", "pgx", "clickhouse", "mssql", "sqlite", "oracle", "snowflake", "trino", "gocosmos", "spanner", "databricks").
 	Description("A database <<drivers, driver>> to use.")
 
 var dsnField = service.NewStringField("dsn").
diff --git a/internal/impl/sql/integration_test.go b/internal/impl/sql/integration_test.go
@@ -750,7 +750,6 @@ func TestIntegrationPostgres(t *testing.T) {
 		"postgres",
 		"pgx",
 	} {
-		driver := driver
 		t.Run(fmt.Sprintf("driver %s", driver), func(t *testing.T) {
 			var db *sql.DB
 			t.Cleanup(func() {
@@ -828,7 +827,6 @@ func TestIntegrationPostgresVector(t *testing.T) {
 		"postgres",
 		"pgx",
 	} {
-		driver := driver
 		t.Run(fmt.Sprintf("driver %s", driver), func(t *testing.T) {
 			var db *sql.DB
 			t.Cleanup(func() {
