feat(sinks): Add on-conflict config support to postgres sink

changelog.d/postgres_sink_on_conflict.enhancement.md

@@ -0,0 +1,3 @@
+Add an `on_conflict` option to the `postgres` sink. Setting this option to `do_nothing` appends `ON CONFLICT DO NOTHING` to the insert, so rows that violate a unique constraint are skipped instead of failing the whole batch. Skipped rows are reported as intentionally discarded events. The default, `error`, keeps the existing behavior.
+
+authors: vaughnw128

src/sinks/postgres/config.rs

@@ -32,6 +32,22 @@ const fn default_pool_size() -> u32 {
     5
 }
 
+/// How the sink handles rows that violate a unique or exclusion constraint on insert.
+#[configurable_component]
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
+#[serde(rename_all = "snake_case")]
+pub enum OnConflict {
+    /// Fail the whole batch when any row conflicts with an existing row.
+    #[default]
+    Error,
+
+    /// Skip rows that conflict with an existing row and insert the rest of the batch.
+    ///
+    /// This appends `ON CONFLICT DO NOTHING` to the insert statement. Skipped rows are reported
+    /// as intentionally discarded events.
+    DoNothing,
+}
+
 /// Configuration for the `postgres` sink.
 #[configurable_component(sink("postgres", "Deliver log data to a PostgreSQL database."))]
 #[derive(Clone, Default, Debug)]
