mirror of
https://github.com/t8y2/dbx.git
synced 2026-10-19 16:16:45 +00:00
fix(transfer): replace existing mysql schema objects during transfer
This commit is contained in:
@@ -31,7 +31,7 @@ import { ensureReadOnlyWriteAccess } from "@/lib/database/readOnlyWriteAccess";
|
||||
import { useProductionSafetyStore } from "@/stores/productionSafetyStore";
|
||||
import * as api from "@/lib/backend/api";
|
||||
import type { TransferContent, TransferObjectKind, TransferTableNameCase } from "@/lib/backend/api";
|
||||
import { crossFamilyTransferableKinds, isSameTransferFamily, isTransferPairSupported, transferObjectKindsForDatabase } from "@/lib/database/transferObjectKinds";
|
||||
import { crossFamilyTransferableKinds, isSameTransferFamily, isTransferPairSupported, transferObjectFamily, transferObjectKindsForDatabase } from "@/lib/database/transferObjectKinds";
|
||||
import ObjectSelectionTree from "@/components/transfer/ObjectSelectionTree.vue";
|
||||
import TransferTaskTree from "@/components/transfer/TransferTaskTree.vue";
|
||||
import DataTransferProgressDialog from "@/components/transfer/DataTransferProgressDialog.vue";
|
||||
@@ -199,7 +199,10 @@ const pendingTargetSchemaPrefill = ref("");
|
||||
// Options
|
||||
const selectedStrategy = ref<TransferStrategy>("append");
|
||||
const targetTableStrategy = computed<TransferStrategy>({
|
||||
get: () => (transferContent.value === "structureOnly" && selectedStrategy.value !== "rebuild" ? "append" : selectedStrategy.value),
|
||||
// Structure-only transfers have no table-data operation, but overwrite is
|
||||
// still meaningful for selected schema objects (for example MySQL routines).
|
||||
// Keep upsert unavailable in this mode while preserving an explicit overwrite.
|
||||
get: () => (transferContent.value === "structureOnly" && (selectedStrategy.value === "upsert" || (selectedStrategy.value === "overwrite" && !structureOverwriteSupported())) ? "append" : selectedStrategy.value),
|
||||
set: (value) => {
|
||||
selectedStrategy.value = value;
|
||||
},
|
||||
@@ -285,6 +288,12 @@ function requestedTableFilters(): Record<string, string> | undefined {
|
||||
const sourceTransferType = computed(() => transferDatabaseTypeForConnection(store.getConfig(sourceConnectionId.value)));
|
||||
const targetTransferType = computed(() => transferDatabaseTypeForConnection(store.getConfig(targetConnectionId.value)));
|
||||
const isXuguTransferPair = computed(() => sourceTransferType.value === "xugu" && targetTransferType.value === "xugu");
|
||||
|
||||
function structureOverwriteSupported(): boolean {
|
||||
if (transferContent.value !== "structureOnly") return true;
|
||||
const hasNonTableObject = Object.entries(selectedObjects.value).some(([kind, names]) => kind !== "TABLE" && (names?.size ?? 0) > 0);
|
||||
return transferObjectFamily(sourceTransferType.value) === "mysql" && transferObjectFamily(targetTransferType.value) === "mysql" && hasNonTableObject;
|
||||
}
|
||||
// The table-name-case selector is hidden for Xugu pairs; reset any earlier choice so the
|
||||
// blocking xuguPreserveTableNames hint cannot strand the user without a visible control.
|
||||
watch(isXuguTransferPair, (isXuguPair) => {
|
||||
@@ -1195,7 +1204,10 @@ function onSelectedTaskIdUpdate(id: string | null) {
|
||||
// ---- confirmation for the immutable, previewed transfer request ----
|
||||
|
||||
function transferStrategyLabel(request: api.TransferRequest): string {
|
||||
if (request.content === "structureOnly") return t(request.dropTargetBeforeCreate ? "transfer.modeRebuildStructure" : "transfer.modeKeepExisting");
|
||||
if (request.content === "structureOnly") {
|
||||
if (request.dropTargetBeforeCreate) return t("transfer.modeRebuildStructure");
|
||||
return t(request.mode === "overwrite" ? "transfer.modeReplaceExisting" : "transfer.modeKeepExisting");
|
||||
}
|
||||
const strategy = resolveTransferStrategy(request);
|
||||
const keys = { append: "modeAppend", overwrite: "modeOverwrite", upsert: "modeUpsert", rebuild: "modeRebuild" };
|
||||
return t(`transfer.${keys[strategy]}`);
|
||||
@@ -1265,7 +1277,7 @@ const confirmationDetails = computed(() => {
|
||||
lines.push(t("transfer.structurePlanSqlPreview"));
|
||||
|
||||
const unexpandedObjects = request.objects.filter((selection) => selection.objectType !== "TABLE" && selection.names.length > 0);
|
||||
if (unexpandedObjects.length > 0) {
|
||||
if (unexpandedObjects.length > 0 && !preview.structure.schemaObjectsExpanded) {
|
||||
lines.push(t("transfer.structurePreviewUnexpandedObjects", { objects: unexpandedObjects.map((selection) => selection.names.join(", ")).join("; ") }));
|
||||
}
|
||||
lines.push(t("transfer.structurePreviewRevalidation"));
|
||||
@@ -1599,7 +1611,7 @@ async function saveConfigTask() {
|
||||
</SelectTrigger>
|
||||
<SelectContent>
|
||||
<SelectItem value="append">{{ t(transferContent === "structureOnly" ? "transfer.modeKeepExisting" : "transfer.modeAppend") }}</SelectItem>
|
||||
<SelectItem v-if="transferContent !== 'structureOnly'" value="overwrite">{{ t("transfer.modeOverwrite") }}</SelectItem>
|
||||
<SelectItem v-if="transferContent !== 'structureOnly' || structureOverwriteSupported()" value="overwrite">{{ t(transferContent === "structureOnly" ? "transfer.modeReplaceExisting" : "transfer.modeOverwrite") }}</SelectItem>
|
||||
<SelectItem v-if="transferContent !== 'structureOnly'" value="upsert" :disabled="!upsertSupported">{{ t("transfer.modeUpsert") }}</SelectItem>
|
||||
<SelectItem value="rebuild" :disabled="!!rebuildDisabledReason">{{ t(transferContent === "structureOnly" ? "transfer.modeRebuildStructure" : "transfer.modeRebuild") }}</SelectItem>
|
||||
</SelectContent>
|
||||
|
||||
+1
-1
@@ -27,7 +27,7 @@ function structureRequest(overrides: Partial<TransferRequest> = {}): TransferReq
|
||||
}
|
||||
|
||||
function structurePreview(sql = '-- orders -> reporting.orders\nCREATE TABLE "reporting"."orders" ("id" integer);'): TransferOwnershipPreview["structure"] {
|
||||
return { sql, tables: [{ sourceTable: "orders", targetTable: "reporting.orders", preexisting: false, sql }], operations: [] };
|
||||
return { sql, tables: [{ sourceTable: "orders", targetTable: "reporting.orders", preexisting: false, sql }], operations: [], schemaObjectsExpanded: false };
|
||||
}
|
||||
|
||||
function preview(overrides: Partial<TransferOwnershipPreview> = {}): TransferOwnershipPreview {
|
||||
|
||||
@@ -7604,6 +7604,7 @@ export default {
|
||||
transferMode: "Transfer mode",
|
||||
modeAppend: "Append (INSERT)",
|
||||
modeOverwrite: "Overwrite (TRUNCATE + INSERT)",
|
||||
modeReplaceExisting: "Replace existing objects",
|
||||
modeUpsert: "Upsert (INSERT or UPDATE by PK)",
|
||||
targetTableNameCase: "Target table name case",
|
||||
quoteTargetColumnNames: "Quote target column names",
|
||||
|
||||
@@ -6820,6 +6820,7 @@ export default withEnglishFallback({
|
||||
transferMode: "Modo de transferencia",
|
||||
modeAppend: "Agregar (INSERT)",
|
||||
modeOverwrite: "Sobrescribir (TRUNCATE + INSERT)",
|
||||
modeReplaceExisting: "Reemplazar objetos existentes",
|
||||
modeUpsert: "Upsert (INSERT o UPDATE por PK)",
|
||||
targetTableNameCase: "Mayúsculas/minúsculas de tabla destino",
|
||||
quoteTargetColumnNames: "Citar columnas de destino",
|
||||
|
||||
@@ -6637,6 +6637,7 @@ export default withEnglishFallback({
|
||||
transferMode: "Modalità trasferimento",
|
||||
modeAppend: "Accoda (INSERT)",
|
||||
modeOverwrite: "Sovrascrivi (TRUNCATE + INSERT)",
|
||||
modeReplaceExisting: "Sostituisci oggetti esistenti",
|
||||
modeUpsert: "Aggiorna/Inserisci (INSERT o UPDATE per chiave primaria)",
|
||||
targetTableNameCase: "Maiuscole/minuscole tabella destinazione",
|
||||
quoteTargetColumnNames: "Quota colonne di destinazione",
|
||||
|
||||
@@ -6657,6 +6657,7 @@ export default withEnglishFallback({
|
||||
transferMode: "転送モード",
|
||||
modeAppend: "追加 (INSERT)",
|
||||
modeOverwrite: "上書き (TRUNCATE + INSERT)",
|
||||
modeReplaceExisting: "既存のオブジェクトを置き換え",
|
||||
modeUpsert: "アップサート (主キーでINSERT or UPDATE)",
|
||||
targetTableNameCase: "対象テーブル名の大小文字",
|
||||
quoteTargetColumnNames: "対象列名を引用",
|
||||
|
||||
@@ -6436,6 +6436,7 @@ export default withEnglishFallback({
|
||||
transferMode: "이전 모드",
|
||||
modeAppend: "추가 (INSERT)",
|
||||
modeOverwrite: "덮어쓰기 (TRUNCATE + INSERT)",
|
||||
modeReplaceExisting: "기존 객체 교체",
|
||||
modeUpsert: "Upsert (PK 기준 INSERT 또는 UPDATE)",
|
||||
targetTableNameCase: "대상 테이블 이름 대소문자",
|
||||
quoteTargetColumnNames: "대상 열 이름 인용",
|
||||
|
||||
@@ -6637,6 +6637,7 @@ export default withEnglishFallback({
|
||||
transferMode: "Modo de transferência",
|
||||
modeAppend: "Anexar (INSERT)",
|
||||
modeOverwrite: "Sobrescrever (TRUNCATE + INSERT)",
|
||||
modeReplaceExisting: "Substituir objetos existentes",
|
||||
modeUpsert: "Upsert (INSERT ou UPDATE por PK)",
|
||||
targetTableNameCase: "Maiúsculas/minúsculas da tabela de destino",
|
||||
quoteTargetColumnNames: "Citar colunas de destino",
|
||||
|
||||
@@ -7870,6 +7870,7 @@ export default withEnglishFallback({
|
||||
transferMode: "Режим переноса",
|
||||
modeAppend: "Добавить (INSERT)",
|
||||
modeOverwrite: "Перезаписать (TRUNCATE + INSERT)",
|
||||
modeReplaceExisting: "Заменить существующие объекты",
|
||||
modeUpsert: "Upsert (INSERT или UPDATE по PK)",
|
||||
targetTableNameCase: "Регистр имён целевых таблиц",
|
||||
quoteTargetColumnNames: "Заключать имена целевых столбцов в кавычки",
|
||||
|
||||
@@ -8260,6 +8260,7 @@ export default withEnglishFallback({
|
||||
transferMode: "Mode การถ่ายโอน",
|
||||
modeAppend: "Append (INSERT)",
|
||||
modeOverwrite: "Overwrite (TRUNCATE + INSERT)",
|
||||
modeReplaceExisting: "แทนที่ออบเจ็กต์ที่มีอยู่",
|
||||
modeUpsert: "Upsert (INSERT หรือ UPDATE ด้วย PK)",
|
||||
targetTableNameCase: "ตัวพิมพ์เล็กใหญ่ของชื่อ Table เป้าหมาย",
|
||||
quoteTargetColumnNames: "ใส่ Quote ชื่อ Column เป้าหมาย",
|
||||
|
||||
@@ -7586,6 +7586,7 @@ export default withEnglishFallback({
|
||||
transferMode: "传输模式",
|
||||
modeAppend: "追加 (INSERT)",
|
||||
modeOverwrite: "覆盖 (TRUNCATE + INSERT)",
|
||||
modeReplaceExisting: "存在则替换对象",
|
||||
modeUpsert: "更新插入 (按主键 INSERT 或 UPDATE)",
|
||||
targetTableNameCase: "目标表名大小写",
|
||||
quoteTargetColumnNames: "引用目标字段名",
|
||||
|
||||
@@ -5968,6 +5968,7 @@ export default withEnglishFallback({
|
||||
transferMode: "傳輸模式",
|
||||
modeAppend: "追加 (INSERT)",
|
||||
modeOverwrite: "覆蓋 (TRUNCATE + INSERT)",
|
||||
modeReplaceExisting: "已有物件則取代",
|
||||
modeUpsert: "更新插入 (按主鍵 INSERT 或 UPDATE)",
|
||||
batchSize: "每批筆數",
|
||||
operationTimeout: "操作逾時(秒)",
|
||||
|
||||
@@ -5894,6 +5894,8 @@ export interface TransferStructurePreview {
|
||||
sql: string;
|
||||
tables: TransferStructurePreviewTable[];
|
||||
operations: TransferStructureOperation[];
|
||||
/** Whether selected non-table schema objects are expanded in the SQL preview. */
|
||||
schemaObjectsExpanded: boolean;
|
||||
}
|
||||
|
||||
export interface TransferOwnershipPreview {
|
||||
|
||||
@@ -34,7 +34,9 @@ use crate::db;
|
||||
use crate::db::agent_driver::AgentTableReadStartParams;
|
||||
use crate::db::mongo_driver::MongoDocumentResult;
|
||||
use crate::models::connection::{ConnectionConfig, DatabaseType};
|
||||
use crate::object_source_sql::{build_executable_object_source_statements, EditableObjectSourceSqlInput};
|
||||
use crate::object_source_sql::{
|
||||
build_executable_object_source_statements, build_mysql_routine_replacement_statements, EditableObjectSourceSqlInput,
|
||||
};
|
||||
use crate::query::{
|
||||
agent_execute_query_params, is_dbx_query_timeout_error, pool_error_action, query_timeout_duration,
|
||||
should_discard_pool_after_query_timeout, wait_for_query_opt, PoolErrorAction, QueryExecutionOptions,
|
||||
@@ -393,6 +395,9 @@ pub struct TransferStructurePreview {
|
||||
pub sql: String,
|
||||
pub tables: Vec<TransferStructurePreviewTable>,
|
||||
pub operations: Vec<TransferStructureOperation>,
|
||||
/// Whether selected non-table schema objects are expanded in `sql`.
|
||||
#[serde(rename = "schemaObjectsExpanded")]
|
||||
pub schema_objects_expanded: bool,
|
||||
}
|
||||
|
||||
/// One user-facing logical operation produced by the structure transfer planner.
|
||||
@@ -1068,6 +1073,30 @@ pub fn validate_transfer_database_pair(
|
||||
if transfer_support_level(target_db_type).is_none() {
|
||||
return Err(format!("{} 暂未声明数据传输兼容级别", target_db_type.as_str()));
|
||||
}
|
||||
if !matches!(source_db_type, DatabaseType::Xugu)
|
||||
&& !matches!(target_db_type, DatabaseType::Xugu)
|
||||
&& matches!(request.content, TransferContent::StructureOnly)
|
||||
&& request.mode == TransferMode::Overwrite
|
||||
&& request.objects.as_ref().map_or_else(
|
||||
|| {
|
||||
// The legacy PostgreSQL→PostgreSQL request shape omitted
|
||||
// `objects` but still means "all schema objects".
|
||||
transfer_object_family(source_db_type) == Some(TransferObjectFamily::Postgres)
|
||||
&& transfer_object_family(target_db_type) == Some(TransferObjectFamily::Postgres)
|
||||
},
|
||||
|objects| {
|
||||
objects
|
||||
.iter()
|
||||
.any(|selection| selection.object_type != TransferObjectKind::Table && !selection.names.is_empty())
|
||||
},
|
||||
)
|
||||
&& !matches!(
|
||||
(transfer_object_family(source_db_type), transfer_object_family(target_db_type)),
|
||||
(Some(TransferObjectFamily::Mysql), Some(TransferObjectFamily::Mysql))
|
||||
)
|
||||
{
|
||||
return Err("结构-only 的“存在则替换”当前仅支持 MySQL 同族的非表对象".to_string());
|
||||
}
|
||||
if !matches!(source_db_type, DatabaseType::Xugu) && !matches!(target_db_type, DatabaseType::Xugu) {
|
||||
return Ok(());
|
||||
}
|
||||
@@ -1365,7 +1394,7 @@ pub fn target_object_exists_sql(
|
||||
/// Remove `DEFINER=`user`@`host`` tokens from MySQL DDL (they are not
|
||||
/// transferable and frequently reference accounts that don't exist on target).
|
||||
pub fn strip_mysql_definer(ddl: &str) -> String {
|
||||
let re = Regex::new(r"(?i)\bDEFINER\s*=\s*`[^`]*`@`[^`]*`\s*").unwrap();
|
||||
let re = Regex::new(r"(?i)\bDEFINER\s*=\s*`(?:``|[^`])*`@`(?:``|[^`])*`\s*").unwrap();
|
||||
re.replace_all(ddl, "").to_string()
|
||||
}
|
||||
|
||||
@@ -1375,8 +1404,10 @@ pub fn rewrite_mysql_schema_qualifier(ddl: &str, source_schema: &str, target_sch
|
||||
if source_schema == target_schema || source_schema.is_empty() {
|
||||
return ddl.to_string();
|
||||
}
|
||||
let re = Regex::new(&format!(r"`{}`\.", regex::escape(source_schema))).unwrap();
|
||||
re.replace_all(ddl, &format!("`{}`.", target_schema)).to_string()
|
||||
let source = quote_identifier(source_schema, &DatabaseType::Mysql);
|
||||
let target = quote_identifier(target_schema, &DatabaseType::Mysql);
|
||||
let re = Regex::new(&format!(r"{}\.", regex::escape(&source))).unwrap();
|
||||
re.replace_all(ddl, |_: ®ex::Captures<'_>| format!("{target}.")).to_string()
|
||||
}
|
||||
|
||||
pub fn mysql_trigger_ddl(
|
||||
@@ -1387,8 +1418,11 @@ pub fn mysql_trigger_ddl(
|
||||
table: &str,
|
||||
statement: &str,
|
||||
) -> String {
|
||||
let name = quote_identifier(name, &DatabaseType::Mysql);
|
||||
let schema = quote_identifier(schema, &DatabaseType::Mysql);
|
||||
let table = quote_identifier(table, &DatabaseType::Mysql);
|
||||
format!(
|
||||
"CREATE TRIGGER `{name}` {timing} {manipulation} ON `{schema}`.`{table}` FOR EACH ROW {statement}",
|
||||
"CREATE TRIGGER {name} {timing} {manipulation} ON {schema}.{table} FOR EACH ROW {statement}",
|
||||
name = name,
|
||||
timing = timing,
|
||||
manipulation = manipulation,
|
||||
@@ -1399,8 +1433,9 @@ pub fn mysql_trigger_ddl(
|
||||
}
|
||||
|
||||
pub fn mysql_event_ddl(_schema: &str, name: &str, status: &str, schedule: &str, body: &str) -> String {
|
||||
let name = quote_identifier(name, &DatabaseType::Mysql);
|
||||
format!(
|
||||
"CREATE EVENT `{name}` ON SCHEDULE {schedule} {status} DO {body}",
|
||||
"CREATE EVENT {name} ON SCHEDULE {schedule} {status} DO {body}",
|
||||
name = name,
|
||||
schedule = schedule,
|
||||
status = status,
|
||||
@@ -1416,10 +1451,12 @@ pub fn mysql_event_ddl(_schema: &str, name: &str, status: &str, schedule: &str,
|
||||
pub fn mysql_object_source_query(kind: &TransferObjectKind, database: &str, name: &str) -> Result<String, String> {
|
||||
let db = quote_string_literal(database);
|
||||
let n = quote_string_literal(name);
|
||||
let database_identifier = quote_identifier(database, &DatabaseType::Mysql);
|
||||
let object_identifier = quote_identifier(name, &DatabaseType::Mysql);
|
||||
let ddl = match kind {
|
||||
TransferObjectKind::View => format!("SHOW CREATE VIEW `{database}`.`{name}`"),
|
||||
TransferObjectKind::Procedure => format!("SHOW CREATE PROCEDURE `{database}`.`{name}`"),
|
||||
TransferObjectKind::Function => format!("SHOW CREATE FUNCTION `{database}`.`{name}`"),
|
||||
TransferObjectKind::View => format!("SHOW CREATE VIEW {database_identifier}.{object_identifier}"),
|
||||
TransferObjectKind::Procedure => format!("SHOW CREATE PROCEDURE {database_identifier}.{object_identifier}"),
|
||||
TransferObjectKind::Function => format!("SHOW CREATE FUNCTION {database_identifier}.{object_identifier}"),
|
||||
TransferObjectKind::Trigger => format!(
|
||||
"SELECT TRIGGER_NAME, ACTION_TIMING, EVENT_MANIPULATION, EVENT_OBJECT_TABLE, ACTION_STATEMENT \
|
||||
FROM information_schema.TRIGGERS WHERE TRIGGER_SCHEMA = {db} AND TRIGGER_NAME = {n}"
|
||||
@@ -1477,6 +1514,175 @@ pub fn mysql_object_ddl_from_result(
|
||||
}
|
||||
}
|
||||
|
||||
/// Build the statements needed to replace an existing MySQL schema object.
|
||||
fn mysql_object_replacement_statements(
|
||||
kind: &TransferObjectKind,
|
||||
target_database: &str,
|
||||
name: &str,
|
||||
ddl: &str,
|
||||
validation_name: Option<&str>,
|
||||
) -> Result<Vec<String>, String> {
|
||||
let source_kind = match kind {
|
||||
TransferObjectKind::Procedure => Some(db::ObjectSourceKind::Procedure),
|
||||
TransferObjectKind::Function => Some(db::ObjectSourceKind::Function),
|
||||
_ => None,
|
||||
};
|
||||
|
||||
if let Some(source_kind) = source_kind {
|
||||
let input = EditableObjectSourceSqlInput {
|
||||
database_type: DatabaseType::Mysql,
|
||||
object_type: source_kind,
|
||||
schema: Some(target_database.to_string()),
|
||||
name: name.to_string(),
|
||||
source: ddl.to_string(),
|
||||
};
|
||||
let statements = match validation_name {
|
||||
Some(validation_name) => build_mysql_routine_replacement_statements(input, validation_name)?,
|
||||
None => build_executable_object_source_statements(input)?,
|
||||
};
|
||||
// SHOW CREATE should return the selected routine name. Refuse to
|
||||
// replace anything if a malformed/renamed source DDL would make the
|
||||
// generated DROP target a different routine; otherwise a failed CREATE
|
||||
// could leave the requested target missing without a safe rollback.
|
||||
if validation_name.is_some()
|
||||
&& !statements.get(2).is_some_and(|statement| {
|
||||
mysql_replacement_statement_drops_target(kind, target_database, name, statement)
|
||||
})
|
||||
{
|
||||
return Err(format!("MySQL {kind:?} source DDL name does not match selected object {name:?}"));
|
||||
}
|
||||
return Ok(statements);
|
||||
}
|
||||
|
||||
let keyword = match kind {
|
||||
TransferObjectKind::View => "VIEW",
|
||||
TransferObjectKind::Trigger => "TRIGGER",
|
||||
TransferObjectKind::Event => "EVENT",
|
||||
_ => return Err(format!("MySQL object replacement not supported for {kind:?}")),
|
||||
};
|
||||
let qualified_name = format!(
|
||||
"{}.{}",
|
||||
quote_identifier(target_database, &DatabaseType::Mysql),
|
||||
quote_identifier(name, &DatabaseType::Mysql)
|
||||
);
|
||||
Ok(vec![format!("DROP {keyword} IF EXISTS {qualified_name}"), ddl.to_string()])
|
||||
}
|
||||
|
||||
fn mysql_object_drop_statement(kind: &TransferObjectKind, target_database: &str, name: &str) -> Result<String, String> {
|
||||
let keyword = match kind {
|
||||
TransferObjectKind::View => "VIEW",
|
||||
TransferObjectKind::Procedure => "PROCEDURE",
|
||||
TransferObjectKind::Function => "FUNCTION",
|
||||
TransferObjectKind::Trigger => "TRIGGER",
|
||||
TransferObjectKind::Event => "EVENT",
|
||||
_ => return Err(format!("MySQL object replacement not supported for {kind:?}")),
|
||||
};
|
||||
let qualified_name = format!(
|
||||
"{}.{}",
|
||||
quote_identifier(target_database, &DatabaseType::Mysql),
|
||||
quote_identifier(name, &DatabaseType::Mysql)
|
||||
);
|
||||
Ok(format!("DROP {keyword} IF EXISTS {qualified_name};"))
|
||||
}
|
||||
|
||||
fn mysql_transfer_validation_routine_name(
|
||||
transfer_id: &str,
|
||||
target_database: &str,
|
||||
kind: &TransferObjectKind,
|
||||
name: &str,
|
||||
) -> String {
|
||||
// Scope the temporary name to one transfer run and keep it below MySQL's
|
||||
// routine-name limit without exposing request text in executable SQL.
|
||||
let mut hash = 0xcbf29ce484222325u64;
|
||||
let kind_name = format!("{kind:?}");
|
||||
for value in [transfer_id, target_database, kind_name.as_str(), name] {
|
||||
for byte in value.bytes() {
|
||||
hash ^= byte as u64;
|
||||
hash = hash.wrapping_mul(0x100000001b3);
|
||||
}
|
||||
hash ^= 0xff;
|
||||
hash = hash.wrapping_mul(0x100000001b3);
|
||||
}
|
||||
format!("dbx_transfer_check_{hash:016x}")
|
||||
}
|
||||
|
||||
fn mysql_object_restore_statements(
|
||||
kind: &TransferObjectKind,
|
||||
target_database: &str,
|
||||
name: &str,
|
||||
ddl: &str,
|
||||
) -> Result<Vec<String>, String> {
|
||||
Ok(vec![mysql_object_drop_statement(kind, target_database, name)?, ddl.to_string()])
|
||||
}
|
||||
|
||||
fn normalize_mysql_statement_for_compare(statement: &str) -> String {
|
||||
statement.trim().trim_end_matches(';').trim().to_ascii_lowercase()
|
||||
}
|
||||
|
||||
fn mysql_replacement_statement_drops_target(
|
||||
kind: &TransferObjectKind,
|
||||
target_database: &str,
|
||||
target_name: &str,
|
||||
statement: &str,
|
||||
) -> bool {
|
||||
let Ok(expected) = mysql_object_drop_statement(kind, target_database, target_name) else {
|
||||
return false;
|
||||
};
|
||||
normalize_mysql_statement_for_compare(statement) == normalize_mysql_statement_for_compare(&expected)
|
||||
}
|
||||
|
||||
fn mysql_replacement_statement_drops_validation(
|
||||
kind: &TransferObjectKind,
|
||||
target_database: &str,
|
||||
statement: &str,
|
||||
validation_name: Option<&str>,
|
||||
) -> bool {
|
||||
let Some(validation_name) = validation_name else {
|
||||
return false;
|
||||
};
|
||||
let Ok(expected) = mysql_object_drop_statement(kind, target_database, validation_name) else {
|
||||
return false;
|
||||
};
|
||||
normalize_mysql_statement_for_compare(statement) == normalize_mysql_statement_for_compare(&expected)
|
||||
}
|
||||
|
||||
fn mysql_replacement_statement_creates_validation(
|
||||
kind: &TransferObjectKind,
|
||||
statement: &str,
|
||||
validation_name: Option<&str>,
|
||||
) -> bool {
|
||||
let Some(validation_name) = validation_name else {
|
||||
return false;
|
||||
};
|
||||
let keyword = match kind {
|
||||
TransferObjectKind::Procedure => "procedure",
|
||||
TransferObjectKind::Function => "function",
|
||||
_ => return false,
|
||||
};
|
||||
let prefix =
|
||||
Regex::new(&format!(r"(?is)^\s*CREATE\s+(?:DEFINER\s*=\s*.+?\s+)?{keyword}\s+(?:IF\s+NOT\s+EXISTS\s+)?"))
|
||||
.expect("valid MySQL routine validation prefix");
|
||||
let Some(captures) = prefix.find(statement) else {
|
||||
return false;
|
||||
};
|
||||
statement[captures.end()..]
|
||||
.trim_start()
|
||||
.to_ascii_lowercase()
|
||||
.starts_with("e_identifier(validation_name, &DatabaseType::Mysql).to_ascii_lowercase())
|
||||
}
|
||||
|
||||
async fn mysql_existing_object_ddl(
|
||||
state: &AppState,
|
||||
target_pool_key: &str,
|
||||
target_database: &str,
|
||||
kind: &TransferObjectKind,
|
||||
name: &str,
|
||||
) -> Result<String, String> {
|
||||
let query = mysql_object_source_query(kind, target_database, name)?;
|
||||
let result = execute_on_pool(state, target_pool_key, &query).await?;
|
||||
mysql_object_ddl_from_result(kind, target_database, &result.rows)
|
||||
}
|
||||
|
||||
pub fn oracle_object_source_query(kind: &TransferObjectKind, schema: &str, name: &str) -> Result<String, String> {
|
||||
let object_type = match kind {
|
||||
TransferObjectKind::View => "VIEW",
|
||||
@@ -9327,6 +9533,9 @@ where
|
||||
F: FnMut(TransferProgress),
|
||||
{
|
||||
let mut outcome = TransferObjectOutcome::default();
|
||||
// Overwrite is the explicit opt-in for replacing selected MySQL schema
|
||||
// objects; append keeps the historical skip-existing behavior.
|
||||
let replace_existing = request.mode == TransferMode::Overwrite;
|
||||
let source_db = &request.source_database;
|
||||
let target_db =
|
||||
if request.target_database.trim().is_empty() { source_db.as_str() } else { request.target_database.as_str() };
|
||||
@@ -9355,24 +9564,196 @@ where
|
||||
// skip if the target already has it
|
||||
let exists_sql = target_object_exists_sql(&DatabaseType::Mysql, target_db, &name, &kind)?;
|
||||
let exists = !execute_on_pool(state, target_pool_key, &exists_sql).await?.rows.is_empty();
|
||||
if exists {
|
||||
if exists && !replace_existing {
|
||||
outcome.skipped.push(format!("{kind:?}:{name}"));
|
||||
progress(&mut outcome, TransferStatus::Running, None);
|
||||
continue;
|
||||
}
|
||||
let validation_name = if exists
|
||||
&& replace_existing
|
||||
&& matches!(kind, TransferObjectKind::Procedure | TransferObjectKind::Function)
|
||||
{
|
||||
let validation_name =
|
||||
mysql_transfer_validation_routine_name(&request.transfer_id, target_db, &kind, &name);
|
||||
let validation_exists_sql =
|
||||
target_object_exists_sql(&DatabaseType::Mysql, target_db, &validation_name, &kind)?;
|
||||
let validation_exists =
|
||||
!execute_on_pool(state, target_pool_key, &validation_exists_sql).await?.rows.is_empty();
|
||||
if validation_exists {
|
||||
let error = format!(
|
||||
"Temporary MySQL routine validation name already exists: {target_db}.{validation_name}; retry the transfer"
|
||||
);
|
||||
outcome.failed.push(format!("{kind:?}:{name}"));
|
||||
progress(&mut outcome, TransferStatus::Error, Some(error));
|
||||
continue;
|
||||
}
|
||||
Some(validation_name)
|
||||
} else {
|
||||
None
|
||||
};
|
||||
let replacement_backup = if exists && replace_existing {
|
||||
match mysql_existing_object_ddl(state, target_pool_key, target_db, &kind, &name).await {
|
||||
Ok(ddl) => Some(ddl),
|
||||
Err(error) => {
|
||||
outcome.failed.push(format!("{kind:?}:{name}"));
|
||||
progress(&mut outcome, TransferStatus::Error, Some(error));
|
||||
continue;
|
||||
}
|
||||
}
|
||||
} else {
|
||||
None
|
||||
};
|
||||
let query = mysql_object_source_query(&kind, source_db, &name)?;
|
||||
let result = execute_on_pool(state, source_pool_key, &query).await?;
|
||||
let raw_ddl = mysql_object_ddl_from_result(&kind, source_db, &result.rows)?;
|
||||
let ddl = strip_mysql_definer(&raw_ddl);
|
||||
let ddl = rewrite_mysql_schema_qualifier(&ddl, source_db, target_db);
|
||||
match execute_on_pool(state, target_pool_key, &ddl).await {
|
||||
Ok(_) => {
|
||||
let statements = if exists && replace_existing {
|
||||
match mysql_object_replacement_statements(&kind, target_db, &name, &ddl, validation_name.as_deref()) {
|
||||
Ok(statements) => statements,
|
||||
Err(error) => {
|
||||
outcome.failed.push(format!("{kind:?}:{name}"));
|
||||
progress(&mut outcome, TransferStatus::Error, Some(error));
|
||||
continue;
|
||||
}
|
||||
}
|
||||
} else {
|
||||
vec![ddl]
|
||||
};
|
||||
let mut execution_error = None;
|
||||
let mut target_drop_attempted = false;
|
||||
let mut target_drop_completed = false;
|
||||
let mut validation_created = false;
|
||||
let mut validation_cleaned = false;
|
||||
for statement in statements {
|
||||
let creates_validation =
|
||||
mysql_replacement_statement_creates_validation(&kind, &statement, validation_name.as_deref());
|
||||
let drops_validation = mysql_replacement_statement_drops_validation(
|
||||
&kind,
|
||||
target_db,
|
||||
&statement,
|
||||
validation_name.as_deref(),
|
||||
);
|
||||
let drops_target = mysql_replacement_statement_drops_target(&kind, target_db, &name, &statement);
|
||||
target_drop_attempted |= drops_target;
|
||||
if let Err(error) = execute_on_pool(state, target_pool_key, &statement).await {
|
||||
execution_error = Some(error);
|
||||
break;
|
||||
}
|
||||
validation_created |= creates_validation;
|
||||
validation_cleaned |= drops_validation;
|
||||
target_drop_completed |= drops_target;
|
||||
}
|
||||
match execution_error {
|
||||
None => {
|
||||
outcome.transferred.push(format!("{kind:?}:{name}"));
|
||||
progress(&mut outcome, TransferStatus::Running, None);
|
||||
}
|
||||
Err(e) => {
|
||||
Some(mut error) => {
|
||||
if target_drop_attempted && !target_drop_completed {
|
||||
match target_object_exists_sql(&DatabaseType::Mysql, target_db, &name, &kind) {
|
||||
Ok(exists_sql) => match execute_on_pool(state, target_pool_key, &exists_sql).await {
|
||||
Ok(result) if result.rows.is_empty() => target_drop_completed = true,
|
||||
Ok(_) => {}
|
||||
Err(verify_error) => {
|
||||
error = format!(
|
||||
"{error}; failed to verify whether the original object still exists: {verify_error}"
|
||||
);
|
||||
}
|
||||
},
|
||||
Err(verify_error) => {
|
||||
error = format!(
|
||||
"{error}; failed to build verification for the original object: {verify_error}"
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
if validation_created && !validation_cleaned {
|
||||
if let Some(validation_name) = validation_name.as_deref() {
|
||||
match mysql_object_drop_statement(&kind, target_db, validation_name) {
|
||||
Ok(cleanup) => {
|
||||
if let Err(cleanup_error) = execute_on_pool(state, target_pool_key, &cleanup).await
|
||||
{
|
||||
error = format!(
|
||||
"{error}; failed to clean up temporary validation routine: {cleanup_error}"
|
||||
);
|
||||
}
|
||||
}
|
||||
Err(cleanup_error) => {
|
||||
error = format!(
|
||||
"{error}; failed to build temporary validation cleanup: {cleanup_error}"
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
if target_drop_completed {
|
||||
if let Some(backup_ddl) = replacement_backup.as_deref() {
|
||||
match mysql_object_restore_statements(&kind, target_db, &name, backup_ddl) {
|
||||
Ok(restore_statements) => {
|
||||
let mut restore_error = None;
|
||||
for statement in restore_statements {
|
||||
if let Err(restore) = execute_on_pool(state, target_pool_key, &statement).await
|
||||
{
|
||||
restore_error = Some(restore);
|
||||
break;
|
||||
}
|
||||
}
|
||||
if let Some(restore) = restore_error {
|
||||
// A target routine may have a DEFINER account that the
|
||||
// transfer user cannot recreate. Retry with the same
|
||||
// definer-stripped form used for normal transfers before
|
||||
// reporting that the original object is lost.
|
||||
let stripped_backup = strip_mysql_definer(backup_ddl);
|
||||
if stripped_backup != backup_ddl {
|
||||
match mysql_object_restore_statements(
|
||||
&kind,
|
||||
target_db,
|
||||
&name,
|
||||
&stripped_backup,
|
||||
) {
|
||||
Ok(fallback_statements) => {
|
||||
let mut fallback_error = None;
|
||||
for statement in fallback_statements {
|
||||
if let Err(fallback) =
|
||||
execute_on_pool(state, target_pool_key, &statement).await
|
||||
{
|
||||
fallback_error = Some(fallback);
|
||||
break;
|
||||
}
|
||||
}
|
||||
if let Some(fallback) = fallback_error {
|
||||
error = format!(
|
||||
"{error}; failed to restore the original object: {restore}; fallback restore also failed: {fallback}"
|
||||
);
|
||||
} else {
|
||||
error = format!(
|
||||
"{error}; the original object was restored without its original DEFINER"
|
||||
);
|
||||
}
|
||||
}
|
||||
Err(fallback) => {
|
||||
error = format!(
|
||||
"{error}; failed to restore the original object: {restore}; fallback restore could not be built: {fallback}"
|
||||
);
|
||||
}
|
||||
}
|
||||
} else {
|
||||
error =
|
||||
format!("{error}; failed to restore the original object: {restore}");
|
||||
}
|
||||
} else {
|
||||
error = format!("{error}; the original object was restored");
|
||||
}
|
||||
}
|
||||
Err(restore) => {
|
||||
error = format!("{error}; failed to build restore statements: {restore}");
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
outcome.failed.push(format!("{kind:?}:{name}"));
|
||||
progress(&mut outcome, TransferStatus::Error, Some(e));
|
||||
progress(&mut outcome, TransferStatus::Error, Some(error));
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -14920,6 +15301,30 @@ CREATE TABLE "Other"."prefix""Source"."NAME" ("ID" INT);"#;
|
||||
assert!(error.contains("gbase"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn structure_only_object_overwrite_is_limited_to_mysql_family() {
|
||||
let request = TransferRequest {
|
||||
content: TransferContent::StructureOnly,
|
||||
mode: TransferMode::Overwrite,
|
||||
objects: Some(vec![TransferObjectSelection {
|
||||
object_type: TransferObjectKind::Procedure,
|
||||
names: vec!["refresh_cache".into()],
|
||||
}]),
|
||||
..base_request()
|
||||
};
|
||||
|
||||
assert!(validate_transfer_database_pair(&request, &DatabaseType::Mysql, &DatabaseType::Mysql).is_ok());
|
||||
let error = validate_transfer_database_pair(&request, &DatabaseType::Postgres, &DatabaseType::Postgres)
|
||||
.expect_err("non-MySQL object replacement must be rejected rather than silently skipped");
|
||||
assert!(error.contains("MySQL"), "{error}");
|
||||
|
||||
let legacy_request = TransferRequest { objects: None, ..request };
|
||||
assert!(
|
||||
validate_transfer_database_pair(&legacy_request, &DatabaseType::Postgres, &DatabaseType::Postgres,)
|
||||
.is_err()
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn xugu_structure_transfer_rejects_renamed_tables_and_non_table_objects() {
|
||||
let request = TransferRequest { target_table_name_case: TransferTableNameCase::Upper, ..base_request() };
|
||||
@@ -15475,6 +15880,10 @@ CREATE TABLE "Other"."prefix""Source"."NAME" ("ID" INT);"#;
|
||||
strip_mysql_definer("CREATE ALGORITHM=UNDEFINED DEFINER=`root`@`localhost` VIEW v AS SELECT 1"),
|
||||
"CREATE ALGORITHM=UNDEFINED VIEW v AS SELECT 1"
|
||||
);
|
||||
assert_eq!(
|
||||
strip_mysql_definer("CREATE DEFINER=`u``ser`@`host``name` PROCEDURE p() BEGIN END"),
|
||||
"CREATE PROCEDURE p() BEGIN END"
|
||||
);
|
||||
assert_eq!(strip_mysql_definer("CREATE PROCEDURE p() BEGIN END"), "CREATE PROCEDURE p() BEGIN END");
|
||||
}
|
||||
|
||||
@@ -15484,6 +15893,14 @@ CREATE TABLE "Other"."prefix""Source"."NAME" ("ID" INT);"#;
|
||||
rewrite_mysql_schema_qualifier("CREATE VIEW `src`.`v` AS SELECT 1 FROM `src`.`t`", "src", "dst"),
|
||||
"CREATE VIEW `dst`.`v` AS SELECT 1 FROM `dst`.`t`"
|
||||
);
|
||||
assert_eq!(
|
||||
rewrite_mysql_schema_qualifier(
|
||||
"CREATE VIEW `shop``tenant`.`v` AS SELECT 1 FROM `shop``tenant`.`t`",
|
||||
"shop`tenant",
|
||||
"dst$schema",
|
||||
),
|
||||
"CREATE VIEW `dst$schema`.`v` AS SELECT 1 FROM `dst$schema`.`t`"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
@@ -15495,6 +15912,10 @@ CREATE TABLE "Other"."prefix""Source"."NAME" ("ID" INT);"#;
|
||||
);
|
||||
let event = mysql_event_ddl("shop", "ev1", "ENABLE", "EVERY 1 DAY", "DELETE FROM logs");
|
||||
assert_eq!(event, "CREATE EVENT `ev1` ON SCHEDULE EVERY 1 DAY ENABLE DO DELETE FROM logs");
|
||||
|
||||
let quoted = mysql_trigger_ddl("shop`tenant", "trg`1", "BEFORE", "INSERT", "user`events", "SET NEW.id = 1");
|
||||
assert!(quoted.contains("CREATE TRIGGER `trg``1`"));
|
||||
assert!(quoted.contains("ON `shop``tenant`.`user``events`"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
@@ -15755,6 +16176,8 @@ CREATE TABLE "Other"."prefix""Source"."NAME" ("ID" INT);"#;
|
||||
fn builds_mysql_object_source_queries() {
|
||||
let sql = mysql_object_source_query(&TransferObjectKind::View, "shop", "v1").unwrap();
|
||||
assert!(sql.contains("SHOW CREATE VIEW"));
|
||||
let quoted = mysql_object_source_query(&TransferObjectKind::View, "tenant`schema", "active`view").unwrap();
|
||||
assert!(quoted.contains("SHOW CREATE VIEW `tenant``schema`.`active``view`"), "{quoted}");
|
||||
let sql = mysql_object_source_query(&TransferObjectKind::Trigger, "shop", "trg1").unwrap();
|
||||
assert!(sql.contains("information_schema.TRIGGERS"));
|
||||
assert!(sql.contains("TRIGGER_NAME = 'trg1'"));
|
||||
@@ -15887,6 +16310,149 @@ CREATE TABLE "Other"."prefix""Source"."NAME" ("ID" INT);"#;
|
||||
assert_eq!(views, vec!["v1", "v2"]);
|
||||
assert!(selected_object_names(&selections, &TransferObjectKind::Event).is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn mysql_overwrite_builds_drop_and_create_for_existing_objects() {
|
||||
let view = mysql_object_replacement_statements(
|
||||
&TransferObjectKind::View,
|
||||
"target_db",
|
||||
"v1",
|
||||
"CREATE VIEW `target_db`.`v1` AS SELECT 1",
|
||||
None,
|
||||
)
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
view,
|
||||
vec![
|
||||
"DROP VIEW IF EXISTS `target_db`.`v1`".to_string(),
|
||||
"CREATE VIEW `target_db`.`v1` AS SELECT 1".to_string(),
|
||||
]
|
||||
);
|
||||
let trigger = mysql_object_replacement_statements(
|
||||
&TransferObjectKind::Trigger,
|
||||
"target_db",
|
||||
"trg1",
|
||||
"CREATE TRIGGER `trg1` BEFORE INSERT ON `target_db`.`items` FOR EACH ROW SET NEW.id = 1",
|
||||
None,
|
||||
)
|
||||
.unwrap();
|
||||
assert_eq!(trigger[0], "DROP TRIGGER IF EXISTS `target_db`.`trg1`".to_string());
|
||||
let event = mysql_object_replacement_statements(
|
||||
&TransferObjectKind::Event,
|
||||
"target_db",
|
||||
"ev1",
|
||||
"CREATE EVENT `ev1` ON SCHEDULE EVERY 1 DAY ENABLE DO SELECT 1",
|
||||
None,
|
||||
)
|
||||
.unwrap();
|
||||
assert_eq!(event[0], "DROP EVENT IF EXISTS `target_db`.`ev1`".to_string());
|
||||
|
||||
let validation_name =
|
||||
mysql_transfer_validation_routine_name("transfer-1", "target_db", &TransferObjectKind::Procedure, "p1");
|
||||
let routine = mysql_object_replacement_statements(
|
||||
&TransferObjectKind::Procedure,
|
||||
"target_db",
|
||||
"p1",
|
||||
"CREATE PROCEDURE `p1`() BEGIN SELECT 1; END",
|
||||
Some(&validation_name),
|
||||
)
|
||||
.unwrap();
|
||||
assert_eq!(routine.len(), 4);
|
||||
assert!(routine[0].contains(&format!("CREATE PROCEDURE `{validation_name}`")));
|
||||
assert!(routine[1].contains(&format!("DROP PROCEDURE IF EXISTS `target_db`.`{validation_name}`")));
|
||||
assert!(routine[2].contains("DROP PROCEDURE IF EXISTS `target_db`.`p1`"));
|
||||
assert!(!routine[0].to_ascii_uppercase().contains("DROP"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn restore_statements_keep_original_definer_until_fallback() {
|
||||
let backup = "CREATE DEFINER=`owner`@`localhost` PROCEDURE `p1`() BEGIN END";
|
||||
let restore =
|
||||
mysql_object_restore_statements(&TransferObjectKind::Procedure, "target_db", "p1", backup).unwrap();
|
||||
assert_eq!(restore[0], "DROP PROCEDURE IF EXISTS `target_db`.`p1`;".to_string());
|
||||
assert_eq!(restore[1], backup);
|
||||
assert_eq!(strip_mysql_definer(backup), "CREATE PROCEDURE `p1`() BEGIN END");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn validation_drop_is_not_treated_as_target_drop() {
|
||||
let validation_name = "dbx_transfer_check_0123456789abcdef";
|
||||
let validation_drop = format!("DROP PROCEDURE IF EXISTS `target_db`.`{validation_name}`;");
|
||||
let target_drop = "DROP PROCEDURE IF EXISTS `target_db`.`p1`;";
|
||||
assert!(!mysql_replacement_statement_drops_target(
|
||||
&TransferObjectKind::Procedure,
|
||||
"target_db",
|
||||
"p1",
|
||||
&validation_drop,
|
||||
));
|
||||
assert!(mysql_replacement_statement_drops_target(
|
||||
&TransferObjectKind::Procedure,
|
||||
"target_db",
|
||||
"p1",
|
||||
target_drop,
|
||||
));
|
||||
let dotted_name = "routine.name";
|
||||
let dotted_drop = "DROP PROCEDURE IF EXISTS `target_db`.`routine.name`;";
|
||||
assert!(mysql_replacement_statement_drops_target(
|
||||
&TransferObjectKind::Procedure,
|
||||
"target_db",
|
||||
dotted_name,
|
||||
dotted_drop,
|
||||
));
|
||||
let validation_create = format!("CREATE PROCEDURE `{validation_name}`() BEGIN END;");
|
||||
assert!(mysql_replacement_statement_creates_validation(
|
||||
&TransferObjectKind::Procedure,
|
||||
&validation_create,
|
||||
Some(validation_name),
|
||||
));
|
||||
assert!(mysql_replacement_statement_drops_validation(
|
||||
&TransferObjectKind::Procedure,
|
||||
"target_db",
|
||||
&validation_drop,
|
||||
Some(validation_name),
|
||||
));
|
||||
|
||||
let differently_cased_target_drop = "DROP PROCEDURE IF EXISTS `target_db`.`P1`;";
|
||||
assert!(mysql_replacement_statement_drops_target(
|
||||
&TransferObjectKind::Procedure,
|
||||
"target_db",
|
||||
"p1",
|
||||
differently_cased_target_drop,
|
||||
));
|
||||
|
||||
let current_user_validation_create =
|
||||
format!("CREATE DEFINER=CURRENT_USER() PROCEDURE `{validation_name}`() BEGIN END;");
|
||||
assert!(mysql_replacement_statement_creates_validation(
|
||||
&TransferObjectKind::Procedure,
|
||||
¤t_user_validation_create,
|
||||
Some(validation_name),
|
||||
));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn mysql_routine_replacement_rejects_a_source_name_mismatch() {
|
||||
let validation_name =
|
||||
mysql_transfer_validation_routine_name("transfer-1", "target_db", &TransferObjectKind::Procedure, "p1");
|
||||
let error = mysql_object_replacement_statements(
|
||||
&TransferObjectKind::Procedure,
|
||||
"target_db",
|
||||
"p1",
|
||||
"CREATE PROCEDURE `p2`() BEGIN END",
|
||||
Some(&validation_name),
|
||||
)
|
||||
.expect_err("a source routine with a different name must not replace the requested target");
|
||||
assert!(error.contains("does not match"), "{error}");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn transfer_validation_scopes_routine_validation_names_to_the_run() {
|
||||
let first =
|
||||
mysql_transfer_validation_routine_name("transfer-1", "target_db", &TransferObjectKind::Procedure, "p1");
|
||||
let second =
|
||||
mysql_transfer_validation_routine_name("transfer-2", "target_db", &TransferObjectKind::Procedure, "p1");
|
||||
assert_ne!(first, second);
|
||||
assert!(first.len() < 64);
|
||||
}
|
||||
}
|
||||
|
||||
mod transfer_oracle_executor_tests {
|
||||
|
||||
@@ -28,11 +28,7 @@ pub async fn clear_foreign_key_linked_overwrite_targets(
|
||||
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)
|
||||
{
|
||||
if !overwrite_clear_prepass_enabled(request, tables.len(), &target_db_type) {
|
||||
return Ok(HashSet::new());
|
||||
}
|
||||
|
||||
@@ -141,6 +137,20 @@ pub async fn clear_foreign_key_linked_overwrite_targets(
|
||||
Ok(cleared)
|
||||
}
|
||||
|
||||
fn overwrite_clear_prepass_enabled(
|
||||
request: &TransferRequest,
|
||||
table_count: usize,
|
||||
target_db_type: &DatabaseType,
|
||||
) -> bool {
|
||||
request.mode == TransferMode::Overwrite
|
||||
// Structure-only transfers use overwrite for selected schema objects
|
||||
// (for example MySQL routines), but must never mutate table rows.
|
||||
&& should_copy_data(&request.content)
|
||||
&& !request.drop_target_before_create
|
||||
&& table_count >= 2
|
||||
&& !is_mongodb_transfer_type(target_db_type)
|
||||
}
|
||||
|
||||
pub(super) fn overwrite_pre_pass_clear_sql(
|
||||
table: &str,
|
||||
schema: &str,
|
||||
@@ -159,6 +169,32 @@ pub(super) fn overwrite_pre_pass_clear_sql(
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
fn request(content: TransferContent) -> TransferRequest {
|
||||
TransferRequest {
|
||||
transfer_id: "transfer-test".into(),
|
||||
source_connection_id: "source".into(),
|
||||
source_database: "source".into(),
|
||||
source_schema: "source".into(),
|
||||
source_catalog: None,
|
||||
target_connection_id: "target".into(),
|
||||
target_database: "target".into(),
|
||||
target_schema: "target".into(),
|
||||
target_catalog: None,
|
||||
tables: vec!["parent".into(), "child".into()],
|
||||
create_table: true,
|
||||
content,
|
||||
objects: None,
|
||||
mode: TransferMode::Overwrite,
|
||||
target_table_name_case: TransferTableNameCase::Preserve,
|
||||
quote_target_column_names: true,
|
||||
ownership_policy: TransferOwnershipPolicy::Preserve,
|
||||
batch_size: 1000,
|
||||
table_filters: HashMap::new(),
|
||||
drop_target_before_create: false,
|
||||
drop_target_confirmed: false,
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn referenced_targets_are_deleted_instead_of_truncated() {
|
||||
assert_eq!(
|
||||
@@ -174,4 +210,12 @@ mod tests {
|
||||
"TRUNCATE TABLE \"APP\".\"EVENTS\" IMMEDIATE"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn structure_only_overwrite_does_not_clear_table_rows() {
|
||||
let structure_request = request(TransferContent::StructureOnly);
|
||||
assert!(!overwrite_clear_prepass_enabled(&structure_request, 2, &DatabaseType::Mysql));
|
||||
let data_request = request(TransferContent::StructureAndData);
|
||||
assert!(overwrite_clear_prepass_enabled(&data_request, 2, &DatabaseType::Mysql));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -26,7 +26,7 @@ const CREATE_TABLE_DISABLED_NOTE: &str =
|
||||
"-- This transfer has structure DDL disabled (createTable = false): no structure statements will run.";
|
||||
|
||||
/// Emitted when the transfer also moves non-table schema objects whose DDL this preview
|
||||
/// deliberately does not expand (views, routines, triggers, standalone sequences, ...).
|
||||
/// does not expand (except for the MySQL overwrite path, which is rendered explicitly).
|
||||
const UNEXPANDED_OBJECTS_NOTE: &str = "\
|
||||
-- Additional schema objects selected by this request (or the legacy PostgreSQL default) are transferred.
|
||||
-- Their generated DDL is not expanded in this preview.";
|
||||
@@ -49,11 +49,25 @@ pub(super) async fn build_structure_preview(
|
||||
source_pool_key: &str,
|
||||
target_pool_key: &str,
|
||||
) -> Result<TransferStructurePreview, String> {
|
||||
let schema_object_preview = mysql_schema_object_preview_sql(
|
||||
state,
|
||||
request,
|
||||
source_db_type,
|
||||
target_db_type,
|
||||
source_pool_key,
|
||||
target_pool_key,
|
||||
)
|
||||
.await?;
|
||||
if !request.create_table {
|
||||
let mut sections = vec![CREATE_TABLE_DISABLED_NOTE.to_string()];
|
||||
if let Some(sql) = &schema_object_preview {
|
||||
sections.push(sql.clone());
|
||||
}
|
||||
return Ok(TransferStructurePreview {
|
||||
sql: CREATE_TABLE_DISABLED_NOTE.to_string(),
|
||||
sql: sections.join("\n\n"),
|
||||
tables: Vec::new(),
|
||||
operations: Vec::new(),
|
||||
schema_objects_expanded: schema_object_preview.is_some(),
|
||||
});
|
||||
}
|
||||
|
||||
@@ -165,9 +179,94 @@ pub(super) async fn build_structure_preview(
|
||||
));
|
||||
}
|
||||
operations.extend(deferred_fk_operations);
|
||||
sections.extend(unexpanded_schema_object_notes(source_db_type, target_db_type, request));
|
||||
if let Some(sql) = schema_object_preview {
|
||||
sections.push(sql);
|
||||
} else {
|
||||
sections.extend(unexpanded_schema_object_notes(source_db_type, target_db_type, request));
|
||||
}
|
||||
|
||||
Ok(TransferStructurePreview { sql: sections.join("\n\n"), tables: planned_tables, operations })
|
||||
Ok(TransferStructurePreview {
|
||||
sql: sections.join("\n\n"),
|
||||
tables: planned_tables,
|
||||
operations,
|
||||
schema_objects_expanded: request.mode == TransferMode::Overwrite
|
||||
&& transfer_object_family(source_db_type) == Some(TransferObjectFamily::Mysql)
|
||||
&& transfer_object_family(target_db_type) == Some(TransferObjectFamily::Mysql)
|
||||
&& request.object_selection_mode().selections().iter().any(|selection| {
|
||||
selection.object_type != TransferObjectKind::Table
|
||||
&& transfer_object_kinds_for_family(&TransferObjectFamily::Mysql).contains(&selection.object_type)
|
||||
&& !selection.names.is_empty()
|
||||
}),
|
||||
})
|
||||
}
|
||||
|
||||
/// Expands the MySQL overwrite path in the read-only structure preview. The
|
||||
/// executor re-reads the same source DDL and target existence state, but the
|
||||
/// preview intentionally never executes any generated DDL.
|
||||
async fn mysql_schema_object_preview_sql(
|
||||
state: &Arc<AppState>,
|
||||
request: &TransferRequest,
|
||||
source_db_type: &DatabaseType,
|
||||
target_db_type: &DatabaseType,
|
||||
source_pool_key: &str,
|
||||
target_pool_key: &str,
|
||||
) -> Result<Option<String>, String> {
|
||||
if request.mode != TransferMode::Overwrite
|
||||
|| transfer_object_family(source_db_type) != Some(TransferObjectFamily::Mysql)
|
||||
|| transfer_object_family(target_db_type) != Some(TransferObjectFamily::Mysql)
|
||||
{
|
||||
return Ok(None);
|
||||
}
|
||||
let selections = request.object_selection_mode().selections();
|
||||
let supported_kinds = transfer_object_kinds_for_family(&TransferObjectFamily::Mysql);
|
||||
if !selections.iter().any(|selection| {
|
||||
selection.object_type != TransferObjectKind::Table
|
||||
&& supported_kinds.contains(&selection.object_type)
|
||||
&& !selection.names.is_empty()
|
||||
}) {
|
||||
return Ok(None);
|
||||
}
|
||||
|
||||
let source_database = &request.source_database;
|
||||
let target_database = if request.target_database.trim().is_empty() {
|
||||
source_database.as_str()
|
||||
} else {
|
||||
request.target_database.as_str()
|
||||
};
|
||||
let mut sections = Vec::new();
|
||||
let order = ordered_transfer_object_kinds(
|
||||
selections
|
||||
.iter()
|
||||
.filter(|selection| {
|
||||
selection.object_type != TransferObjectKind::Table && supported_kinds.contains(&selection.object_type)
|
||||
})
|
||||
.map(|selection| selection.object_type)
|
||||
.collect(),
|
||||
);
|
||||
for kind in order {
|
||||
for name in selected_object_names(selections, &kind) {
|
||||
let exists_sql = target_object_exists_sql(&DatabaseType::Mysql, target_database, &name, &kind)?;
|
||||
let exists = !execute_on_pool(state, target_pool_key, &exists_sql).await?.rows.is_empty();
|
||||
let query = mysql_object_source_query(&kind, source_database, &name)?;
|
||||
let result = execute_on_pool(state, source_pool_key, &query).await?;
|
||||
let raw_ddl = mysql_object_ddl_from_result(&kind, source_database, &result.rows)?;
|
||||
let ddl = rewrite_mysql_schema_qualifier(&strip_mysql_definer(&raw_ddl), source_database, target_database);
|
||||
let validation_name =
|
||||
if exists && matches!(kind, TransferObjectKind::Procedure | TransferObjectKind::Function) {
|
||||
Some(mysql_transfer_validation_routine_name(&request.transfer_id, target_database, &kind, &name))
|
||||
} else {
|
||||
None
|
||||
};
|
||||
let statements = if exists {
|
||||
mysql_object_replacement_statements(&kind, target_database, &name, &ddl, validation_name.as_deref())?
|
||||
} else {
|
||||
vec![ddl]
|
||||
};
|
||||
let operation = if exists { "Replace" } else { "Create" };
|
||||
sections.push(format!("-- {operation} MySQL {kind:?} {name}\n{}", statement_block(&statements)));
|
||||
}
|
||||
}
|
||||
Ok((!sections.is_empty()).then(|| sections.join("\n\n")))
|
||||
}
|
||||
|
||||
struct PlannedTableStructure {
|
||||
|
||||
@@ -471,6 +471,32 @@ pub fn rewrite_oracle_table_references(input: OracleTableReferenceRewriteInput)
|
||||
}
|
||||
|
||||
pub fn build_executable_object_source_statements(input: EditableObjectSourceSqlInput) -> Result<Vec<String>, String> {
|
||||
build_executable_object_source_statements_with_mysql_validation_name(input, None, true)
|
||||
}
|
||||
|
||||
/// Builds executable MySQL routine replacement statements using a caller-owned
|
||||
/// validation name. Transfer callers pass a per-run name and preflight its
|
||||
/// absence, so validation never drops an unrelated user routine.
|
||||
pub fn build_mysql_routine_replacement_statements(
|
||||
input: EditableObjectSourceSqlInput,
|
||||
validation_name: &str,
|
||||
) -> Result<Vec<String>, String> {
|
||||
if !is_mysql_like(input.database_type)
|
||||
|| !matches!(input.object_type, ObjectSourceKind::Procedure | ObjectSourceKind::Function)
|
||||
{
|
||||
return Err("MySQL routine replacement is supported only for MySQL procedures and functions.".to_string());
|
||||
}
|
||||
if validation_name.trim().is_empty() {
|
||||
return Err("MySQL routine validation name must not be empty.".to_string());
|
||||
}
|
||||
build_executable_object_source_statements_with_mysql_validation_name(input, Some(validation_name), false)
|
||||
}
|
||||
|
||||
fn build_executable_object_source_statements_with_mysql_validation_name(
|
||||
input: EditableObjectSourceSqlInput,
|
||||
mysql_validation_name: Option<&str>,
|
||||
drop_mysql_validation_before_create: bool,
|
||||
) -> Result<Vec<String>, String> {
|
||||
if input.database_type == DatabaseType::OceanbaseOracle && input.object_type == ObjectSourceKind::Sequence {
|
||||
return Ok(vec![oceanbase_sequence_edit_sql(&input)?]);
|
||||
}
|
||||
@@ -544,7 +570,12 @@ pub fn build_executable_object_source_statements(input: EditableObjectSourceSqlI
|
||||
if is_mysql_like(input.database_type)
|
||||
&& matches!(input.object_type, ObjectSourceKind::Function | ObjectSourceKind::Procedure)
|
||||
{
|
||||
return Ok(executable_mysql_routine_statements(&input, source));
|
||||
return Ok(executable_mysql_routine_statements(
|
||||
&input,
|
||||
source,
|
||||
mysql_validation_name,
|
||||
drop_mysql_validation_before_create,
|
||||
));
|
||||
}
|
||||
|
||||
let create_statement = ensure_semicolon(source);
|
||||
@@ -1261,7 +1292,12 @@ fn drop_informix_view_if_exists(name: &str) -> String {
|
||||
format!("DROP VIEW IF EXISTS {};", name.trim())
|
||||
}
|
||||
|
||||
fn executable_mysql_routine_statements(input: &EditableObjectSourceSqlInput, source: &str) -> Vec<String> {
|
||||
fn executable_mysql_routine_statements(
|
||||
input: &EditableObjectSourceSqlInput,
|
||||
source: &str,
|
||||
custom_validation_name: Option<&str>,
|
||||
drop_validation_before_create: bool,
|
||||
) -> Vec<String> {
|
||||
if !mysql_source_starts_with_create_routine(source) {
|
||||
return vec![ensure_semicolon(source)];
|
||||
}
|
||||
@@ -1275,18 +1311,26 @@ fn executable_mysql_routine_statements(input: &EditableObjectSourceSqlInput, sou
|
||||
// MySQL has no cross-version CREATE OR REPLACE for stored routines. Validate the CREATE
|
||||
// body under a temporary name first (same idea as Informix view saves) so a syntax error
|
||||
// cannot leave the original routine deleted after DROP.
|
||||
let validation_name = mysql_validation_routine_name(create_name);
|
||||
if let Some(validation_source) = replace_mysql_routine_declaration_name(source, &validation_name) {
|
||||
statements.push(mysql_drop_routine_if_exists(
|
||||
input.object_type.clone(),
|
||||
input.schema.as_deref(),
|
||||
&validation_name,
|
||||
));
|
||||
let owned_validation_name;
|
||||
let validation_name = if let Some(validation_name) = custom_validation_name {
|
||||
validation_name
|
||||
} else {
|
||||
owned_validation_name = mysql_validation_routine_name(create_name);
|
||||
&owned_validation_name
|
||||
};
|
||||
if let Some(validation_source) = replace_mysql_routine_declaration_name(source, validation_name) {
|
||||
if drop_validation_before_create {
|
||||
statements.push(mysql_drop_routine_if_exists(
|
||||
input.object_type.clone(),
|
||||
input.schema.as_deref(),
|
||||
validation_name,
|
||||
));
|
||||
}
|
||||
statements.push(ensure_semicolon(&validation_source));
|
||||
statements.push(mysql_drop_routine_if_exists(
|
||||
input.object_type.clone(),
|
||||
input.schema.as_deref(),
|
||||
&validation_name,
|
||||
validation_name,
|
||||
));
|
||||
}
|
||||
|
||||
@@ -2943,6 +2987,27 @@ mod tests {
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn mysql_transfer_routine_replacement_uses_scoped_validation_name() {
|
||||
let statements = build_mysql_routine_replacement_statements(
|
||||
EditableObjectSourceSqlInput {
|
||||
database_type: DatabaseType::Mysql,
|
||||
object_type: ObjectSourceKind::Procedure,
|
||||
schema: Some("app".to_string()),
|
||||
name: "refresh_cache".to_string(),
|
||||
source: "CREATE PROCEDURE `refresh_cache`() BEGIN SELECT 1; END".to_string(),
|
||||
},
|
||||
"dbx_transfer_check_0123456789abcdef",
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(statements.len(), 4);
|
||||
assert!(statements[0].contains("CREATE PROCEDURE `dbx_transfer_check_0123456789abcdef`"));
|
||||
assert!(statements[1].contains("DROP PROCEDURE IF EXISTS `app`.`dbx_transfer_check_0123456789abcdef`"));
|
||||
assert!(statements[2].contains("DROP PROCEDURE IF EXISTS `app`.`refresh_cache`"));
|
||||
assert!(!statements[0].contains("DROP"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn mysql_function_save_replaces_existing_routine() {
|
||||
let source = "CREATE DEFINER=CURRENT_USER FUNCTION `active_count`() RETURNS INT RETURN 1";
|
||||
|
||||
Reference in New Issue
Block a user