From 37ea6d4fc88ca070b6a405217a5b5199b941191e Mon Sep 17 00:00:00 2001 From: Jeff Date: Sat, 10 Oct 2026 00:12:38 +0800 Subject: [PATCH] fix(transfer): clear FK-linked target children first on overwrite --- crates/dbx-core/src/data/transfer.rs | 14 +- .../dbx-core/src/data/transfer/iris_tests.rs | 1 + .../src/data/transfer/overwrite_clear.rs | 177 ++++++++++++++++++ .../dbx-core/tests/live_sqlserver_transfer.rs | 140 +++++++++++++- crates/dbx-web/src/routes/transfer.rs | 31 +++ src-tauri/src/commands/transfer.rs | 35 ++++ 6 files changed, 394 insertions(+), 4 deletions(-) create mode 100644 crates/dbx-core/src/data/transfer/overwrite_clear.rs diff --git a/crates/dbx-core/src/data/transfer.rs b/crates/dbx-core/src/data/transfer.rs index 9a984dc860..1778a61384 100644 --- a/crates/dbx-core/src/data/transfer.rs +++ b/crates/dbx-core/src/data/transfer.rs @@ -24,8 +24,11 @@ mod iris_tests; mod db2; mod ddl_plan; +mod overwrite_clear; mod structure_plan; +pub use overwrite_clear::clear_foreign_key_linked_overwrite_targets; + use crate::connection::{config_for_pool_key, AppState, PoolKind}; use crate::db; use crate::db::agent_driver::AgentTableReadStartParams; @@ -11094,6 +11097,7 @@ async fn transfer_table_inner( known_foreign_keys: &HashMap>, pending_fk_alters: &mut Vec<(String, String)>, preexisting_backup_names: Option<&HashMap>, + target_cleared: bool, mut progress_callback: F, mut source_count_callback: C, ) -> Result @@ -11539,8 +11543,9 @@ where // Truncate target if overwrite mode (only when not rebuilding the table). // When drop_target_before_create is true, the target table was just created - // and is already empty, so TRUNCATE is unnecessary. - if request.mode == TransferMode::Overwrite && !request.drop_target_before_create { + // and is already empty, so TRUNCATE is unnecessary. `target_cleared` means + // `clear_foreign_key_linked_overwrite_targets` already emptied it children first. + if request.mode == TransferMode::Overwrite && !request.drop_target_before_create && !target_cleared { let truncate_sql = transfer_clear_table_sql( &target_table, &request.target_schema, @@ -12330,6 +12335,7 @@ where known_foreign_keys, pending_fk_alters, preexisting_backup_names, + false, progress_callback, |_| {}, ) @@ -12337,6 +12343,8 @@ where .map(|result| result.moved_rows) } +/// Pass `target_cleared = true` for tables that [`clear_foreign_key_linked_overwrite_targets`] +/// already emptied, so the overwrite clear is not repeated. #[allow(clippy::too_many_arguments)] pub async fn transfer_table_with_result( state: &Arc, @@ -12350,6 +12358,7 @@ pub async fn transfer_table_with_result( known_foreign_keys: &HashMap>, pending_fk_alters: &mut Vec<(String, String)>, preexisting_backup_names: Option<&HashMap>, + target_cleared: bool, mut progress_callback: F, mut source_count_callback: C, ) -> Result @@ -12395,6 +12404,7 @@ where &known_foreign_keys, &mut task_pending_fk_alters, preexisting_backup_names.as_ref(), + target_cleared, move |progress| { try_send_transfer_progress(&progress_tx, progress); }, diff --git a/crates/dbx-core/src/data/transfer/iris_tests.rs b/crates/dbx-core/src/data/transfer/iris_tests.rs index 5cb8217ae2..55d3dd2775 100644 --- a/crates/dbx-core/src/data/transfer/iris_tests.rs +++ b/crates/dbx-core/src/data/transfer/iris_tests.rs @@ -323,6 +323,7 @@ async fn iris_transfer_agent_preserves_errors_count_fallback_and_cancellation() &HashMap::new(), &mut Vec::new(), None, + false, |progress| { totals.push(progress.total_rows); if failure == "cancel" { diff --git a/crates/dbx-core/src/data/transfer/overwrite_clear.rs b/crates/dbx-core/src/data/transfer/overwrite_clear.rs new file mode 100644 index 0000000000..97d16efad1 --- /dev/null +++ b/crates/dbx-core/src/data/transfer/overwrite_clear.rs @@ -0,0 +1,177 @@ +//! Overwrite pre-pass for target tables linked by foreign keys inside the transfer. +//! +//! `transfer_table` clears each overwrite target right before copying into it, and the +//! transfer loop visits tables parents first. That order cannot clear a referenced table: +//! SQL Server (4712), MySQL (1701) and PostgreSQL refuse to `TRUNCATE` a table another +//! table's foreign key points at, even when the referencing table is already empty, and a +//! parent cannot lose its rows while a child still references them. This pass empties the +//! linked tables children first before the loop starts, and the loop then skips the +//! per-table clear for them. + +use super::*; + +/// Empty the overwrite targets that are linked to each other by foreign keys, children first. +/// +/// Tables referenced by another selected table are cleared with `DELETE`, the rest with the +/// usual per-database clear statement. Returns the source names of the tables that were +/// cleared; the caller passes `target_cleared = true` to `transfer_table_with_result` for +/// those so they are not cleared a second time. +/// +/// Returns an empty set, leaving every table to the per-table clear, when the request is not +/// an overwrite, when no selected target references another, when the foreign key metadata +/// cannot be read, or when a table outside the selection references one of the linked +/// tables: a `DELETE` there could cascade into rows the transfer never selected. +pub async fn clear_foreign_key_linked_overwrite_targets( + state: &Arc, + request: &TransferRequest, + tables: &[String], + target_db_type: DatabaseType, + target_pool_key: &str, +) -> Result, String> { + if request.mode != TransferMode::Overwrite + || request.drop_target_before_create + || tables.len() < 2 + || is_mongodb_transfer_type(&target_db_type) + { + return Ok(HashSet::new()); + } + + let mut targets: Vec<(String, String)> = Vec::new(); + for table in tables { + let ResolvedTransferTargetTable { name, preexisting } = resolve_transfer_target_table_name( + state, + request, + table, + target_pool_key, + &target_db_type, + request.source_catalog.as_deref(), + request.target_catalog.as_deref(), + ) + .await; + if preexisting { + targets.push((table.clone(), name)); + } + } + + let mut edges: Vec<(String, String)> = Vec::new(); + for (_, target_table) in &targets { + let foreign_keys = match crate::schema::list_foreign_keys_core( + state, + &request.target_connection_id, + &request.target_database, + &request.target_schema, + target_table, + ) + .await + { + Ok(foreign_keys) => foreign_keys, + Err(error) => { + log::warn!( + "[transfer] overwrite pre-pass skipped: cannot read foreign keys of {target_table}: {error}" + ); + return Ok(HashSet::new()); + } + }; + for foreign_key in foreign_keys { + let referenced_in_target_schema = foreign_key.ref_schema.as_deref().is_none_or(|schema| { + schema.is_empty() + || schema.eq_ignore_ascii_case(&request.target_schema) + || schema.eq_ignore_ascii_case(&request.target_database) + }); + if referenced_in_target_schema + && foreign_key.ref_table != *target_table + && targets.iter().any(|(_, name)| *name == foreign_key.ref_table) + { + edges.push((target_table.clone(), foreign_key.ref_table)); + } + } + } + edges.sort(); + edges.dedup(); + if edges.is_empty() { + return Ok(HashSet::new()); + } + + let linked: Vec = targets + .iter() + .map(|(_, name)| name.clone()) + .filter(|name| edges.iter().any(|(child, parent)| child == name || parent == name)) + .collect(); + match crate::transfer_rebuild::detect_external_incoming_foreign_keys( + state, + target_pool_key, + &request.target_database, + &request.target_schema, + &linked, + target_db_type, + ) + .await + { + Ok(external) if external.is_empty() => {} + Ok(external) => { + log::warn!( + "[transfer] overwrite pre-pass skipped: {} foreign key(s) from outside the transfer reference the target tables", + external.len() + ); + return Ok(HashSet::new()); + } + Err(error) => { + log::warn!("[transfer] overwrite pre-pass skipped: cannot check incoming foreign keys: {error}"); + return Ok(HashSet::new()); + } + } + + let mut cleared = HashSet::new(); + for target_table in sort_table_names_by_dependencies(&linked, &edges, false) { + let referenced = edges.iter().any(|(_, parent)| *parent == target_table); + let sql = overwrite_pre_pass_clear_sql( + &target_table, + &request.target_schema, + &target_db_type, + request.target_catalog.as_deref(), + referenced, + ); + execute_on_pool(state, target_pool_key, &sql) + .await + .map_err(|error| format!("Failed to clear target table '{target_table}' before overwrite: {error}"))?; + if let Some((source_table, _)) = targets.iter().find(|(_, name)| *name == target_table) { + cleared.insert(source_table.clone()); + } + } + Ok(cleared) +} + +pub(super) fn overwrite_pre_pass_clear_sql( + table: &str, + schema: &str, + db_type: &DatabaseType, + catalog: Option<&str>, + referenced: bool, +) -> String { + if referenced { + format!("DELETE FROM {}", qualified_table(table, schema, db_type, catalog)) + } else { + transfer_clear_table_sql(table, schema, db_type, catalog) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn referenced_targets_are_deleted_instead_of_truncated() { + assert_eq!( + overwrite_pre_pass_clear_sql("QRTZ_JOB_DETAILS", "dbo", &DatabaseType::SqlServer, None, true), + "DELETE FROM [dbo].[QRTZ_JOB_DETAILS]" + ); + assert_eq!( + overwrite_pre_pass_clear_sql("QRTZ_TRIGGERS", "dbo", &DatabaseType::SqlServer, None, false), + "TRUNCATE TABLE [dbo].[QRTZ_TRIGGERS]" + ); + assert_eq!( + overwrite_pre_pass_clear_sql("EVENTS", "APP", &DatabaseType::Db2, None, false), + "TRUNCATE TABLE \"APP\".\"EVENTS\" IMMEDIATE" + ); + } +} diff --git a/crates/dbx-core/tests/live_sqlserver_transfer.rs b/crates/dbx-core/tests/live_sqlserver_transfer.rs index a2a9dffe10..dd3fc610ad 100644 --- a/crates/dbx-core/tests/live_sqlserver_transfer.rs +++ b/crates/dbx-core/tests/live_sqlserver_transfer.rs @@ -1,8 +1,9 @@ use dbx_core::connection::AppState; use dbx_core::models::connection::{ConnectionConfig, DatabaseType}; use dbx_core::transfer::{ - drop_backup_tables, rename_tables_to_backup, transfer_table, TransferContent, TransferMode, - TransferOwnershipPolicy, TransferRequest, TransferTableNameCase, + clear_foreign_key_linked_overwrite_targets, drop_backup_tables, rename_tables_to_backup, + sort_tables_by_fk_dependency_with_foreign_keys, transfer_table, transfer_table_with_result, TransferContent, + TransferMode, TransferOwnershipPolicy, TransferRequest, TransferTableNameCase, }; use std::collections::HashMap; use std::sync::Arc; @@ -1009,3 +1010,138 @@ async fn live_sqlserver_database_export_replays_composite_foreign_keys() { cleanup.expect("drop export databases"); test_result.expect("database export must replay a composite foreign key without 8168"); } + +#[tokio::test] +#[ignore = "requires DBX_LIVE_SQLSERVER_HOST/PORT/USER/PASSWORD pointing at SQL Server"] +async fn live_sqlserver_transfer_overwrite_replaces_tables_linked_by_foreign_keys() { + let database = std::env::var("DBX_LIVE_SQLSERVER_DATABASE").unwrap_or_else(|_| "dbx_sqlserver_demo".to_string()); + let suffix = uuid::Uuid::new_v4().simple().to_string(); + let connection_id = format!("live-sqlserver-11480-{suffix}"); + let source_schema = format!("dbx_11480_src_{}", &suffix[..12]); + let target_schema = format!("dbx_11480_dst_{}", &suffix[..12]); + + let mut client = sqlserver_connect(&database).await; + let mut fixtures = vec![format!("CREATE SCHEMA [{source_schema}]"), format!("CREATE SCHEMA [{target_schema}]")]; + for schema in [&source_schema, &target_schema] { + fixtures.push(format!( + "CREATE TABLE [{schema}].[job_details] (id INT NOT NULL CONSTRAINT [PK_11480_job_{schema}] PRIMARY KEY, name NVARCHAR(64) NOT NULL)" + )); + fixtures.push(format!( + "CREATE TABLE [{schema}].[triggers] (id INT NOT NULL CONSTRAINT [PK_11480_trigger_{schema}] PRIMARY KEY, job_id INT NOT NULL \ + CONSTRAINT [FK_11480_{schema}] FOREIGN KEY REFERENCES [{schema}].[job_details] (id))" + )); + } + fixtures.extend([ + format!("INSERT INTO [{source_schema}].[job_details] (id, name) VALUES (1, N'first'), (3, N'third')"), + format!("INSERT INTO [{source_schema}].[triggers] (id, job_id) VALUES (10, 1), (30, 3)"), + format!("INSERT INTO [{target_schema}].[job_details] (id, name) VALUES (1, N'stale'), (2, N'stale')"), + format!("INSERT INTO [{target_schema}].[triggers] (id, job_id) VALUES (10, 1), (20, 2)"), + ]); + for statement in &fixtures { + dbx_core::db::sqlserver::execute_batch(&mut client, statement).await.expect("create issue #11480 fixtures"); + } + + let dir = std::env::temp_dir().join(format!("dbx-live-sqlserver-11480-{suffix}")); + std::fs::create_dir_all(&dir).expect("create issue #11480 directory"); + let storage = + dbx_core::persistence::test_storage::open(&dir.join("storage.db")).await.expect("open issue #11480 storage"); + let state = Arc::new(AppState::new(storage)); + state.configs.write().await.insert(connection_id.clone(), live_sqlserver_config(&connection_id, &database)); + let pool_key = state.get_or_create_pool(&connection_id, Some(&database)).await.expect("create SQL Server pool"); + let request = TransferRequest { + table_filters: HashMap::new(), + transfer_id: format!("live-sqlserver-11480-transfer-{suffix}"), + source_connection_id: connection_id.clone(), + source_database: database.clone(), + source_schema: source_schema.clone(), + source_catalog: None, + target_connection_id: connection_id.clone(), + target_database: database.clone(), + target_schema: target_schema.clone(), + target_catalog: None, + tables: vec!["triggers".to_string(), "job_details".to_string()], + create_table: false, + drop_target_before_create: false, + drop_target_confirmed: false, + content: TransferContent::DataOnly, + objects: Some(Vec::new()), + mode: TransferMode::Overwrite, + target_table_name_case: TransferTableNameCase::Preserve, + quote_target_column_names: true, + ownership_policy: TransferOwnershipPolicy::Preserve, + batch_size: 1000, + }; + + // Same sequence as the desktop and web transfer loops: parents first, with the + // foreign-key-linked targets emptied children first beforehand. + let test_result = async { + let (tables, known_foreign_keys) = sort_tables_by_fk_dependency_with_foreign_keys( + &state, + &connection_id, + &database, + &source_schema, + &request.tables, + true, + ) + .await?; + let cleared = + clear_foreign_key_linked_overwrite_targets(&state, &request, &tables, DatabaseType::SqlServer, &pool_key) + .await?; + assert_eq!(cleared.len(), 2, "both linked targets are cleared by the pre-pass"); + for (index, table) in tables.iter().enumerate() { + transfer_table_with_result( + &state, + &request, + table, + index, + &DatabaseType::SqlServer, + &DatabaseType::SqlServer, + &pool_key, + &pool_key, + &known_foreign_keys, + &mut Vec::new(), + None, + cleared.contains(table), + |_| {}, + |_| {}, + ) + .await?; + } + + let jobs = dbx_core::db::sqlserver::execute_query( + &mut client, + &format!("SELECT id, name FROM [{target_schema}].[job_details] ORDER BY id"), + ) + .await + .map_err(|error| format!("read target jobs: {error}"))?; + let triggers = dbx_core::db::sqlserver::execute_query( + &mut client, + &format!("SELECT id, job_id FROM [{target_schema}].[triggers] ORDER BY id"), + ) + .await + .map_err(|error| format!("read target triggers: {error}"))?; + let jobs = + jobs.rows.iter().map(|row| (row[0].as_i64(), row[1].as_str().map(str::to_string))).collect::>(); + let triggers = triggers.rows.iter().map(|row| (row[0].as_i64(), row[1].as_i64())).collect::>(); + assert_eq!(jobs, vec![(Some(1), Some("first".to_string())), (Some(3), Some("third".to_string()))]); + assert_eq!(triggers, vec![(Some(10), Some(1)), (Some(30), Some(3))]); + Ok::<_, String>(()) + } + .await; + + let cleanup = dbx_core::db::sqlserver::execute_batch( + &mut client, + &format!( + "IF OBJECT_ID(N'[{target_schema}].[triggers]', N'U') IS NOT NULL DROP TABLE [{target_schema}].[triggers]; \ + IF OBJECT_ID(N'[{target_schema}].[job_details]', N'U') IS NOT NULL DROP TABLE [{target_schema}].[job_details]; \ + IF OBJECT_ID(N'[{source_schema}].[triggers]', N'U') IS NOT NULL DROP TABLE [{source_schema}].[triggers]; \ + IF OBJECT_ID(N'[{source_schema}].[job_details]', N'U') IS NOT NULL DROP TABLE [{source_schema}].[job_details]; \ + IF SCHEMA_ID(N'{target_schema}') IS NOT NULL DROP SCHEMA [{target_schema}]; \ + IF SCHEMA_ID(N'{source_schema}') IS NOT NULL DROP SCHEMA [{source_schema}];" + ), + ) + .await; + let _ = std::fs::remove_dir_all(dir); + cleanup.expect("cleanup issue #11480 fixtures"); + test_result.expect("SQL Server overwrite transfer should replace tables linked by foreign keys"); +} diff --git a/crates/dbx-web/src/routes/transfer.rs b/crates/dbx-web/src/routes/transfer.rs index c633e0c31b..16dd46ce04 100644 --- a/crates/dbx-web/src/routes/transfer.rs +++ b/crates/dbx-web/src/routes/transfer.rs @@ -356,6 +356,36 @@ pub async fn start_transfer( } } + // Overwrite clears each target right before copying it, parents first, which cannot + // clear a table another selected table references. Empty those children first now. + let overwrite_cleared = match transfer::clear_foreign_key_linked_overwrite_targets( + &app, + &req, + &tables, + target_db_type, + &target_pool_key, + ) + .await + { + Ok(cleared) => cleared, + Err(e) => { + let progress = transfer::TransferProgress { + transfer_id: req.transfer_id.clone(), + table: "overwrite pre-pass".to_string(), + table_index: 0, + total_tables: tables.len(), + rows_transferred: 0, + total_rows: None, + status: TransferStatus::Error, + error: Some(e), + terminal: true, + }; + send_transfer_progress(&progress_channel, &progress); + finish_transfer_channel(&state_clone, &req.transfer_id, &progress_channel).await; + return; + } + }; + for (i, table) in tables.iter().enumerate() { if transfer::is_cancelled(&req.transfer_id).await { let progress = transfer::TransferProgress { @@ -398,6 +428,7 @@ pub async fn start_transfer( &known_foreign_keys, &mut pending_fk_alters, backup_names.as_ref(), + overwrite_cleared.contains(table), |progress| { last_rows_transferred = progress.rows_transferred; last_total_rows = progress.total_rows; diff --git a/src-tauri/src/commands/transfer.rs b/src-tauri/src/commands/transfer.rs index df554786fd..72f43a594f 100644 --- a/src-tauri/src/commands/transfer.rs +++ b/src-tauri/src/commands/transfer.rs @@ -277,6 +277,40 @@ pub async fn start_transfer( } } } + // Overwrite clears each target right before copying it, parents first, which cannot + // clear a table another selected table references. Empty those children first now. + let overwrite_cleared = match dbx_core::transfer::clear_foreign_key_linked_overwrite_targets( + &state, + &request, + &sorted_tables, + target_db_type, + &target_pool_key, + ) + .await + { + Ok(cleared) => cleared, + Err(e) => { + emit_terminal_progress( + &app, + history.as_ref(), + TransferProgress { + transfer_id: transfer_id.clone(), + table: "overwrite pre-pass".to_string(), + table_index: 0, + total_tables, + rows_transferred: last_rows_transferred, + total_rows: last_total_rows, + status: TransferStatus::Error, + error: Some(e), + terminal: true, + }, + ) + .await; + dbx_core::transfer::clear_cancelled(&transfer_id).await; + return; + } + }; + for (i, table) in sorted_tables.iter().enumerate() { if dbx_core::transfer::is_cancelled(&transfer_id).await { emit_terminal_progress( @@ -321,6 +355,7 @@ pub async fn start_transfer( &known_foreign_keys, &mut pending_fk_alters, backup_names.as_ref(), + overwrite_cleared.contains(table), |progress| { last_rows_transferred = progress.rows_transferred; last_total_rows = progress.total_rows;