fix(transfer): clear FK-linked target children first on overwrite

This commit is contained in:
Jeff
2026-10-10 00:12:38 +08:00
committed by GitHub
parent 0d2293525e
commit 37ea6d4fc8
6 changed files with 394 additions and 4 deletions
+12 -2
View File
@@ -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<F, C>(
known_foreign_keys: &HashMap<String, Vec<db::ForeignKeyInfo>>,
pending_fk_alters: &mut Vec<(String, String)>,
preexisting_backup_names: Option<&HashMap<String, String>>,
target_cleared: bool,
mut progress_callback: F,
mut source_count_callback: C,
) -> Result<TransferTableResult, String>
@@ -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<F, C>(
state: &Arc<AppState>,
@@ -12350,6 +12358,7 @@ pub async fn transfer_table_with_result<F, C>(
known_foreign_keys: &HashMap<String, Vec<db::ForeignKeyInfo>>,
pending_fk_alters: &mut Vec<(String, String)>,
preexisting_backup_names: Option<&HashMap<String, String>>,
target_cleared: bool,
mut progress_callback: F,
mut source_count_callback: C,
) -> Result<TransferTableResult, String>
@@ -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);
},
@@ -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" {
@@ -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<AppState>,
request: &TransferRequest,
tables: &[String],
target_db_type: DatabaseType,
target_pool_key: &str,
) -> Result<HashSet<String>, 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<String> = 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"
);
}
}
@@ -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::<Vec<_>>();
let triggers = triggers.rows.iter().map(|row| (row[0].as_i64(), row[1].as_i64())).collect::<Vec<_>>();
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");
}
+31
View File
@@ -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;
+35
View File
@@ -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;