From deaba8a5e412581eb939db0c1a952c751ea64816 Mon Sep 17 00:00:00 2001 From: Mina Karam Date: Thu, 8 Oct 2026 04:31:02 +0000 Subject: [PATCH] fix(migration): conflict on the target's key when copying to postgres (#15009) Co-authored-by: Claude Opus 5.5 Co-authored-by: hengfeiyang --- src/migration/adapter/postgres.rs | 127 +++++++++++++++++++++--------- 1 file changed, 89 insertions(+), 38 deletions(-) diff --git a/src/migration/adapter/postgres.rs b/src/migration/adapter/postgres.rs index 771a5f4269..2e0b685cb1 100644 --- a/src/migration/adapter/postgres.rs +++ b/src/migration/adapter/postgres.rs @@ -240,7 +240,7 @@ impl DbAdapter for PostgresAdapter { &self, table: &str, columns: &[String], - primary_keys: &[String], + _primary_keys: &[String], rows: &[Row], ) -> Result { if rows.is_empty() { @@ -250,43 +250,9 @@ impl DbAdapter for PostgresAdapter { // Get column info for proper null type binding let column_infos = self.get_columns(table).await?; - let cols = columns - .iter() - .map(|c| format!("\"{}\"", c)) - .collect::>() - .join(", "); - - let placeholders = (1..=columns.len()) - .map(|i| format!("${}", i)) - .collect::>() - .join(", "); - - let pk_cols = primary_keys - .iter() - .map(|c| format!("\"{}\"", c)) - .collect::>() - .join(", "); - - // Build DO UPDATE SET clause - let updates = columns - .iter() - .filter(|c| !primary_keys.contains(c)) - .map(|c| format!("\"{}\" = EXCLUDED.\"{}\"", c, c)) - .collect::>() - .join(", "); - - // Use OVERRIDING SYSTEM VALUE to allow inserting into IDENTITY columns - let sql = if updates.is_empty() { - format!( - "INSERT INTO \"{}\" ({}) OVERRIDING SYSTEM VALUE VALUES ({}) ON CONFLICT ({}) DO NOTHING", - table, cols, placeholders, pk_cols - ) - } else { - format!( - "INSERT INTO \"{}\" ({}) OVERRIDING SYSTEM VALUE VALUES ({}) ON CONFLICT ({}) DO UPDATE SET {}", - table, cols, placeholders, pk_cols, updates - ) - }; + // Conflict on the target's key: partitioned file_list tables have none, unlike SQLite's id + let target_keys = self.get_primary_keys(table).await?; + let sql = build_upsert_sql(table, columns, &target_keys); let mut count = 0u64; for row in rows { @@ -414,3 +380,88 @@ impl DbAdapter for PostgresAdapter { Ok(()) } } + +/// Per-row INSERT keeping source IDENTITY values; with no target key, conflicting rows are skipped. +fn build_upsert_sql(table: &str, columns: &[String], primary_keys: &[String]) -> String { + let cols = columns + .iter() + .map(|c| format!("\"{}\"", c)) + .collect::>() + .join(", "); + + let placeholders = (1..=columns.len()) + .map(|i| format!("${}", i)) + .collect::>() + .join(", "); + + let insert = + format!("INSERT INTO \"{table}\" ({cols}) OVERRIDING SYSTEM VALUE VALUES ({placeholders})"); + + if primary_keys.is_empty() { + return format!("{insert} ON CONFLICT DO NOTHING"); + } + + let pk_cols = primary_keys + .iter() + .map(|c| format!("\"{}\"", c)) + .collect::>() + .join(", "); + + let updates = columns + .iter() + .filter(|c| !primary_keys.contains(c)) + .map(|c| format!("\"{}\" = EXCLUDED.\"{}\"", c, c)) + .collect::>() + .join(", "); + + if updates.is_empty() { + format!("{insert} ON CONFLICT ({pk_cols}) DO NOTHING") + } else { + format!("{insert} ON CONFLICT ({pk_cols}) DO UPDATE SET {updates}") + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn cols(names: &[&str]) -> Vec { + names.iter().map(|s| s.to_string()).collect() + } + + #[test] + fn upsert_updates_non_key_columns_on_key_conflict() { + let sql = build_upsert_sql("meta", &cols(&["id", "key1", "value"]), &cols(&["id"])); + assert_eq!( + sql, + "INSERT INTO \"meta\" (\"id\", \"key1\", \"value\") OVERRIDING SYSTEM VALUE \ + VALUES ($1, $2, $3) ON CONFLICT (\"id\") DO UPDATE SET \ + \"key1\" = EXCLUDED.\"key1\", \"value\" = EXCLUDED.\"value\"" + ); + } + + #[test] + fn upsert_does_nothing_when_every_column_is_key() { + let sql = build_upsert_sql( + "org_users", + &cols(&["org", "email"]), + &cols(&["org", "email"]), + ); + assert!( + sql.ends_with("ON CONFLICT (\"org\", \"email\") DO NOTHING"), + "{sql}" + ); + } + + #[test] + fn upsert_without_target_key_skips_any_unique_conflict() { + // Partitioned file_list in Postgres has no primary key. + let sql = build_upsert_sql("file_list", &cols(&["id", "stream", "date", "file"]), &[]); + assert_eq!( + sql, + "INSERT INTO \"file_list\" (\"id\", \"stream\", \"date\", \"file\") OVERRIDING \ + SYSTEM VALUE VALUES ($1, $2, $3, $4) ON CONFLICT DO NOTHING" + ); + assert!(!sql.contains("ON CONFLICT ("), "{sql}"); + } +}