xiaoxue_admin
/
dbx
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix(data-compare): create missing target tables during sync

This commit is contained in:
t8y2 2026-06-02 17:46:30 +08:00
parent b154dcceef
commit b10e7cb1cf
10 changed files with 261 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

60
apps/desktop/src/components/diff/DataCompareDialog.vue
View File

@ -74,6 +74,7 @@ interface DataCompareTableResult {
sourceTruncated: boolean;
targetTruncated: boolean;
databaseType?: DatabaseType;
preSyncStatements?: string[];
diff: SelectableDataCompareResult;
expanded: boolean;
showAll: Record<DiffKind, boolean>;
@ -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: {

1
apps/desktop/src/lib/api.ts
View File

@ -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

15
apps/desktop/src/lib/dataCompare.ts
View File

@ -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 {

6
apps/desktop/src/lib/http.ts
View File

@ -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<DataCompareFromTablesPreparation> {
return post("/api/data-compare/prepare-missing-target", options);
}
export async function buildDataCompareSyncPlan(options: DataCompareSyncPlanOptions): Promise<DataCompareSyncPlan> {
return post("/api/data-compare/build-sync-plan", options);
}

6
apps/desktop/src/lib/tauri.ts
View File

@ -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<DataCompareFromTablesPreparation> {
return invoke("prepare_data_compare_missing_target", { options });
}
export async function buildDataCompareSyncPlan(options: DataCompareSyncPlanOptions): Promise<DataCompareSyncPlan> {
return invoke("build_data_compare_sync_plan", { options });
}

161
crates/dbx-core/src/data_compare.rs
View File

@ -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<usize>,
}
#[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<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub fetch_batch_size: Option<usize>,
}
#[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<DatabaseType>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub pre_sync_statements: Vec<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
@ -131,6 +152,8 @@ pub struct DataCompareFromTablesPreparation {
pub result: DataCompareResult,
pub sync_statements: Vec<String>,
pub sync_sql: String,
#[serde(default)]
pub pre_sync_statements: Vec<String>,
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<Da
key_columns: options.key_columns,
diff: result.clone(),
database_type: options.database_type,
pre_sync_statements: Vec::new(),
}],
});
Ok(DataComparePreparation { result, sync_statements: sync_plan.sync_statements, sync_sql: sync_plan.sync_sql })
@ -232,6 +256,7 @@ pub async fn prepare_data_compare_from_tables(
result: preparation.result,
sync_statements: preparation.sync_statements,
sync_sql: preparation.sync_sql,
pre_sync_statements: Vec::new(),
source_row_count,
target_row_count,
source_truncated: false,
@ -239,6 +264,98 @@ pub async fn prepare_data_compare_from_tables(
})
}
pub async fn prepare_data_compare_missing_target(
state: &AppState,
options: DataCompareMissingTargetOptions,
) -> Result<DataCompareFromTablesPreparation, String> {
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::<Vec<_>>();
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<Vec<Value>>) -> 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<DataCompareResult, String> {
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 {

1
crates/dbx-web/src/main.rs
View File

@ -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))

7
crates/dbx-web/src/routes/data_compare.rs
View File

@ -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<Arc<WebState>>,
Json(options): Json<dbx_core::data_compare::DataCompareMissingTargetOptions>,
) -> Result<Json<dbx_core::data_compare::DataCompareFromTablesPreparation>, 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<dbx_core::data_compare::DataCompareSyncPlanOptions>,
) -> Json<dbx_core::data_compare::DataCompareSyncPlan> {

8
src-tauri/src/commands/data_compare.rs
View File

@ -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<AppState>>,
options: dbx_core::data_compare::DataCompareMissingTargetOptions,
) -> Result<dbx_core::data_compare::DataCompareFromTablesPreparation, String> {
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,

1
src-tauri/src/lib.rs
View File

@ -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,