From c3316fcec66151f3db0074f41f93c662b394d7c2 Mon Sep 17 00:00:00 2001 From: Vaughn Woerpel Date: Fri, 9 Oct 2026 13:16:48 -0400 Subject: [PATCH 1/3] feat(postgres on conflict): Add on-conflict to support to the postgres sink config --- .../postgres_sink_on_conflict.enhancement.md | 3 ++ src/sinks/postgres/config.rs | 50 ++++++++++++++++-- src/sinks/postgres/integration_tests.rs | 52 +++++++++++++++++++ src/sinks/postgres/service.rs | 31 +++++++++-- .../components/sinks/generated/postgres.cue | 22 +++++++- .../sinks/postgres/advanced.yaml | 1 + 6 files changed, 151 insertions(+), 8 deletions(-) create mode 100644 changelog.d/postgres_sink_on_conflict.enhancement.md diff --git a/changelog.d/postgres_sink_on_conflict.enhancement.md b/changelog.d/postgres_sink_on_conflict.enhancement.md new file mode 100644 index 0000000000000..49588389dd9d7 --- /dev/null +++ b/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 diff --git a/src/sinks/postgres/config.rs b/src/sinks/postgres/config.rs index 83590b7a6b74b..8bc574f36acac 100644 --- a/src/sinks/postgres/config.rs +++ b/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, + #[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::(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::(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; diff --git a/src/sinks/postgres/integration_tests.rs b/src/sinks/postgres/integration_tests.rs index cc8ef14a0519f..578611e53558f 100644 --- a/src/sinks/postgres/integration_tests.rs +++ b/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::(&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)> = 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())), + ] + ); +} diff --git a/src/sinks/postgres/service.rs b/src/sinks/postgres/service.rs index f33ba3774f167..b23a63f0d7dfc 100644 --- a/src/sinks/postgres/service.rs +++ b/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, table: String, endpoint: String, + on_conflict: OnConflict, } impl PostgresService { - pub const fn new(connection_pool: Pool, table: String, endpoint: String) -> Self { + pub const fn new( + connection_pool: Pool, + table: String, + endpoint: String, + on_conflict: OnConflict, + ) -> Self { Self { connection_pool, table, endpoint, + on_conflict, } } } @@ -143,6 +152,7 @@ impl Service 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 for PostgresService { .collect::, _>>() .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:: { + count: skipped, + reason: "Row conflicts with an existing row.", + }); + } + } + emit!(EndpointBytesSent { byte_size: metadata.request_encoded_size(), protocol: POSTGRES_PROTOCOL, diff --git a/website/cue/reference/components/sinks/generated/postgres.cue b/website/cue/reference/components/sinks/generated/postgres.cue index 17c272a604168..77248d2197fb3 100644 --- a/website/cue/reference/components/sinks/generated/postgres.cue +++ b/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 diff --git a/website/generated/example-configs/sinks/postgres/advanced.yaml b/website/generated/example-configs/sinks/postgres/advanced.yaml index 300611f4d8d33..9ffd2cbe93425 100644 --- a/website/generated/example-configs/sinks/postgres/advanced.yaml +++ b/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 From fd4e5502c74db38d36c17718befde588346ff972 Mon Sep 17 00:00:00 2001 From: Vaughn Woerpel Date: Fri, 9 Oct 2026 22:31:19 -0400 Subject: [PATCH 2/3] fix(docs): fix docs to address deferred --- src/sinks/postgres/config.rs | 3 +++ src/sinks/postgres/service.rs | 3 ++- website/cue/reference/components/sinks/generated/postgres.cue | 3 +++ 3 files changed, 8 insertions(+), 1 deletion(-) diff --git a/src/sinks/postgres/config.rs b/src/sinks/postgres/config.rs index 8bc574f36acac..6a2ad20b24568 100644 --- a/src/sinks/postgres/config.rs +++ b/src/sinks/postgres/config.rs @@ -45,6 +45,9 @@ pub enum OnConflict { /// /// This appends `ON CONFLICT DO NOTHING` to the insert statement. Skipped rows are reported /// as intentionally discarded events. + /// + /// PostgreSQL rejects this statement on a table that has a `DEFERRABLE` unique or exclusion + /// constraint, even when no row conflicts. Keep the default for such tables. DoNothing, } diff --git a/src/sinks/postgres/service.rs b/src/sinks/postgres/service.rs index b23a63f0d7dfc..a2fd7841dc0bd 100644 --- a/src/sinks/postgres/service.rs +++ b/src/sinks/postgres/service.rs @@ -175,7 +175,8 @@ impl Service for PostgresService { .context(PostgresSnafu)?; if service.on_conflict == OnConflict::DoNothing { - let skipped = event_count.saturating_sub(result.rows_affected() as usize); + let inserted = usize::try_from(result.rows_affected()).unwrap_or(usize::MAX); + let skipped = event_count.saturating_sub(inserted); if skipped > 0 { emit!(ComponentEventsDropped:: { count: skipped, diff --git a/website/cue/reference/components/sinks/generated/postgres.cue b/website/cue/reference/components/sinks/generated/postgres.cue index 77248d2197fb3..e61bbfa28388c 100644 --- a/website/cue/reference/components/sinks/generated/postgres.cue +++ b/website/cue/reference/components/sinks/generated/postgres.cue @@ -50,6 +50,9 @@ generated: components: sinks: postgres: configuration: { This appends `ON CONFLICT DO NOTHING` to the insert statement. Skipped rows are reported as intentionally discarded events. + + PostgreSQL rejects this statement on a table that has a `DEFERRABLE` unique or exclusion + constraint, even when no row conflicts. Keep the default behavior for tables with this constraint. """ error: "Fail the whole batch when any row conflicts with an existing row." } From 13babd6b87e6dd02f817ca7052b258008a072590 Mon Sep 17 00:00:00 2001 From: Vaughn Woerpel Date: Fri, 9 Oct 2026 22:40:42 -0400 Subject: [PATCH 3/3] fix: cue --- website/cue/reference/components/sinks/generated/postgres.cue | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/website/cue/reference/components/sinks/generated/postgres.cue b/website/cue/reference/components/sinks/generated/postgres.cue index e61bbfa28388c..d8d66720000d9 100644 --- a/website/cue/reference/components/sinks/generated/postgres.cue +++ b/website/cue/reference/components/sinks/generated/postgres.cue @@ -52,7 +52,7 @@ generated: components: sinks: postgres: configuration: { as intentionally discarded events. PostgreSQL rejects this statement on a table that has a `DEFERRABLE` unique or exclusion - constraint, even when no row conflicts. Keep the default behavior for tables with this constraint. + constraint, even when no row conflicts. Keep the default for such tables. """ error: "Fail the whole batch when any row conflicts with an existing row." }