From b10e7cb1cf06235c9aae803b01c774856e3af4ee Mon Sep 17 00:00:00 2001 From: t8y2 <1156263951@qq.com> Date: Tue, 2 Jun 2026 17:46:30 +0800 Subject: [PATCH] fix(data-compare): create missing target tables during sync --- .../src/components/diff/DataCompareDialog.vue | 60 ++++++- apps/desktop/src/lib/api.ts | 1 + apps/desktop/src/lib/dataCompare.ts | 15 ++ apps/desktop/src/lib/http.ts | 6 + apps/desktop/src/lib/tauri.ts | 6 + crates/dbx-core/src/data_compare.rs | 161 ++++++++++++++++++ crates/dbx-web/src/main.rs | 1 + crates/dbx-web/src/routes/data_compare.rs | 7 + src-tauri/src/commands/data_compare.rs | 8 + src-tauri/src/lib.rs | 1 + 10 files changed, 261 insertions(+), 5 deletions(-) diff --git a/apps/desktop/src/components/diff/DataCompareDialog.vue b/apps/desktop/src/components/diff/DataCompareDialog.vue index 59618b951c..8475051aa4 100644 --- a/apps/desktop/src/components/diff/DataCompareDialog.vue +++ b/apps/desktop/src/components/diff/DataCompareDialog.vue @@ -74,6 +74,7 @@ interface DataCompareTableResult { sourceTruncated: boolean; targetTruncated: boolean; databaseType?: DatabaseType; + preSyncStatements?: string[]; diff: SelectableDataCompareResult; expanded: boolean; showAll: Record; @@ -171,8 +172,7 @@ const canCompare = computed( selectedSourceTableNames.value.length > 0 && targetConnectionId.value && targetDatabase.value && - targetSchema.value && - (!isBatchCompare.value ? !!targetTable.value : true), + targetSchema.value, ); const keyColumns = computed(() => keyColumnsText.value @@ -278,7 +278,7 @@ function buildCompareTasks(): DataCompareTableTask[] { if (!selectedSourceTableNames.value.length) return []; if (!isBatchCompare.value) { const table = selectedSourceTableNames.value[0]; - return targetTable.value ? [{ sourceTable: table, targetTable: targetTable.value }] : []; + return table ? [{ sourceTable: table, targetTable: targetTable.value || table }] : []; } return selectedSourceTableNames.value.map((table) => ({ sourceTable: table, @@ -586,8 +586,15 @@ function buildSyncPlanTables(): DataCompareSyncPlanTableOptions[] { keyColumns: table.keyColumns, diff: buildSelectedDiff(table), databaseType: table.databaseType, + preSyncStatements: table.preSyncStatements ?? [], })) - .filter((table) => table.diff.added.length > 0 || table.diff.removed.length > 0 || table.diff.modified.length > 0); + .filter( + (table) => + table.preSyncStatements.length > 0 || + table.diff.added.length > 0 || + table.diff.removed.length > 0 || + table.diff.modified.length > 0, + ); } async function rebuildSyncPlan() { @@ -641,7 +648,49 @@ async function startCompare() { try { if (!targetTables.value.includes(task.targetTable)) { - throw new Error(t("dataCompare.targetTableMissing", { table: task.targetTable })); + const sourceColumns = await loadColumnsWithCache( + sourceColumnCache, + sourceConnectionId.value, + sourceDatabase.value, + sourceSchema.value, + task.sourceTable, + ); + const resolvedKeys = keyColumns.value.length > 0 ? keyColumns.value : []; + const preparation = await api.prepareDataCompareMissingTarget({ + sourceConnectionId: sourceConnectionId.value, + sourceDatabase: sourceDatabase.value, + sourceSchema: sourceSchema.value, + sourceTable: task.sourceTable, + targetConnectionId: targetConnectionId.value, + targetDatabase: targetDatabase.value, + targetSchema: targetSchema.value, + targetTable: task.targetTable, + keyColumns: resolvedKeys, + }); + results.push({ + sourceTable: task.sourceTable, + targetTable: task.targetTable, + keyColumns: resolvedKeys, + columns: sourceColumns.map((column) => column.name), + status: "different", + added: preparation.result.added.length, + removed: 0, + modified: 0, + sourceRowCount: preparation.sourceRowCount, + targetRowCount: 0, + sourceTruncated: preparation.sourceTruncated, + targetTruncated: false, + databaseType: currentTargetDatabaseType, + preSyncStatements: preparation.preSyncStatements, + diff: toSelectableDiff(preparation.result), + expanded: preparation.result.added.length > 0, + showAll: { + added: false, + removed: false, + modified: false, + }, + }); + continue; } const resolvedKeys = @@ -732,6 +781,7 @@ async function startCompare() { sourceTruncated: false, targetTruncated: false, databaseType: currentTargetDatabaseType, + preSyncStatements: [], diff: { added: [], removed: [], modified: [] }, expanded: false, showAll: { diff --git a/apps/desktop/src/lib/api.ts b/apps/desktop/src/lib/api.ts index ef3cd1446d..f7f691e75f 100644 --- a/apps/desktop/src/lib/api.ts +++ b/apps/desktop/src/lib/api.ts @@ -133,6 +133,7 @@ export const buildExportSqlInsert = forward("buildExportSqlInsert"); export const buildDatabaseSqlExport = forward("buildDatabaseSqlExport"); export const prepareDataCompare = forward("prepareDataCompare"); export const prepareDataCompareFromTables = forward("prepareDataCompareFromTables"); +export const prepareDataCompareMissingTarget = forward("prepareDataCompareMissingTarget"); export const buildDataCompareSyncPlan = forward("buildDataCompareSyncPlan"); // AI diff --git a/apps/desktop/src/lib/dataCompare.ts b/apps/desktop/src/lib/dataCompare.ts index c0d11c74c3..1c512d1682 100644 --- a/apps/desktop/src/lib/dataCompare.ts +++ b/apps/desktop/src/lib/dataCompare.ts @@ -58,7 +58,21 @@ export interface DataCompareFromTablesOptions { fetchBatchSize?: number; } +export interface DataCompareMissingTargetOptions { + sourceConnectionId: string; + sourceDatabase: string; + sourceSchema: string; + sourceTable: string; + targetConnectionId: string; + targetDatabase: string; + targetSchema: string; + targetTable: string; + keyColumns: string[]; + fetchBatchSize?: number; +} + export interface DataCompareFromTablesPreparation extends DataComparePreparation { + preSyncStatements: string[]; sourceRowCount: number; targetRowCount: number; sourceTruncated: boolean; @@ -72,6 +86,7 @@ export interface DataCompareSyncPlanTableOptions { keyColumns: string[]; diff: DataCompareResult; databaseType?: DatabaseType; + preSyncStatements?: string[]; } export interface DataCompareSyncPlanOptions { diff --git a/apps/desktop/src/lib/http.ts b/apps/desktop/src/lib/http.ts index 2dfe623abf..0963547254 100644 --- a/apps/desktop/src/lib/http.ts +++ b/apps/desktop/src/lib/http.ts @@ -720,6 +720,12 @@ export async function prepareDataCompareFromTables( return post("/api/data-compare/prepare-from-tables", options); } +export async function prepareDataCompareMissingTarget( + options: import("@/lib/dataCompare").DataCompareMissingTargetOptions, +): Promise { + return post("/api/data-compare/prepare-missing-target", options); +} + export async function buildDataCompareSyncPlan(options: DataCompareSyncPlanOptions): Promise { return post("/api/data-compare/build-sync-plan", options); } diff --git a/apps/desktop/src/lib/tauri.ts b/apps/desktop/src/lib/tauri.ts index 4a0d5daaeb..c0029d554c 100644 --- a/apps/desktop/src/lib/tauri.ts +++ b/apps/desktop/src/lib/tauri.ts @@ -719,6 +719,12 @@ export async function prepareDataCompareFromTables( return invoke("prepare_data_compare_from_tables", { options }); } +export async function prepareDataCompareMissingTarget( + options: import("@/lib/dataCompare").DataCompareMissingTargetOptions, +): Promise { + return invoke("prepare_data_compare_missing_target", { options }); +} + export async function buildDataCompareSyncPlan(options: DataCompareSyncPlanOptions): Promise { return invoke("build_data_compare_sync_plan", { options }); } diff --git a/crates/dbx-core/src/data_compare.rs b/crates/dbx-core/src/data_compare.rs index 441b97189f..ab776a7ae8 100644 --- a/crates/dbx-core/src/data_compare.rs +++ b/crates/dbx-core/src/data_compare.rs @@ -6,7 +6,9 @@ use serde_json::Value; use crate::connection::AppState; use crate::models::connection::DatabaseType; use crate::query::{execute_sql_statement_with_options, QueryExecutionOptions}; +use crate::schema::get_columns_core; use crate::sql_dialect::{build_count_table_sql, qualified_table_name, quote_table_identifier, uses_fetch_first}; +use crate::transfer::{generate_comment_ddl, generate_create_table_ddl}; #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] @@ -52,6 +54,23 @@ pub struct DataCompareFromTablesOptions { pub fetch_batch_size: Option, } +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct DataCompareMissingTargetOptions { + pub source_connection_id: String, + pub source_database: String, + pub source_schema: String, + pub source_table: String, + pub target_connection_id: String, + pub target_database: String, + pub target_schema: String, + pub target_table: String, + #[serde(default)] + pub key_columns: Vec, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub fetch_batch_size: Option, +} + #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] pub struct DataCompareChangedCell { @@ -105,6 +124,8 @@ pub struct DataCompareSyncPlanTableOptions { pub diff: DataCompareResult, #[serde(default, skip_serializing_if = "Option::is_none")] pub database_type: Option, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub pre_sync_statements: Vec, } #[derive(Debug, Clone, Serialize, Deserialize)] @@ -131,6 +152,8 @@ pub struct DataCompareFromTablesPreparation { pub result: DataCompareResult, pub sync_statements: Vec, pub sync_sql: String, + #[serde(default)] + pub pre_sync_statements: Vec, pub source_row_count: u64, pub target_row_count: u64, pub source_truncated: bool, @@ -152,6 +175,7 @@ pub fn prepare_data_compare(options: DataComparePreparationOptions) -> Result Result { + let source_database_type = connection_database_type(state, &options.source_connection_id).await?; + let target_database_type = connection_database_type(state, &options.target_connection_id).await?; + let fetch_batch_size = options.fetch_batch_size.unwrap_or(1000).max(1); + let source_columns = get_columns_core( + state, + &options.source_connection_id, + &options.source_database, + &options.source_schema, + &options.source_table, + ) + .await?; + let column_names = source_columns.iter().map(|column| column.name.clone()).collect::>(); + + let source_count_sql = + build_count_table_sql(Some(source_database_type), Some(&options.source_schema), &options.source_table); + let source_count_result = execute_sql_statement_with_options( + state, + &options.source_connection_id, + &options.source_database, + &source_count_sql, + Some(&options.source_schema), + None, + QueryExecutionOptions { max_rows: Some(1), ..Default::default() }, + ) + .await?; + let source_row_count = first_count(&source_count_result.rows)?; + let source_rows = fetch_compare_rows( + state, + &options.source_connection_id, + &options.source_database, + &options.source_schema, + &options.source_table, + &column_names, + &options.key_columns, + source_database_type, + fetch_batch_size, + ) + .await?; + let result = missing_target_diff(&column_names, &options.key_columns, source_rows); + let mut pre_sync_statements = Vec::new(); + pre_sync_statements.push(format!( + "{};", + generate_create_table_ddl( + &source_columns, + &options.target_table, + &options.source_schema, + &options.target_schema, + &target_database_type, + &source_database_type, + None, + ) + )); + pre_sync_statements.extend( + generate_comment_ddl( + &source_columns, + &options.target_table, + &options.target_schema, + &target_database_type, + None, + ) + .into_iter() + .map(|statement| format!("{statement};")), + ); + + let sync_plan = build_data_compare_sync_plan(DataCompareSyncPlanOptions { + tables: vec![DataCompareSyncPlanTableOptions { + table_name: options.target_table, + schema: Some(options.target_schema), + columns: column_names, + key_columns: options.key_columns, + diff: result.clone(), + database_type: Some(target_database_type), + pre_sync_statements: pre_sync_statements.clone(), + }], + }); + + Ok(DataCompareFromTablesPreparation { + result, + sync_statements: sync_plan.sync_statements, + sync_sql: sync_plan.sync_sql, + pre_sync_statements, + source_row_count, + target_row_count: 0, + source_truncated: false, + target_truncated: false, + }) +} + pub fn build_data_compare_sync_plan(options: DataCompareSyncPlanOptions) -> DataCompareSyncPlan { let mut sync_statements = Vec::new(); let mut insert_count = 0; @@ -249,6 +366,7 @@ pub fn build_data_compare_sync_plan(options: DataCompareSyncPlanOptions) -> Data insert_count += table.diff.added.len(); update_count += table.diff.modified.len(); delete_count += table.diff.removed.len(); + sync_statements.extend(table.pre_sync_statements); sync_statements.extend(generate_data_sync_statements(&GenerateDataSyncSqlOptions { table_name: table.table_name, schema: table.schema, @@ -264,6 +382,20 @@ pub fn build_data_compare_sync_plan(options: DataCompareSyncPlanOptions) -> Data DataCompareSyncPlan { insert_count, update_count, delete_count, statement_count, sync_statements, sync_sql } } +fn missing_target_diff(columns: &[String], key_columns: &[String], source_rows: Vec>) -> DataCompareResult { + let added = source_rows + .iter() + .enumerate() + .map(|(index, row)| { + let values = row_object(columns, row); + let key = if key_columns.is_empty() { index.to_string() } else { key_for(&values, key_columns) }; + DataCompareRow { key, key_values: key_values(&values, key_columns), values } + }) + .collect(); + + DataCompareResult { added, removed: Vec::new(), modified: Vec::new() } +} + pub fn compare_data_rows(options: CompareDataRowsOptions) -> Result { if options.key_columns.is_empty() { return Err("At least one key column is required for data comparison".to_string()); @@ -724,6 +856,7 @@ mod tests { }], }, database_type: Some(DatabaseType::Postgres), + pre_sync_statements: Vec::new(), }], }); @@ -733,6 +866,34 @@ mod tests { assert_eq!(plan.statement_count, 2); } + #[test] + fn build_sync_plan_keeps_missing_target_create_table_statement() { + let plan = build_data_compare_sync_plan(DataCompareSyncPlanOptions { + tables: vec![DataCompareSyncPlanTableOptions { + table_name: "users".to_string(), + schema: Some("public".to_string()), + columns: vec!["id".to_string(), "name".to_string()], + key_columns: Vec::new(), + diff: DataCompareResult { + added: vec![DataCompareRow { + key: "0".to_string(), + key_values: HashMap::new(), + values: HashMap::from([(String::from("id"), json!(1)), (String::from("name"), json!("Ada"))]), + }], + removed: Vec::new(), + modified: Vec::new(), + }, + database_type: Some(DatabaseType::Postgres), + pre_sync_statements: vec!["CREATE TABLE \"public\".\"users\" (\"id\" integer);".to_string()], + }], + }); + + assert_eq!(plan.insert_count, 1); + assert_eq!(plan.statement_count, 2); + assert!(plan.sync_sql.starts_with("CREATE TABLE")); + assert!(plan.sync_sql.contains("INSERT INTO \"public\".\"users\"")); + } + #[test] fn requires_at_least_one_key_column() { let err = compare_data_rows(CompareDataRowsOptions { diff --git a/crates/dbx-web/src/main.rs b/crates/dbx-web/src/main.rs index 60ce7091ff..0e8f462b3b 100644 --- a/crates/dbx-web/src/main.rs +++ b/crates/dbx-web/src/main.rs @@ -197,6 +197,7 @@ async fn main() { .route("/query/build-database-sql-export", post(routes::query::build_database_sql_export)) .route("/data-compare/prepare", post(routes::data_compare::prepare_data_compare)) .route("/data-compare/prepare-from-tables", post(routes::data_compare::prepare_data_compare_from_tables)) + .route("/data-compare/prepare-missing-target", post(routes::data_compare::prepare_data_compare_missing_target)) .route("/data-compare/build-sync-plan", post(routes::data_compare::build_data_compare_sync_plan)) .route("/query/cancel", post(routes::query::cancel_query)) .route("/query/close-session", post(routes::query::close_query_session)) diff --git a/crates/dbx-web/src/routes/data_compare.rs b/crates/dbx-web/src/routes/data_compare.rs index 5405d0667c..8d58b6cde3 100644 --- a/crates/dbx-web/src/routes/data_compare.rs +++ b/crates/dbx-web/src/routes/data_compare.rs @@ -18,6 +18,13 @@ pub async fn prepare_data_compare_from_tables( dbx_core::data_compare::prepare_data_compare_from_tables(&state.app, options).await.map(Json).map_err(AppError) } +pub async fn prepare_data_compare_missing_target( + State(state): State>, + Json(options): Json, +) -> Result, AppError> { + dbx_core::data_compare::prepare_data_compare_missing_target(&state.app, options).await.map(Json).map_err(AppError) +} + pub async fn build_data_compare_sync_plan( Json(options): Json, ) -> Json { diff --git a/src-tauri/src/commands/data_compare.rs b/src-tauri/src/commands/data_compare.rs index 627d70ffa2..ca353aba3b 100644 --- a/src-tauri/src/commands/data_compare.rs +++ b/src-tauri/src/commands/data_compare.rs @@ -18,6 +18,14 @@ pub async fn prepare_data_compare_from_tables( dbx_core::data_compare::prepare_data_compare_from_tables(&state, options).await } +#[tauri::command] +pub async fn prepare_data_compare_missing_target( + state: State<'_, Arc>, + options: dbx_core::data_compare::DataCompareMissingTargetOptions, +) -> Result { + dbx_core::data_compare::prepare_data_compare_missing_target(&state, options).await +} + #[tauri::command] pub fn build_data_compare_sync_plan( options: dbx_core::data_compare::DataCompareSyncPlanOptions, diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index c1889718bb..158aeca3a8 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -441,6 +441,7 @@ pub fn run() { commands::query::build_database_sql_export, commands::data_compare::prepare_data_compare, commands::data_compare::prepare_data_compare_from_tables, + commands::data_compare::prepare_data_compare_missing_target, commands::data_compare::build_data_compare_sync_plan, commands::sql_file::preview_sql_file, commands::sql_file::execute_sql_file,