@@ -59,13 +75,18 @@ pub struct PostgresConfig {
     /// a single event in the batch can make the whole batch to fail. For example, if a single event within the batch triggers
     /// a unique constraint violation in the destination table, the whole event batch will fail.
     ///
-    /// As a workaround, [triggers](https://www.postgresql.org/docs/current/sql-createtrigger.html) on constraint violations
-    /// can be defined at a database level to change the behavior of the insert operation on specific tables.
+    /// To skip rows that violate a unique constraint without failing the batch, set `on_conflict` to `do_nothing`.
+    ///
+    /// For other kinds of constraint violations, [triggers](https://www.postgresql.org/docs/current/sql-createtrigger.html)
+    /// on constraint violations can be defined at a database level to change the behavior of the insert operation on specific tables.
     /// Alternatively, setting `max_events` batch setting to `1` will make each event to be inserted independently,
     /// so events that trigger a constraint violation will not affect the rest of the events.
     #[serde(default)]
     pub batch: BatchConfig<RealtimeSizeBasedDefaultBatchSettings>,
 
+    #[serde(default)]
+    pub on_conflict: OnConflict,
+
     #[serde(default)]
     pub request: TowerRequestConfig,
 
@@ -199,7 +220,12 @@ impl ValidatedSink for PostgresConfig {
 
         // The endpoint label must not carry credentials or query parameters.
         let endpoint = protocol_endpoint(endpoint_uri.uri).1;
-        let service = PostgresService::new(connection_pool, self.table.clone(), endpoint);
+        let service = PostgresService::new(
+            connection_pool,
+            self.table.clone(),
+            endpoint,
+            self.on_conflict,
+        );
         let service = ServiceBuilder::new()
             .settings(request_settings, PostgresRetryLogic)
             .service(service);
@@ -235,6 +261,24 @@ mod tests {
         assert_eq!(cfg.table, "mytable");
     }
 
+    #[test]
+    fn parse_on_conflict() {
+        let default_cfg = serde_yaml::from_str::<PostgresConfig>(indoc::indoc! {r#"
+            endpoint: "postgres://user:password@localhost/default"
+            table: "mytable"
+        "#})
+        .unwrap();
+        assert_eq!(default_cfg.on_conflict, OnConflict::Error);
+
+        let cfg = serde_yaml::from_str::<PostgresConfig>(indoc::indoc! {r#"
+            endpoint: "postgres://user:password@localhost/default"
+            table: "mytable"
+            on_conflict: do_nothing
+        "#})
+        .unwrap();
+        assert_eq!(cfg.on_conflict, OnConflict::DoNothing);
+    }
+
     #[test]
     fn validate_produces_usable_values() {
         use crate::config::ValidatedSink;

src/sinks/postgres/integration_tests.rs

@@ -514,3 +514,55 @@ async fn insertion_fails_primary_key_violation() {
     )
     .await;
 }
+
+#[tokio::test]
+async fn insertion_skips_conflicting_rows_with_on_conflict_do_nothing() {
+    trace_init();
+
+    let table = random_table_name();
+    let endpoint = pg_url();
+    let config_str = format!(
+        r#"
+            endpoint = "{endpoint}"
+            table = "{table}"
+            on_conflict = "do_nothing"
+        "#,
+    );
+    let (config, _) = load_sink::<PostgresConfig>(&config_str).unwrap();
+    let mut connection = PgConnection::connect(endpoint.as_str())
+        .await
+        .expect("Failed to connect to Postgres");
+    let (sink, _hc) = config.build(SinkContext::default()).await.unwrap();
+    let create_table_sql = format!(
+        "CREATE TABLE {table} (id BIGINT PRIMARY KEY, host TEXT, timestamp TIMESTAMPTZ, message TEXT, payload JSONB)"
+    );
+    sqlx::query(&create_table_sql)
+        .execute(&mut connection)
+        .await
+        .unwrap();
+    let insert_existing_sql = format!("INSERT INTO {table} (id, host) VALUES (1, 'existing')");
+    sqlx::query(&insert_existing_sql)
+        .execute(&mut connection)
+        .await
+        .unwrap();
+
+    let mut events = [0, 1, 2, 2].map(create_event).to_vec();
+    let mut receiver = BatchNotifier::apply_to(&mut events);
+
+    run_and_assert_sink_compliance(sink, stream::iter(events), &POSTGRES_SINK_TAGS).await;
+    assert_eq!(receiver.try_recv(), Ok(BatchStatus::Delivered));
+
+    let select_sql = format!("SELECT id, host FROM {table} ORDER BY id");
+    let rows: Vec<(i64, Option<String>)> = sqlx::query_as(&select_sql)
+        .fetch_all(&mut connection)
+        .await
+        .unwrap();
+    assert_eq!(
+        rows,
+        vec![
+            (0, Some("example.com".to_owned())),
+            (1, Some("existing".to_owned())),
+            (2, Some("example.com".to_owned())),
+        ]
+    );
+}

src/sinks/postgres/service.rs

@@ -11,10 +11,12 @@ use vector_lib::{
     EstimatedJsonEncodedSizeOf,
     codecs::JsonSerializerConfig,
     event::{Event, EventFinalizers, EventStatus, Finalizable},
+    internal_event::{ComponentEventsDropped, INTENTIONAL},
     request_metadata::{GroupedCountByteSize, MetaDescriptive, RequestMetadata},
     stream::DriverResponse,
 };
 
+use super::config::OnConflict;
 use crate::{
     internal_events::EndpointBytesSent,
     sinks::prelude::{RequestMetadataBuilder, RetryLogic},
@@ -50,14 +52,21 @@ pub struct PostgresService {
     connection_pool: Pool<Postgres>,
     table: String,
     endpoint: String,
+    on_conflict: OnConflict,
 }
 
 impl PostgresService {
-    pub const fn new(connection_pool: Pool<Postgres>, table: String, endpoint: String) -> Self {
+    pub const fn new(
+        connection_pool: Pool<Postgres>,
+        table: String,
+        endpoint: String,
+        on_conflict: OnConflict,
+    ) -> Self {
         Self {
             connection_pool,
             table,
             endpoint,
+            on_conflict,
         }
     }
 }
@@ -143,6 +152,7 @@ impl Service<PostgresRequest> for PostgresService {
         let future = async move {
             let table = service.table;
             let metadata = request.metadata;
+            let event_count = request.events.len();
             let json_serializer = JsonSerializerConfig::default().build();
             let serialized_values = request
                 .events
@@ -151,14 +161,29 @@ impl Service<PostgresRequest> for PostgresService {
                 .collect::<Result<Vec<_>, _>>()
                 .context(VectorCommonSnafu)?;
 
-            sqlx::query(&format!(
-                "INSERT INTO {table} SELECT * FROM jsonb_populate_recordset(NULL::{table}, $1)"
+            let conflict_clause = match service.on_conflict {
+                OnConflict::Error => "",
+                OnConflict::DoNothing => " ON CONFLICT DO NOTHING",
+            };
+
+            let result = sqlx::query(&format!(
+                "INSERT INTO {table} SELECT * FROM jsonb_populate_recordset(NULL::{table}, $1){conflict_clause}"
             ))
             .bind(Json(serialized_values))
             .execute(&service.connection_pool)
             .await
             .context(PostgresSnafu)?;
 
+            if service.on_conflict == OnConflict::DoNothing {
+                let skipped = event_count.saturating_sub(result.rows_affected() as usize);
+                if skipped > 0 {
+                    emit!(ComponentEventsDropped::<INTENTIONAL> {
+                        count: skipped,
+                        reason: "Row conflicts with an existing row.",
+                    });
+                }
+            }
+
             emit!(EndpointBytesSent {
                 byte_size: metadata.request_encoded_size(),
                 protocol: POSTGRES_PROTOCOL,

website/cue/reference/components/sinks/generated/postgres.cue

@@ -20,8 +20,10 @@ generated: components: sinks: postgres: configuration: {
 			a single event in the batch can make the whole batch to fail. For example, if a single event within the batch triggers
 			a unique constraint violation in the destination table, the whole event batch will fail.
 
-			As a workaround, [triggers](https://www.postgresql.org/docs/current/sql-createtrigger.html) on constraint violations
-			can be defined at a database level to change the behavior of the insert operation on specific tables.
+			To skip rows that violate a unique constraint without failing the batch, set `on_conflict` to `do_nothing`.
+
+			For other kinds of constraint violations, [triggers](https://www.postgresql.org/docs/current/sql-createtrigger.html)
+			on constraint violations can be defined at a database level to change the behavior of the insert operation on specific tables.
 			Alternatively, setting `max_events` batch setting to `1` will make each event to be inserted independently,
 			so events that trigger a constraint violation will not affect the rest of the events.
 			"""
@@ -37,6 +39,22 @@ generated: components: sinks: postgres: configuration: {
 		required: true
 		type: string: {}
 	}
+	on_conflict: {
+		description: "How the sink handles rows that violate a unique or exclusion constraint on insert."
+		required:    false
+		type: string: {
+			default: "error"
+			enum: {
+				do_nothing: """
+					Skip rows that conflict with an existing row and insert the rest of the batch.
+
+					This appends `ON CONFLICT DO NOTHING` to the insert statement. Skipped rows are reported
+					as intentionally discarded events.
+					"""
+				error: "Fail the whole batch when any row conflicts with an existing row."
+			}
+		}
+	}
 	pool_size: {
 		description: """
 			The postgres connection pool size. See [this](https://docs.rs/sqlx/latest/sqlx/struct.Pool.html#why-use-a-pool) for more

website/generated/example-configs/sinks/postgres/advanced.yaml

@@ -4,5 +4,6 @@ sinks:
     inputs:
       - my-source-or-transform-id
     endpoint: postgres://localhost:5432
+    on_conflict: error
     pool_size: 5
     table: my_table